feat: Add span filtering feature

Signed-off-by: Cagri Yonca <cagri@ibm.com>

Author: CagriYonca

src/instana/agent/host.py

@@ -22,7 +22,7 @@
 from instana.options import StandardOptions
 from instana.util import to_json
 from instana.util.runtime import get_py_source, log_runtime_env_info
-from instana.util.span_utils import get_operation_specifiers
+from instana.util.span_utils import matches_rule
 from instana.version import VERSION
 
 if TYPE_CHECKING:
@@ -357,51 +357,61 @@ def report_spans(self, payload: Dict[str, Any]) -> Optional[Response]:
 
     def filter_spans(self, spans: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         """
-        Filters given span list using ignore-endpoint variable and returns the list of filtered spans.
+        Filters span list using new hierarchical filtering rules.
         """
         filtered_spans = []
-        endpoint = ""
+
         for span in spans:
-            if (hasattr(span, "n") or hasattr(span, "name")) and hasattr(span, "data"):
-                service = span.n
-                operation_specifier_key, service_specifier_key = (
-                    get_operation_specifiers(service)
-                )
-                if service == "kafka":
-                    endpoint = span.data[service][service_specifier_key]
-                method = span.data[service][operation_specifier_key]
-                if isinstance(method, str) and self.__is_endpoint_ignored(
-                    service, method, endpoint
-                ):
-                    continue
-                else:
-                    filtered_spans.append(span)
-            else:
+            if not (hasattr(span, "n") or hasattr(span, "name")) or not hasattr(
+                span, "data"
+            ):
                 filtered_spans.append(span)
+                continue
+
+            # Set the service name
+            service_name = getattr(span, "n", getattr(span, "name", ""))
+
+            # Set span attributes for filtering
+            attributes_to_check = {
+                "type": service_name,
+                "kind": "entry"
+                if span.k == 1
+                else "exit"
+                if span.k == 2
+                else "intermediate",
+            }
+
+            # Add operation specifiers to the attributes
+            if service_name in span.data:
+                service_data = span.data[service_name]
+                for key, value in service_data.items():
+                    attributes_to_check[f"{service_name}.{key}"] = value
+                    attributes_to_check[key] = value
+
+            # Check if the span need to be ignored
+            if self.__is_endpoint_ignored(attributes_to_check):
+                continue
+
+            filtered_spans.append(span)
+
         return filtered_spans
 
-    def __is_endpoint_ignored(
-        self,
-        service: str,
-        method: str = "",
-        endpoint: str = "",
-    ) -> bool:
-        """Check if the given service and endpoint combination should be ignored."""
-        service = service.lower()
-        method = method.lower()
-        endpoint = endpoint.lower()
-        filter_rules = [
-            f"{service}.{method}",  # service.method
-            f"{service}.*",  # service.*
-        ]
-
-        if service == "kafka" and endpoint:
-            filter_rules += [
-                f"{service}.{method}.{endpoint}",  # service.method.endpoint
-                f"{service}.*.{endpoint}",  # service.*.endpoint
-                f"{service}.{method}.*",  # service.method.*
-            ]
-        return any(rule in self.options.ignore_endpoints for rule in filter_rules)
+    def __is_endpoint_ignored(self, span_attributes: dict) -> bool:
+        filters = self.options.ignore_endpoints
+        if not filters:
+            return False
+
+        # Check include rules
+        for rule in filters.get("include", []):
+            if matches_rule(rule.get("attributes", []), span_attributes):
+                return False
+
+        # Check exclude rules
+        for rule in filters.get("exclude", []):
+            if matches_rule(rule.get("attributes", []), span_attributes):
+                return True
+
+        return False
 
     def handle_agent_tasks(self, task: Dict[str, Any]) -> None:
         """

src/instana/instrumentation/kafka/confluent_kafka_python.py

@@ -74,10 +74,15 @@ def trace_kafka_produce(
         # Get the topic from either args or kwargs
         topic = args[0] if args else kwargs.get("topic", "")
 
+        attributes_to_check = {
+            "type": "kafka",
+            "kind": "exit",
+            "kafka.service": topic,
+            "kafka.access": "produce",
+        }
+
         is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
-            "kafka",
-            "produce",
-            topic,
+            attributes_to_check
         )
 
         with tracer.start_as_current_span(
@@ -137,10 +142,14 @@ def create_span(
             is_suppressed = False
 
             if topic:
+                attributes_to_check = {
+                    "type": "kafka",
+                    "kind": "entry",
+                    "kafka.service": topic,
+                    "kafka.access": span_type,
+                }
                 is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
-                    "kafka",
-                    span_type,
-                    topic,
+                    attributes_to_check
                 )
 
             if not is_suppressed and headers:

src/instana/instrumentation/kafka/kafka_python.py

@@ -39,20 +39,26 @@ def trace_kafka_send(
 
         # Get the topic from either args or kwargs
         topic = args[0] if args else kwargs.get("topic", "")
+        attributes_to_check = {
+            "type": "kafka",
+            "kind": "exit",
+            "kafka.service": topic,
+            "kafka.access": "send",
+        }
 
         is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
-            "kafka",
-            "send",
-            topic,
+            attributes_to_check
         )
+
         with tracer.start_as_current_span(
             "kafka-producer", span_context=parent_context, kind=SpanKind.PRODUCER
         ) as span:
             span.set_attribute("kafka.service", topic)
             span.set_attribute("kafka.access", "send")
 
-            # context propagation
+            # Context propagation
             headers = kwargs.get("headers", [])
+
             if not is_suppressed and ("x_instana_l_s", b"0") in headers:
                 is_suppressed = True
 
@@ -70,6 +76,7 @@ def trace_kafka_send(
 
             if tracer.exporter.options.kafka_trace_correlation:
                 kwargs["headers"] = headers
+
             try:
                 res = wrapped(*args, **kwargs)
                 return res
@@ -94,10 +101,14 @@ def create_span(
 
             is_suppressed = False
             if topic:
+                attributes_to_check = {
+                    "type": "kafka",
+                    "kind": "entry",
+                    "kafka.service": topic,
+                    "kafka.access": span_type,
+                }
                 is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
-                    "kafka",
-                    span_type,
-                    topic,
+                    attributes_to_check
                 )
 
             if not is_suppressed and headers: