api-analytics

name	api-analytics
description	Implement API analytics to measure usage, performance, errors, and consumer behaviour. Outputs instrumentation design, metric taxonomy, dashboard specifications, and alerting strategy.
argument-hint	["API type","traffic volume","consumer types","observability stack","business goals"]
allowed-tools	Read, Write

API Analytics

API analytics gives you visibility into how your API is being used: which endpoints are popular, which consumers are most active, where latency is highest, and which errors are most common. This data drives product decisions, SLA negotiations, and capacity planning.

Metric Taxonomy

REQUEST METRICS
  api_requests_total{endpoint, method, status_code, consumer_id, version}
  api_request_duration_seconds{endpoint, method, version} — histogram
  api_request_size_bytes{endpoint, method}
  api_response_size_bytes{endpoint, method}

CONSUMER METRICS
  api_consumer_requests_total{consumer_id, endpoint}
  api_consumer_rate_limit_hits_total{consumer_id}
  api_consumer_errors_total{consumer_id, error_type}

BUSINESS METRICS
  api_revenue_generated{endpoint, consumer_id}   -- if trackable
  api_unique_consumers_active{endpoint, period}
  api_feature_adoption{feature_flag, endpoint}

ERROR METRICS
  api_errors_total{endpoint, error_code, error_type}
  api_timeout_total{endpoint, upstream}
  api_validation_errors_total{endpoint, field}

Instrumentation Middleware

import time
import hashlib
 prometheus_client  Counter, Histogram, Gauge
 fastapi  FastAPI, Request, Response

app = FastAPI()


requests_total = Counter(
    ,
    ,
    [, , , , ]
)

request_duration = Histogram(
    ,
    ,
    [, , ],
    buckets=[, , , , , , , , , , ]
)


  ():
    start = time.monotonic()
    correlation_id = request.headers.get(, )

    
    consumer_tier =  get_consumer_tier(request)
    version = extract_api_version(request.url.path)

    
    endpoint_template = normalise_path(request.url.path)

    response =  call_next(request)

    duration = time.monotonic() - start

    
    requests_total.labels(
        endpoint=endpoint_template,
        method=request.method,
        status_code=(response.status_code),
        consumer_tier=consumer_tier,
        version=version,
    ).inc()

    request_duration.labels(
        endpoint=endpoint_template,
        method=request.method,
        version=version,
    ).observe(duration)

    
     structlog
    structlog.get_logger().info(
        ,
        endpoint=endpoint_template,
        method=request.method,
        status_code=response.status_code,
        duration_ms=(duration * , ),
        consumer_tier=consumer_tier,
        version=version,
        correlation_id=correlation_id,
        
        consumer_hash=hashlib.sha256(
            request.headers.get(, ).encode()
        ).hexdigest()[:],
    )

     response

 () -> :
    
     re
    path = re.sub(, , path)
    path = re.sub(, , path)
     path