diff --git a/services/fraud/app.py b/services/fraud/app.py index 4281417..dfdbaf8 100644 --- a/services/fraud/app.py +++ b/services/fraud/app.py @@ -2,6 +2,7 @@ import httpx from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel +from opentelemetry import trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from common.http import downstream_client @@ -15,6 +16,7 @@ from .scoring import score SERVICE_NAME = "fraud" logger = configure_logging(SERVICE_NAME) +tracer = trace.get_tracer(__name__) setup_telemetry(service_name=SERVICE_NAME) app = FastAPI(title=SERVICE_NAME) @@ -64,23 +66,34 @@ async def score_transaction(body: ScoreRequest): logger.info(f"blocking_event_loop chaos engaged for request_id={body.request_id}") chaos.synchronous_cpu_work() - async with downstream_client(body.request_id) as client: - try: - resp = await client.post( - f"{settings.sanctions_url}/internal/lookup", - json={"holder_name": body.holder_name, "request_id": body.request_id}, - ) - except httpx.TimeoutException: - logger.exception(f"sanctions lookup timed out for request_id={body.request_id}") - return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"}) - except httpx.RequestError as exc: - logger.exception(f"sanctions lookup failed for request_id={body.request_id}: {exc}") + with tracer.start_as_current_span( + "fraud.score_transaction", + attributes={ + "ledgerline.account.id": body.account_id, + "ledgerline.tx.amount_minor": body.amount_minor, + }, + ) as span: + if body.category_id is not None: + span.set_attribute("ledgerline.merchant.category_id", body.category_id) + async with downstream_client(body.request_id) as client: + try: + resp = await client.post( + f"{settings.sanctions_url}/internal/lookup", + json={"holder_name": body.holder_name, "request_id": body.request_id}, + ) + except httpx.TimeoutException: + logger.exception(f"sanctions lookup timed out for request_id={body.request_id}") + return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"}) + except httpx.RequestError as exc: + logger.exception(f"sanctions lookup failed for request_id={body.request_id}: {exc}") + return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) + + if resp.status_code >= 300: + logger.info(f"sanctions returned {resp.status_code} for request_id={body.request_id}") return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) - if resp.status_code >= 300: - logger.info(f"sanctions returned {resp.status_code} for request_id={body.request_id}") - return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) - - matched = resp.json().get("matched", False) - risk_score = score(body.category_id, matched) - return {"risk_score": risk_score, "matched": matched} + matched = resp.json().get("matched", False) + risk_score = score(body.category_id, matched) + span.set_attribute("ledgerline.sanctions.matched", matched) + span.set_attribute("ledgerline.risk.score", risk_score) + return {"risk_score": risk_score, "matched": matched} diff --git a/services/gateway/app.py b/services/gateway/app.py index 959f092..3ce85af 100644 --- a/services/gateway/app.py +++ b/services/gateway/app.py @@ -8,6 +8,7 @@ from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, field_validator from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor +from opentelemetry.instrumentation.utils import suppress_instrumentation from common.chaos_flags import ALL_FLAGS, get_flags, set_flag, set_profile from common.logging_config import configure_logging @@ -133,7 +134,10 @@ async def health_all(): except httpx.RequestError: return name, "down" - results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()]) + # The route itself is in excluded_urls; without this its five downstream + # /healthz calls would each still be a parentless root trace. + with suppress_instrumentation(): + results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()]) services = {"gateway": "ok", **dict(results)} return {"services": services} diff --git a/services/insights/analytics.py b/services/insights/analytics.py index 086437f..89fda66 100644 --- a/services/insights/analytics.py +++ b/services/insights/analytics.py @@ -1,4 +1,5 @@ import sqlalchemy as sa +from opentelemetry import trace # Deliberately the same shape of query the test suite's own EXPLAIN check runs # against the schema directly (tests/test_pathologies.py, 6.4) -- kept here @@ -6,6 +7,8 @@ import sqlalchemy as sa _N_PLUS_ONE_ROW_CAP = 1500 +tracer = trace.get_tracer(__name__) + async def account_exists(session, account_id: str) -> bool: result = await session.execute(sa.text("SELECT 1 FROM accounts WHERE id = :id"), {"id": account_id}) @@ -13,29 +16,34 @@ async def account_exists(session, account_id: str) -> bool: async def compute_summary(session, account_id: str, *, n_plus_one: bool) -> dict: - if n_plus_one: - rows = await _n_plus_one_summary(session, account_id) - else: - rows = await _joined_summary(session, account_id) + with tracer.start_as_current_span( + "insights.aggregate_120d", + attributes={"ledgerline.account.id": account_id, "ledgerline.summary.window_days": 120}, + ) as span: + if n_plus_one: + rows = await _n_plus_one_summary(session, account_id) + else: + rows = await _joined_summary(session, account_id) - total_count = 0 - total_amount = 0 - by_category: dict = {} - for r in rows: - total_count += 1 - total_amount += r["amount_minor"] - label = r["label"] or "uncategorized" - entry = by_category.setdefault(label, {"count": 0, "total_amount_minor": 0}) - entry["count"] += 1 - entry["total_amount_minor"] += r["amount_minor"] + total_count = 0 + total_amount = 0 + by_category: dict = {} + for r in rows: + total_count += 1 + total_amount += r["amount_minor"] + label = r["label"] or "uncategorized" + entry = by_category.setdefault(label, {"count": 0, "total_amount_minor": 0}) + entry["count"] += 1 + entry["total_amount_minor"] += r["amount_minor"] - return { - "account_id": account_id, - "window_days": 120, - "transaction_count": total_count, - "total_amount_minor": total_amount, - "by_category": by_category, - } + span.set_attribute("ledgerline.summary.transaction_count", total_count) + return { + "account_id": account_id, + "window_days": 120, + "transaction_count": total_count, + "total_amount_minor": total_amount, + "by_category": by_category, + } async def _joined_summary(session, account_id: str) -> list[dict]: diff --git a/services/ledger/app.py b/services/ledger/app.py index 0c89e05..9b02abb 100644 --- a/services/ledger/app.py +++ b/services/ledger/app.py @@ -1,6 +1,7 @@ from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, field_validator +from opentelemetry import trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor @@ -13,6 +14,7 @@ from . import repository SERVICE_NAME = "ledger" logger = configure_logging(SERVICE_NAME) +tracer = trace.get_tracer(__name__) setup_telemetry(service_name=SERVICE_NAME) app = FastAPI(title=SERVICE_NAME) @@ -133,13 +135,19 @@ async def settle(tx_id: str, body: SettleRequest): starved = await chaos.pool_starvation_enabled() row_lock = await chaos.row_lock_contention_enabled() - async with session_factory(starved=starved) as session: - tx = await repository.get_transaction(session, tx_id) - if tx is None: - return JSONResponse(status_code=404, content={"error": "transaction not found"}) - account = await repository.get_account(session, tx.account_id) - use_lock = row_lock and bool(account.is_hot) - balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock) + with tracer.start_as_current_span( + "ledger.settle", + attributes={"ledgerline.tx.id": tx_id, "ledgerline.risk.score": body.risk_score}, + ) as span: + async with session_factory(starved=starved) as session: + tx = await repository.get_transaction(session, tx_id) + if tx is None: + return JSONResponse(status_code=404, content={"error": "transaction not found"}) + span.set_attribute("ledgerline.account.id", str(tx.account_id)) + span.set_attribute("ledgerline.tx.amount_minor", tx.amount_minor) + account = await repository.get_account(session, tx.account_id) + use_lock = row_lock and bool(account.is_hot) + balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock) logger.info(f"settled transaction {tx_id} request_id={body.request_id}") return {"status": "settled", "balance_minor": balance} diff --git a/services/sanctions/app.py b/services/sanctions/app.py index 723539a..b1332b8 100644 --- a/services/sanctions/app.py +++ b/services/sanctions/app.py @@ -3,6 +3,7 @@ import asyncio from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel +from opentelemetry import trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor @@ -15,6 +16,7 @@ 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) @@ -66,7 +68,12 @@ async def lookup(body: LookupRequest): 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) + # 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"]) if match: return {"matched": True, **match} return {"matched": False, "country": None, "list_source": None}