Usage Tracking & Metering

Overview

Usage tracking and metering is the process of measuring, recording, and aggregating resource consumption in a SaaS platform so that customers can be billed accurately and operators can understand capacity demands. Unlike monitoring (which tracks system health) or audit logging (which tracks user actions), metering tracks quantitative consumption: how many queries a tenant ran, how many tokens their AI answers consumed, how many bytes of storage they used, and how many hours their clusters ran. These usage metrics are the source of truth for billing — every dollar on the invoice traces back to a metered event.

In a SaaS management platform, usage metering tracks three primary dimensions: queries executed (broken down by type — search, analytics, admin), AI-generated answers (measured in tokens consumed by the LLM inference pipeline), and storage consumed (bytes stored across all of a tenant’s clusters). Usage events are collected in real time, aggregated into time buckets (hourly and daily), and exposed through both a customer-facing usage dashboard and an internal billing pipeline. A background cleanup job prunes granular usage records after aggregation, keeping storage costs manageable while preserving the aggregated totals needed for invoicing.

This topic covers the architecture of a usage metering pipeline, time-bucketed aggregation, idempotent event recording, billing integration, and the cleanup strategies that keep the metering system sustainable at scale.

Core Concepts

How It Works

The platform’s metering pipeline has four stages: capture (recording individual usage events), aggregate (rolling events into time buckets), serve (exposing aggregated data to dashboards and billing), and clean (purging stale granular data). Each stage operates independently, connected by the database as the shared state store.

Usage Event Schema

-- Granular usage events — high volume, short retention.
CREATE TABLE usage_events (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id   UUID NOT NULL,
    event_type  VARCHAR(50) NOT NULL,  -- "query", "answer", "storage_sample"
    metric      VARCHAR(50) NOT NULL,  -- "query_count", "tokens", "bytes_stored"
    value       BIGINT NOT NULL,       -- the measured quantity
    dimensions  JSONB,                 -- {"query_type":"search","model":"gpt-4"}
    event_id    VARCHAR(255) UNIQUE,   -- idempotency key
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Aggregated hourly buckets — medium volume, long retention.
CREATE TABLE usage_hourly (
    tenant_id   UUID NOT NULL,
    metric      VARCHAR(50) NOT NULL,
    bucket      TIMESTAMPTZ NOT NULL,  -- start of the hour
    value       BIGINT NOT NULL DEFAULT 0,
    dimensions  JSONB,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (tenant_id, metric, bucket, dimensions)
);

-- Aggregated daily buckets — low volume, billing retention.
CREATE TABLE usage_daily (
    tenant_id   UUID NOT NULL,
    metric      VARCHAR(50) NOT NULL,
    bucket      DATE NOT NULL,
    value       BIGINT NOT NULL DEFAULT 0,
    dimensions  JSONB,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (tenant_id, metric, bucket, dimensions)
);

CREATE INDEX idx_usage_events_tenant_time ON usage_events (tenant_id, created_at);
CREATE INDEX idx_usage_hourly_tenant ON usage_hourly (tenant_id, bucket DESC);
CREATE INDEX idx_usage_daily_tenant ON usage_daily (tenant_id, bucket DESC);

Recording Usage Events

// internal/metering/recorder.go
package metering

import (
    "context"
    "time"
)

type Recorder struct {
    events chan UsageEvent
    store  Store
    logger *zap.Logger
}

type UsageEvent struct {
    TenantID   string
    EventType  string
    Metric     string
    Value      int64
    Dimensions map[string]string
    EventID    string // idempotency key
}

func NewRecorder(store Store, logger *zap.Logger, bufSize int) *Recorder {
    return &Recorder{
        events: make(chan UsageEvent, bufSize),
        store:  store,
        logger: logger.Named("metering"),
    }
}

// Record enqueues a usage event. Non-blocking — drops if buffer full.
func (r *Recorder) Record(event UsageEvent) {
    select {
    case r.events <- event:
    default:
        r.logger.Warn("usage event dropped: buffer full",
            zap.String("tenant_id", event.TenantID),
            zap.String("metric", event.Metric),
        )
    }
}

// Run processes events in batches and writes them to the store.
func (r *Recorder) Run(ctx context.Context) {
    batch := make([]UsageEvent, 0, 200)
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case ev := <-r.events:
            batch = append(batch, ev)
            if len(batch) >= 200 {
                r.flush(ctx, batch)
                batch = batch[:0]
            }
        case <-ticker.C:
            if len(batch) > 0 {
                r.flush(ctx, batch)
                batch = batch[:0]
            }
        case <-ctx.Done():
            // Drain on shutdown.
            close(r.events)
            for ev := range r.events {
                batch = append(batch, ev)
            }
            if len(batch) > 0 {
                r.flush(context.Background(), batch)
            }
            return
        }
    }
}

func (r *Recorder) flush(ctx context.Context, batch []UsageEvent) {
    if err := r.store.InsertEvents(ctx, batch); err != nil {
        r.logger.Error("failed to flush usage events",
            zap.Int("batch_size", len(batch)),
            zap.Error(err),
        )
    }
}

Hourly Aggregation Job

// internal/metering/aggregator.go
func (a *Aggregator) AggregateHour(ctx context.Context, hour time.Time) error {
    // Truncate to the start of the hour.
    bucket := hour.Truncate(time.Hour)

    // Aggregate raw events into hourly buckets using an upsert.
    query := `
        INSERT INTO usage_hourly (tenant_id, metric, bucket, value, dimensions, updated_at)
        SELECT
            tenant_id,
            metric,
            date_trunc('hour', created_at) AS bucket,
            SUM(value) AS value,
            dimensions,
            NOW()
        FROM usage_events
        WHERE created_at >= $1 AND created_at < $2
        GROUP BY tenant_id, metric, date_trunc('hour', created_at), dimensions
        ON CONFLICT (tenant_id, metric, bucket, dimensions)
        DO UPDATE SET
            value = usage_hourly.value + EXCLUDED.value,
            updated_at = NOW()
    `
    _, err := a.db.ExecContext(ctx, query, bucket, bucket.Add(time.Hour))
    if err != nil {
        return fmt.Errorf("aggregate hour %s: %w", bucket, err)
    }

    a.logger.Info("hourly aggregation completed",
        zap.Time("bucket", bucket),
    )
    return nil
}

Instrumenting the Query Path

// internal/api/query_handler.go
func (h *QueryHandler) HandleQuery(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    actor := auth.ActorFromContext(ctx)
    req := parseQueryRequest(r)

    result, err := h.queryEngine.Execute(ctx, req)
    if err != nil {
        respondError(w, err)
        return
    }

    // Record usage: query count.
    h.metering.Record(metering.UsageEvent{
        TenantID:  actor.TenantID,
        EventType: "query",
        Metric:    "query_count",
        Value:     1,
        Dimensions: map[string]string{
            "query_type": req.Type, // "search", "analytics", "admin"
        },
        EventID: middleware.RequestIDFromContext(ctx),
    })

    // Record usage: tokens consumed (if AI answer was generated).
    if result.TokensUsed > 0 {
        h.metering.Record(metering.UsageEvent{
            TenantID:  actor.TenantID,
            EventType: "answer",
            Metric:    "tokens",
            Value:     int64(result.TokensUsed),
            Dimensions: map[string]string{
                "model": result.ModelUsed,
            },
            EventID: middleware.RequestIDFromContext(ctx) + ":tokens",
        })
    }

    respondJSON(w, result)
}

Cleanup Job

// internal/metering/cleanup.go
func (c *Cleaner) PurgeOldEvents(ctx context.Context) error {
    cutoff := time.Now().UTC().Add(-c.eventRetention) // e.g., -7 days

    result, err := c.db.ExecContext(ctx,
        "DELETE FROM usage_events WHERE created_at < $1", cutoff,
    )
    if err != nil {
        return fmt.Errorf("purge events: %w", err)
    }

    rows, _ := result.RowsAffected()
    c.logger.Info("usage event cleanup completed",
        zap.Int64("deleted", rows),
        zap.Time("cutoff", cutoff),
    )
    return nil
}

Practical Applications

The platform’s metering data flows into three downstream systems. The customer-facing Usage Dashboard shows real-time and historical usage broken down by metric type, with hourly granularity for the current billing period and daily granularity for previous periods. Customers use this to understand their consumption patterns, forecast upcoming bills, and identify unexpected usage spikes — for example, a runaway integration that fires 100x more queries than expected.

The internal billing pipeline reads daily aggregated usage at the end of each billing cycle, applies the tenant’s pricing plan (per-query rates, token pricing tiers, storage per-GB rates), and generates an invoice. The pipeline includes a reconciliation step that compares metered usage against Prometheus metrics as a sanity check — if the metered query count diverges from the Prometheus request counter by more than 1%, the billing run is flagged for manual review.

The capacity planning system uses aggregated usage trends to forecast infrastructure needs. By analyzing weekly growth rates in queries, tokens, and storage across all tenants, the platform team can predict when additional database nodes, GPU instances, or storage volumes will be needed — weeks before capacity limits are hit.

Typical Metered Dimensions

• Query count — number of search, analytics, and admin queries per tenant per hour.
• Token consumption — input and output tokens consumed by AI answer generation.
• Storage bytes — peak daily storage across all of a tenant’s clusters.
• Data transfer — bytes transferred in API responses (egress metering).
• Compute hours — wall-clock time that a tenant’s clusters were running.

Implementation Patterns

Idempotent Upserts for At-Least-Once Delivery

When usage events arrive through a message queue or are retried after transient failures, duplicates are inevitable. Use the request ID or a deterministic event ID as an idempotency key with ON CONFLICT DO NOTHING for raw events. For aggregation upserts, use ON CONFLICT DO UPDATE SET value = value + EXCLUDED.value — this is safe because the aggregation query runs over the same time window each time (the raw events are idempotent, so re-aggregating produces the same delta).

Separate Read and Write Paths

The write path (recording events, aggregating buckets) and the read path (dashboard queries, billing exports) have very different access patterns. The write path is high-throughput, append-only, and latency-insensitive. The read path is lower-throughput but requires fast, indexed lookups by tenant and time range. Use separate database connections (or connection pool priorities) for reads and writes to prevent write-heavy aggregation jobs from starving dashboard queries.

Billing Reconciliation

Metering systems can drift from reality due to bugs, dropped events, or race conditions. Cross-check metered totals against independent data sources — Prometheus request counters, API gateway access logs, or infrastructure provider billing data. Build the reconciliation check into the billing pipeline so discrepancies are caught before invoices are sent, not after.

// internal/billing/reconcile.go
func (r *Reconciler) Check(ctx context.Context, tenantID string, period BillingPeriod) error {
    metered, err := r.meteringStore.QueryTotal(ctx, tenantID, "query_count", period)
    if err != nil {
        return err
    }

    prometheus, err := r.prometheusClient.QueryRange(ctx,
        fmt.Sprintf(`sum(increase(mydb_api_requests_total{tenant="%s"}[1h]))`, tenantID),
        period.Start, period.End,
    )
    if err != nil {
        return err
    }

    drift := math.Abs(float64(metered-prometheus)) / float64(prometheus)
    if drift > 0.01 { // >1% discrepancy
        return fmt.Errorf("metering drift %.2f%% for tenant %s: metered=%d prometheus=%d",
            drift*100, tenantID, metered, prometheus)
    }
    return nil
}

Common Pitfalls

Double-Counting on Retries

Without idempotent event recording, a network retry that re-sends the same usage event inflates the customer’s bill. This is the most common and most damaging metering bug. Every usage event must have a unique, deterministic ID derived from the operation that generated it (the request ID is ideal). Use database-level uniqueness constraints — application-level dedup checks are vulnerable to race conditions.

Time Zone Confusion in Bucket Boundaries

If the aggregation job uses local time for bucket boundaries and the server’s time zone changes (DST transition, VM migration), buckets overlap or have gaps. Always use UTC for bucket timestamps. Store all timestamps as TIMESTAMPTZ in PostgreSQL and truncate to hour/day boundaries in UTC. Display local time in the customer-facing dashboard, but never use local time internally.

Aggregating Before Events Are Complete

If the hourly aggregation job runs at 15:00:00 for the 14:00–15:00 bucket, events that arrive with a few seconds of clock skew or network delay after 15:00:00 are missed. Add a lag window: aggregate the 14:00–15:00 bucket at 15:05:00, giving late-arriving events a 5-minute grace period. Use the upsert pattern so re-running aggregation for the same bucket is safe.

Unbounded Dimensions in Usage Events

If dimensions include high-cardinality values (individual query IDs, user IDs, or full query text), the aggregated usage table explodes in size. Dimensions should be low-cardinality grouping keys — query type (search, analytics, admin), model name (gpt-4, claude-3), storage tier (hot, cold). High-cardinality details belong in the raw event (which is cleaned up quickly), not in the aggregated buckets.

No Backfill Strategy

When the metering pipeline is down (database outage, deployment bug), usage events are lost. Without a backfill strategy, the billing period has gaps that either undercharge the customer (revenue loss) or require manual adjustment (operational overhead). Retain enough data in secondary sources (API gateway logs, Prometheus metrics) to reconstruct metered usage for backfill. Build and test the backfill procedure before you need it.

Further Reading