Input pagination without caches

[kuthulas]

Hello,

I'm looking to implement a Redis scan input. I wrote an example that involved a cache (based on #752 (comment)), and that worked well.

However, I do not prefer using a cache, so I rewrote it as follows. But the message to the processors after the generate input is always empty. So the last cursor never gets used in the subsequent scan calls. It seems to me that the generate input does not propagate messages from the previous iteration of read_until. I'd appreciate some help in fixing this.

input:
  read_until:
    restart_input: true
    check: this.cursor == "0"
    input:
      generate:
        interval: ""
        mapping: root = (this|{})
      processors:
        - redis:
            url: tcp://localhost:6379
            command: scan
            args_mapping: root = [ this.cursor, "match", "*" ]
        - mapping: |
            root.cursor = this.index(0)
            root.items = this.index(1)

[kuthulas]

I just found #1986. Is there no way around using a cache?

[mihaitodor]

Hey @kuthulas, messages in Benthos aren't connected in any way, so you can't reference the previous message from the current one. Also, using an in-memory cache is perfectly reasonable (although you might want to use an external one if you need fault tolerance). Why are you looking to avoid it?

Having said that, since you're fetching data using a processor (namely redis), you can start with a generate input which generates a single message and, in the processors section, you can use a while processor with at_least_once: true and your check condition. Then, inside the processors section of this while processor you can have your redis processor followed by a mapping processor which alters the message metadata such that you keep appending the result to a metadata field which holds an array of messages. Something like this: meta messages = @messages.append(this) (or meta messages = @messages.append(content()) if the data you get from the redis processor isn't a JSON object). Finally, after the while processor, you can add a mapping processor with root = @messages followed by an unarchive processor with format: json_array so you'll explode your messages array into a batch of messages (that you can then split into smaller batches using the split processor if needed).

[kuthulas]

Hey @mihaitodor, thanks for your response and the suggestion.

I'd like to avoid caches because I'm using Benthos in streams mode, and I don't prefer to keep this specific cache resource in the global configuration file for logistics reasons. I'd like to keep a stream's resources in its own configuration file.

And, I think your suggested workaround would look like this:

input:
  generate:
    interval: ""
    mapping: root = ""
  processors:
    - mapping: meta messages = []
    - while:
        at_least_once: true
        check: this.cursor != "0"
        processors:
          - redis:
              url: tcp://localhost:6379
              command: scan
              args_mapping: root = [ this.cursor, "match", "input_key*" ]
          - mapping: |
              root.cursor = this.index(0)
              meta messages = @messages.append(this)
    - mapping: root = @messages
    - unarchive:
        format: json_array

This works, but this would mean that the pipeline starts only after the scan is complete. This is a hot path and there might be a million keys in the database, so I cannot have it this way either.

So my next question would be, is there a way the while processor could be made to yield messages to the pipeline during the iterations?

Thanks for your time.

[mihaitodor]

Hey @kuthulas, you're welcome!

I'd like to avoid caches because I'm using Benthos in streams mode, and I don't prefer to keep this specific cache resource in the global configuration file for logistics reasons. I'd like to keep a stream's resources in its own configuration file.

OK, fair enough.

This works, but this would mean that the pipeline starts only after the scan is complete. This is a hot path and there might be a million keys in the database, so I cannot have it this way either.

That sucks... Unfortunately, you can't stream data from a processor.

In this case, I think the best thing to do is to write your own custom input similar to the examples that you can find here. You can find more inspiration in the existing Benthos components which are implemented here. As a really dirty hack which I don't recommend, you could write a simple custom processor which holds state about the previous message. It could be unpredictable if you use it in the pipeline section or at the output level, since those can end up executing in parallel, depending on pipeline.threads or max_in_flight, but it should be OK if you use it under input.read_until.input.processors like you have in your initial config, since those don't run in parallel.

[kuthulas]

Thanks @mihaitodor. I wrote an input plugin, and it works well.