Benthos + Consul

benthos.yaml

input:
  sequence:
    inputs:
      - generate:
          count: 1
        processors:
          - http:
              verb: GET
              url: ${! env("CONSUL_ENDPOINT") }/v1/kv/${! env("CONSUL_KV_KEY") }
              headers:
                x-consul-token: ${! env("CONSUL_TOKEN") }
          - mapping: |
              root = {
                "stdout": {},
                "processors": {
                  "redis": {
                    "url": json("0.Value").decode("base64").string(),
                    "command": "mset"
                  }
                }
              }
          - http:
              verb: POST
              url: http://localhost:4195/outputs/x
          - mapping: root = deleted()

      - redis_streams:
          url: ${RECORD_STREAMS_CONN}
          streams: [records]

pipeline:
  processors: [] # TODO

output:
  dynamic:
    outputs: {}