aforo-grpc-metering

aforo-grpc-metering

Meter every gRPC call with a server interceptor — one Aforo event per RPC, with the gRPC status mapped to a readable label, call type, and duration. Streaming RPCs are metered with one explicit record() call.

Version: 1.2.2 · Apache-2.0 · Changelog · User guide

Install

Intended public install:

pip install aforo-grpc-metering
pip install "aforo-grpc-metering[httpx]"     # or [aiohttp] — faster HTTP flush than stdlib urllib

Install 1.2.2 or later. 1.0.0 on PyPI was built from an older copy of this code and lacks the fixes listed in the changelog. If 1.2.2 is not on PyPI yet, install from source:

git clone https://github.com/aforoai/SDKs.git
cd SDKs/aforo-metering-sdks/python-grpc     # folder holding setup.py
pip install -e .
pip install -e ".[httpx]"              # or [aiohttp]

The one required dependency is grpcio>=1.50.

Quickstart — unary interceptor

Best when your service is mostly unary RPCs and you want per-call billing with no handler changes.

import os, grpc
from concurrent import futures
from aforo_grpc_metering import AforoGrpcBilling, AforoGrpcInterceptor

billing = AforoGrpcBilling(
    tenant_id="tenant_acme",
    product_id="prod_grpc_user_svc",
    api_key=os.environ["AFORO_API_KEY"],
    ingestor_url="https://api.aforo.ai",
    service_name="acme.v1.UserService",
)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[AforoGrpcInterceptor(billing)],
)
# add_UserServiceServicer_to_server(servicer, server)
server.add_insecure_port("[::]:50051")
server.start()
server.wait_for_termination()

Every unary RPC is now metered — one grpc_api.rpc_calls event with grpcStatusCode, grpcCallType=UNARY, and executionDurationMs, POSTed to https://api.aforo.ai/v1/ingest/batch with X-API-Key: <api_key> and X-Tenant-Id: <tenant_id>.

⚠ Events are sent to the ingestor's /v1/ingest/batch path as {"events": [...]}, at most 1000 events per request (larger buffers are split). Set ingestor_url to the host only — the SDK appends the path.

The interceptor auto-wraps unary RPCs only. For server-stream / client-stream / bidi, call billing.record(...) yourself at the end of the handler (see the user guide). tenant_id is fixed from config; the default extractor reads x-customer-id from invocation metadata, and calls with no resolvable customer ID are not metered.

Configuration

Constructor arguments for AforoGrpcBilling(...):

Status mapping: GRPC_STATUS_LABELS maps numeric codes to descriptor labels (e.g. OK, NOT_FOUND, UNAVAILABLE); the interceptor records the label as grpcStatusCode. Retry is fixed at 3 attempts (1s / 2s backoff between them); 408 and 5xx are retried, 429 waits for Retry-After (capped at 60 s), and any other 4xx is not retried and the batch is dropped with reason rejected.

Execution status (executionStatus)

Each event can carry an execution status. OUTCOME_BASED rate plans bill each event at the weight set for its status; events without one bill at full price.

The SDK derives it from the gRPC status code:

The interceptor reads the code the handler set with context.abort(...) or context.set_code(...) (grpcio 1.38+); without one, an uncaught exception counts as INTERNAL and a normal return as OK.

The mapping is exported as outcome_from_grpc_status(...) (takes a grpc.StatusCode, its int value, or its name). A value you pass yourself always wins:

billing.record("ListUsers", "SERVER_STREAM", customer_id, "OK", count, duration_ms,
               execution_status="PARTIAL")

# Interceptor: return a status from the handler's context (None keeps the derived one)
AforoGrpcInterceptor(billing, execution_status_resolver=lambda ctx: current_outcome.get(None))

execution_status is keyword-only. The SDK trims and upper-cases it and leaves it off the event when it's blank. If status isn't a gRPC status name or code, and you pass no value, the event goes out without one. Accepted values: SUCCESS, PARTIAL, TIMEOUT, ERROR, VALIDATION_FAILED, FAILED, FAILURE, CANCELLED, PENDING, BLOCKED, HITL_REQUIRED — any other value is logged as a warning and left off the event, since the ingestor would reject the event.

Dropped events

An event that will never reach Aforo is counted in billing.dropped_count, logged at WARNING, and passed to the opt-in on_drop(events, reason) hook. The events keep their idempotency keys, so sending them again later cannot double-bill.

def on_drop(events, reason):
    dead_letter.write(reason, events)

billing = AforoGrpcBilling(..., on_drop=on_drop)

When a 202 response reports refused events without a usable index, they are counted in dropped_count but not passed to the hook, since the SDK cannot tell which events they were. Exceptions raised by the hook are swallowed. An unknown executionStatus is not a drop: the field is left off and the event is still sent.

Walk me through it

Install → add the interceptor → call an RPC → confirm the event in Aforo, plus the streaming pattern, is in USER_GUIDE.md.

What this doesn't cover

The interceptor only auto-meters unary RPCs — streaming RPCs need a manual record() (the SDK can't know when a stream ends or how many messages flowed). It meters per-call, not per-message, unless you pass message_count/data_bytes to record(). It doesn't enforce quotas or abort RPCs. Pricing and metric mapping live in the Aforo console.