feat: Add KafkaPropagator to handle context propagation.

Signed-off-by: Paulo Vital <[email protected]>

Author: pvital

src/instana/propagators/format.py

@@ -50,3 +50,15 @@ class Format(object):
     should use a prefix or other convention to distinguish tracer-specific
     key:value pairs.
     """
+
+    KAFKA_HEADERS = "kafka_headers"
+    """
+    The KAFKA_HEADERS format represents :class:`SpanContext`\\ s in a python
+    ``dict`` mapping from character-restricted strings to strings.
+
+    Keys and values in the KAFKA_HEADERS carrier must be suitable for use as
+    HTTP headers (without modification or further escaping). That is, the
+    keys have a greatly restricted character set, casing for the keys may not
+    be preserved by various intermediaries, and the values should be
+    URL-escaped.
+    """

src/instana/propagators/kafka_propagator.py

@@ -0,0 +1,61 @@
+# (c) Copyright IBM Corp. 2025
+from typing import TYPE_CHECKING
+
+from opentelemetry.trace.span import format_span_id
+
+from instana.log import logger
+from instana.propagators.base_propagator import BasePropagator, CarrierT
+from instana.util.ids import hex_id_limited
+
+if TYPE_CHECKING:
+    from instana.span_context import SpanContext
+
+
+class KafkaPropagator(BasePropagator):
+    """
+    Instana Propagator for Format.KAFKA_HEADERS.
+
+    The KAFKA_HEADERS format deals with key-values with string to string mapping.
+    The character set should be restricted to HTTP compatible.
+    """
+
+    def __init__(self) -> None:
+        super(KafkaPropagator, self).__init__()
+
+    def inject(
+        self,
+        span_context: "SpanContext",
+        carrier: CarrierT,
+        disable_w3c_trace_context: bool = True,
+    ) -> None:
+        trace_id = span_context.trace_id
+        span_id = span_context.span_id
+        dictionary_carrier = self.extract_headers_dict(carrier)
+
+        if dictionary_carrier:
+            # Suppression `level` made in the child context or in the parent context
+            # has priority over any non-suppressed `level` setting
+            child_level = int(
+                self.extract_instana_headers(dictionary_carrier)[2] or "1"
+            )
+            span_context.level = min(child_level, span_context.level)
+
+        serializable_level = str(span_context.level)
+
+        def inject_key_value(carrier, key, value):
+            if isinstance(carrier, list):
+                carrier.append((key, value))
+            elif isinstance(carrier, dict) or "__setitem__" in dir(carrier):
+                carrier[key] = value
+            else:
+                raise Exception(
+                    f"KafkaPropagator: Unsupported carrier type {type(carrier)}",
+                )
+
+        try:
+            inject_key_value(carrier, "X_INSTANA_L_S", serializable_level)
+            inject_key_value(carrier, "X_INSTANA_T", hex_id_limited(trace_id))
+            inject_key_value(carrier, "X_INSTANA_S", format_span_id(span_id))
+
+        except Exception:
+            logger.debug("KafkaPropagator - inject error:", exc_info=True)

src/instana/tracer.py

@@ -26,6 +26,7 @@
 from instana.propagators.exceptions import UnsupportedFormatException
 from instana.propagators.format import Format
 from instana.propagators.http_propagator import HTTPPropagator
+from instana.propagators.kafka_propagator import KafkaPropagator
 from instana.propagators.text_propagator import TextPropagator
 from instana.recorder import StanRecorder
 from instana.sampling import InstanaSampler, Sampler
@@ -53,6 +54,7 @@ def __init__(
         self._propagators[Format.HTTP_HEADERS] = HTTPPropagator()
         self._propagators[Format.TEXT_MAP] = TextPropagator()
         self._propagators[Format.BINARY] = BinaryPropagator()
+        self._propagators[Format.KAFKA_HEADERS] = KafkaPropagator()
 
     def get_tracer(
         self,
@@ -118,7 +120,9 @@ def start_span(
         record_exception: bool = True,
         set_status_on_exception: bool = True,
     ) -> InstanaSpan:
-        parent_context = span_context if span_context else get_current_span().get_span_context()
+        parent_context = (
+            span_context if span_context else get_current_span().get_span_context()
+        )
 
         if parent_context and not isinstance(parent_context, SpanContext):
             raise TypeError("parent_context must be an Instana SpanContext or None.")
@@ -224,9 +228,13 @@ def _create_span_context(self, parent_context: SpanContext) -> SpanContext:
             level=(parent_context.level if parent_context else 1),
             synthetic=(parent_context.synthetic if parent_context else False),
             trace_parent=(parent_context.trace_parent if parent_context else None),
-            instana_ancestor=(parent_context.instana_ancestor if parent_context else None),
+            instana_ancestor=(
+                parent_context.instana_ancestor if parent_context else None
+            ),
             long_trace_id=(parent_context.long_trace_id if parent_context else None),
-            correlation_type=(parent_context.correlation_type if parent_context else None),
+            correlation_type=(
+                parent_context.correlation_type if parent_context else None
+            ),
             correlation_id=(parent_context.correlation_id if parent_context else None),
             traceparent=(parent_context.traceparent if parent_context else None),
             tracestate=(parent_context.tracestate if parent_context else None),
@@ -237,7 +245,9 @@ def _create_span_context(self, parent_context: SpanContext) -> SpanContext:
     def inject(
         self,
         span_context: SpanContext,
-        format: Union[Format.BINARY, Format.HTTP_HEADERS, Format.TEXT_MAP],
+        format: Union[
+            Format.BINARY, Format.HTTP_HEADERS, Format.TEXT_MAP, Format.KAFKA_HEADERS
+        ],
         carrier: "CarrierT",
         disable_w3c_trace_context: bool = False,
     ) -> Optional["CarrierT"]:
@@ -250,7 +260,9 @@ def inject(
 
     def extract(
         self,
-        format: Union[Format.BINARY, Format.HTTP_HEADERS, Format.TEXT_MAP],
+        format: Union[
+            Format.BINARY, Format.HTTP_HEADERS, Format.TEXT_MAP, Format.KAFKA_HEADERS
+        ],
         carrier: "CarrierT",
         disable_w3c_trace_context: bool = False,
     ) -> Optional[Context]:

tests/test_tracer.py

@@ -27,7 +27,7 @@ def test_tracer_defaults(tracer_provider: InstanaTracerProvider) -> None:
     assert isinstance(tracer._sampler, InstanaSampler)
     assert isinstance(tracer.span_processor, StanRecorder)
     assert isinstance(tracer.exporter, HostAgent)
-    assert len(tracer._propagators) == 3
+    assert len(tracer._propagators) == 4
 
 
 def test_tracer_start_span(

tests/test_tracer_provider.py

@@ -7,6 +7,7 @@
 from instana.propagators.binary_propagator import BinaryPropagator
 from instana.propagators.format import Format
 from instana.propagators.http_propagator import HTTPPropagator
+from instana.propagators.kafka_propagator import KafkaPropagator
 from instana.propagators.text_propagator import TextPropagator
 from instana.recorder import StanRecorder
 from instana.sampling import InstanaSampler
@@ -18,10 +19,11 @@ def test_tracer_provider_defaults() -> None:
     assert isinstance(provider.sampler, InstanaSampler)
     assert isinstance(provider._span_processor, StanRecorder)
     assert isinstance(provider._exporter, HostAgent)
-    assert len(provider._propagators) == 3
+    assert len(provider._propagators) == 4
     assert isinstance(provider._propagators[Format.HTTP_HEADERS], HTTPPropagator)
     assert isinstance(provider._propagators[Format.TEXT_MAP], TextPropagator)
     assert isinstance(provider._propagators[Format.BINARY], BinaryPropagator)
+    assert isinstance(provider._propagators[Format.KAFKA_HEADERS], KafkaPropagator)
 
 
 def test_tracer_provider_get_tracer() -> None: