Generating and ingesting sample data with Benthos

input.yaml

input:
  generate:
    interval: "50ms" # Generate an event every 50ms
    mapping: |
      let max_subjects = 100

      let event_type = "api-calls"
      let source = "api-gateway"
      let methods = ["GET", "POST"]
      let paths = ["/", "/about", "/contact", "/pricing", "/docs"]

      let subject = "subject-%d".format(random_int(seed: timestamp_unix_nano()) % $max_subjects)
      let time = (now().ts_sub_iso8601("P3D").ts_unix() + random_int(min: 60, max: 60 * 60 * 24 * 3)).ts_format()

      let method = $methods.index(random_int(seed: timestamp_unix_nano()) % $methods.length())
      let path = $paths.index(random_int(seed: timestamp_unix_nano()) % $paths.length())
      let duration = random_int(seed: timestamp_unix_nano(), max: 1000)

      root = {
        "id": uuid_v4(),
        "specversion": "1.0",
        "type": $event_type,
        "source": $source,
        "subject": $subject,
        "time": $time,
        "data": {
          "method": $method,
          "path": $path,
          "duration_ms": $duration,
        },
      }

output.yaml

# input:
# ...

output:
  http_client:
    url: ${OPENMETER_BASE_URL:http://127.0.0.1:8888}/api/v1/events
    verb: POST
    headers:
      Content-Type: application/cloudevents+json
      Authorization: "Bearer ${OPENMETER_TOKEN:}"
    max_in_flight: 64