Collectors

The Metering & Billing Collector is an open-source software you can run in your own infrastructure to meter external systems, like Kubernetes and GPUs, and import usage from these data sources into Metering & Billing. While collectors provide a higher abstraction than the API, you can always use the API to integrate new data sources.

The following collectors are available:

The Metering & Billing Collector is built on top of Redpanda Connect, a robust in-process stream processing tool that can connect to a wide range of data sources. The Metering & Billing Collector uses Redpanda Connect because of its strong delivery guarantees and it’s ability to retry failed messages.

Architecture

The Collector connects data sources to sinks through a pipeline comprising various processing steps. It reads messages from a streaming source, processes them (for example, validation, transformation, and enrichment), and then sends them to Metering & Billing. It does so with strong delivery guarantees, ensuring that messages are recovered and will be retried until they are successfully delivered to their destination.

The three major components of the Collector pipeline:

• Inputs: Read usage data from various sources
• Processors: Validate, transform, and filter usage events
• Output: Send data to Metering & Billing with retry and buffer

Ingesting data

When using the collector, the output of the pipeline is always a Metering & Billing instance:

output:
  openmeter:
    url: https://us.api.konghq.com/v3/openmeter # default
    token: 'YOUR KONNECT SYSTEM ACCESS TOKEN' # required

Replace YOUR KONNECT SYSTEM ACCESS TOKEN with your own system access token.

As of today, Metering & Billing requires ingested data to be in the CloudEvents format. Redpanda Connect offers a range of processors that can transform, validate, and enrich messages. The most commonly used processor is known as mapping, which uses Redpanda Connect’s mapping language, bloblang.

Let’s assume you have access logs are in the following format:

{
  "timestamp": "2021-01-01T00:00:00.000Z",
  "method": "GET",
  "path": "/app",
  "user": "USERID",
  "duration": 100
}

You can use the mapping processor to transform the data into a usage event:

pipeline:
  processors:
    - mapping: |
        root = {
          "id": uuid_v4(),
          "specversion": "1.0",
          "type": "api-call",
          "source": "api-gateway",
          "time": this.timestamp,
          "subject": this.user,
          "data": {
            "duration": this.duration,
          },
        }

Install the Collector

The Metering & Billing Collector (a custom Redpanda Connect distribution) is available via the following distribution strategies:

• Binaries can be downloaded from the GitHub Releases page.
• Container images are available on ghcr.io.
• A Helm chart is also available on GitHub Packages.

The Metering & Billing Collector is designed with strong delivery guarantees and capability to retry failed messages. You can handle network outages without loosing any data by using event buffering.

Event buffering

The Metering & Billing Collector uses a persistent queue to buffer events. This allows the collector to store events on a persistent disk and retry them later in case of network failures or other issues. Based on the load and disk size, the collector can buffer events for extended periods of time, such as hours or days, and safely replay them to Metering & Billing when the network is restored.

How does event buffering work?

Installing the Metering & Billing Collector with the buffer enabled lets you send your usage events to the Collector instead of sending events directly to Metering & Billing. In this configuration, your app will first send the event to the Collector, which will forward the events to Metering & Billing and buffer them, retrying in the case of network failure.

To increase the resilience of your metering pipeline, the Collector comes with:

• Buffering
• Retries and backoff
• Deduplication
• OpenTelemetry metrics and logging

When buffering is enabled and there are connectivity issues, the Collector stores events on an attached persistent volume until the network recovers. The available space on the attached disk determines the size of the buffer.

The Collector also provides visibility into the buffer and processing states by exposing Prometheus metrics.

Get started with buffering

You can enable buffering with the Collector by doing the following:

1. Install the Collector:

   # Set your token
   export OPENMETER_TOKEN=om_...
   
   # Get the latest version
   export LATEST_VERSION=$(curl -s https://api.github.com/repos/openmeterio/openmeter/releases/ latest | jq -r '.tag_name' | cut -c2-)
   
   # Install the collector in the openmeter-collector namespace
   helm upgrade openmeter-collector oci://ghcr.io/openmeterio/helm-charts/benthos-collector \
     --version=${LATEST_VERSION} \
     --install --wait --create-namespace \
     --namespace openmeter-collector \
     --set fullnameOverride=openmeter-collector \
     --set openmeter.token=${OPENMETER_TOKEN} \
     --set service.enabled=true \
     --set storage.enabled=true \
     --set preset=http-server
   

   Copied!

2. Point your API to the installed collector: Send events to the collector’s URL in your infrastructure, for example http://openmeter-collector.svc.

   curl -X POST http://openmeter-collector.svc \
     -H 'Content-Type: application/cloudevents+json' \
     -H 'Authorization: Bearer <API_TOKEN>' \
     --data '
     {
       "specversion" : "1.0",
       "type": "prompt",
       "id": "e4bacd05-5abd-4e1c-be6f-daf6b8e9dd92",
       "time": "2026-02-01T16:41:37.188Z",
       "source": "my-app",
       "subject": "customer-1",
       "data": {
         "tokens": "12345",
         "provider": "openai",
         "model": "gpt-5-nano",
         "type": "output"
       }
     }
     '
   

   Copied!

Retries

The Metering & Billing Collector will retry failed messages up to three times by default, but you can configure this setting. If the message still fails after three retries, it will be dropped.

Event duplication processing

Metering & Billing will process duplicate events based on the id and source fields. If a message with the same id and source is received multiple times, only the first occurrence will be processed.

API and SDK passthrough

The API can be pointed to the Metering & Billing Collector endpoint to enrich events with additional metadata or increase the reliability of event delivery by leveraging the collector’s buffer and retry capabilities.

The Metering & Billing Collector exports OpenTelemetry metrics and logs for observability. The metrics are available on the GET /metrics endpoint if you want to scrape them with OpenTelemetry-compatible tools like Prometheus.

Observability metrics

The Collector exports the following observability metrics:

Batch metrics:

Buffer metrics:

Other metrics:

• Processor metrics
• Output metrics

Observability logs

The Collector logs the following:

• Collector startup and shutdown
• Collector errors
• Collector configuration errors
• Collector event source errors
• Collector event output errors
• Collector logs