Files
Movses AslanyanandClaude Opus 5.5 920712f32b 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>
2026-10-19 16:05:40 +03:30

91 lines
3.1 KiB
Python

import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from opentelemetry import trace
from opentelemetry.trace import StatusCode
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
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
from .repository import is_match
SERVICE_NAME = "sanctions"
logger = configure_logging(SERVICE_NAME)
tracer = trace.get_tracer(__name__)
setup_telemetry(service_name=SERVICE_NAME)
app = FastAPI(title=SERVICE_NAME)
FastAPIInstrumentor.instrument_app(
app,
excluded_urls="/healthz",
exclude_spans=["receive", "send"],
)
SQLAlchemyInstrumentor().instrument(engines=[engine.sync_engine])
observe_db_pools({"default": engine})
class LookupRequest(BaseModel):
holder_name: str
@app.get("/")
async def root():
return {"service": SERVICE_NAME, "version": "0.1.0"}
@app.get("/healthz")
async def healthz(request: Request):
if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
logger.info("healthz_probe_received", probe="chatty-health")
return {"status": "ok", "service": SERVICE_NAME}
@app.get("/readyz")
async def readyz():
try:
async with engine.connect() as conn:
await conn.exec_driver_sql("SELECT 1")
except Exception as exc:
span = trace.get_current_span()
span.record_exception(exc)
span.set_status(StatusCode.ERROR, str(exc))
logger.exception("readyz_check_failed")
return JSONResponse(status_code=503, content={"status": "unavailable"})
return {"status": "ok"}
@app.post("/internal/lookup")
async def lookup(body: LookupRequest):
# holder_name is PII: logged by omission, not by redaction.
logger.info("lookup_received")
if await chaos.sanctions_errors_enabled() and chaos.should_error():
logger.info("chaos_engaged", chaos_flag="sanctions_errors")
return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})
if await chaos.slow_sanctions_enabled():
delay = chaos.sample_delay_seconds()
logger.info("chaos_engaged", chaos_flag="slow_sanctions", delay_ms=round(delay * 1000, 1))
await asyncio.sleep(delay)
# holder_name is PII: it stays off the span; only the outcome is recorded.
with tracer.start_as_current_span("sanctions.lookup_watchlist") as span:
match = await is_match(body.holder_name)
span.set_attribute("ledgerline.sanctions.matched", match is not None)
if match:
span.set_attribute("ledgerline.sanctions.list_source", match["list_source"])
logger.info(
"holder_screened",
matched=match is not None,
list_source=match["list_source"] if match else None,
)
if match:
return {"matched": True, **match}
return {"matched": False, "country": None, "list_source": None}