Optimal Config for Openmeter collector

[nishantmehta-innovaccer]

I am Nishant Mehta, an engineer at Innovaccer, currently building a solution for metering and usage-based billing. We are evaluating Openmeter Cloud and the Openmeter Collector as part of our POC, and I would appreciate some guidance on optimizing the collector configuration for our use case.

Here are some issues we have encountered during testing:

1. During load tests (10–20 RPS, 1,000–3,000 concurrent requests), we are seeing latency of around one second per request when batching and buffering are enabled. Disabling these settings makes updates almost real-time.
2. With the default configuration, the collector consumes about 20 millicores CPU and 67 MB memory. I am unsure whether this scales automatically based on the cluster or needs configuration tuning.
3. We are looking for the best configuration for batching and buffering — the key requirement is zero event loss (as events are tied to billing), and a 5–10 minute delay is acceptable (not more than that).
4. Is it possible for the collector to accumulate single-event requests and batch them before sending to Openmeter Cloud? I noticed the ingestion API supports arrays of events, so this seems technically viable.

Attaching my sample collector config which i am using
sample_openmeter_collector_doc.yml

Any insights or configuration recommendations would be extremely helpful in finalizing our setup.

Thank you,
Nishant Mehta
Innovaccer

[tothandras]

Hi @nishantmehta-innovaccer,
We've learned that the post_processors in the buffer can cause some issues, can you try the following config / Helm chart values.yaml?

# Replica count in the StatefulSet
replicaCount: 1

# Kubernetes service configuration
service:
  enabled: true
  type: ClusterIP
  port: 80

# Persistent storage for the buffer
storage:
  enabled: true
  size: 1Gi

# Path to the buffer file
env:
  BUFFER_PATH: /data/buffer.db
  LOG_LEVEL: INFO

# Docs: https://docs.redpanda.com/redpanda-connect/components
# This config defines the data pipeline for collecting usage metrics from various sources
config:
  logger:
    level: '${LOG_LEVEL:INFO}'
    format: '${LOG_FORMAT:json}'
    # Log fields that are constant across all events
    static_fields:
      service: 'openmeter-collector'
      instance: '${K8S_APP_INSTANCE:}'
      version: '${K8S_APP_VERSION:}'

  # Delay before shutting down the collector
  shutdown_delay: 5s
  # Timeout before shutting down the collector
  shutdown_timeout: 10s

  # HTTP server for metrics
  http:
    enabled: true
    address: 0.0.0.0:4195
    debug_endpoints: false

  # Define the HTTP server input
  input:
    http_server:
      address: 0.0.0.0:8080
      path: /api/v1/events
      allowed_verbs:
        - POST
      timeout: 10s
      sync_response:
        status: '${! meta("http_status_code").or("204") }'
        headers:
          Content-Type: '${! meta("content_type").or("application/json") }'
    processors:
      # Track metrics on received messages
      - metric:
          type: counter
          name: openmeter_messages_received
          value: 1
      - mapping: |
          meta received_at = meta("received_at").or(now())
      # Validate that mapped events conform to CloudEvents schema
      - label: 'validation'
        json_schema:
          schema_path: file://./cloudevents.spec.json
      - catch:
          - log:
              level: ERROR
              message: 'schema validation failed due to: ${!error()}'
              fields_mapping: |-
                root = this
          - metric:
              type: counter
              name: openmeter_messages_dropped
              labels:
                reason: schema_validation_failed
              value: 1
          # Return a 400 Bad Request response for invalid events
          - mapping: |
              meta http_status_code = "400"
              meta content_type = "application/problem+json"

              root = {
                "type": "about:blank",
                "title": "Bad Request",
                "status": 400,
                "detail": "invalid event: %s".format(error()),
              }
          - sync_response: {}
          # Drop messages that fail validation
          - mapping: root = deleted()
      # Return a 204 No Content response for valid events
      - mapping: |
          meta http_status_code = "204"
      - sync_response: {}
      # Track metrics on successfully processed messages by type
      - metric:
          type: counter
          name: openmeter_messages_processed
          labels:
            type: ${! type }
          value: 1

  # Persistent buffer to ensure data isn't lost during processing
  buffer:
    sqlite:
      path: '${BUFFER_PATH:./buffer.db}'

  # Send processed events to OpenMeter
  output:
    label: 'openmeter'
    http_client:
      url: '${OPENMETER_URL:https://openmeter.cloud}/api/v1/events'
      verb: POST
      headers:
        Authorization: 'Bearer ${OPENMETER_TOKEN:}'
        Content-Type: 'application/json'
      timeout: 30s
      retry_period: 15s
      drop_on:
        - 400 # Bad Request
      retries: 3
      max_retry_backoff: 1m
      # Maximum number of concurrent requests
      max_in_flight: 64
      batch_as_multipart: false
      # Batch settings for efficient API usage
      batching:
        # Send up to 50 events in a single request
        count: 50
        # Or send after 1 second, whichever comes first
        period: 1s
        processors:
          # Track metrics on sent events
          - metric:
              type: counter
              name: openmeter_events_sent
              labels:
                type: ${! type }
                subject: ${! subject }
              value: 1
          - metric:
              type: timing
              name: openmeter_events_sent_duration_ns
              labels:
                type: ${! type }
              value: ${! timestamp_unix_nano() - meta("received_at").ts_unix_nano() }
          - metric:
              type: timing
              name: openmeter_events_sent_delay_ns
              labels:
                type: ${! type }
              value: ${! timestamp_unix_nano() - time.ts_unix_nano() }
          # Split large batches into 50 event batches
          - split:
              size: 50
          # Convert batch to JSON array format
          - archive:
              format: json_array
      dump_request_log_level: DEBUG

  # Expose internal metrics in Prometheus format on /metrics endpoint
  metrics:
    prometheus:
      add_process_metrics: true
    mapping: |
      meta version = env("K8S_APP_VERSION").or("-")
      meta instance = env("K8S_APP_INSTANCE").or("-")

export OPENMETER_URL=https://openmeter.cloud
export OPENMETER_TOKEN=om_...
export VERSION=1.0.0-beta.225

helm upgrade openmeter-collector oci://ghcr.io/openmeterio/helm-charts/benthos-collector --version ${VERSION} --install --wait --create-namespace \
  --namespace openmeter-collector \
  --set fullnameOverride=openmeter-collector \
  --set image.tag=v${VERSION} \
  --set openmeter.url=${OPENMETER_URL} \
  --set openmeter.token=${OPENMETER_TOKEN} \
  -f ./values.yaml

[nishantmehta-innovaccer]

Hi @tothandras ,
I wanted to thank you for the config. It worked well for the few load tests which i performed.

Just to clarify,

For example,
For 100 rps and total events just 1000 ~ to be pushed in 10 + few additional seconds, openmeter_events_sent_delay_ns_sum came as 211 seconds. Is it expected or can we further fine tune for better results? (We have not specified the resources which means it would take as much as required from the cluster.)

Also, another metrics which i wanted to highlight:
buffer_latency_ns {instance="openmeter-collector",label="",path="root.buffer",version="v1.0.0-beta.225",quantile="0.99"} 1.093208303e+09
buffer_latency_ns_sum {instance="openmeter-collector",label="",path="root.buffer",version="v1.0.0-beta.225"} 1.37127418444e+11
buffer_latency_ns_count {instance="openmeter-collector",label="",path="root.buffer",version="v1.0.0-beta.225"} 1000

From the above metrics, p99 for buffer latency is 1.09 seconds and total 137 seconds (as a sum for all 1000) where requests were in buffer. can it be improved?

Again thanks for your help.

[tothandras]

You can try to remove the buffer and see if that causes the latency, or you can tune values of the batching config:

  # Persistent buffer to ensure data isn't lost during processing
  #  buffer:
  #    sqlite:
  #      path: '${BUFFER_PATH:./buffer.db}'
  # ...
        # Send up to 50 events in a single request
        count: 50
        # Or send after 1 second, whichever comes first
        period: 1s