diff --git a/services/audit/consumer.py b/services/audit/consumer.py index b1e1e52..5b429f3 100644 --- a/services/audit/consumer.py +++ b/services/audit/consumer.py @@ -1,8 +1,8 @@ import asyncio import json -import logging import sqlalchemy as sa +import structlog from opentelemetry import propagate, trace from opentelemetry.instrumentation.utils import suppress_instrumentation from opentelemetry.trace import SpanKind, StatusCode @@ -11,7 +11,7 @@ from common.db import Session from common.redis_client import redis_client from common.settings import settings -logger = logging.getLogger("audit") +logger = structlog.get_logger("audit") tracer = trace.get_tracer(__name__) GROUP = "audit" @@ -46,7 +46,7 @@ async def run(): proves the async fork fans out to genuinely separate consumers, not one consumer doing both jobs.""" await ensure_group() - logger.info("audit consumer started") + logger.info("consumer_started", consumer_group=GROUP) while True: try: # The blocking poll would otherwise be a ~2s root trace every loop. @@ -61,7 +61,7 @@ async def run(): except asyncio.CancelledError: raise except Exception: - logger.exception("audit consumer read failed") + logger.exception("stream_read_failed", consumer_group=GROUP) await asyncio.sleep(1) continue @@ -85,8 +85,19 @@ async def run(): ) as span: try: await _handle(fields) + logger.info( + "stream_message_processed", + consumer_group=GROUP, + message_id=msg_id, + transaction_id=fields.get("transaction_id"), + ) except Exception as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"audit failed to process message {msg_id}") + logger.exception( + "stream_message_failed", + consumer_group=GROUP, + message_id=msg_id, + transaction_id=fields.get("transaction_id"), + ) await redis_client.xack(settings.stream_name, GROUP, msg_id) diff --git a/services/fraud/app.py b/services/fraud/app.py index 08b8969..9cfadc6 100644 --- a/services/fraud/app.py +++ b/services/fraud/app.py @@ -43,7 +43,7 @@ async def root(): @app.get("/healthz") async def healthz(request: Request): if request.headers.get("X-Ledgerline-Probe") == "chatty-health": - logger.info("received chatty-health healthz probe from gateway") + logger.info("healthz_probe_received", probe="chatty-health") return {"status": "ok", "service": SERVICE_NAME} @@ -55,18 +55,23 @@ async def readyz(): span = trace.get_current_span() span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"readyz check failed: {exc}") + logger.exception("readyz_check_failed") return JSONResponse(status_code=503, content={"status": "unavailable"}) return {"status": "ok"} @app.post("/internal/score") async def score_transaction(body: ScoreRequest): - logger.info(f"scoring account {body.account_id}") + logger.info( + "score_requested", + account_id=body.account_id, + amount_minor=body.amount_minor, + category_id=body.category_id, + ) await redis_client.incr("fraud:calls_total") if await chaos.blocking_event_loop_enabled(): - logger.info(f"blocking_event_loop chaos engaged for account {body.account_id}") + logger.info("chaos_engaged", chaos_flag="blocking_event_loop", account_id=body.account_id) chaos.synchronous_cpu_work() with tracer.start_as_current_span( @@ -87,20 +92,25 @@ async def score_transaction(body: ScoreRequest): except httpx.TimeoutException as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, "sanctions lookup timed out") - logger.exception(f"sanctions lookup timed out for account {body.account_id}") + logger.exception("sanctions_check_timed_out", account_id=body.account_id) return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"}) except httpx.RequestError as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, "sanctions lookup failed") - logger.exception(f"sanctions lookup failed for account {body.account_id}: {exc}") + logger.exception("sanctions_check_failed", account_id=body.account_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 account {body.account_id}") + logger.warning( + "sanctions_check_failed", account_id=body.account_id, status_code=resp.status_code + ) return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"}) 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) + logger.info( + "transaction_scored", account_id=body.account_id, risk_score=risk_score, matched=matched + ) return {"risk_score": risk_score, "matched": matched} diff --git a/services/gateway/app.py b/services/gateway/app.py index 24603fb..c0004d5 100644 --- a/services/gateway/app.py +++ b/services/gateway/app.py @@ -118,6 +118,14 @@ async def submit_transaction(body: TransactionRequest): try: result = await orchestrator.create_transaction(body.model_dump()) except UpstreamFailure as exc: + log = logger.error if exc.status_code >= 500 else logger.warning + log( + "transaction_failed", + account_id=body.account_id, + status_code=exc.status_code, + reason=exc.message, + cause=repr(exc.__cause__) if exc.__cause__ else None, + ) return JSONResponse( status_code=exc.status_code, content={"error": exc.message, "request_id": request_id} ) diff --git a/services/gateway/orchestrator.py b/services/gateway/orchestrator.py index 170b01a..b8a340a 100644 --- a/services/gateway/orchestrator.py +++ b/services/gateway/orchestrator.py @@ -1,8 +1,8 @@ -import logging import time import uuid import httpx +import structlog from opentelemetry import propagate, trace from opentelemetry.trace import SpanKind @@ -12,7 +12,7 @@ from common.settings import settings from . import chaos -logger = logging.getLogger("gateway") +logger = structlog.get_logger("gateway") tracer = trace.get_tracer(__name__) REJECT_THRESHOLD = 0.8 @@ -76,7 +76,12 @@ async def _run_chatty_health_probe(client: httpx.AsyncClient) -> None: async def create_transaction(payload: dict) -> dict: start = time.perf_counter() - logger.info(f"start transaction for account {payload['account_id']}") + logger.info( + "transaction_started", + account_id=payload["account_id"], + kind=payload["kind"], + amount_minor=payload["amount_minor"], + ) retry_storm = await chaos.retry_storm_enabled() no_batching = await chaos.no_batching_enabled() @@ -155,7 +160,7 @@ async def create_transaction(payload: dict) -> dict: msg_id = await redis_client.xadd(settings.stream_name, fields) span.set_attribute("messaging.message.id", msg_id) except Exception: - logger.exception(f"failed to publish transaction {tx_id} to the stream") + logger.exception("stream_publish_failed", transaction_id=tx_id) summary = {} try: @@ -165,10 +170,16 @@ async def create_transaction(payload: dict) -> dict: if summary_resp.status_code == 200: summary = summary_resp.json() except httpx.RequestError: - logger.exception(f"insights summary unavailable for transaction {tx_id}") + logger.exception("insights_summary_unavailable", transaction_id=tx_id) elapsed_ms = (time.perf_counter() - start) * 1000 - logger.info(f"finished transaction {tx_id} status={settled['status']}") + logger.info( + "transaction_finished", + transaction_id=tx_id, + status=settled["status"], + risk_score=risk_score, + duration_ms=round(elapsed_ms, 1), + ) return { "transaction_id": tx_id, diff --git a/services/insights/app.py b/services/insights/app.py index 9fac44d..c3d90fd 100644 --- a/services/insights/app.py +++ b/services/insights/app.py @@ -59,7 +59,7 @@ async def root(): @app.get("/healthz") async def healthz(request: Request): if request.headers.get("X-Ledgerline-Probe") == "chatty-health": - logger.info("received chatty-health healthz probe from gateway") + logger.info("healthz_probe_received", probe="chatty-health") return {"status": "ok", "service": SERVICE_NAME} @@ -72,7 +72,7 @@ async def readyz(): span = trace.get_current_span() span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"readyz check failed: {exc}") + logger.exception("readyz_check_failed") return JSONResponse(status_code=503, content={"status": "unavailable"}) return {"status": "ok"} diff --git a/services/ledger/app.py b/services/ledger/app.py index 99f598e..d442a57 100644 --- a/services/ledger/app.py +++ b/services/ledger/app.py @@ -52,7 +52,7 @@ class SettleRequest(BaseModel): @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): - logger.exception(f"unhandled exception on {request.url.path}: {exc}") + logger.exception("unhandled_exception", path=request.url.path) return JSONResponse(status_code=500, content={"error": "internal error"}) @@ -64,7 +64,7 @@ async def root(): @app.get("/healthz") async def healthz(request: Request): if request.headers.get("X-Ledgerline-Probe") == "chatty-health": - logger.info("received chatty-health healthz probe from gateway") + logger.info("healthz_probe_received", probe="chatty-health") return {"status": "ok", "service": SERVICE_NAME} @@ -77,7 +77,7 @@ async def readyz(): span = trace.get_current_span() span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"readyz check failed: {exc}") + logger.exception("readyz_check_failed") return JSONResponse(status_code=503, content={"status": "unavailable"}) return {"status": "ok"} @@ -91,7 +91,12 @@ async def list_accounts(): @app.post("/internal/transactions") async def create_transaction(body: CreateTransactionRequest): - logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor}") + logger.info( + "transaction_received", + account_id=body.account_id, + kind=body.kind, + amount_minor=body.amount_minor, + ) starved = await chaos.pool_starvation_enabled() poison_enabled = await chaos.poison_data_enabled() @@ -151,7 +156,7 @@ async def settle(tx_id: str, body: SettleRequest): 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}") + logger.info("transaction_settled", transaction_id=tx_id, risk_score=body.risk_score) return {"status": "settled", "balance_minor": balance} @@ -165,5 +170,5 @@ async def reject(tx_id: str, body: SettleRequest): return JSONResponse(status_code=404, content={"error": "transaction not found"}) balance = await repository.reject_transaction(session, tx=tx, risk_score=body.risk_score) - logger.info(f"rejected transaction {tx_id}") + logger.info("transaction_rejected", transaction_id=tx_id, risk_score=body.risk_score) return {"status": "rejected", "balance_minor": balance} diff --git a/services/notifier/app.py b/services/notifier/app.py index 5013cf8..d40be10 100644 --- a/services/notifier/app.py +++ b/services/notifier/app.py @@ -46,7 +46,7 @@ async def root(): @app.get("/healthz") async def healthz(request: Request): if request.headers.get("X-Ledgerline-Probe") == "chatty-health": - logger.info("received chatty-health healthz probe from gateway") + logger.info("healthz_probe_received", probe="chatty-health") return {"status": "ok", "service": SERVICE_NAME} @@ -60,6 +60,6 @@ async def readyz(): span = trace.get_current_span() span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"readyz check failed: {exc}") + logger.exception("readyz_check_failed") return JSONResponse(status_code=503, content={"status": "unavailable"}) return {"status": "ok"} diff --git a/services/notifier/consumer.py b/services/notifier/consumer.py index bf35299..4fbc085 100644 --- a/services/notifier/consumer.py +++ b/services/notifier/consumer.py @@ -1,9 +1,9 @@ import asyncio -import logging import random import uuid import sqlalchemy as sa +import structlog from opentelemetry import propagate, trace from opentelemetry.instrumentation.utils import suppress_instrumentation from opentelemetry.trace import SpanKind, StatusCode @@ -14,7 +14,7 @@ from common.settings import settings from . import chaos -logger = logging.getLogger("notifier") +logger = structlog.get_logger("notifier") tracer = trace.get_tracer(__name__) GROUP = "notifier" @@ -53,7 +53,7 @@ async def run(): independent group (`audit`) reads the same stream in services/audit/worker.py. Redis Streams carry no automatic context propagation, deliberately.""" await ensure_group() - logger.info("notifier consumer started") + logger.info("consumer_started", consumer_group=GROUP) while True: try: # The blocking poll would otherwise be a ~2s root trace every loop. @@ -68,7 +68,7 @@ async def run(): except asyncio.CancelledError: raise except Exception: - logger.exception("notifier consumer read failed") + logger.exception("stream_read_failed", consumer_group=GROUP) await asyncio.sleep(1) continue @@ -92,8 +92,19 @@ async def run(): ) as span: try: await _handle(fields) + logger.info( + "stream_message_processed", + consumer_group=GROUP, + message_id=msg_id, + transaction_id=fields.get("transaction_id"), + ) except Exception as exc: span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"notifier failed to process message {msg_id}") + logger.exception( + "stream_message_failed", + consumer_group=GROUP, + message_id=msg_id, + transaction_id=fields.get("transaction_id"), + ) await redis_client.xack(settings.stream_name, GROUP, msg_id) diff --git a/services/sanctions/app.py b/services/sanctions/app.py index ab49ce3..3babf05 100644 --- a/services/sanctions/app.py +++ b/services/sanctions/app.py @@ -40,7 +40,7 @@ async def root(): @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") + logger.info("healthz_probe_received", probe="chatty-health") return {"status": "ok", "service": SERVICE_NAME} @@ -53,22 +53,23 @@ async def readyz(): span = trace.get_current_span() span.record_exception(exc) span.set_status(StatusCode.ERROR, str(exc)) - logger.exception(f"readyz check failed: {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): - logger.info(f"lookup for holder_name={body.holder_name!r}") + # 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("sanctions_errors chaos triggered") + 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(f"slow_sanctions chaos sleeping {delay:.3f}s") + 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. @@ -77,6 +78,11 @@ async def lookup(body: LookupRequest): 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}