diff --git a/common/metrics.py b/common/metrics.py new file mode 100644 index 0000000..648d9f4 --- /dev/null +++ b/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", + ) diff --git a/services/audit/worker.py b/services/audit/worker.py index 61d7321..f05d0b5 100644 --- a/services/audit/worker.py +++ b/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()) diff --git a/services/fraud/app.py b/services/fraud/app.py index 9cfadc6..1ca60e9 100644 --- a/services/fraud/app.py +++ b/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 ) diff --git a/services/gateway/app.py b/services/gateway/app.py index c0004d5..2f7f1a4 100644 --- a/services/gateway/app.py +++ b/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}) diff --git a/services/insights/app.py b/services/insights/app.py index c3d90fd..359a042 100644 --- a/services/insights/app.py +++ b/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 diff --git a/services/ledger/app.py b/services/ledger/app.py index d442a57..19b1b58 100644 --- a/services/ledger/app.py +++ b/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): diff --git a/services/notifier/app.py b/services/notifier/app.py index d40be10..f734d04 100644 --- a/services/notifier/app.py +++ b/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("/") diff --git a/services/sanctions/app.py b/services/sanctions/app.py index 3babf05..a6df0f1 100644 --- a/services/sanctions/app.py +++ b/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