From 091c18d278b1f1e75a28649228c22391523d9c9e Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Thu, 12 Feb 2026 18:07:49 +0100 Subject: [PATCH] fix: passing dlq msg as a payload, not in headers --- backend/app/core/providers.py | 2 + backend/app/core/tracing/__init__.py | 4 - backend/app/core/tracing/utils.py | 29 +---- backend/app/dlq/manager.py | 8 -- backend/app/dlq/models.py | 1 - backend/app/events/core/producer.py | 32 ++--- backend/app/events/handlers.py | 186 +++++++++------------------ 7 files changed, 75 insertions(+), 187 deletions(-) diff --git a/backend/app/core/providers.py b/backend/app/core/providers.py index 9ec70f95..033132a9 100644 --- a/backend/app/core/providers.py +++ b/backend/app/core/providers.py @@ -7,6 +7,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler from dishka import Provider, Scope, from_context, provide from faststream.kafka import KafkaBroker +from faststream.kafka.opentelemetry import KafkaTelemetryMiddleware from kubernetes_asyncio import client as k8s_client from kubernetes_asyncio import config as k8s_config from kubernetes_asyncio.client.rest import ApiException @@ -96,6 +97,7 @@ async def get_broker( logger=logger, client_id=f"integr8scode-{settings.SERVICE_NAME}", request_timeout_ms=settings.KAFKA_REQUEST_TIMEOUT_MS, + middlewares=(KafkaTelemetryMiddleware(),), ) logger.info("Kafka broker created") try: diff --git a/backend/app/core/tracing/__init__.py b/backend/app/core/tracing/__init__.py index e12cabe2..2116b5b2 100644 --- a/backend/app/core/tracing/__init__.py +++ b/backend/app/core/tracing/__init__.py @@ -19,9 +19,7 @@ # Import utilities and decorators from app.core.tracing.utils import ( add_span_attributes, - extract_trace_context, get_tracer, - inject_trace_context, trace_span, ) @@ -38,9 +36,7 @@ "init_tracing", # Utilities and decorators "add_span_attributes", - "extract_trace_context", "get_tracer", - "inject_trace_context", "trace_span", # OpenTelemetry types "context", diff --git a/backend/app/core/tracing/utils.py b/backend/app/core/tracing/utils.py index 963ea91d..54e84f83 100644 --- a/backend/app/core/tracing/utils.py +++ b/backend/app/core/tracing/utils.py @@ -2,7 +2,7 @@ from contextlib import contextmanager from typing import Any -from opentelemetry import context, propagate, trace +from opentelemetry import trace from opentelemetry.trace import SpanKind, Status, StatusCode @@ -45,33 +45,6 @@ def trace_span( raise -def inject_trace_context(headers: dict[str, str]) -> dict[str, str]: - """ - Inject current trace context into headers for propagation. - - Args: - headers: Existing headers dictionary - - Returns: - Headers with trace context injected - """ - propagation_headers = headers.copy() - propagate.inject(propagation_headers) - return propagation_headers - - -def extract_trace_context(headers: dict[str, str]) -> context.Context: - """ - Extract trace context from headers. - - Args: - headers: Headers containing trace context - - Returns: - Extracted OpenTelemetry context - """ - return propagate.extract(headers) - def add_span_attributes(**attributes: Any) -> None: """ diff --git a/backend/app/dlq/manager.py b/backend/app/dlq/manager.py index ff638e8c..323a12c4 100644 --- a/backend/app/dlq/manager.py +++ b/backend/app/dlq/manager.py @@ -5,7 +5,6 @@ from faststream.kafka import KafkaBroker from app.core.metrics import DLQMetrics -from app.core.tracing import inject_trace_context from app.db.repositories import DLQRepository from app.dlq.models import ( DLQBatchRetryResult, @@ -127,17 +126,10 @@ async def retry_message(self, message: DLQMessage) -> None: FastStream handles JSON serialization of Pydantic models natively. """ - hdrs: dict[str, str] = { - "event_type": message.event.event_type, - } - hdrs = inject_trace_context(hdrs) - - # Publish directly to original topic - FastStream serializes Pydantic to JSON await self._broker.publish( message=message.event, topic=message.original_topic, key=message.event.event_id.encode(), - headers=hdrs, ) self.metrics.record_dlq_message_retried(message.original_topic, message.event.event_type, "success") diff --git a/backend/app/dlq/models.py b/backend/app/dlq/models.py index 8dd6a5b8..c4fb5168 100644 --- a/backend/app/dlq/models.py +++ b/backend/app/dlq/models.py @@ -49,7 +49,6 @@ class DLQMessage(BaseModel): dlq_offset: int | None = None dlq_partition: int | None = None last_error: str | None = None - headers: dict[str, str] = Field(default_factory=dict) @dataclass diff --git a/backend/app/events/core/producer.py b/backend/app/events/core/producer.py index 1af6deb8..a1e5caf7 100644 --- a/backend/app/events/core/producer.py +++ b/backend/app/events/core/producer.py @@ -6,9 +6,8 @@ from faststream.kafka import KafkaBroker from app.core.metrics import EventMetrics -from app.core.tracing import inject_trace_context from app.db.repositories import EventRepository -from app.dlq.models import DLQMessageStatus +from app.dlq.models import DLQMessage, DLQMessageStatus from app.domain.enums import KafkaTopic from app.domain.events import DomainEvent from app.infrastructure.kafka.mappings import EVENT_TYPE_TO_TOPIC @@ -41,16 +40,10 @@ async def produce(self, event_to_produce: DomainEvent, key: str) -> None: await self._event_repository.store_event(event_to_produce) topic = f"{self._topic_prefix}{EVENT_TYPE_TO_TOPIC[event_to_produce.event_type]}" try: - headers = inject_trace_context({ - "event_type": event_to_produce.event_type, - }) - - # FastStream handles Pydantic → JSON serialization natively await self._broker.publish( message=event_to_produce, topic=topic, key=key.encode(), - headers=headers, ) self._event_metrics.record_kafka_message_produced(topic) @@ -72,23 +65,20 @@ async def send_to_dlq( dlq_topic = f"{self._topic_prefix}{KafkaTopic.DEAD_LETTER_QUEUE}" - headers = inject_trace_context({ - "event_type": original_event.event_type, - "original_topic": original_topic, - "error_type": type(error).__name__, - "error": str(error), - "retry_count": str(retry_count), - "failed_at": datetime.now(timezone.utc).isoformat(), - "status": DLQMessageStatus.PENDING, - "producer_id": producer_id, - }) + dlq_msg = DLQMessage( + event=original_event, + original_topic=original_topic, + error=str(error), + retry_count=retry_count, + failed_at=datetime.now(timezone.utc), + status=DLQMessageStatus.PENDING, + producer_id=producer_id, + ) - # FastStream handles Pydantic → JSON serialization natively await self._broker.publish( - message=original_event, + message=dlq_msg, topic=dlq_topic, key=original_event.event_id.encode(), - headers=headers, ) self._event_metrics.record_kafka_message_produced(dlq_topic) diff --git a/backend/app/events/handlers.py b/backend/app/events/handlers.py index c9165cc5..c8bfc4e7 100644 --- a/backend/app/events/handlers.py +++ b/backend/app/events/handlers.py @@ -1,17 +1,15 @@ import asyncio import logging -from collections.abc import Awaitable, Callable from datetime import datetime, timezone from typing import Any from dishka.integrations.faststream import FromDishka -from faststream import AckPolicy, StreamMessage +from faststream import AckPolicy from faststream.kafka import KafkaBroker, KafkaMessage -from opentelemetry.trace import SpanKind +from faststream.message import decode_message -from app.core.tracing import EventAttributes, extract_trace_context, get_tracer from app.dlq.manager import DLQManager -from app.dlq.models import DLQMessage, DLQMessageStatus +from app.dlq.models import DLQMessage from app.domain.enums import EventType, GroupId, KafkaTopic from app.domain.events import ( CreatePodCommandEvent, @@ -35,41 +33,15 @@ from app.settings import Settings -def _extract_headers(msg: StreamMessage[Any]) -> dict[str, str]: - """Decode raw Kafka headers into a string dict for OTel extraction.""" - return {k: v.decode() if isinstance(v, bytes) else v for k, v in (msg.raw_message.headers or [])} - - -async def _with_trace( - msg: StreamMessage[Any], - span_name: str, - body: DomainEvent, - handler: Callable[[], Awaitable[None]], -) -> None: - """Run handler inside an OTel consumer span linked to the producer's trace context.""" - headers = _extract_headers(msg) - ctx = extract_trace_context(headers) - with get_tracer().start_as_current_span( - name=span_name, - context=ctx, - kind=SpanKind.CONSUMER, - attributes={ - EventAttributes.EVENT_TYPE: body.event_type, - EventAttributes.EVENT_ID: body.event_id, - }, - ): - await handler() - - async def with_idempotency( event: DomainEvent, - handler: Callable[..., Awaitable[None]], + handler: Any, idem: IdempotencyManager, key_strategy: KeyStrategy, ttl_seconds: int, logger: logging.Logger, ) -> None: - """Run *handler* inside an idempotency guard (check → execute → mark).""" + """Run *handler* inside an idempotency guard (check -> execute -> mark).""" result = await idem.check_and_reserve( event=event, key_strategy=key_strategy, ttl_seconds=ttl_seconds, ) @@ -93,6 +65,11 @@ def _topics(settings: Settings, group_id: GroupId) -> list[str]: ] +def _event_type_filter(msg: Any, expected: str) -> bool: + """Body-based event_type filter for @sub(filter=...) lambdas.""" + return decode_message(msg).get("event_type") == expected # type: ignore[union-attr] + + def register_coordinator_subscriber(broker: KafkaBroker, settings: Settings) -> None: sub = broker.subscriber( *_topics(settings, GroupId.EXECUTION_COORDINATOR), @@ -100,53 +77,49 @@ def register_coordinator_subscriber(broker: KafkaBroker, settings: Settings) -> ack_policy=AckPolicy.ACK, ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_REQUESTED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_REQUESTED)) async def on_execution_requested( body: ExecutionRequestedEvent, - msg: KafkaMessage, coordinator: FromDishka[ExecutionCoordinator], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "coordinator.execution_requested", body, lambda: with_idempotency( + await with_idempotency( body, coordinator.handle_execution_requested, idem, KeyStrategy.EVENT_BASED, 7200, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_COMPLETED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_COMPLETED)) async def on_execution_completed( body: ExecutionCompletedEvent, - msg: KafkaMessage, coordinator: FromDishka[ExecutionCoordinator], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "coordinator.execution_completed", body, lambda: with_idempotency( + await with_idempotency( body, coordinator.handle_execution_completed, idem, KeyStrategy.EVENT_BASED, 7200, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_FAILED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_FAILED)) async def on_execution_failed( body: ExecutionFailedEvent, - msg: KafkaMessage, coordinator: FromDishka[ExecutionCoordinator], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "coordinator.execution_failed", body, lambda: with_idempotency( + await with_idempotency( body, coordinator.handle_execution_failed, idem, KeyStrategy.EVENT_BASED, 7200, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_CANCELLED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_CANCELLED)) async def on_execution_cancelled( body: ExecutionCancelledEvent, - msg: KafkaMessage, coordinator: FromDishka[ExecutionCoordinator], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "coordinator.execution_cancelled", body, lambda: with_idempotency( + await with_idempotency( body, coordinator.handle_execution_cancelled, idem, KeyStrategy.EVENT_BASED, 7200, logger, - )) + ) @sub async def on_unhandled(body: DomainEvent) -> None: @@ -160,29 +133,27 @@ def register_k8s_worker_subscriber(broker: KafkaBroker, settings: Settings) -> N ack_policy=AckPolicy.ACK, ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.CREATE_POD_COMMAND) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.CREATE_POD_COMMAND)) async def on_create_pod( body: CreatePodCommandEvent, - msg: KafkaMessage, worker: FromDishka[KubernetesWorker], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "k8s_worker.create_pod", body, lambda: with_idempotency( + await with_idempotency( body, worker.handle_create_pod_command, idem, KeyStrategy.CONTENT_HASH, 3600, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.DELETE_POD_COMMAND) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.DELETE_POD_COMMAND)) async def on_delete_pod( body: DeletePodCommandEvent, - msg: KafkaMessage, worker: FromDishka[KubernetesWorker], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "k8s_worker.delete_pod", body, lambda: with_idempotency( + await with_idempotency( body, worker.handle_delete_pod_command, idem, KeyStrategy.CONTENT_HASH, 3600, logger, - )) + ) @sub async def on_unhandled(body: DomainEvent) -> None: @@ -198,41 +169,38 @@ def register_result_processor_subscriber(broker: KafkaBroker, settings: Settings auto_offset_reset="earliest", ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_COMPLETED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_COMPLETED)) async def on_execution_completed( body: ExecutionCompletedEvent, - msg: KafkaMessage, processor: FromDishka[ResultProcessor], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "result_processor.execution_completed", body, lambda: with_idempotency( + await with_idempotency( body, processor.handle_execution_completed, idem, KeyStrategy.CONTENT_HASH, 7200, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_FAILED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_FAILED)) async def on_execution_failed( body: ExecutionFailedEvent, - msg: KafkaMessage, processor: FromDishka[ResultProcessor], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "result_processor.execution_failed", body, lambda: with_idempotency( + await with_idempotency( body, processor.handle_execution_failed, idem, KeyStrategy.CONTENT_HASH, 7200, logger, - )) + ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_TIMEOUT) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_TIMEOUT)) async def on_execution_timeout( body: ExecutionTimeoutEvent, - msg: KafkaMessage, processor: FromDishka[ResultProcessor], idem: FromDishka[IdempotencyManager], logger: FromDishka[logging.Logger], ) -> None: - await _with_trace(msg, "result_processor.execution_timeout", body, lambda: with_idempotency( + await with_idempotency( body, processor.handle_execution_timeout, idem, KeyStrategy.CONTENT_HASH, 7200, logger, - )) + ) @sub async def on_unhandled(body: DomainEvent) -> None: @@ -246,37 +214,33 @@ def register_saga_subscriber(broker: KafkaBroker, settings: Settings) -> None: ack_policy=AckPolicy.ACK, ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_REQUESTED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_REQUESTED)) async def on_execution_requested( body: ExecutionRequestedEvent, - msg: KafkaMessage, orchestrator: FromDishka[SagaOrchestrator], ) -> None: - await _with_trace(msg, "saga.execution_requested", body, lambda: orchestrator.handle_execution_requested(body)) + await orchestrator.handle_execution_requested(body) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_COMPLETED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_COMPLETED)) async def on_execution_completed( body: ExecutionCompletedEvent, - msg: KafkaMessage, orchestrator: FromDishka[SagaOrchestrator], ) -> None: - await _with_trace(msg, "saga.execution_completed", body, lambda: orchestrator.handle_execution_completed(body)) + await orchestrator.handle_execution_completed(body) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_FAILED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_FAILED)) async def on_execution_failed( body: ExecutionFailedEvent, - msg: KafkaMessage, orchestrator: FromDishka[SagaOrchestrator], ) -> None: - await _with_trace(msg, "saga.execution_failed", body, lambda: orchestrator.handle_execution_failed(body)) + await orchestrator.handle_execution_failed(body) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_TIMEOUT) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_TIMEOUT)) async def on_execution_timeout( body: ExecutionTimeoutEvent, - msg: KafkaMessage, orchestrator: FromDishka[SagaOrchestrator], ) -> None: - await _with_trace(msg, "saga.execution_timeout", body, lambda: orchestrator.handle_execution_timeout(body)) + await orchestrator.handle_execution_timeout(body) @sub async def on_unhandled(body: DomainEvent) -> None: @@ -309,31 +273,26 @@ def register_notification_subscriber(broker: KafkaBroker, settings: Settings) -> auto_offset_reset="latest", ) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_COMPLETED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_COMPLETED)) async def on_execution_completed( body: ExecutionCompletedEvent, - msg: KafkaMessage, service: FromDishka[NotificationService], ) -> None: - await _with_trace( - msg, "notification.execution_completed", body, lambda: service.handle_execution_completed(body), - ) + await service.handle_execution_completed(body) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_FAILED) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_FAILED)) async def on_execution_failed( body: ExecutionFailedEvent, - msg: KafkaMessage, service: FromDishka[NotificationService], ) -> None: - await _with_trace(msg, "notification.execution_failed", body, lambda: service.handle_execution_failed(body)) + await service.handle_execution_failed(body) - @sub(filter=lambda msg: msg.headers["event_type"] == EventType.EXECUTION_TIMEOUT) + @sub(filter=lambda msg: _event_type_filter(msg, EventType.EXECUTION_TIMEOUT)) async def on_execution_timeout( body: ExecutionTimeoutEvent, - msg: KafkaMessage, service: FromDishka[NotificationService], ) -> None: - await _with_trace(msg, "notification.execution_timeout", body, lambda: service.handle_execution_timeout(body)) + await service.handle_execution_timeout(body) @sub async def on_unhandled(body: DomainEvent) -> None: @@ -343,8 +302,8 @@ async def on_unhandled(body: DomainEvent) -> None: def register_dlq_subscriber(broker: KafkaBroker, settings: Settings) -> None: """Register a DLQ subscriber that consumes dead-letter messages. - DLQ messages are JSON-encoded DomainEvents (Pydantic serialization via FastStream). - DLQ metadata (original_topic, error, retry_count, etc.) lives in Kafka headers. + DLQ messages are JSON-encoded DLQMessage models (Pydantic serialization via FastStream). + All DLQ metadata is in the message body — no Kafka headers needed. """ topic_name = f"{settings.KAFKA_TOPIC_PREFIX}{KafkaTopic.DEAD_LETTER_QUEUE}" @@ -355,45 +314,22 @@ def register_dlq_subscriber(broker: KafkaBroker, settings: Settings) -> None: auto_offset_reset="earliest", ) async def on_dlq_message( - body: DomainEvent, + body: DLQMessage, msg: KafkaMessage, manager: FromDishka[DLQManager], logger: FromDishka[logging.Logger], ) -> None: start = asyncio.get_running_loop().time() - headers = _extract_headers(msg) raw = msg.raw_message - assert not isinstance(raw, tuple) # single-message consumer, never batch - - dlq_msg = DLQMessage( - event=body, - original_topic=headers["original_topic"], - error=headers["error"], - retry_count=int(headers["retry_count"]), - failed_at=datetime.fromisoformat(headers["failed_at"]), - status=DLQMessageStatus(headers["status"]), - producer_id=headers["producer_id"], - dlq_offset=raw.offset, - dlq_partition=raw.partition, - headers=headers, - ) + assert not isinstance(raw, tuple) + body.dlq_offset = raw.offset + body.dlq_partition = raw.partition + + await manager.handle_message(body) - ctx = extract_trace_context(headers) - with get_tracer().start_as_current_span( - name="dlq.consume", - context=ctx, - kind=SpanKind.CONSUMER, - attributes={ - EventAttributes.KAFKA_TOPIC: manager.dlq_topic, - EventAttributes.EVENT_TYPE: body.event_type, - EventAttributes.EVENT_ID: body.event_id, - }, - ): - await manager.handle_message(dlq_msg) - - manager.metrics.record_dlq_message_received(dlq_msg.original_topic, body.event_type) + manager.metrics.record_dlq_message_received(body.original_topic, body.event.event_type) manager.metrics.record_dlq_message_age( - (datetime.now(timezone.utc) - dlq_msg.failed_at).total_seconds() + (datetime.now(timezone.utc) - body.failed_at).total_seconds() ) manager.metrics.record_dlq_processing_duration( asyncio.get_running_loop().time() - start, "process"