Filter A365 exports to only include genAI spans

Author: AlexLu2016

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter.py

@@ -18,6 +18,7 @@
 from opentelemetry.trace import StatusCode
 
 from .utils import (
+    INFERENCE_OPERATION_TYPE_NAMES,
     build_export_url,
     get_validated_domain_override,
     hex_span_id,
@@ -28,6 +29,7 @@
     status_name,
     truncate_span,
 )
+from ..constants import CHAT_OPERATION_NAME, GEN_AI_OPERATION_NAME_KEY
 
 # ---- Exporter ---------------------------------------------------------------
 
@@ -277,6 +279,14 @@ def _map_span(self, sp: ReadableSpan) -> dict[str, Any]:
         # attributes
         attrs = dict(sp.attributes or {})
 
+        # Normalize gen_ai.operation.name from any InferenceOperationType enum
+        # value (Chat, TextCompletion, GenerateContent) to the canonical "chat"
+        # value the ingest service accepts. This is applied only on the export
+        # payload; the underlying span attribute is left untouched.
+        op_name = attrs.get(GEN_AI_OPERATION_NAME_KEY)
+        if isinstance(op_name, str) and op_name in INFERENCE_OPERATION_TYPE_NAMES:
+            attrs[GEN_AI_OPERATION_NAME_KEY] = CHAT_OPERATION_NAME
+
         # events
         events = []
         for ev in sp.events:

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/utils.py

@@ -14,16 +14,47 @@
 from opentelemetry.trace import SpanKind, StatusCode
 
 from ..constants import (
+    CHAT_OPERATION_NAME,
     ENABLE_A365_OBSERVABILITY_EXPORTER,
+    EXECUTE_TOOL_OPERATION_NAME,
     GEN_AI_AGENT_ID_KEY,
+    GEN_AI_OPERATION_NAME_KEY,
+    INVOKE_AGENT_OPERATION_NAME,
+    OUTPUT_MESSAGES_OPERATION_NAME,
     TENANT_ID_KEY,
 )
+from ..inference_operation_type import InferenceOperationType
 
 logger = logging.getLogger(__name__)
 
 # Maximum allowed span size in bytes (250KB)
 MAX_SPAN_SIZE_BYTES = 250 * 1024
 
+# Operation names that identify a span as a genAI span eligible for export to
+# the Agent 365 observability ingest service. Spans without a known
+# gen_ai.operation.name are filtered out of the export batch.
+GEN_AI_OPERATION_NAMES: frozenset[str] = frozenset(
+    {
+        INVOKE_AGENT_OPERATION_NAME,
+        EXECUTE_TOOL_OPERATION_NAME,
+        OUTPUT_MESSAGES_OPERATION_NAME,
+        CHAT_OPERATION_NAME,
+        InferenceOperationType.CHAT.value,
+        InferenceOperationType.TEXT_COMPLETION.value,
+        InferenceOperationType.GENERATE_CONTENT.value,
+    }
+)
+
+# Inference operation type values that the ingest service expects to be
+# normalized to the canonical "chat" gen_ai.operation.name.
+INFERENCE_OPERATION_TYPE_NAMES: frozenset[str] = frozenset(
+    {
+        InferenceOperationType.CHAT.value,
+        InferenceOperationType.TEXT_COMPLETION.value,
+        InferenceOperationType.GENERATE_CONTENT.value,
+    }
+)
+
 
 def hex_trace_id(value: int) -> str:
     # 128-bit -> 32 hex chars
@@ -131,18 +162,41 @@ def partition_by_identity(
     spans: Sequence[ReadableSpan],
 ) -> dict[tuple[str, str], list[ReadableSpan]]:
     """
-    Extract (tenantId, agentId). Prefer attributes; if you also stamp baggage
-    into attributes via a processor, they'll be here already.
+    Partition spans by (tenantId, agentId).
+
+    Only genAI spans (those with a known ``gen_ai.operation.name``) are
+    included; non-genAI spans (e.g. HTTP, DB) are filtered out. Spans
+    without both tenant and agent identity are also skipped.
     """
     groups: dict[tuple[str, str], list[ReadableSpan]] = {}
+    non_gen_ai_count = 0
+    missing_identity_count = 0
     for sp in spans:
         attrs = sp.attributes or {}
+        operation_name = as_str(attrs.get(GEN_AI_OPERATION_NAME_KEY))
+        if not operation_name or operation_name not in GEN_AI_OPERATION_NAMES:
+            non_gen_ai_count += 1
+            continue
         tenant = as_str(attrs.get(TENANT_ID_KEY))
         agent = as_str(attrs.get(GEN_AI_AGENT_ID_KEY))
         if not tenant or not agent:
+            missing_identity_count += 1
             continue
         key = (tenant, agent)
         groups.setdefault(key, []).append(sp)
+
+    if non_gen_ai_count > 0:
+        logger.info(f"[Agent365Exporter] {non_gen_ai_count} non-genAI spans filtered out")
+    if missing_identity_count > 0:
+        logger.warning(
+            f"[Agent365Exporter] {missing_identity_count} spans skipped due to "
+            "missing tenant or agent ID"
+        )
+    skipped = non_gen_ai_count + missing_identity_count
+    logger.info(
+        f"[Agent365Exporter] Partitioned into {len(groups)} identity groups "
+        f"({skipped} spans skipped)"
+    )
     return groups
 
 

tests/observability/core/test_agent365_exporter.py

@@ -6,7 +6,12 @@
 import unittest
 from unittest.mock import Mock, patch
 
-from microsoft_agents_a365.observability.core.constants import GEN_AI_AGENT_ID_KEY, TENANT_ID_KEY
+from microsoft_agents_a365.observability.core.constants import (
+    GEN_AI_AGENT_ID_KEY,
+    GEN_AI_OPERATION_NAME_KEY,
+    INVOKE_AGENT_OPERATION_NAME,
+    TENANT_ID_KEY,
+)
 from microsoft_agents_a365.observability.core.exporters.agent365_exporter import (
     DEFAULT_ENDPOINT_URL,
     _Agent365Exporter,
@@ -54,6 +59,7 @@ def _create_mock_span(
         scope_version: str = "1.0.0",
         tenant_id: str = "test-tenant-123",
         agent_id: str = "test-agent-456",
+        operation_name: str | None = INVOKE_AGENT_OPERATION_NAME,
     ) -> ReadableSpan:
         """Create a mock ReadableSpan for testing."""
         mock_span = Mock(spec=ReadableSpan)
@@ -85,6 +91,8 @@ def _create_mock_span(
                 TENANT_ID_KEY: tenant_id,
                 GEN_AI_AGENT_ID_KEY: agent_id,
             })
+        if operation_name is not None and GEN_AI_OPERATION_NAME_KEY not in span_attributes:
+            span_attributes[GEN_AI_OPERATION_NAME_KEY] = operation_name
 
         mock_span.attributes = span_attributes
         mock_span.events = []
@@ -657,6 +665,89 @@ def test_export_no_fallback_when_default_succeeds(self):
             self.assertEqual(result, SpanExportResult.SUCCESS)
             mock_post.assert_called_once()
 
+    def test_export_filters_out_non_genai_spans(self):
+        """Spans without a known gen_ai.operation.name are filtered out."""
+        # Arrange: one genAI span and two non-genAI spans (no/unknown operation name)
+        genai_span = self._create_mock_span("genai_span", trace_id=1, span_id=2)
+        no_op_span = self._create_mock_span("http_span", trace_id=3, span_id=4, operation_name=None)
+        unknown_op_span = self._create_mock_span(
+            "db_span", trace_id=5, span_id=6, operation_name="some_random_op"
+        )
+
+        with patch.object(self.exporter, "_post_with_retries", return_value=True) as mock_post:
+            # Act
+            result = self.exporter.export([genai_span, no_op_span, unknown_op_span])
+
+            # Assert: only the genAI span is exported
+            self.assertEqual(result, SpanExportResult.SUCCESS)
+            mock_post.assert_called_once()
+            _, body, _ = mock_post.call_args[0]
+            request_data = json.loads(body)
+            spans_out = request_data["resourceSpans"][0]["scopeSpans"][0]["spans"]
+            self.assertEqual(len(spans_out), 1)
+            self.assertEqual(spans_out[0]["name"], "genai_span")
+
+    def test_export_filters_out_only_non_genai_spans_returns_success(self):
+        """When all spans are filtered out, export returns SUCCESS without HTTP call."""
+        # Arrange
+        spans = [
+            self._create_mock_span("http_span", operation_name=None),
+            self._create_mock_span("db_span", operation_name="other"),
+        ]
+
+        with patch.object(self.exporter, "_post_with_retries", return_value=True) as mock_post:
+            # Act
+            result = self.exporter.export(spans)
+
+            # Assert
+            self.assertEqual(result, SpanExportResult.SUCCESS)
+            mock_post.assert_not_called()
+
+    def test_export_includes_inference_operation_type_spans(self):
+        """Spans with InferenceOperationType enum values are kept and normalized."""
+        # Arrange
+        chat_span = self._create_mock_span(
+            "chat_span", trace_id=1, span_id=2, operation_name="Chat"
+        )
+        text_completion_span = self._create_mock_span(
+            "text_completion_span", trace_id=3, span_id=4, operation_name="TextCompletion"
+        )
+        generate_content_span = self._create_mock_span(
+            "generate_content_span", trace_id=5, span_id=6, operation_name="GenerateContent"
+        )
+
+        with patch.object(self.exporter, "_post_with_retries", return_value=True) as mock_post:
+            # Act
+            result = self.exporter.export([chat_span, text_completion_span, generate_content_span])
+
+            # Assert: all three are exported and normalized to "chat"
+            self.assertEqual(result, SpanExportResult.SUCCESS)
+            mock_post.assert_called_once()
+            _, body, _ = mock_post.call_args[0]
+            request_data = json.loads(body)
+            spans_out = request_data["resourceSpans"][0]["scopeSpans"][0]["spans"]
+            self.assertEqual(len(spans_out), 3)
+            for span in spans_out:
+                self.assertEqual(span["attributes"]["gen_ai.operation.name"], "chat")
+
+    def test_export_does_not_normalize_canonical_operation_names(self):
+        """invoke_agent / execute_tool / output_messages / chat are not rewritten."""
+        cases = ["invoke_agent", "execute_tool", "output_messages", "chat"]
+        for op in cases:
+            with self.subTest(operation_name=op):
+                span = self._create_mock_span(
+                    f"{op}_span", trace_id=1, span_id=2, operation_name=op
+                )
+                with patch.object(
+                    self.exporter, "_post_with_retries", return_value=True
+                ) as mock_post:
+                    result = self.exporter.export([span])
+                    self.assertEqual(result, SpanExportResult.SUCCESS)
+                    _, body, _ = mock_post.call_args[0]
+                    request_data = json.loads(body)
+                    span_out = request_data["resourceSpans"][0]["scopeSpans"][0]["spans"][0]
+                    self.assertEqual(span_out["attributes"]["gen_ai.operation.name"], op)
+
 
 if __name__ == "__main__":
     unittest.main()