feat: added suppression downstream headers

Signed-off-by: Cagri Yonca <[email protected]>

Author: CagriYonca

src/instana/instrumentation/kafka/confluent_kafka_python.py

@@ -58,6 +58,11 @@ def trace_kafka_produce(
 
         tracer, parent_span, _ = get_tracer_tuple()
         parent_context = parent_span.get_span_context() if parent_span else None
+        is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
+            "kafka",
+            "produce",
+            args[0],
+        )
 
         with tracer.start_as_current_span(
             "kafka-producer", span_context=parent_context, kind=SpanKind.PRODUCER
@@ -73,15 +78,21 @@ def trace_kafka_produce(
             # dictionary. To maintain compatibility with the headers for the
             # Kafka Python library, we will use a list of tuples.
             headers = args[6] if len(args) > 6 else kwargs.get("headers", [])
+            suppression_header = {"x_instana_l_s": "0" if is_suppressed else "1"}
+            headers.append(suppression_header)
+
             tracer.inject(
                 span.context,
                 Format.KAFKA_HEADERS,
                 headers,
                 disable_w3c_trace_context=True,
             )
 
-            try:
+            headers.remove(suppression_header)
+
+            if tracer.exporter.options.kafka_trace_correlation:
                 kwargs["headers"] = headers
+            try:
                 res = wrapped(*args, **kwargs)
             except Exception as exc:
                 span.record_exception(exc)

src/instana/instrumentation/kafka/kafka_python.py

@@ -30,7 +30,11 @@ def trace_kafka_send(
 
         tracer, parent_span, _ = get_tracer_tuple()
         parent_context = parent_span.get_span_context() if parent_span else None
-
+        is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
+            "kafka",
+            "send",
+            args[0],
+        )
         with tracer.start_as_current_span(
             "kafka-producer", span_context=parent_context, kind=SpanKind.PRODUCER
         ) as span:
@@ -39,15 +43,21 @@ def trace_kafka_send(
 
             # context propagation
             headers = kwargs.get("headers", [])
+            suppression_header = {"x_instana_l_s": "0" if is_suppressed else "1"}
+            headers.append(suppression_header)
+
             tracer.inject(
                 span.context,
                 Format.KAFKA_HEADERS,
                 headers,
                 disable_w3c_trace_context=True,
             )
 
-            try:
+            headers.remove(suppression_header)
+
+            if tracer.exporter.options.kafka_trace_correlation:
                 kwargs["headers"] = headers
+            try:
                 res = wrapped(*args, **kwargs)
             except Exception as exc:
                 span.record_exception(exc)

src/instana/options.py

@@ -37,36 +37,9 @@ def __init__(self, **kwds: Dict[str, Any]) -> None:
         self.extra_http_headers = None
         self.allow_exit_as_root = False
         self.ignore_endpoints = []
+        self.kafka_trace_correlation = True
 
-        if "INSTANA_DEBUG" in os.environ:
-            self.log_level = logging.DEBUG
-            self.debug = True
-
-        if "INSTANA_EXTRA_HTTP_HEADERS" in os.environ:
-            self.extra_http_headers = (
-                str(os.environ["INSTANA_EXTRA_HTTP_HEADERS"]).lower().split(";")
-            )
-
-        if "INSTANA_IGNORE_ENDPOINTS_PATH" in os.environ:
-            self.ignore_endpoints = parse_ignored_endpoints_from_yaml(
-                os.environ["INSTANA_IGNORE_ENDPOINTS_PATH"]
-            )
-        else:
-            if "INSTANA_IGNORE_ENDPOINTS" in os.environ:
-                self.ignore_endpoints = parse_ignored_endpoints(
-                    os.environ["INSTANA_IGNORE_ENDPOINTS"]
-                )
-            else:
-                if (
-                    isinstance(config.get("tracing"), dict)
-                    and "ignore_endpoints" in config["tracing"]
-                ):
-                    self.ignore_endpoints = parse_ignored_endpoints(
-                        config["tracing"]["ignore_endpoints"],
-                    )
-
-        if os.environ.get("INSTANA_ALLOW_EXIT_AS_ROOT", None) == "1":
-            self.allow_exit_as_root = True
+        self.set_trace_configurations()
 
         # Defaults
         self.secrets_matcher = "contains-ignore-case"
@@ -87,6 +60,56 @@ def __init__(self, **kwds: Dict[str, Any]) -> None:
 
         self.__dict__.update(kwds)
 
+    def set_trace_configurations(self) -> None:
+        """
+        Set tracing configurations from the environment variables and config file.
+        @return: None
+        """
+        # Use self.configurations to not read local configuration file
+        # in set_tracing method
+        if "INSTANA_DEBUG" in os.environ:
+            self.log_level = logging.DEBUG
+            self.debug = True
+
+        if "INSTANA_EXTRA_HTTP_HEADERS" in os.environ:
+            self.extra_http_headers = (
+                str(os.environ["INSTANA_EXTRA_HTTP_HEADERS"]).lower().split(";")
+            )
+
+        if "1" in [
+            os.environ.get("INSTANA_ALLOW_EXIT_AS_ROOT", None),  # deprecated
+            os.environ.get("INSTANA_ALLOW_ROOT_EXIT_SPAN", None),
+        ]:
+            self.allow_exit_as_root = True
+
+        # The priority is as follows:
+        # environment variables > in-code configuration >
+        # > agent config (configuration.yaml) > default value
+        if "INSTANA_IGNORE_ENDPOINTS" in os.environ:
+            self.ignore_endpoints = parse_ignored_endpoints(
+                os.environ["INSTANA_IGNORE_ENDPOINTS"]
+            )
+        elif "INSTANA_IGNORE_ENDPOINTS_PATH" in os.environ:
+            self.ignore_endpoints = parse_ignored_endpoints_from_yaml(
+                os.environ["INSTANA_IGNORE_ENDPOINTS_PATH"]
+            )
+        elif (
+            isinstance(config.get("tracing"), dict)
+            and "ignore_endpoints" in config["tracing"]
+        ):
+            self.ignore_endpoints = parse_ignored_endpoints(
+                config["tracing"]["ignore_endpoints"],
+            )
+
+        if "INSTANA_KAFKA_TRACE_CORRELATION" in os.environ:
+            self.kafka_trace_correlation = (
+                os.environ["INSTANA_KAFKA_TRACE_CORRELATION"].lower() == "true"
+            )
+        elif isinstance(config.get("tracing"), dict) and "kafka" in config["tracing"]:
+            self.kafka_trace_correlation = config["tracing"]["kafka"].get(
+                "trace_correlation", True
+            )
+
 
 class StandardOptions(BaseOptions):
     """The options class used when running directly on a host/node with an Instana agent"""
@@ -132,12 +155,30 @@ def set_tracing(self, tracing: Dict[str, Any]) -> None:
         @param tracing: tracing configuration dictionary
         @return: None
         """
-        if (
-            "ignore-endpoints" in tracing
-            and "INSTANA_IGNORE_ENDPOINTS" not in os.environ
-            and "tracing" not in config
-        ):
+        if "ignore-endpoints" in tracing and not self.ignore_endpoints:
             self.ignore_endpoints = parse_ignored_endpoints(tracing["ignore-endpoints"])
+
+        if "kafka" in tracing:
+            if (
+                "INSTANA_KAFKA_TRACE_CORRELATION" not in os.environ
+                and not (
+                    isinstance(config.get("tracing"), dict)
+                    and "kafka" in config["tracing"]
+                )
+                and "trace-correlation" in tracing["kafka"]
+            ):
+                self.kafka_trace_correlation = (
+                    str(tracing["kafka"].get("trace-correlation", True)) == "true"
+                )
+
+            if (
+                "header-format" in tracing["kafka"]
+                and tracing["kafka"]["header-format"] == "binary"
+            ):
+                logger.warning(
+                    "Binary header format for Kafka is deprecated. Please use string header format."
+                )
+
         if "extra-http-headers" in tracing:
             self.extra_http_headers = tracing["extra-http-headers"]
 
@@ -156,6 +197,7 @@ def set_from(self, res_data: Dict[str, Any]) -> None:
 
         if "tracing" in res_data:
             self.set_tracing(res_data["tracing"])
+
         else:
             if "extraHeaders" in res_data:
                 self.set_extra_headers(res_data["extraHeaders"])

src/instana/propagators/kafka_propagator.py

@@ -73,8 +73,8 @@ def extract(
                 disable_w3c_trace_context=disable_w3c_trace_context,
             )
 
-        except Exception:
-            logger.debug("kafka_propagator extract error:", exc_info=True)
+        except Exception as e:
+            logger.debug(f"kafka_propagator extract error: {e}", exc_info=True)
 
     # Assisted by watsonx Code Assistant
     def inject(
@@ -98,15 +98,12 @@ def inject(
         span_id = span_context.span_id
         dictionary_carrier = self.extract_carrier_headers(carrier)
 
+        suppression_level = 1
         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)
+            suppression_level = int(self.extract_instana_headers(dictionary_carrier)[2])
+            span_context.level = min(suppression_level, span_context.level)
 
         def inject_key_value(carrier, key, value):
             if isinstance(carrier, list):
@@ -122,18 +119,18 @@ def inject_key_value(carrier, key, value):
             inject_key_value(
                 carrier,
                 self.KAFKA_HEADER_KEY_L_S,
-                serializable_level.encode("utf-8"),
-            )
-            inject_key_value(
-                carrier,
-                self.KAFKA_HEADER_KEY_T,
-                hex_id_limited(trace_id).encode("utf-8"),
+                str(suppression_level).encode("utf-8"),
             )
-            inject_key_value(
-                carrier,
-                self.KAFKA_HEADER_KEY_S,
-                format_span_id(span_id).encode("utf-8"),
-            )
-
+            if suppression_level == 1:
+                inject_key_value(
+                    carrier,
+                    self.KAFKA_HEADER_KEY_T,
+                    hex_id_limited(trace_id).encode("utf-8"),
+                )
+                inject_key_value(
+                    carrier,
+                    self.KAFKA_HEADER_KEY_S,
+                    format_span_id(span_id).encode("utf-8"),
+                )
         except Exception:
             logger.debug("KafkaPropagator - inject error:", exc_info=True)