Read from input periodically

[sagikazarmark]

I'm trying to read from an input (let's say HTTP endpoint) periodically (meaning I want to essentially re-read a list of items every X seconds).

The closest thing I found to a periodic input is the generate input, but it doesn't fit this use case well (although the documentation mentions it could be used with an HTTP processor).

The idea I'm working on right now is a schedule input (based on the generate input) that would essentially block for the interval (using a timer) and would read from a child input.

I don't actually need batching in this case, I just want to read all, so I'll omit the batching configuration.

First of all: is this a good idea? (I was kinda surprised there is no such input in benthos which made me doubt if I'm on the right path)

I found #580, but I don't see too much activity on it lately.

Secondly, I'm not really sure about the semantics of ReadBatch: in my case, it would block on a timer and then return a set of messages. Would ReadBatch immediately be called again? (Otherwise the timer is kinda pointless/unreliable).

(I could run Benthos as a cron job, but this service would run in Kubernetes and I don't trust it entirely, especially when it's run at low (5-10 second) intervals)

Any insights are greatly appreciated.

[mihaitodor]

Hey @sagikazarmark, can you offer some more details about what you're trying to achieve? It's not clear why the generate + http processor doesn't suit your needs. Are you looking to consume more than one message from the input? If yes, what is the stop condition? While it's a bit convoluted, the sequence input with two child inputs might help in such cases (the first child is a generate with a sleep processor and the second child is a read_until with a http_client or something like that). The cork input proposal from #1069 might be worth resurrecting in this case.

Secondly, I'm not really sure about the semantics of ReadBatch: in my case, it would block on a timer and then return a set of messages. Would ReadBatch immediately be called again? (Otherwise the timer is kinda pointless/unreliable).

Are you asking about building a custom input plugin? That should be the case, yeah, although the batcher supports multiple modes and you can configure it to wait until it reads n messages or until a certain time duration passes or until it reads a certain number or bytes or various combinations of these parameters.

LE: See below for details about back pressure and throttling.

[sagikazarmark]

Honestly, it just doesn't feel very intuitive. It's hard to reason about after writing similar pipelines with an input that serves as a datasource (instead of a processor in the recommended setup).

Also, I may not actually be able to use the HTTP client/processor (I may need some custom logic).

I ended up implementing the schedule input as I described it above. It's actually not that different from the cork input, except it only works with a timer, there are no other signals.

Here is the config:

input:
  schedule:
    input:
      # ...
    interval: "3s"

Again, I find this to be more intuitive and easier to understand than the generate+http processor solution.

Are you asking about building a custom input plugin?

Precisely. I'm more interested in how this is going to work with a time.Ticker inside ReadBatch (similarly to how the generate input is implemented): the channel under the hood is a buffered channel with 1 as the buffer size which means I may lose some ticks if the channel is not continuously consumed (which happens on calls to ReadBatch).

I suspect that once ReadBatch returns, Benthos calls it basically immediately again. Is that correct? Does that also depend on the batching config (given there is one)?

[sagikazarmark]

Appreciate the quick reply by the way. Wish you a happy new year!

[mihaitodor]

Happy New Year to you too!

Honestly, it just doesn't feel very intuitive. It's hard to reason about after writing similar pipelines with an input that serves as a datasource (instead of a processor in the recommended setup).

Feel free to submit a new proposal or resurrect the cork one. I'll have to defer this to @Jeffail.

Also, I may not actually be able to use the HTTP client/processor (I may need some custom logic).

Sure, but you can write your own custom processor and use that in combination with generate.

I suspect that once ReadBatch returns, Benthos calls it basically immediately again. Is that correct? Does that also depend on the batching config (given there is one)?

One thing to keep in mind is back pressure. See here and here, so I wouldn't really count on ReadBatch to be called immediately after it returns. I forgot to mention this in my previous answer, sorry about that!

Also, there can be rate limits and sleep processors. You can try using a buffer to avoid some of those issues, but there are no guarantees. In my experience, like you mentioned, even external schedulers such as Kubernetes can stall.

[sagikazarmark]

One thing to keep in mind is back pressure.

That's a good point, thanks. I guess generate (and cork) is affected by that as well.

I guess the right solution is fine tuning configuration depending on the expectations.

Thanks again for the help!

[Jeffail]

you can also stick a rate limit processor within the input processors and itll apply back pressure to the input, it'd be the exact same functionality as having some sort of scheduled input wrapper with an interval.

[sagikazarmark]

Thank you @mihaitodor and @Jeffail for your answers!

I created the aforementioned schedule input and I used it with success in several cases.

One issue I've recently experienced though is with inputs like sql_select. I believe there are actually two issues:

1. Combined with my schedule input, sql_select only returns a single row. I think this is because sql_select itself is not a batch input. I'm currently trying read_until and batched inputs to see if I can get Benthos to read until the end from sql_select which may work, but leads to issue number 2.
2. Once the input is read to the end, it shuts down. That's expected from sql_select, but I think it would make more sense for schedule to restart the input (or at least make that an option), otherwise it's not really a schedule.

What I'm trying to do now is add an option to schedule to suppress input end errors and always restart the input. That combined with a read_until input with check: true should do the trick.

I realize this is probably dangerous, but in this particular case, where I know there is an end, I think it should be okay. WDYT?

I also found #2301 which is somewhat similar and @Jeffail suggested a similar input implementation.