From d1c158c835930ffeba0eab6e436a20f09efebecf Mon Sep 17 00:00:00 2001 From: Movses Aslanyan Date: Tue, 13 Oct 2026 17:46:51 +0330 Subject: [PATCH] Carry trace context across the Redis Stream by hand Redis Streams have no headers, so the gateway injects traceparent into the message fields inside a PRODUCER span. The notifier and audit consumers extract it and open one CONSUMER span per message, so their DB and XACK spans join the transaction's trace. The idle XREADGROUP poll runs under suppress_instrumentation (it was a ~2s root trace every loop), and audit strips the trace fields so its stored payload is unchanged. The gateway also stops tracing its frontend's /api/health/ polling. Co-Authored-By: Claude Opus 5.5 --- services/audit/consumer.py | 48 +++++++++++++++++++++++--------- services/gateway/app.py | 2 +- services/gateway/orchestrator.py | 35 ++++++++++++++++------- services/notifier/consumer.py | 43 ++++++++++++++++++++-------- 4 files changed, 92 insertions(+), 36 deletions(-) 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)