Usage Metering Middleware

Introduction

Note: This middleware is only available with Kipchak Enterprise.

Records what each consumer of your API used, so that you can bill for it, charge it back to internal teams, or enforce plans. Every metered request produces one CloudEvents 1.0 event saying:

• who made it: the consumer it is billed to and, optionally, the tenant;
• what it called: the method and the route pattern (/v1/users/{id}, not /v1/users/42);
• how it ended: status, duration, and request and response size;
• what it consumed: one requests unit, plus any units you add, such as tokens, rows or credits.

Events go to Kafka (through the Kafka driver), to your logs for a log shipper to collect, or to a sink of your own.

Recording usage never affects the response. Kafka events are queued without waiting for the broker, so a slow or unreachable cluster does not slow requests, and any event that cannot be delivered is written to the log in full so that it can be replayed.

Installation

To install this middleware, you need access to the Enterprise Composer repository at https://php.pkgs.1x.ax/.

composer require kipchak/middleware-metering

To send events to Kafka, install the Kafka driver too (version 1.10 or later), and initialise it in drivers/drivers.php:

composer require kipchak/driver-kafka

Then add the middleware in middlewares/middlewares.php. Call it before your authentication middlewares:

use Kipchak\Middleware\Error\Error;
use Kipchak\Middleware\Metering\Metering;
use Kipchak\Middleware\Auth\Key\Key;
use Kipchak\Middleware\Auth\HMAC\HMAC;

Error::initialise($app);
Quotas::initialise($app);     // if you use Quotas: before Metering
Metering::initialise($app);   // before the auth middlewares
Key::initialise($app);
HMAC::initialise($app);

Slim runs the middleware added last first. Initialised before authentication, Metering runs after it, so it can see the identity that authentication attached to the request, and requests that authentication rejects are not billed.

Finally, copy sample.config.php to config/kipchak.metering.php.

The usage event

{
  "specversion": "1.0",
  "id": "01a115ce-3dfa-7b96-97e3-12f3fd8c79a0",
  "source": "kipchak/billing-api",
  "type": "dev.kipchak.usage.request",
  "time": "2026-10-07T10:00:00.250Z",
  "subject": "client-a",
  "datacontenttype": "application/json",
  "data": {
    "consumer": "client-a",
    "tenant": "acme",
    "method": "POST",
    "route": "/v1/completions/{model}",
    "status": 200,
    "duration_ms": 41.207,
    "request_bytes": 312,
    "response_bytes": 2048,
    "units": { "requests": 1, "tokens": 1250 },
    "dimensions": { "region": "eu-west-1", "model": "large" }
  }
}

• id is a UUID version 7. It is unique across every worker and host, so a billing pipeline should deduplicate on it: Kafka can deliver a message more than once if a broker fails mid-acknowledgement. Within one worker, ids also increase in the order requests were recorded.
• subject is the consumer, so tools that read CloudEvents usage, such as OpenMeter, attribute the event without a mapping. Anonymous events have no subject.
• time is when the request started. duration_ms covers the route and every middleware after Metering.
• route is null when no route matched (a 404).
• request_bytes and response_bytes come from Content-Length, or the body size when that is absent, and are null when neither is known (for example, a streamed response).
• units and dimensions are always JSON objects, even when empty.

On Kafka, each message carries the content-type: application/cloudevents+json; charset=UTF-8 header (the CloudEvents Kafka binding's structured mode) and is keyed by consumer, so one consumer's events stay in order on one partition.

Configuration

Configuration errors, such as an unknown source, an invalid unit name or a Kafka connection that does not exist, stop the API at startup with a message naming the key, rather than failing on the first request.

By default every status is recorded, including errors, so that billing decides what to charge for. Paths are not recorded by default because they carry ids and sometimes personal data.

Identifying the consumer

Who each request is billed to comes from Consumer Identity: config/kipchak.identity.php, shared with Quotas so that a client is the same consumer on the bill and in the limits.

// config/kipchak.identity.php
return [
    'consumer' => [
        ['source' => 'token', 'claim' => 'client_id'],              // auth-jwks / auth-jwt
        ['source' => 'attribute', 'name' => 'kipchak.hmac.key_id'],  // auth-hmac
        ['source' => 'api_key'],                                     // auth-key client names
    ],
    'tenant' => [
        ['source' => 'token', 'claim' => 'org_id'],
    ],
];

The sources (token, api_key, attribute, header, callable, and map on any of them), and examples for each authentication middleware, are on the Consumer Identity page.

To bill differently from how quotas limit, set consumer or tenant in kipchak.metering.php itself. It replaces the shared definition for metering only, key by key.

For tokens from an identity provider, run auth-jwks or auth-jwt globally (initialised after Metering in middlewares/middlewares.php), so that the verified token is on the request when Metering reads it.

Recording what a request consumed

Metering attaches a Usage object to every metered request. A route can add to it while it runs:

use Kipchak\Middleware\Metering\Usage;

$usage = Usage::of($request);
$usage?->add('tokens', $completion->usage->totalTokens);   // repeated calls accumulate
$usage?->dimension('model', $model);                     // the last value wins
$usage?->setConsumer($token->sub);                       // overrides the configured sources
$usage?->setTenant($token->org);
$usage?->skip();                                         // record nothing for this request

Usage::of() returns null when the request is not metered, hence ?->. Meter and dimension names are 1–64 letters, digits, _, . or -. Quantities are numbers of zero or more. requests is reserved.

Units per route

Units that a route always consumes can be configured instead of coded. They are added to whatever the route records itself:

'units' => [
    'POST /v1/reports' => ['credits' => 5],
    '/v1/search' => ['credits' => 1],   // any method
],

Keys are route patterns exactly as defined in routes/, optionally preceded by the method. A key with the method takes precedence over one without.

Sinks

Kafka

'sink' => [
    'type' => 'kafka',
    'connection' => 'default',    // a connection in kipchak.kafka
    'topic' => 'kipchak.usage',
    'key' => 'consumer',          // 'consumer', 'tenant' or 'none'
],

Events are queued with the Kafka driver's produce() and delivered from librdkafka's background thread, so a request does not wait for the broker.

• Short outages lose nothing. Queued events are sent when the brokers return, as long as that is within the connection's delivery_timeout_ms.
• Longer outages are logged. An event that is not delivered in time, or that cannot be queued because the local queue is full, is logged at error level with the message Usage event was not delivered to Kafka and the complete event under usage_event, ready to replay.
• Graceful shutdown flushes. Events still queued when a worker stops are sent before it exits. A worker that is killed outright loses what was still queued, so keep the connection's linger_ms low.

The Kafka connection can also carry your API's own messages. Metering reports only failures on its topic.

Log

'sink' => ['type' => 'log', 'level' => 'info'],

Writes each event to the Kipchak logger (JSON on stdout) under the context key usage_event, for Vector, Fluent Bit or the OpenTelemetry Collector to route to your billing system.

Your own sink

'sink' => ['type' => 'custom', 'class' => App\Billing\UsageSink::class],

The class implements Kipchak\Middleware\Metering\Sink\Sink, with one method, emit(UsageEvent $event). It is built through the container when the container can build it, so it can have dependencies. emit() runs before the response is sent, so it should queue or buffer rather than make a network call. If it throws, the exception is logged with the event and the response is unaffected.

Per route instead of globally

Leave enabled as false and add Meter to the routes or groups to meter:

use Kipchak\Middleware\Metering\Handlers\Meter;

$app->post('/v1/completions/{model}', [Completions::class, 'create'])
    ->add(new Meter($app->getContainer()));

Testing your API

Put a MemorySink in the container before the first request, and assert on what a route was billed:

use Kipchak\Middleware\Metering\Sink\MemorySink;
use Kipchak\Middleware\Metering\Sink\SinkFactory;

$sink = new MemorySink();
$container->set(SinkFactory::CONTAINER_KEY, $sink);

// ... handle a request ...

$this->assertSame(1250, $sink->last()->units['tokens']);

Worker mode

Metering is safe in FrankenPHP worker mode. Its settings and sink are built once per worker; each request gets a new Usage object. A global Meter runs before Slim's routing, so it looks the route up itself, using Slim's dispatcher, which is built once per worker.

Git Repository

The source code for this middleware is hosted internally.