mohsen_hassani/LedgerLine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Carry trace context across the Redis Stream by hand

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Movses Aslanyan
2026-10-13 17:46:51 +03:30
co-authored by Claude Opus 5.5
parent 14938ab36b
commit d1c158c835
4 changed files with 92 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/audit/consumer.py
+35 -13
View File
@@ -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)
services/gateway/app.py
+1 -1
View File
@@ -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"],
)
services/gateway/orchestrator.py
+25 -10
View File
@@ -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")
services/notifier/consumer.py
+31 -12
View File
@@ -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)