Rewrite every log call as an event name with typed fields

Prose f-strings become past-tense events with consistent field names
across services (account_id, transaction_id, amount_minor, risk_score,
status_code, consumer_group, message_id, chaos_flag). The orchestrator
and both consumers move from logging.getLogger to structlog so they can
pass fields.

New events cover what was silent before: the gateway's UpstreamFailure
path (transaction_failed, with the underlying cause), fraud's score and
sanctions' screening outcomes, and each successfully processed stream
message. holder_name is no longer logged at all.

"lookup" and "healthz" still appear in exactly one event each, so the
pathology tests' log-line counts are unchanged until they move to LogQL.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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)

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}

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}
         )

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,

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"}
 

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}

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"}

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)

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}