diff --git a/A365_DOCUMENTATION.md b/A365_DOCUMENTATION.md index 723aafea..72321245 100644 --- a/A365_DOCUMENTATION.md +++ b/A365_DOCUMENTATION.md @@ -254,6 +254,45 @@ with ( ... ``` +To copy an application-specific baggage value onto Agent365 GenAI spans, +explicitly opt in each key with `custom_attribute()` or `custom_attributes()`: + +```python +with ( + BaggageBuilder() + .tenant_id("contoso-tenant") + .agent_id("weather-agent-001") + .custom_attribute("customer.tier", "gold") + .custom_attributes({"customer.region": "west"}) + .build() +): + with InvokeAgentScope.start(...) as scope: + ... +``` + +`set_pairs()` only sets baggage. It does not opt arbitrary baggage keys into +span attributes; use `custom_attribute()` for any custom key that should appear +on recognized Agent365 GenAI spans. Baggage propagation never overwrites +attributes already present on the current span. Baggage-propagated values are +applied when the span starts, so they precede later `record_attributes()` calls +when duplicate-key protection is also present; direct/current span attributes +remain authoritative. + +A span is recognized as GenAI at span start by evaluating these signals in +order: a supported `gen_ai.operation.name` attribute; if that attribute is +present but unrecognized (`chain`, `embeddings`, `text_completion`, +`generate_content`, `create_agent`, ...) it is authoritative, so baggage and +span-name inference are skipped and only the instrumentation scope can still +classify the span; otherwise a recognized `gen_ai.operation.name` baggage entry, +then a span name matching a supported operation (`invoke_agent ...`, +`chat ...`, ...) or a known pre-rename name (`chat.completions ...`), then a +supported GenAI instrumentation scope (`Agent365Sdk`, `semantic_kernel.*`, +`agent_framework`, `microsoft.opentelemetry._genai.*`, +`opentelemetry.instrumentation.openai_v2`, +`opentelemetry.instrumentation.openai_agents`). Spans classified only by +instrumentation scope are GenAI with an unknown operation: opted-in custom +baggage applies to them, but `invoke_agent`-only attributes never do. + ### From TurnContext (Hosting Framework) ```python diff --git a/CHANGELOG.md b/CHANGELOG.md index ec45c246..c23ad83e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # Release History # Unreleased ### Features Added +- Add explicit custom baggage APIs and propagate opted-in attributes to supported GenAI spans, including unmodeled operations. ([#264](https://github.com/microsoft/opentelemetry-distro-python/pull/264)) - Add Python-native `InvokeAgentScope` request and response parameter models that emit OpenTelemetry GenAI semantic attributes, including structured system instructions and cache read/write token counts, introduced by .NET diff --git a/samples/langchain/validate_traces.py b/samples/langchain/validate_traces.py index 2147717d..254582ee 100644 --- a/samples/langchain/validate_traces.py +++ b/samples/langchain/validate_traces.py @@ -153,9 +153,7 @@ def check(label, condition, detail=""): print(f"\n{'='*60}") print("RESPONSES-API FAKE SPAN CHECKS") print(f"{'='*60}") - responses_spans = [ - s for s in llm_spans if s.attributes.get("gen_ai.response.id") == "resp_abc123" - ] + responses_spans = [s for s in llm_spans if s.attributes.get("gen_ai.response.id") == "resp_abc123"] check("Responses-API fake LLM span found", len(responses_spans) == 1, f"found {len(responses_spans)}") if responses_spans: attrs = responses_spans[0].attributes diff --git a/samples/microsoft_agent_framework/sample_maf_agent.py b/samples/microsoft_agent_framework/sample_maf_agent.py index cfbcd166..171a88d5 100644 --- a/samples/microsoft_agent_framework/sample_maf_agent.py +++ b/samples/microsoft_agent_framework/sample_maf_agent.py @@ -60,4 +60,4 @@ async def main(): if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file + asyncio.run(main()) diff --git a/src/microsoft/opentelemetry/a365/README.md b/src/microsoft/opentelemetry/a365/README.md index dce127ba..251c9af9 100644 --- a/src/microsoft/opentelemetry/a365/README.md +++ b/src/microsoft/opentelemetry/a365/README.md @@ -20,7 +20,7 @@ Core tracing primitives — scopes, configuration, data models, and internal uti | `__init__.py` | Public API surface. Re-exports scope classes, data models, enums, and related core types. | | `agent_details.py` | `AgentDetails` dataclass — metadata about an AI agent (ID, name, description, blueprint/platform IDs, tenant, version). | | `channel.py` | `Channel` dataclass — channel context (name, link) for agent execution. | -| `constants.py` | Core-level constants for span operations, OTel conventions, feature switches, and error types. | +| `constants.py` | Core-level constants for span operations, GenAI processor operation names, baggage metadata, OTel conventions, feature switches, and error types. | | `gen_ai_request_parameters.py` | GenAI request parameter models for generation settings and related semantic attributes. | | `gen_ai_response_parameters.py` | GenAI response parameter models for finish reasons and token usage attributes. | | `execute_tool_scope.py` | `ExecuteToolScope` — tracing scope for AI tool executions. Records tool name, arguments, call ID, type, and endpoint. | @@ -47,7 +47,7 @@ Span export pipeline — processors and exporters for Agent365 and Spectra backe | `agent365_exporter_options.py` | `Agent365ExporterOptions` — configuration for the Agent365 exporter (cluster category, token resolver, endpoint flags, batch settings). | | `enriched_span.py` | `EnrichedReadableSpan` — wrapper allowing extra attributes on immutable `ReadableSpan` objects. | | `enriching_span_processor.py` | Span enrichment support with registration for platform instrumentors (LangChain, Semantic Kernel, OpenAI Agents). `_EnrichingBatchSpanProcessor` applies enrichers before batching. | -| `span_processor.py` | `A365SpanProcessor` — propagates OpenTelemetry baggage entries onto spans as attributes, with special handling for invoke_agent spans. | +| `span_processor.py` | `A365SpanProcessor` — propagates documented OpenTelemetry baggage entries onto spans, keeps invoke_agent-specific handling, and copies only opted-in custom baggage keys onto recognized GenAI spans. | | `spectra_exporter_options.py` | `SpectraExporterOptions` — configuration for OTLP export to a Spectra Collector sidecar (gRPC or HTTP, tuned for Kubernetes). | | `utils.py` | Exporter utilities: hex encoding for trace/span IDs, span size truncation, span partitioning, environment variable handling, payload building helpers. | @@ -55,7 +55,7 @@ Span export pipeline — processors and exporters for Agent365 and Spectra backe | File | Description | |------|-------------| -| `baggage_builder.py` | `BaggageBuilder` — fluent API for setting per-request baggage values (tenant ID, agent ID, caller/user details, session/conversation IDs, channel, endpoints). Provides context manager for baggage scope. | +| `baggage_builder.py` | `BaggageBuilder` — fluent API for setting per-request baggage values (tenant ID, agent ID, caller/user details, session/conversation IDs, channel, endpoints) plus `custom_attribute()` / `custom_attributes()` opt-in for custom GenAI span attributes. Provides context manager for baggage scope. | ### `core/models/` @@ -150,7 +150,7 @@ below are available via `from microsoft.opentelemetry.a365.core import ...`. | Symbol | Kind | Description | |--------|------|-------------| -| `BaggageBuilder` | class | Fluent API for setting per-request baggage (tenant, agent, user, channel, session, conversation). Call `.build()` to get a context manager. | +| `BaggageBuilder` | class | Fluent API for setting per-request baggage (tenant, agent, user, channel, session, conversation) and opting custom baggage keys into recognized GenAI span attributes. Call `.build()` to get a context manager. | #### Data Classes diff --git a/src/microsoft/opentelemetry/a365/core/constants.py b/src/microsoft/opentelemetry/a365/core/constants.py index ca24cb87..150fda42 100644 --- a/src/microsoft/opentelemetry/a365/core/constants.py +++ b/src/microsoft/opentelemetry/a365/core/constants.py @@ -7,6 +7,8 @@ shared across the Agent365 core scopes and exporters. """ +from microsoft.opentelemetry.a365.core.inference_operation_type import InferenceOperationType + # --- Span operation names --- INVOKE_AGENT_OPERATION_NAME = "invoke_agent" EXECUTE_TOOL_OPERATION_NAME = "execute_tool" @@ -14,6 +16,20 @@ CHAT_OPERATION_NAME = "chat" APPLY_GUARDRAIL_OPERATION_NAME = "apply_guardrail" +GEN_AI_PROCESSOR_OPERATION_NAMES: frozenset[str] = frozenset( + { + INVOKE_AGENT_OPERATION_NAME, + EXECUTE_TOOL_OPERATION_NAME, + OUTPUT_MESSAGES_OPERATION_NAME, + CHAT_OPERATION_NAME, + APPLY_GUARDRAIL_OPERATION_NAME, + } + | {operation.value for operation in InferenceOperationType} +) + +# --- Baggage metadata --- +CUSTOM_KEYS_BAGGAGE_KEY = "_internal.custom_keys" + # --- OpenTelemetry semantic conventions --- ERROR_TYPE_KEY = "error.type" ERROR_MESSAGE_KEY = "error.message" @@ -21,6 +37,28 @@ AZURE_RP_NAMESPACE_VALUE = "Microsoft.CognitiveServices" SOURCE_NAME = "Agent365Sdk" +# Identify GenAI spans before instrumentations set ``gen_ai.operation.name``. +# Scope matching accepts an exact root or dotted child. +GEN_AI_INSTRUMENTATION_SCOPE_ROOTS: tuple[str, ...] = ( + SOURCE_NAME, + "agent_framework", + "semantic_kernel", + "microsoft.opentelemetry._genai", + "opentelemetry.instrumentation.openai_v2", + "opentelemetry.instrumentation.openai_agents", +) + +# Initial names used by supported GenAI instrumentations before span renaming. +GEN_AI_INITIAL_SPAN_NAMES: frozenset[str] = frozenset( + { + "chat.completions", + "chat.streaming_completions", + "text.completions", + "text.streaming_completions", + "text_completions", + } +) + # --- Feature switches --- ENABLE_OPENTELEMETRY_SWITCH = "Azure.Experimental.EnableActivitySource" TRACE_CONTENTS_SWITCH = "Azure.Experimental.TraceGenAIMessageContent" diff --git a/src/microsoft/opentelemetry/a365/core/exporters/_gen_ai_span_classifier.py b/src/microsoft/opentelemetry/a365/core/exporters/_gen_ai_span_classifier.py new file mode 100644 index 00000000..94368289 --- /dev/null +++ b/src/microsoft/opentelemetry/a365/core/exporters/_gen_ai_span_classifier.py @@ -0,0 +1,106 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Classify spans for Agent365 GenAI baggage propagation.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from microsoft.opentelemetry.a365.core.constants import ( + GEN_AI_INITIAL_SPAN_NAMES, + GEN_AI_INSTRUMENTATION_SCOPE_ROOTS, + GEN_AI_OPERATION_NAME_KEY, + GEN_AI_PROCESSOR_OPERATION_NAMES, +) + + +@dataclass(frozen=True) +class _GenAISpanClassification: + """Whether a span is GenAI and, when identifiable, the operation it represents.""" + + is_gen_ai_span: bool + operation_name: str | None = None + + +def _recognized_operation_name(value: object | None) -> str | None: + return value if isinstance(value, str) and value in GEN_AI_PROCESSOR_OPERATION_NAMES else None + + +def _span_name(span: Any) -> str | None: + span_name = getattr(span, "name", None) + return span_name if isinstance(span_name, str) else None + + +def _operation_name_from_span_name(span: Any) -> str | None: + span_name = _span_name(span) + if span_name is None: + return None + + for operation_name in GEN_AI_PROCESSOR_OPERATION_NAMES: + if span_name == operation_name or span_name.startswith(f"{operation_name} "): + return operation_name + return None + + +def _has_known_initial_span_name(span: Any) -> bool: + """Match span names supported instrumentations use before renaming the span.""" + span_name = _span_name(span) + if span_name is None: + return False + + return any(span_name == known or span_name.startswith(f"{known} ") for known in GEN_AI_INITIAL_SPAN_NAMES) + + +def _instrumentation_scope_name(span: Any) -> str | None: + """Read the tracer (source) name recorded on a ReadWriteSpan.""" + # pylint: disable=broad-exception-caught + for attribute_name in ("instrumentation_scope", "instrumentation_info"): + try: + scope = getattr(span, attribute_name, None) + scope_name = getattr(scope, "name", None) if scope is not None else None + except Exception: + continue + if isinstance(scope_name, str) and scope_name: + return scope_name + return None + + +def _is_supported_gen_ai_scope(span: Any) -> bool: + scope_name = _instrumentation_scope_name(span) + if scope_name is None: + return False + + return any(scope_name == root or scope_name.startswith(f"{root}.") for root in GEN_AI_INSTRUMENTATION_SCOPE_ROOTS) + + +def _classify_gen_ai_span( + span: Any, + existing_attributes: Mapping[str, object], + baggage_map: Mapping[str, object], +) -> _GenAISpanClassification: + """Resolve whether a span is GenAI and which operation it represents. + + An explicit ``gen_ai.operation.name`` attribute is authoritative over the + baggage and span-name inference signals, including when it holds a value + this processor does not model (``chain``, ``embeddings``, + ``text_completion``, ``generate_content``, ``create_agent``). Such a span is + still GenAI when a supported instrumentation emitted it, but its operation + stays unknown so invoke_agent-only attributes are withheld. + + Without an explicit operation attribute, a recognized span-name operation + takes precedence over inherited operation baggage. + """ + if GEN_AI_OPERATION_NAME_KEY in existing_attributes: + explicit_operation_name = _recognized_operation_name(existing_attributes.get(GEN_AI_OPERATION_NAME_KEY)) + if explicit_operation_name is not None: + return _GenAISpanClassification(True, explicit_operation_name) + return _GenAISpanClassification(_is_supported_gen_ai_scope(span)) + + operation_name = _operation_name_from_span_name(span) + if operation_name is None: + operation_name = _recognized_operation_name(baggage_map.get(GEN_AI_OPERATION_NAME_KEY)) + if operation_name is not None: + return _GenAISpanClassification(True, operation_name) + + return _GenAISpanClassification(_has_known_initial_span_name(span) or _is_supported_gen_ai_scope(span)) diff --git a/src/microsoft/opentelemetry/a365/core/exporters/span_processor.py b/src/microsoft/opentelemetry/a365/core/exporters/span_processor.py index fef0d4e0..543b04d1 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/span_processor.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/span_processor.py @@ -12,6 +12,31 @@ * For each documented key with a truthy value not already present as a span attribute, add it via span.set_attribute * Never overwrites existing attributes + +Custom baggage is propagated only to recognized GenAI spans. A span is +recognized as GenAI by evaluating these signals in order at ``on_start``: + + 1. An explicit ``gen_ai.operation.name`` attribute holding a recognized + operation: GenAI with a known operation. + 2. An explicit but *unrecognized* ``gen_ai.operation.name`` attribute: the + attribute is authoritative, so the baggage and span-name inference of + signals 3 and 4 is skipped. The span is still GenAI when a supported + instrumentation emitted it (signal 5), with an unknown operation. + 3. A recognized ``gen_ai.operation.name`` baggage entry. + 4. A span name that is (or starts with) a recognized operation name, or a + name a supported instrumentation is known to use before it renames the + span (Semantic Kernel ``chat.completions ``). + 5. The instrumentation scope (source) name of a supported GenAI + instrumentation: GenAI with an unknown operation. + +Signals 4 and 5 exist because most GenAI instrumentations apply +``gen_ai.operation.name`` *after* the span starts: LangChain chat spans start +as ``ChatOpenAI`` and the OpenAI Agents processor starts workflow spans as +``Agent workflow``. Signal 5 also keeps spans whose operation this processor +does not model (``chain``, ``embeddings``, ``text_completion``, +``generate_content``, ``create_agent``) from being dropped. Only signals 1, 3 +and 4 identify *which* operation a span represents, which is what gates the +invoke_agent-only attributes. """ from __future__ import annotations @@ -19,9 +44,11 @@ from opentelemetry import baggage, context from opentelemetry.sdk.trace import SpanProcessor as BaseSpanProcessor -from microsoft.opentelemetry.a365.constants import ( +from microsoft.opentelemetry.a365.core.exporters._gen_ai_span_classifier import _classify_gen_ai_span +from microsoft.opentelemetry.a365.core.constants import ( CHANNEL_LINK_KEY, CHANNEL_NAME_KEY, + CUSTOM_KEYS_BAGGAGE_KEY, CUSTOM_PARENT_SPAN_ID_KEY, CUSTOM_SPAN_NAME_KEY, GEN_AI_AGENT_AUID_KEY, @@ -98,6 +125,21 @@ ] +def _custom_baggage_keys(baggage_map) -> list[str]: + metadata = baggage_map.get(CUSTOM_KEYS_BAGGAGE_KEY) + if not metadata: + return [] + + keys: list[str] = [] + for raw_key in str(metadata).split(","): + key = raw_key.strip() + if not key or key == CUSTOM_KEYS_BAGGAGE_KEY: + continue + if key not in keys: + keys.append(key) + return keys + + # pylint: disable=broad-exception-caught, too-many-branches, useless-parent-delegation # pylint: disable=global-statement class A365SpanProcessor(BaseSpanProcessor): @@ -151,18 +193,17 @@ def on_start(self, span, parent_context=None): # type: ignore[override] except Exception: baggage_map = {} - operation_name = existing.get(GEN_AI_OPERATION_NAME_KEY) - is_invoke_agent = False - if operation_name == INVOKE_AGENT_OPERATION_NAME: - is_invoke_agent = True - elif isinstance(getattr(span, "name", None), str) and span.name.startswith(INVOKE_AGENT_OPERATION_NAME): - is_invoke_agent = True + classification = _classify_gen_ai_span(span, existing, baggage_map) target_keys = list(COMMON_ATTRIBUTES) - if is_invoke_agent: + if classification.operation_name == INVOKE_AGENT_OPERATION_NAME: for k in INVOKE_AGENT_ATTRIBUTES: if k not in target_keys: target_keys.append(k) + if classification.is_gen_ai_span: + for k in _custom_baggage_keys(baggage_map): + if k not in target_keys: + target_keys.append(k) for key in target_keys: if key in existing: diff --git a/src/microsoft/opentelemetry/a365/core/middleware/baggage_builder.py b/src/microsoft/opentelemetry/a365/core/middleware/baggage_builder.py index 2b96c586..64f2ade5 100644 --- a/src/microsoft/opentelemetry/a365/core/middleware/baggage_builder.py +++ b/src/microsoft/opentelemetry/a365/core/middleware/baggage_builder.py @@ -4,6 +4,7 @@ # Per request baggage builder for OpenTelemetry context propagation. import logging +from collections.abc import Iterable, Mapping from typing import Any from opentelemetry import baggage, context @@ -11,6 +12,7 @@ from microsoft.opentelemetry.a365.core.constants import ( CHANNEL_LINK_KEY, CHANNEL_NAME_KEY, + CUSTOM_KEYS_BAGGAGE_KEY, GEN_AI_AGENT_AUID_KEY, GEN_AI_AGENT_BLUEPRINT_ID_KEY, GEN_AI_AGENT_DESCRIPTION_KEY, @@ -57,9 +59,10 @@ class BaggageBuilder: # Baggage is restored after exiting the context """ - def __init__(self): + def __init__(self) -> None: """Initialize the baggage builder.""" self._pairs: dict[str, str] = {} + self._custom_keys: list[str] = [] def operation_source(self, value: str | None) -> "BaggageBuilder": """Set the operation source baggage value. @@ -235,7 +238,29 @@ def set_pairs(self, pairs: Any) -> "BaggageBuilder": for k, v in iterator: if v is None: continue - self._set(str(k), str(v)) + key = str(k) + if key == CUSTOM_KEYS_BAGGAGE_KEY: + continue + self._set(key, str(v)) + return self + + def custom_attribute(self, key: str, value: str | None) -> "BaggageBuilder": + """Set a custom baggage value that should propagate to GenAI spans.""" + normalized_key = self._validate_custom_key(key) + if value is not None and value.strip(): + self._pairs[normalized_key] = value + if normalized_key not in self._custom_keys: + self._custom_keys.append(normalized_key) + return self + + def custom_attributes( + self, + attributes: Mapping[str, str | None] | Iterable[tuple[str, str | None]], + ) -> "BaggageBuilder": + """Set custom baggage values that should propagate to GenAI spans.""" + iterator = attributes.items() if isinstance(attributes, Mapping) else attributes + for key, value in iterator: + self.custom_attribute(key, value) return self def build(self) -> "BaggageScope": @@ -244,7 +269,11 @@ def build(self) -> "BaggageScope": Returns: A context manager that restores the previous baggage on exit """ - return BaggageScope(self._pairs) + pairs = self._pairs.copy() + custom_keys = [key for key in self._custom_keys if pairs.get(key, "").strip()] + if custom_keys: + pairs[CUSTOM_KEYS_BAGGAGE_KEY] = ",".join(custom_keys) + return BaggageScope(pairs) def _set(self, key: str, value: str | None) -> None: """Add a baggage key/value if the value is not None or whitespace. @@ -256,6 +285,17 @@ def _set(self, key: str, value: str | None) -> None: if value is not None and value.strip(): self._pairs[key] = value + @staticmethod + def _validate_custom_key(key: str) -> str: + normalized_key = key.strip() + if not normalized_key: + raise ValueError("custom baggage key must not be blank") + if "," in normalized_key: + raise ValueError("custom baggage key must not contain commas") + if normalized_key == CUSTOM_KEYS_BAGGAGE_KEY: + raise ValueError(f"{CUSTOM_KEYS_BAGGAGE_KEY} is reserved") + return normalized_key + class BaggageScope: """Context manager for baggage scope. @@ -287,6 +327,11 @@ def __enter__(self) -> "BaggageScope": new_context = self._previous_context for key, value in self._pairs.items(): if value and value.strip(): + if key == CUSTOM_KEYS_BAGGAGE_KEY: + inherited_value = baggage.get_baggage(key, context=new_context) + inherited_keys = str(inherited_value).split(",") if inherited_value else [] + current_keys = value.split(",") + value = ",".join(dict.fromkeys(inherited_keys + current_keys)) new_context = baggage.set_baggage(key, value, context=new_context) # Attach the new context diff --git a/tests/a365/test_baggage_builder.py b/tests/a365/test_baggage_builder.py new file mode 100644 index 00000000..7949de87 --- /dev/null +++ b/tests/a365/test_baggage_builder.py @@ -0,0 +1,86 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import pytest +from opentelemetry import baggage, context + +from microsoft.opentelemetry.a365.core.constants import CUSTOM_KEYS_BAGGAGE_KEY +from microsoft.opentelemetry.a365.core.middleware.baggage_builder import BaggageBuilder + + +def test_custom_attribute_sets_value_and_metadata(): + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + assert baggage.get_baggage("customer.tier") == "gold" + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) == "customer.tier" + + +def test_custom_attributes_track_multiple_keys_in_order_without_duplicates(): + attributes = [ + ("customer.tier", "gold"), + ("customer.region", "west"), + ("customer.tier", "platinum"), + ] + + with BaggageBuilder().custom_attributes(attributes).build(): + assert baggage.get_baggage("customer.tier") == "platinum" + assert baggage.get_baggage("customer.region") == "west" + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) == "customer.tier,customer.region" + + +def test_blank_custom_values_are_skipped_without_metadata(): + builder = BaggageBuilder().custom_attribute("customer.tier", " ") + + with builder.build(): + assert baggage.get_baggage("customer.tier") is None + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) is None + + +@pytest.mark.parametrize("key", ["", "bad,key", CUSTOM_KEYS_BAGGAGE_KEY]) +def test_custom_attribute_rejects_invalid_keys(key): + with pytest.raises(ValueError): + BaggageBuilder().custom_attribute(key, "value") + + +def test_set_pairs_does_not_mark_custom_metadata(): + with BaggageBuilder().set_pairs({"customer.tier": "gold"}).build(): + assert baggage.get_baggage("customer.tier") == "gold" + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) is None + + +def test_set_pairs_does_not_accept_custom_metadata(): + pairs = { + "customer.tier": "gold", + CUSTOM_KEYS_BAGGAGE_KEY: "customer.tier", + } + + with BaggageBuilder().set_pairs(pairs).build(): + assert baggage.get_baggage("customer.tier") == "gold" + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) is None + + +def test_nested_custom_attributes_preserve_outer_metadata(): + outer_attributes = { + "customer.tier": "gold", + "customer.region": "west", + } + inner_attributes = { + "customer.region": "east", + "customer.segment": "enterprise", + } + + with BaggageBuilder().custom_attributes(outer_attributes).build(): + with BaggageBuilder().custom_attributes(inner_attributes).build(): + assert baggage.get_baggage("customer.tier") == "gold" + assert baggage.get_baggage("customer.region") == "east" + assert baggage.get_baggage("customer.segment") == "enterprise" + assert baggage.get_baggage(CUSTOM_KEYS_BAGGAGE_KEY) == "customer.tier,customer.region,customer.segment" + + +def test_baggage_scope_restores_previous_context(): + token = context.attach(baggage.set_baggage("customer.tier", "silver")) + try: + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + assert baggage.get_baggage("customer.tier") == "gold" + assert baggage.get_baggage("customer.tier") == "silver" + finally: + context.detach(token) diff --git a/tests/a365/test_gen_ai_span_classifier.py b/tests/a365/test_gen_ai_span_classifier.py new file mode 100644 index 00000000..41110476 --- /dev/null +++ b/tests/a365/test_gen_ai_span_classifier.py @@ -0,0 +1,57 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from types import SimpleNamespace + +import pytest + +from microsoft.opentelemetry.a365.core.constants import ( + GEN_AI_INSTRUMENTATION_SCOPE_ROOTS, + GEN_AI_OPERATION_NAME_KEY, +) +from microsoft.opentelemetry.a365.core.exporters._gen_ai_span_classifier import ( + _classify_gen_ai_span, +) + + +def test_classifies_supported_scope_with_unknown_operation_as_gen_ai(): + span = SimpleNamespace( + name="chain RunnableSequence", + instrumentation_scope=SimpleNamespace(name="microsoft.opentelemetry._genai._langchain"), + ) + + classification = _classify_gen_ai_span( + span, + {GEN_AI_OPERATION_NAME_KEY: "chain"}, + {}, + ) + + assert classification.is_gen_ai_span + assert classification.operation_name is None + + +@pytest.mark.parametrize("root", GEN_AI_INSTRUMENTATION_SCOPE_ROOTS) +@pytest.mark.parametrize("suffix", ["", ".instrumentation"]) +def test_classifies_every_supported_scope_root_and_dotted_child(root, suffix): + span = SimpleNamespace( + name="unmodeled operation", + instrumentation_scope=SimpleNamespace(name=f"{root}{suffix}"), + ) + + classification = _classify_gen_ai_span(span, {}, {}) + + assert classification.is_gen_ai_span + assert classification.operation_name is None + + +@pytest.mark.parametrize("root", GEN_AI_INSTRUMENTATION_SCOPE_ROOTS) +def test_rejects_scope_root_prefix_without_dotted_boundary(root): + span = SimpleNamespace( + name="unmodeled operation", + instrumentation_scope=SimpleNamespace(name=f"{root}_helpers"), + ) + + classification = _classify_gen_ai_span(span, {}, {}) + + assert not classification.is_gen_ai_span + assert classification.operation_name is None diff --git a/tests/a365/test_span_processor.py b/tests/a365/test_span_processor.py index e9c21aa9..da1d8e5b 100644 --- a/tests/a365/test_span_processor.py +++ b/tests/a365/test_span_processor.py @@ -3,6 +3,7 @@ # pylint: disable=no-member import unittest +from types import SimpleNamespace from unittest.mock import MagicMock from opentelemetry import baggage, context @@ -12,6 +13,24 @@ COMMON_ATTRIBUTES, INVOKE_AGENT_ATTRIBUTES, ) +from microsoft.opentelemetry.a365.core.middleware.baggage_builder import BaggageBuilder +from microsoft.opentelemetry.a365.core.constants import GEN_AI_OPERATION_NAME_KEY + +LANGCHAIN_SCOPE = "microsoft.opentelemetry._genai._langchain._tracer_instrumentor" +OPENAI_AGENTS_SCOPE = "microsoft.opentelemetry._genai._openai_agents._trace_instrumentor" +UPSTREAM_OPENAI_AGENTS_SCOPE = "opentelemetry.instrumentation.openai_agents" +OPENAI_V2_SCOPE = "opentelemetry.instrumentation.openai_v2" +SEMANTIC_KERNEL_SCOPE = "semantic_kernel.utils.telemetry.model_diagnostics.decorators" +AGENT_FRAMEWORK_SCOPE = "agent_framework" + + +def _mock_span(name, attributes=None, scope_name=None): + """Build a mock ReadWriteSpan with a controllable instrumentation scope.""" + span = MagicMock() + span.name = name + span.attributes = dict(attributes or {}) + span.instrumentation_scope = SimpleNamespace(name=scope_name, version=None) if scope_name else None + return span class TestA365SpanProcessor(unittest.TestCase): @@ -129,6 +148,335 @@ def test_invoke_agent_attributes_not_propagated_for_other_spans(self): for call in span.set_attribute.call_args_list: self.assertNotEqual(call[0][0], "microsoft.a365.caller.agent.id") + def test_custom_baggage_attribute_propagated_to_genai_span(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "invoke_agent Travel_Assistant" + span.attributes = {"gen_ai.operation.name": "invoke_agent"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_set_pairs_custom_baggage_is_not_propagated_without_opt_in(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "invoke_agent Travel_Assistant" + span.attributes = {"gen_ai.operation.name": "invoke_agent"} + + with BaggageBuilder().set_pairs({"customer.tier": "gold"}).build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_custom_baggage_attribute_does_not_overwrite_span_attribute(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "invoke_agent Travel_Assistant" + span.attributes = {"gen_ai.operation.name": "invoke_agent", "customer.tier": "direct"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_custom_baggage_attribute_ignored_on_non_genai_span(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "http request" + span.attributes = {"gen_ai.operation.name": "not_genai"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_custom_baggage_attribute_ignored_for_false_operation_prefixes(self): + processor = A365SpanProcessor() + + for span_name in ("chatbot_loop", "execute_toolbox"): + with self.subTest(span_name=span_name): + span = MagicMock() + span.name = span_name + span.attributes = {} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_custom_baggage_attribute_uses_recognized_baggage_operation_name(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "http request" + span.attributes = {} + + with ( + BaggageBuilder() + .set_pairs({GEN_AI_OPERATION_NAME_KEY: "chat"}) + .custom_attribute("customer.tier", "gold") + .build() + ): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_attribute_honors_unrecognized_explicit_operation_name(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "chat gpt-4" + span.attributes = {GEN_AI_OPERATION_NAME_KEY: "not_genai"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_custom_baggage_attribute_propagated_to_text_completion_span(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "TextCompletion summarize" + span.attributes = {"gen_ai.operation.name": "TextCompletion"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_attribute_propagated_to_generate_content_span(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "GenerateContent image" + span.attributes = {"gen_ai.operation.name": "GenerateContent"} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_attribute_propagated_to_known_initial_span_name(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "chat.completions gpt-4o" + span.attributes = {} + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_attribute_propagated_to_supported_scope_child(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "ChatOpenAI" + span.attributes = {} + span.instrumentation_scope = SimpleNamespace(name="microsoft.opentelemetry._genai._langchain") + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_attribute_ignored_for_scope_prefix_without_boundary(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "helper span" + span.attributes = {} + span.instrumentation_scope = SimpleNamespace(name="semantic_kernel_helpers") + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + # -- explicit but unrecognized operation names -- + + def test_custom_baggage_propagated_for_explicit_chain_operation_on_openai_agents_scope(self): + """OpenAI Agents emits ``chain`` spans this processor does not model.""" + processor = A365SpanProcessor() + span = _mock_span( + "chain RunnableSequence", + {GEN_AI_OPERATION_NAME_KEY: "chain"}, + scope_name=OPENAI_AGENTS_SCOPE, + ) + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_propagated_for_explicit_embeddings_operation_on_agent_framework_scope(self): + processor = A365SpanProcessor() + span = _mock_span( + "embeddings text-embedding-3-small", + {GEN_AI_OPERATION_NAME_KEY: "embeddings"}, + scope_name=AGENT_FRAMEWORK_SCOPE, + ) + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_propagated_for_explicit_text_completion_operation_on_openai_v2_scope(self): + processor = A365SpanProcessor() + span = _mock_span( + "text_completion gpt-4o", + {GEN_AI_OPERATION_NAME_KEY: "text_completion"}, + scope_name=OPENAI_V2_SCOPE, + ) + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_propagated_for_other_unrecognized_operations_on_supported_scopes(self): + for scope_name, operation_name in ( + (UPSTREAM_OPENAI_AGENTS_SCOPE, "create_agent"), + (LANGCHAIN_SCOPE, "generate_content"), + (SEMANTIC_KERNEL_SCOPE, "embeddings"), + ): + with self.subTest(scope_name=scope_name, operation_name=operation_name): + processor = A365SpanProcessor() + span = _mock_span( + f"{operation_name} target", + {GEN_AI_OPERATION_NAME_KEY: operation_name}, + scope_name=scope_name, + ) + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + + def test_custom_baggage_ignored_for_unrecognized_explicit_operation_on_unrelated_scope(self): + processor = A365SpanProcessor() + span = _mock_span( + "POST /v1/chain", + {GEN_AI_OPERATION_NAME_KEY: "chain"}, + scope_name="opentelemetry.instrumentation.requests", + ) + + with BaggageBuilder().custom_attribute("customer.tier", "gold").build(): + processor.on_start(span, parent_context=context.get_current()) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "customer.tier") + + def test_unrecognized_explicit_operation_with_invoke_agent_span_name_is_not_invoke_agent(self): + """An unrecognized operation stays unknown, so invoke-only keys are withheld.""" + processor = A365SpanProcessor() + span = _mock_span( + "invoke_agent Travel_Assistant", + {GEN_AI_OPERATION_NAME_KEY: "create_agent"}, + scope_name=OPENAI_AGENTS_SCOPE, + ) + + with ( + BaggageBuilder() + .set_pairs( + { + "microsoft.a365.caller.agent.id": "caller-1", + "server.address": "agent.contoso.com", + } + ) + .custom_attribute("customer.tier", "gold") + .build() + ): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + for call in span.set_attribute.call_args_list: + self.assertNotIn(call[0][0], ("microsoft.a365.caller.agent.id", "server.address")) + + def test_unrecognized_explicit_operation_ignores_invoke_agent_baggage_operation(self): + """Baggage inference is skipped once an explicit operation is present.""" + processor = A365SpanProcessor() + span = _mock_span( + "chain RunnableSequence", + {GEN_AI_OPERATION_NAME_KEY: "chain"}, + scope_name=LANGCHAIN_SCOPE, + ) + + ctx = baggage.set_baggage(GEN_AI_OPERATION_NAME_KEY, "invoke_agent", context.get_current()) + ctx = baggage.set_baggage("microsoft.a365.caller.agent.id", "caller-1", ctx) + + processor.on_start(span, parent_context=ctx) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "microsoft.a365.caller.agent.id") + + def test_invoke_agent_attributes_use_recognized_baggage_operation_name(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "http request" + span.attributes = {} + + ctx = context.get_current() + ctx = baggage.set_baggage(GEN_AI_OPERATION_NAME_KEY, "invoke_agent", ctx) + ctx = baggage.set_baggage("microsoft.a365.caller.agent.id", "caller-1", ctx) + + processor.on_start(span, parent_context=ctx) + + span.set_attribute.assert_any_call("microsoft.a365.caller.agent.id", "caller-1") + + def test_span_name_operation_takes_precedence_over_inherited_baggage_operation(self): + processor = A365SpanProcessor() + span = _mock_span("execute_tool get_weather") + + with ( + BaggageBuilder() + .set_pairs( + { + GEN_AI_OPERATION_NAME_KEY: "invoke_agent", + "microsoft.a365.caller.agent.id": "caller-1", + "server.address": "agent.contoso.com", + } + ) + .custom_attribute("customer.tier", "gold") + .build() + ): + processor.on_start(span, parent_context=context.get_current()) + + span.set_attribute.assert_any_call("customer.tier", "gold") + for call in span.set_attribute.call_args_list: + self.assertNotIn(call[0][0], ("microsoft.a365.caller.agent.id", "server.address")) + + def test_invoke_agent_attributes_ignored_for_raw_prefix_without_boundary(self): + processor = A365SpanProcessor() + + span = MagicMock() + span.name = "invoke_agent.debug" + span.attributes = {} + + ctx = context.get_current() + ctx = baggage.set_baggage("microsoft.a365.caller.agent.id", "caller-1", ctx) + + processor.on_start(span, parent_context=ctx) + + for call in span.set_attribute.call_args_list: + self.assertNotEqual(call[0][0], "microsoft.a365.caller.agent.id") + def test_empty_baggage(self): processor = A365SpanProcessor()