diff --git a/services/audit/consumer.py b/services/audit/consumer.py index e28dd16..ae96f5b 100644 --- a/services/audit/consumer.py +++ b/services/audit/consumer.py @@ -3,12 +3,16 @@ import json import logging import sqlalchemy as sa +from opentelemetry import propagate, trace +from opentelemetry.instrumentation.utils import suppress_instrumentation +from opentelemetry.trace import SpanKind from common.db import Session from common.redis_client import redis_client from common.settings import settings logger = logging.getLogger("audit") +tracer = trace.get_tracer(__name__) GROUP = "audit" CONSUMER = "audit-1" @@ -23,13 +27,16 @@ async def ensure_group(): async def _handle(fields: dict): + # Store the business fields only, not the propagated trace context. + trace_fields = propagate.get_global_textmap().fields + payload = {k: v for k, v in fields.items() if k not in trace_fields} async with Session() as session: await session.execute( sa.text( "INSERT INTO audit_events (transaction_id, event_type, payload, recorded_at) " "VALUES (:tx_id, 'transaction_processed', CAST(:payload AS jsonb), now())" ), - {"tx_id": fields.get("transaction_id"), "payload": json.dumps(fields)}, + {"tx_id": fields.get("transaction_id"), "payload": json.dumps(payload)}, ) await session.commit() @@ -42,13 +49,15 @@ async def run(): logger.info("audit consumer started") while True: try: - resp = await redis_client.xreadgroup( - groupname=GROUP, - consumername=CONSUMER, - streams={settings.stream_name: ">"}, - count=10, - block=2000, - ) + # The blocking poll would otherwise be a ~2s root trace every loop. + with suppress_instrumentation(): + resp = await redis_client.xreadgroup( + groupname=GROUP, + consumername=CONSUMER, + streams={settings.stream_name: ">"}, + count=10, + block=2000, + ) except asyncio.CancelledError: raise except Exception: @@ -61,8 +70,21 @@ async def run(): for _, messages in resp: for msg_id, fields in messages: - try: - await _handle(fields) - except Exception: - logger.exception(f"audit failed to process message {msg_id}") - await redis_client.xack(settings.stream_name, GROUP, msg_id) + # Parent is the gateway's PRODUCER span, carried in the message. + with tracer.start_as_current_span( + f"process {settings.stream_name}", + context=propagate.extract(fields), + kind=SpanKind.CONSUMER, + attributes={ + "messaging.system": "redis", + "messaging.destination.name": settings.stream_name, + "messaging.operation.type": "process", + "messaging.consumer.group.name": GROUP, + "messaging.message.id": msg_id, + }, + ): + try: + await _handle(fields) + except Exception: + logger.exception(f"audit failed to process message {msg_id}") + await redis_client.xack(settings.stream_name, GROUP, msg_id) diff --git a/services/gateway/app.py b/services/gateway/app.py index c60bb2f..959f092 100644 --- a/services/gateway/app.py +++ b/services/gateway/app.py @@ -24,7 +24,7 @@ app = FastAPI(title=SERVICE_NAME) FastAPIInstrumentor.instrument_app( app, - excluded_urls="/healthz", + excluded_urls="/healthz,/api/health/", exclude_spans=["receive", "send"], ) diff --git a/services/gateway/orchestrator.py b/services/gateway/orchestrator.py index 066bc2b..59f42bf 100644 --- a/services/gateway/orchestrator.py +++ b/services/gateway/orchestrator.py @@ -3,6 +3,8 @@ import time import uuid import httpx +from opentelemetry import propagate, trace +from opentelemetry.trace import SpanKind from common.http import downstream_client from common.redis_client import redis_client @@ -11,6 +13,7 @@ from common.settings import settings from . import chaos logger = logging.getLogger("gateway") +tracer = trace.get_tracer(__name__) REJECT_THRESHOLD = 0.8 @@ -129,17 +132,29 @@ async def create_transaction(payload: dict, request_id: str) -> dict: settled = settle_resp.json() try: - await redis_client.xadd( - settings.stream_name, - { - "transaction_id": tx_id, - "account_id": created["account_id"], - "kind": created["kind"], - "amount_minor": str(created["amount_minor"]), - "status": settled["status"], - "risk_score": str(risk_score), + fields = { + "transaction_id": tx_id, + "account_id": created["account_id"], + "kind": created["kind"], + "amount_minor": str(created["amount_minor"]), + "status": settled["status"], + "risk_score": str(risk_score), + } + with tracer.start_as_current_span( + f"send {settings.stream_name}", + kind=SpanKind.PRODUCER, + attributes={ + "messaging.system": "redis", + "messaging.destination.name": settings.stream_name, + "messaging.operation.type": "send", }, - ) + ) as span: + # Redis Streams carry no trace context, so it travels in the + # message itself. Injected inside the PRODUCER span, so the + # consumers' parent is this span. + propagate.inject(fields) + 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") diff --git a/services/notifier/consumer.py b/services/notifier/consumer.py index eda5fcb..aa7aa4b 100644 --- a/services/notifier/consumer.py +++ b/services/notifier/consumer.py @@ -4,6 +4,9 @@ import random import uuid import sqlalchemy as sa +from opentelemetry import propagate, trace +from opentelemetry.instrumentation.utils import suppress_instrumentation +from opentelemetry.trace import SpanKind from common.db import Session from common.redis_client import redis_client @@ -12,6 +15,7 @@ from common.settings import settings from . import chaos logger = logging.getLogger("notifier") +tracer = trace.get_tracer(__name__) GROUP = "notifier" CONSUMER = "notifier-1" @@ -52,13 +56,15 @@ async def run(): logger.info("notifier consumer started") while True: try: - resp = await redis_client.xreadgroup( - groupname=GROUP, - consumername=CONSUMER, - streams={settings.stream_name: ">"}, - count=10, - block=2000, - ) + # The blocking poll would otherwise be a ~2s root trace every loop. + with suppress_instrumentation(): + resp = await redis_client.xreadgroup( + groupname=GROUP, + consumername=CONSUMER, + streams={settings.stream_name: ">"}, + count=10, + block=2000, + ) except asyncio.CancelledError: raise except Exception: @@ -71,8 +77,21 @@ async def run(): for _, messages in resp: for msg_id, fields in messages: - try: - await _handle(fields) - except Exception: - logger.exception(f"notifier failed to process message {msg_id}") - await redis_client.xack(settings.stream_name, GROUP, msg_id) + # Parent is the gateway's PRODUCER span, carried in the message. + with tracer.start_as_current_span( + f"process {settings.stream_name}", + context=propagate.extract(fields), + kind=SpanKind.CONSUMER, + attributes={ + "messaging.system": "redis", + "messaging.destination.name": settings.stream_name, + "messaging.operation.type": "process", + "messaging.consumer.group.name": GROUP, + "messaging.message.id": msg_id, + }, + ): + try: + await _handle(fields) + except Exception: + logger.exception(f"notifier failed to process message {msg_id}") + await redis_client.xack(settings.stream_name, GROUP, msg_id)