[request] add schedule processor

[artemklevtsov]

Hi,
It would be nice to add schedule processor to consume the child input with defined interval or cron.

Use case: read paginated http response with interval.

I can do it only with external scheduler (like cron) now. Or maybe is it possible with current capabilities?

Example:

input:
  # need repeat this block every hour
  read_until:
    check: '@offset >= 5'
    input:
      generate:
        count: 0
        interval: 1s
        mapping: 'root = null'
      processors:
        - mapping: |
            meta n = 5
            meta offset = counter(min: 0)
        - http:
            url: https://httpbin.org/links/${! metadata("n") }/${! metadata("offset") }
            verb: GET

pipeline:
  processors:
    - xml:
        operator: to_json
    - mapping: |
        root = json("html.body.a")
    - unarchive:
        format: json_array
    - mutation: |
        root.page = @offset

output:
  stdout: {}

Possible solutions:

• add input field to the generate input
• add schedule input

[mihaitodor]

Hey @artemklevtsov

Use case: read paginated http response with interval.

You can use cron expressions with the generate input.

Then, for pagination, you can use a while processor to wrap the http processor and use metadata to keep track of the page state. It's also possible to use the http_client input like so: redpanda-data/connect#2005 (comment), but that feels more complex.

Also, you might want to look into rate limits? There's a rate_limit field on the http processor in case you need to limit how many requests are made.

[artemklevtsov]

Thank you for the reply.
I know about rate limit functionality.
Fix me if I'm wrong but while don't produce any messages before it exit.

[artemklevtsov]

I tried the following without success:

input:
  generate:
    count: 0
    interval: 120s
    mapping: |
      meta n = 5
      meta offset = 0
  processors:
    - while:
        check: '@offset < @n'
        processors:
          - http:
              url: https://httpbin.org/links/${! metadata("n") }/${! metadata("offset") }
              verb: GET
              dump_request_log_level: info
          - mapping: |
              meta offset = @offset + 1

pipeline:
  processors:
    - xml:
        operator: to_json
    - mapping: |
        root = json("html.body.a")
    - unarchive:
        format: json_array
    - mutation: |
        root.page = @offset

output:
  stdout: {}

@mihaitodor can you suggest solution for the this case?

[mihaitodor]

@artemklevtsov the pagination from that config requires you to accumulate the results somehow. I normally do it via a branch processor and stuff them in metadata. Something like this:

input:
  generate:
    count: 0
    interval: 120s
    mapping: |
      meta n = 5
      meta offset = 0
      meta results = []
  processors:
    - while:
        check: '@offset < @n'
        processors:
          - branch:
              processors:
                - http:
                    url: https://httpbin.org/links/${! metadata("n") }/${! metadata("offset") }
                    verb: GET
                    dump_request_log_level: info
                - xml:
                    operator: to_json
                - mapping: |
                    root = json("html.body.a")
              result_map: |
                meta results = @results.append(this.merge({"page": @offset}))
          - mapping: |
              meta offset = @offset + 1
    - mapping: |
        root = @results

output:
  stdout: {}

[artemklevtsov]

Thank you for solution. This solve the example case. Although I would not call the solution simple and obvious.

Small fixes for your code (for others):

input:
  generate:
    count: 0
    interval: 120s
    mapping: |
      meta n = 5
      meta offset = 0
      meta results = []
  processors:
    - while:
        check: '@offset < @n'
        processors:
          - branch:
              processors:
                - http:
                    url: https://httpbin.org/links/${! metadata("n") }/${! metadata("offset") }
                    verb: GET
                    # dump_request_log_level: info
                - xml:
                    operator: to_json
                - mapping: |
                    root = json("html.body.a")
              result_map: |
                meta results = @results.concat(this.map_each(item -> item.merge({"page": @offset})))
          - mapping: |
              meta offset = @offset + 1
    - mapping: |
        root = @results
    - unarchive:
        format: json_array

output:
  stdout: {}

P.S.: I still think that a separate scheduler input could be useful for simpler solutions. This already discussed here: redpanda-data/connect/issues/580 and here: redpanda-data/connect/discussions/2314. Implementation as plugin can be found: openmeterio/openmeter/blob/main/collector/benthos/input/schedule.go.

[artemklevtsov]

Close this as not planned if you think that another component (schedule input) will be superfluous.

[mihaitodor]

It's hard to know what users expect from such a scheduler input besides what generate already does. I haven't seen much consensus from the issues you pointed out (and others). This is why Benthos has all these building blocks which users can mix & match, but there's a learning curve...

The above implementation isn't very scalable, since it collects all the data in memory before sending it to the output. Also, if something goes wrong, the progress will be lost. When dealing with larger volumes of data, one might want a sequence with 2 generate inputs: One to handle the cron and another one to kick off the actual http calls and use a cache to keep track of pages. One limitation of sequence is that it doesn't restart automatically, so that could be a small enhancement to add a restart_on_finish flag for it to avoid having to wrap it in a read_until.

Note: You can also wrap that entire input logic in a template (and even register it in your custom build so you don't have to specify it when running the CLI).

[artemklevtsov]

The above implementation isn't very scalable, since it collects all the data in memory before sending it to the output. Also, if something goes wrong, the progress will be lost.

Thanks for the explanation.

One limitation of sequence is that it doesn't restart automatically, so that could be a small enhancement to add a restart_on_finish flag for it to avoid having to wrap it in a read_until.

Nice idea.

Note: You can also wrap that entire input logic in a template (and even register it in your custom build so you don't have to specify it when running the CLI).

Very interesting. Can you provide example or details. I read the docs, but I can't correlate this with the current context.

[mihaitodor]

Very interesting. Can you provide example or details.

Sure, you can use the service.MustRegisterTemplateYAML API. See an example here: https://github.com/redpanda-data/connect/blob/main/internal/impl/twitter/init.go#L31

The template could look something like this (I haven't tested it and you'll likely want to expose more fields):

name: scheduled_crawler
type: input
status: experimental
categories: [ Services ]
summary: TODO
description: |
  TODO

fields:
  - name: base_url
    description: TODO
    type: string
    kind: scalar
    default: https://httpbin.org


mapping: |
  #!blobl

  root.generate = {
    "count": 0,
    "interval": "120s",
    mapping: """
      meta n = 5
      meta offset = 0
      meta results = []
    """
  }

  root.processors = []
  root.processors."-" = """
    while:
      check: '@offset < @n'
      processors:
        - branch:
            processors:
              - http:
                  url: %s/links/${! metadata("n") }/${! metadata("offset") }
                  verb: GET
                  # dump_request_log_level: info
              - xml:
                  operator: to_json
              - mapping: |
                  root = json("html.body.a")
            result_map: |
              meta results = @results.concat(this.map_each(item -> item.merge({"page": @offset})))
        - mapping: |
            meta offset = @offset + 1
  """.format(this.base_url).parse_yaml() # You can also use `replace_all_many` and have some placeholders in the YAML

  root.processors."-" = {
    "mapping": """
      root = @results
    """
  }

  root.processors."-" = {
    "unarchive": {
      "format": "json_array"
    }
  }

[artemklevtsov]

@mihaitodor

so that could be a small enhancement to add a restart_on_finish flag for it to avoid having to wrap it in a read_until.

There any plans to add the restart_on_finish flag or restart_interval fields for the read_until? I have many cases for this.