Add the five Ledgerline application metrics

- ledgerline.transactions.total (counter: kind, status, currency) and
  ledgerline.transaction.duration (histogram, seconds: kind) in gateway,
  the only place that sees the final outcome, failures included. currency
  is client input, so values outside the seeded set are recorded as
  "other".
- ledgerline.fraud.score (histogram, buckets in 0.1 steps to 1.0) in fraud.
- ledgerline.stream.lag (observable gauge: consumer_group) in audit and
  notifier, read from XINFO GROUPS through a separate sync Redis client,
  because callbacks run on the scrape thread; the call is untraced so
  scrapes don't create traces.
- ledgerline.db.pool.in_use (observable up-down counter: pool) in every
  service with a DB engine. The SQLAlchemy instrumentation's own pool
  metric names a pool by DSN, which merges the normal and starved engines.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

common/metrics.py

@@ -0,0 +1,58 @@
+"""Observable (polled) instruments shared by more than one service.
+
+Their callbacks run on prometheus_client's scrape thread, not on the event
+loop, so they only do thread-safe reads: a pool's checked-out count, and a
+query through a separate synchronous Redis client (the shared asyncio one
+belongs to the loop)."""
+import redis
+from opentelemetry import metrics
+from opentelemetry.instrumentation.utils import suppress_instrumentation
+from opentelemetry.metrics import CallbackOptions, Observation
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from .settings import settings
+
+meter = metrics.get_meter(__name__)
+
+
+def observe_db_pools(pools: dict[str, AsyncEngine]) -> None:
+    """ledgerline.db.pool.in_use, one series per named pool. The SQLAlchemy
+    instrumentation's own db.client.connections.usage names a pool by its
+    connection string, which a service's normal and starved engines share, so
+    it sums the two into one series."""
+
+    def callback(_options: CallbackOptions):
+        return [
+            Observation(engine.sync_engine.pool.checkedout(), {"pool": name})
+            for name, engine in pools.items()
+        ]
+
+    meter.create_observable_up_down_counter(
+        "ledgerline.db.pool.in_use",
+        callbacks=[callback],
+        unit="{connection}",
+        description="Connections currently checked out of the pool",
+    )
+
+
+def observe_stream_lag(consumer_group: str) -> None:
+    """ledgerline.stream.lag for this process's consumer group: stream entries
+    not yet delivered to the group (XINFO GROUPS `lag`, Redis 7+)."""
+    client = redis.Redis.from_url(settings.redis_url, decode_responses=True, socket_timeout=1)
+
+    def callback(_options: CallbackOptions):
+        # Untraced, or every scrape would be a one-span root trace in Tempo.
+        with suppress_instrumentation():
+            groups = client.xinfo_groups(settings.stream_name)
+        return [
+            Observation(group["lag"], {"consumer_group": consumer_group})
+            for group in groups
+            if group["name"] == consumer_group and group["lag"] is not None
+        ]
+
+    meter.create_observable_gauge(
+        "ledgerline.stream.lag",
+        callbacks=[callback],
+        unit="{message}",
+        description="Stream entries not yet delivered to the consumer group",
+    )

services/audit/worker.py

@@ -4,6 +4,7 @@ from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
 
 from common.db import engine
 from common.logging_config import configure_logging
+from common.metrics import observe_db_pools, observe_stream_lag
 from common.telemetry import setup_telemetry
 
 from . import consumer
@@ -11,6 +12,8 @@ from . import consumer
 configure_logging("audit")
 setup_telemetry(service_name="audit")
 SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine])
+observe_db_pools({"default": engine})
+observe_stream_lag(consumer.GROUP)
 
 if __name__ == "__main__":
     asyncio.run(consumer.run())

services/fraud/app.py

@@ -2,7 +2,7 @@ import httpx
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
-from opentelemetry import trace
+from opentelemetry import metrics, trace
 from opentelemetry.trace import StatusCode
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 
@@ -27,6 +27,13 @@ FastAPIInstrumentor.instrument_app(
     exclude_spans=["receive", "send"],
 )
 
+meter = metrics.get_meter(__name__)
+fraud_score = meter.create_histogram(
+    "ledgerline.fraud.score",
+    description="Risk score per scoring call (0-1; gateway rejects above 0.8)",
+    explicit_bucket_boundaries_advisory=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
+)
+
 
 class ScoreRequest(BaseModel):
     account_id: str
@@ -110,6 +117,7 @@ async def score_transaction(body: ScoreRequest):
         risk_score = score(body.category_id, matched)
         span.set_attribute("ledgerline.sanctions.matched", matched)
         span.set_attribute("ledgerline.risk.score", risk_score)
+        fraud_score.record(risk_score)
         logger.info(
             "transaction_scored", account_id=body.account_id, risk_score=risk_score, matched=matched
         )

services/gateway/app.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 import uuid
 from pathlib import Path
 
@@ -7,7 +8,7 @@ from fastapi import FastAPI, Request
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel, field_validator
-from opentelemetry import trace
+from opentelemetry import metrics, trace
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.utils import suppress_instrumentation
 
@@ -30,6 +31,25 @@ FastAPIInstrumentor.instrument_app(
     exclude_spans=["receive", "send"],
 )
 
+meter = metrics.get_meter(__name__)
+transactions_total = meter.create_counter(
+    "ledgerline.transactions.total",
+    unit="{transaction}",
+    description="Transactions submitted, by final outcome",
+)
+transaction_duration = meter.create_histogram(
+    "ledgerline.transaction.duration",
+    unit="s",
+    description="End-to-end time to process a submitted transaction",
+    # Seconds, not the SDK's millisecond-shaped defaults (0, 5, 10, ... 10000).
+    explicit_bucket_boundaries_advisory=[
+        0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10
+    ],
+)
+# currency is unvalidated client input; only the seeded
+# set becomes a label value, so a bad client can't mint new series.
+KNOWN_CURRENCIES = {"EUR", "USD", "GBP"}
+
 STATIC_DIR = Path(__file__).parent / "static"
 app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
 
@@ -115,8 +135,11 @@ async def account_insights(account_id: str):
 @app.post("/api/transactions", status_code=201)
 async def submit_transaction(body: TransactionRequest):
     request_id = _request_id()
+    start = time.perf_counter()
+    status = "failed"
     try:
         result = await orchestrator.create_transaction(body.model_dump())
+        status = result["status"]
     except UpstreamFailure as exc:
         log = logger.error if exc.status_code >= 500 else logger.warning
         log(
@@ -129,6 +152,10 @@ async def submit_transaction(body: TransactionRequest):
         return JSONResponse(
             status_code=exc.status_code, content={"error": exc.message, "request_id": request_id}
         )
+    finally:
+        transaction_duration.record(time.perf_counter() - start, {"kind": body.kind})
+        currency = body.currency if body.currency in KNOWN_CURRENCIES else "other"
+        transactions_total.add(1, {"kind": body.kind, "status": status, "currency": currency})
     return JSONResponse(status_code=201, content={**result, "request_id": request_id})
 
 

services/insights/app.py

@@ -11,6 +11,7 @@ from opentelemetry.trace import StatusCode
 
 from common.chaos_flags import flags_fingerprint
 from common.logging_config import configure_logging
+from common.metrics import observe_db_pools
 from common.redis_client import redis_client
 from common.settings import settings
 from common.telemetry import setup_telemetry
@@ -34,6 +35,7 @@ FastAPIInstrumentor.instrument_app(
 SQLAlchemyInstrumentor().instrument(
     engines=[normal_engine.sync_engine, starved_engine.sync_engine]
 )
+observe_db_pools({"normal": normal_engine, "starved": starved_engine})
 
 # Cache-stampede protection for the Redis-cached branch below: a burst of
 # concurrent requests for the same not-yet-cached (account, flags) key would

services/ledger/app.py

@@ -7,6 +7,7 @@ from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
 
 from common.logging_config import configure_logging
+from common.metrics import observe_db_pools
 from common.telemetry import setup_telemetry
 
 from . import chaos
@@ -27,6 +28,7 @@ FastAPIInstrumentor.instrument_app(
 SQLAlchemyInstrumentor().instrument(
     engines=[normal_engine.sync_engine, starved_engine.sync_engine]
 )
+observe_db_pools({"normal": normal_engine, "starved": starved_engine})
 
 
 class CreateTransactionRequest(BaseModel):

services/notifier/app.py

@@ -11,6 +11,7 @@ from opentelemetry.trace import StatusCode
 
 from common.db import engine
 from common.logging_config import configure_logging
+from common.metrics import observe_db_pools, observe_stream_lag
 from common.redis_client import redis_client
 from common.telemetry import setup_telemetry
 
@@ -36,6 +37,8 @@ FastAPIInstrumentor.instrument_app(
     exclude_spans=["receive", "send"],
 )
 SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine])
+observe_db_pools({"default": engine})
+observe_stream_lag(consumer.GROUP)
 
 
 @app.get("/")

services/sanctions/app.py

@@ -10,6 +10,7 @@ from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
 
 from common.db import engine
 from common.logging_config import configure_logging
+from common.metrics import observe_db_pools
 from common.telemetry import setup_telemetry
 
 from . import chaos
@@ -27,6 +28,7 @@ FastAPIInstrumentor.instrument_app(
     exclude_spans=["receive", "send"],
 )
 SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine])
+observe_db_pools({"default": engine})
 
 class LookupRequest(BaseModel):
     holder_name: str