Events

An event can be anything you need to track accurately over time for billing or analytics purposes. For example, a CI/CD product can include active or parallel jobs, build minutes, network traffic, storage used, or other product-related actions.

Metering & Billing uses a stream processing architecture to collect usage events and turn them into metered consumption. This guide explains how Metering & Billing ingests events via Kafka and transfers them into ClickHouse, the columnar database used as long-term storage.

Stream processing pipeline

First, the Metering & Billing API accepts events in the CloudEvents format and publishes them to Kafka topics before further processing them. This allows Metering & Billing to process events in batches and handle traffic spikes efficiently.

The events are then processed by a custom Kafka Consumer written in Go, which validates events and ensures consistent deduplication and exactly-once inserts into ClickHouse. The Kafka Consumer scales horizontally by Kafka partitions, allowing for parallel processing of events and ensuring high availability.

 
flowchart LR
    A[Konnect API] --> B[Kafka]
    B --> C[Go worker]
    B --> D[Go worker]
    B --> E[Go worker]
    C --> F[ClickHouse]
    D --> F
    E --> F
  

Metering & Billing leverages the CloudEvents specification, which offers a standardized and flexible way to describe event data, making it easier to connect your services and tools seamlessly.

To ingest events into Metering & Billing, send them to the Konnect API:

curl -X POST "https://us.api.konghq.com/v3/openmeter/events" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/cloudevents+json" \
     --json '{
       "specversion": "1.0",
       "type": "request",
       "id": "00001",
       "source": "service-0",
       "time": "2023-01-01T00:00:00.001Z",
       "subject": "customer-1",
       "data": {
         "method": "GET",
         "route": "/hello"
       }
     }'

An event contains the following properties:

Metering & Billing continuously processes usage events, allowing you to update meters in real time. Once an event is ingested, Metering & Billing aggregates the data based on your defined meters. For example, you can define meters called “Parallel jobs”, and Metering & Billing will aggregate the maximum number of jobs by each customer over a given time period.

Let’s say you want to track serverless execution duration by endpoint and you defined the following meter:

curl -X POST "https://us.api.konghq.com/v3/openmeter/meters" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Accept: application/json, application/problem+json" \
     --json '{
       "name": "Total API requests",
       "key": "api_requests_total",
       "description": "API Requests",
       "event_type": "request",
       "value_property": "$.duration_seconds",
       "aggregation": "sum",
       "dimensions": {
         "method": "$.method",
         "route": "$.route"
       }
     }'

$.duration_seconds is a JSONPath expression to access the data.duration_seconds property, providing powerful capabilities to extract values from nested data properties.

The meter config above tells Metering & Billing to expect CloudEvents with type=request where the usage value is stored in the data.duration_seconds, and to sum them up by data.route. Metering & Billing will track the usage value for every time window when at least one event was reported, and for every subject and groupBy permutation.

For example, when sending the following event:

curl -X POST "https://us.api.konghq.com/v3/openmeter/events" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/cloudevents+json" \
     --json '{
       "specversion": "1.0",
       "type": "request",
       "id": "00001",
       "source": "service-0",
       "time": "2023-01-01T00:00:00.001Z",
       "subject": "customer-1",
       "data": {
         "method": "GET",
         "route": "/hello"
       }
     }'

Metering & Billing will track the usage value for the time window and customer as:

windowstart   = "2024-01-01T00:00"
windowend     = "2024-01-01T00:01"
subject       = "customer-1"
duration_seconds   = 10
method        = "GET"
route         = "/hello"

When sending a second event (with a different id and duration_seconds value):

curl -X POST "https://us.api.konghq.com/v3/openmeter/events" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/cloudevents+json" \
     --json '{
       "specversion": "1.0",
       "type": "request",
       "id": "00002",
       "source": "service-0",
       "time": "2024-01-01T00:00:00.001Z",
       "subject": "customer-1",
       "data": {
         "duration_seconds": 20,
         "method": "GET",
         "route": "/hello"
       }
     }'

Metering & Billing will increase sum of the duration for the two events for the same time window (windowstart, windowend), method, route, and subject:

windowstart   = "2024-01-01T00:00"
windowend     = "2024-01-01T00:01"
subject       = "customer-1"
duration_seconds      = 30
method        = "GET"
route         = "/hello"

CloudEvents are unique by id and source. For more information, see CloudEvents specification.

Producers must ensure that the source and id combination is unique for each distinct event. If a duplicate event is re-sent (for example, due to a network error) it may have the same id. Consumers may assume that events with identical source and id are duplicates.

Metering & Billing deduplicates events by id and source. This ensures that if multiple events with the same id and source are sent, they are only processed once. This is useful when you want to retry or replay events in your infrastructure.

You may need to pre-process events before they are ingested into Metering & Billing to normalize data, enrich events, or calculate derived fields like cost for example.

To pre-process events, use Collectors, which support Bloblang mapping for transformations.