feat(RHINENG-28313): add OpenTelemetry distributed tracing

common/logging.py

@@ -9,6 +9,7 @@
 import watchtower
 from boto3 import client
 from botocore.exceptions import ClientError
+from opentelemetry import trace as _otel_trace
 
 from .config import Config
 
@@ -80,13 +81,32 @@ def setup_cw_logging(main_logger):
             stream_name=CFG.hostname,
         )
         handler.setFormatter(CloudWatchLogFormatterCustom())
+        handler.addFilter(ContextualFilter())
     except ClientError:
         logger.exception("Unable to enable CloudWatch logging: ")
     else:  # pragma: no cover
         main_logger.addHandler(handler)
         logger.info("CloudWatch logging ENABLED!")
 
 
+class ContextualFilter(logging.Filter):
+    """Logging filter that injects OpenTelemetry trace context into log records.
+
+    Adds hex-encoded trace_id (32 chars) and span_id (16 chars) to every log
+    record. When no active span exists, emits None without crashing.
+    """
+
+    def filter(self, log_record):
+        try:
+            span_context = _otel_trace.get_current_span().get_span_context()
+            log_record.trace_id = format(span_context.trace_id, "032x") if span_context.is_valid else None
+            log_record.span_id = format(span_context.span_id, "016x") if span_context.is_valid else None
+        except Exception:
+            log_record.trace_id = None
+            log_record.span_id = None
+        return True
+
+
 def init_logging(num_servers=1):
     """Setup root logger handler."""
     logger = logging.getLogger()
@@ -103,6 +123,7 @@ def init_logging(num_servers=1):
         handler = logging.StreamHandler()
         formatter = OneLineExceptionFormatter(log_fmt)
         handler.setFormatter(formatter)
+        handler.addFilter(ContextualFilter())
         logger.addHandler(handler)
         setup_cw_logging(logger)
 

common/mqueue.py

@@ -8,13 +8,17 @@
 from aiokafka import AIOKafkaProducer
 from aiokafka.errors import KafkaError
 from aiokafka.partitioner import DefaultPartitioner
+from opentelemetry.trace import SpanKind
 
 from common.config import Config
 from common.logging import get_logger
+from common.telemetry import get_tracer
+from common.telemetry import inject_context_to_msg_headers
 
 LOGGER = get_logger(__name__)
 
 CFG = Config()
+TRACER = get_tracer(__name__)
 
 
 class Partitioners:
@@ -142,8 +146,18 @@ async def send_one(self, msg, key=None, headers=None):
         await self.start()
         try:
             data = bytes(json.dumps(msg).encode("utf-8"))
-            res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers)
-            LOGGER.debug(res)
+            with TRACER.start_as_current_span(
+                f"{self.topic} send",
+                kind=SpanKind.PRODUCER,
+                attributes={
+                    "messaging.system": "kafka",
+                    "messaging.operation.name": "send",
+                    "messaging.destination.name": self.topic,
+                },
+            ):
+                headers = inject_context_to_msg_headers(headers)
+                res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers)
+                LOGGER.debug(res)
         except KafkaError:
             self.connected = False
 
@@ -153,7 +167,17 @@ async def send_many(self, msg_list, key=None, headers=None):
         try:
             for msg in msg_list:
                 data = bytes(json.dumps(msg).encode("utf-8"))
-                res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=headers)
+                with TRACER.start_as_current_span(
+                    f"{self.topic} send",
+                    kind=SpanKind.PRODUCER,
+                    attributes={
+                        "messaging.system": "kafka",
+                        "messaging.operation.name": "send",
+                        "messaging.destination.name": self.topic,
+                    },
+                ):
+                    msg_headers = inject_context_to_msg_headers(headers)
+                    res = await self.client.send_and_wait(self.topic, value=data, key=self._serialize_key(key), headers=msg_headers)
                 LOGGER.debug(res)
         except KafkaError:
             self.connected = False
@@ -162,8 +186,18 @@ async def send_raw(self, msg: bytes, key=None, headers=None):
         """Logic around sending raw message"""
         await self.start()
         try:
-            res = await self.client.send_and_wait(self.topic, value=msg, key=self._serialize_key(key), headers=headers)
-            LOGGER.debug(res)
+            with TRACER.start_as_current_span(
+                f"{self.topic} send",
+                kind=SpanKind.PRODUCER,
+                attributes={
+                    "messaging.system": "kafka",
+                    "messaging.operation.name": "send",
+                    "messaging.destination.name": self.topic,
+                },
+            ):
+                headers = inject_context_to_msg_headers(headers)
+                res = await self.client.send_and_wait(self.topic, value=msg, key=self._serialize_key(key), headers=headers)
+                LOGGER.debug(res)
         except KafkaError:
             self.connected = False