Movses AslanyanandClaude Opus 5.5
ec6ddb6a26
sanctions, notifier, ledger, insights and audit now call setup_telemetry and instrument their engines with one SQLAlchemyInstrumentor call per process (a second instrument() call is ignored, leaving that engine with connect spans only). ledger and insights pass both their normal and starved engines. The FastAPI services also get the server span, with the same /healthz and receive/send exclusions as sanctions. Compose: OTEL_SEMCONV_STABILITY_OPT_IN=http,database for the stable attribute names, the OTLP endpoint moves to the shared env, and the four newly instrumented services get the auth header from .env.observability. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
import asyncio
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
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.telemetry import setup_telemetry
|
|
|
|
from . import chaos
|
|
from .repository import is_match
|
|
|
|
SERVICE_NAME = "sanctions"
|
|
logger = configure_logging(SERVICE_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])
|
|
|
|
class LookupRequest(BaseModel):
|
|
holder_name: str
|
|
request_id: str = "unknown"
|
|
|
|
|
|
@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(f"received chatty-health healthz probe from gateway")
|
|
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:
|
|
logger.exception(f"readyz check failed: {exc}")
|
|
return JSONResponse(status_code=503, content={"status": "unavailable"})
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/internal/lookup")
|
|
async def lookup(body: LookupRequest):
|
|
logger.info(f"lookup for holder_name={body.holder_name!r} request_id={body.request_id}")
|
|
|
|
if await chaos.sanctions_errors_enabled() and chaos.should_error():
|
|
logger.info(f"sanctions_errors chaos triggered for request_id={body.request_id}")
|
|
return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})
|
|
|
|
if await chaos.slow_sanctions_enabled():
|
|
delay = chaos.sample_delay_seconds()
|
|
logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s for request_id={body.request_id}")
|
|
await asyncio.sleep(delay)
|
|
|
|
match = await is_match(body.holder_name)
|
|
if match:
|
|
return {"matched": True, **match}
|
|
return {"matched": False, "country": None, "list_source": None}
|