diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/CHANGELOG.md index e04229104..c0cb68363 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/CHANGELOG.md @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Avoid enriching non-MAF spans that use overlapping GenAI operation names. - Preserve parent-child context for legacy Microsoft Agent Framework streaming agent spans. +- Stop exporting process-cumulative GenAI ObservableGauges and use Microsoft + Agent Framework's native metric instruments instead. +- Keep non-streaming MAF spans current for nested work and scope Robin provider + suppression to the active LLM call when that optional integration is present. ## Version 0.7.0 (2026-07-03) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/README.rst b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/README.rst index 843fce423..ddef7d7e5 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/README.rst +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/README.rst @@ -15,11 +15,14 @@ through ``opentelemetry-util-genai`` before spans are ended: recording. This ensures exporter snapshots include ``gen_ai.span.kind``, ``gen_ai.operation.name``, normalized provider names, token usage, finish reasons, and streaming TTFT where MAF exposes enough - data. + data. Non-streaming spans remain current for the wrapped call, and nested + Robin provider instrumentation is suppressed when its optional context key + is available. * ``MAFSemanticProcessor`` remains as a compatibility layer for MAF workflow, - MCP, private-prefix attribute normalization, and the in-process ARMS gauges. - Successful spans keep the OpenTelemetry default ``UNSET`` status; failed - spans keep MAF's ``ERROR`` status. + MCP, and private-prefix attribute normalization. Metrics come from MAF's + native counter and histogram instruments; this package does not export + process-cumulative ObservableGauges. Successful spans keep the OpenTelemetry + default ``UNSET`` status; failed spans keep MAF's ``ERROR`` status. * ``react_step_patch`` (opt-in via ``ARMS_MAF_REACT_STEP_ENABLED=true``) wraps ``FunctionInvocationLayer.get_response`` so that each LLM round-trip inside the ReAct loop emits one ``react step`` span via @@ -56,6 +59,4 @@ Env Default Description ``ARMS_MAF_INSTRUMENTATION_ENABLED`` ``true`` Master switch; ``false`` disables instrumentation. ``ARMS_MAF_SENSITIVE_DATA_ENABLED`` ``false`` Capture inputs/outputs (linked to MAF's sensitive-data option). ``ARMS_MAF_REACT_STEP_ENABLED`` ``false`` Emit ``react step`` spans for non-streaming ReAct tool loops. -``ARMS_MAF_METRICS_ENABLED`` ``true`` Aggregate ARMS GenAI gauges in-process. -``ARMS_MAF_SLOW_THRESHOLD_MS`` ``1000`` Slow-call threshold in ms. ====================================== ========== ============================================================== diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/__init__.py b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/__init__.py index 15930c667..13053ae14 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/__init__.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/__init__.py @@ -18,7 +18,8 @@ with ``force=True`` so a sticky user-disable does not block us), bridges MAF's native span helpers through ``opentelemetry-util-genai`` finish helpers, and registers :class:`~.span_processor.MAFSemanticProcessor` for workflow/MCP -normalization plus metrics aggregation. +normalization. Microsoft Agent Framework's native counter and histogram +instruments remain the metric source. The optional ReAct step patch (``ARMS_MAF_REACT_STEP_ENABLED=true``) wraps the ``FunctionInvocationLayer.get_response`` ReAct loop with @@ -42,9 +43,7 @@ from opentelemetry.trace import get_tracer_provider from .config import ( - get_slow_threshold_ms, is_instrumentation_enabled, - is_metrics_enabled, is_react_step_enabled, is_sensitive_data_enabled, ) @@ -116,20 +115,19 @@ def _instrument(self, **kwargs: Any) -> None: # invocation finish helpers. This keeps MAF's span lifetime and # streaming cleanup behavior, but writes AGENT/LLM/TOOL semantic # attributes before span.end() creates the exporter snapshot. - apply_util_genai_bridge() + apply_util_genai_bridge( + tracer_provider=tracer_provider, + meter_provider=meter_provider, + ) # 3) Register the semantic SpanProcessor. MAF uses the standard OTel # TracerProvider (it does not have its own multi-processor), so # ``add_span_processor`` is the right hook. - processor = MAFSemanticProcessor( - meter_provider=meter_provider, - slow_threshold_ms=get_slow_threshold_ms(), - metrics_enabled=is_metrics_enabled(default=True), - capture_sensitive_data=sensitive, - ) + processor = MAFSemanticProcessor(capture_sensitive_data=sensitive) try: tracer_provider.add_span_processor(processor) except Exception as exc: + revert_util_genai_bridge() logger.warning("add_span_processor failed: %s", exc) raise diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/span_processor.py b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/span_processor.py index 9b6e10bdd..bab4a9a7b 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/span_processor.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/span_processor.py @@ -29,9 +29,8 @@ 3. Backfills ``gen_ai.response.time_to_first_token`` from the first streaming chunk event timestamp. 4. Normalizes ``gen_ai.provider.name`` (``azure_openai`` → ``openai``). -5. Aggregates 6 ARMS gauges (``genai_calls_count`` / ``genai_calls_duration_seconds`` - / ``genai_calls_error_count`` / ``genai_calls_slow_count`` / ``genai_llm_first_token_seconds`` - / ``genai_llm_usage_tokens``) in-process, exposed via observable gauges. +5. Leaves metric emission to Microsoft Agent Framework's native counter and + histogram instruments instead of re-exporting process-cumulative gauges. Truncation / JSON serialization are reused from ``opentelemetry.util.genai.utils`` (``gen_ai_json_dumps``) — aligned with the pattern at @@ -46,31 +45,24 @@ import logging import threading import time -from collections import defaultdict from typing import Any, Dict, Mapping, Optional, Tuple from opentelemetry.context import Context -from opentelemetry.metrics import ObservableGauge, get_meter from opentelemetry.sdk.trace import ( SpanProcessor, TracerProvider, # noqa: F401 (typing hint) ) from opentelemetry.trace import Span as OtelSpan -from opentelemetry.trace import SpanKind, StatusCode +from opentelemetry.trace import SpanKind from opentelemetry.trace.span import TraceState # noqa: F401 from opentelemetry.util.genai.utils import gen_ai_json_dumps from .semantic_conventions import ( - ERROR_TYPE, GEN_AI_OPERATION_NAME, GEN_AI_PROVIDER_NAME, - GEN_AI_REQUEST_MODEL, GEN_AI_RESPONSE_FINISH_REASONS, - GEN_AI_RESPONSE_MODEL, GEN_AI_RESPONSE_TTFT, GEN_AI_SPAN_KIND, - GEN_AI_USAGE_INPUT_TOKENS, - GEN_AI_USAGE_OUTPUT_TOKENS, MAF_ATTR_RENAME_MAP, MAF_LIVE_SPAN_MARKER, MAF_PROVIDER_NAME, @@ -820,49 +812,6 @@ def _is_exception_event(event: Any) -> bool: ) -# ---------- Metrics aggregation ---------- - - -class _Counters: - """In-process counters keyed by ``(model, span_kind[, usage_type])``. - - Designed to be cheap to update in ``on_end`` and read out by observable - gauge callbacks. Single-process only — multi-process deployments need - per-process reporters (documented in README). - """ - - __slots__ = ( - "calls_count", - "calls_duration_ns_sum", - "calls_error_count", - "calls_slow_count", - "llm_first_token_ns_sum", - "llm_first_token_count", - "llm_usage_input_tokens", - "llm_usage_output_tokens", - ) - - def __init__(self) -> None: - self.calls_count: Dict[Tuple[str, str], int] = defaultdict(int) - self.calls_duration_ns_sum: Dict[Tuple[str, str], int] = defaultdict( - int - ) - self.calls_error_count: Dict[Tuple[str, str], int] = defaultdict(int) - self.calls_slow_count: Dict[Tuple[str, str], int] = defaultdict(int) - self.llm_first_token_ns_sum: Dict[Tuple[str, str], int] = defaultdict( - int - ) - self.llm_first_token_count: Dict[Tuple[str, str], int] = defaultdict( - int - ) - self.llm_usage_input_tokens: Dict[Tuple[str, str], int] = defaultdict( - int - ) - self.llm_usage_output_tokens: Dict[Tuple[str, str], int] = defaultdict( - int - ) - - class MAFSemanticProcessor(SpanProcessor): """SpanProcessor that injects ARMS GenAI semantic conventions into MAF spans.""" @@ -873,187 +822,13 @@ def __init__( metrics_enabled: bool = True, capture_sensitive_data: bool = False, ) -> None: + # Deprecated compatibility parameters. MAF emits native metrics; + # retain the old constructor surface for direct processor users. + _ = meter_provider, slow_threshold_ms, metrics_enabled self._live_spans: Dict[str, OtelSpan] = {} self._span_parents: Dict[str, Optional[str]] = {} self._live_span_lock = threading.Lock() - self._slow_threshold_ns = int(slow_threshold_ms) * 1_000_000 self._capture_sensitive = capture_sensitive_data - self._counters = _Counters() - self._counter_lock = threading.Lock() - self._meter = None - self._gauges: list[ObservableGauge] = [] - self._metrics_enabled = metrics_enabled - if metrics_enabled: - try: - self._init_metrics(meter_provider) - except Exception as exc: # pragma: no cover - defensive - logger.warning("MAF metrics disabled: %s", exc) - self._metrics_enabled = False - - # ---- metrics ---- - def _init_metrics(self, meter_provider: Any) -> None: - self._meter = get_meter( - "opentelemetry.instrumentation.microsoft_agent_framework", - meter_provider=meter_provider, - ) - c = self._counters - - def _calls_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), count in c.calls_count.items(): - observations.append( - _obs( - count, - { - "modelName": model or "unknown", - "spanKind": kind, - }, - ) - ) - yield from observations - - def _duration_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), total in c.calls_duration_ns_sum.items(): - count = max(c.calls_count.get((model, kind), 0), 1) - observations.append( - _obs( - total / count / 1e9, - { - "modelName": model or "unknown", - "spanKind": kind, - }, - ) - ) - yield from observations - - def _error_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), count in c.calls_error_count.items(): - observations.append( - _obs( - count, - { - "modelName": model or "unknown", - "spanKind": kind, - }, - ) - ) - yield from observations - - def _slow_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), count in c.calls_slow_count.items(): - observations.append( - _obs( - count, - { - "modelName": model or "unknown", - "spanKind": kind, - }, - ) - ) - yield from observations - - def _ttft_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), total in c.llm_first_token_ns_sum.items(): - count = max( - c.llm_first_token_count.get((model, kind), 0), 1 - ) - observations.append( - _obs( - total / count / 1e9, - { - "modelName": model or "unknown", - "spanKind": kind, - }, - ) - ) - yield from observations - - def _tokens_input_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), total in c.llm_usage_input_tokens.items(): - observations.append( - _obs( - total, - { - "modelName": model or "unknown", - "spanKind": kind, - "usageType": "input", - }, - ) - ) - yield from observations - - def _tokens_output_cb(options): - observations = [] - with self._counter_lock: - for (model, kind), total in c.llm_usage_output_tokens.items(): - observations.append( - _obs( - total, - { - "modelName": model or "unknown", - "spanKind": kind, - "usageType": "output", - }, - ) - ) - yield from observations - - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_calls_count", - callbacks=[_calls_cb], - description="Number of GenAI calls.", - ) - ) - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_calls_duration_seconds", - callbacks=[_duration_cb], - description="Average GenAI call duration in seconds.", - unit="s", - ) - ) - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_calls_error_count", - callbacks=[_error_cb], - description="Number of failed GenAI calls.", - ) - ) - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_calls_slow_count", - callbacks=[_slow_cb], - description="Number of slow GenAI calls (threshold-configurable).", - ) - ) - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_llm_first_token_seconds", - callbacks=[_ttft_cb], - description="Average LLM time-to-first-token in seconds.", - unit="s", - ) - ) - self._gauges.append( - self._meter.create_observable_gauge( - name="genai_llm_usage_tokens", - callbacks=[_tokens_input_cb, _tokens_output_cb], - description="LLM token usage (input / output).", - unit="{token}", - ) - ) # ---- SpanProcessor interface ---- def on_start( @@ -1137,10 +912,6 @@ def on_end(self, span: Any) -> None: # 9) error.type already set by MAF via capture_exception; nothing to do. - # 10) Aggregate metrics - if self._metrics_enabled: - self._aggregate_metrics(span, span_kind, op_name) - except Exception as exc: # pragma: no cover - defensive logger.warning("MAFSemanticProcessor.on_end failed: %s", exc) finally: @@ -1258,60 +1029,6 @@ def _apply_semantic_attributes( return span_kind, op_name - def _aggregate_metrics( - self, readable: Any, span_kind: str, op_name: str - ) -> None: - try: - model = _attr_value(readable, GEN_AI_REQUEST_MODEL) - if not model: - model = _attr_value(readable, GEN_AI_RESPONSE_MODEL) - model = model if isinstance(model, str) else "unknown" - key = (model, span_kind) - with self._counter_lock: - self._counters.calls_count[key] += 1 - - start = getattr(readable, "start_time", None) - end = getattr(readable, "end_time", None) - if start is not None and end is not None: - try: - duration_ns = int(end - start) - self._counters.calls_duration_ns_sum[key] += ( - duration_ns - ) - if duration_ns >= self._slow_threshold_ns: - self._counters.calls_slow_count[key] += 1 - except (TypeError, ValueError): - pass - - current_status = getattr(readable, "status", None) - status_code = getattr(current_status, "status_code", None) - if status_code == StatusCode.ERROR: - self._counters.calls_error_count[key] += 1 - elif _attr_value(readable, ERROR_TYPE): - self._counters.calls_error_count[key] += 1 - - if span_kind == GenAISpanKind.LLM: - ttft = _attr_value(readable, GEN_AI_RESPONSE_TTFT) - if isinstance(ttft, (int, float)) and ttft > 0: - self._counters.llm_first_token_ns_sum[key] += int(ttft) - self._counters.llm_first_token_count[key] += 1 - input_tokens = _attr_value( - readable, GEN_AI_USAGE_INPUT_TOKENS - ) - if isinstance(input_tokens, (int, float)): - self._counters.llm_usage_input_tokens[key] += int( - input_tokens - ) - output_tokens = _attr_value( - readable, GEN_AI_USAGE_OUTPUT_TOKENS - ) - if isinstance(output_tokens, (int, float)): - self._counters.llm_usage_output_tokens[key] += int( - output_tokens - ) - except Exception as exc: # pragma: no cover - defensive - logger.debug("MAF metrics aggregation failed: %s", exc) - def shutdown(self) -> None: with self._live_span_lock: self._live_spans.clear() @@ -1342,11 +1059,4 @@ def _sweep_stale_live_spans( self._span_parents.pop(key, None) -def _obs(value: float, attrs: Dict[str, str]): - """Build an Observation compatible with OTel callbacks.""" - from opentelemetry.metrics import Observation - - return Observation(value, attrs) - - __all__ = ["MAFSemanticProcessor"] diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/util_genai_bridge.py b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/util_genai_bridge.py index c5c08e287..b8f757b9f 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/util_genai_bridge.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/src/opentelemetry/instrumentation/microsoft_agent_framework/util_genai_bridge.py @@ -33,10 +33,16 @@ from time import perf_counter, time_ns from typing import Any, AsyncGenerator, Callable, Generator, Mapping, Optional +from opentelemetry import context as otel_context from opentelemetry import trace as otel_trace from opentelemetry.trace import Span as OtelSpan from opentelemetry.trace import SpanKind +try: + from aliyun.sdk.extension.arms.semconv import _SUPPRESS_LLM_SDK_KEY +except ImportError: + _SUPPRESS_LLM_SDK_KEY = None + try: from opentelemetry.util.genai.extended_span_utils import ( _apply_embedding_finish_attributes, @@ -104,8 +110,11 @@ _original_get_function_span: Any = None _original_create_mcp_client_span: Any = None _original_tools_get_function_span: Any = None +_original_tools_get_meter: Any = None _original_mcp_create_mcp_client_span: Any = None _original_agent_trace_invocation: Any = None +_original_get_tracer: Any = None +_original_get_meter: Any = None _legacy_response_stream_originals: dict[str, Any] = {} _FINALIZED_ATTR = "_loongsuite_util_genai_finalized" @@ -124,7 +133,11 @@ def _mark_maf_live_span(span: OtelSpan | None) -> None: pass -def apply_util_genai_bridge() -> None: +def apply_util_genai_bridge( + *, + tracer_provider: Any = None, + meter_provider: Any = None, +) -> None: """Patch MAF span helpers so GenAI spans are finalized by util-genai.""" global _applied global _original_create_mcp_client_span @@ -134,7 +147,10 @@ def apply_util_genai_bridge() -> None: global _original_get_span global _original_start_streaming_span global _original_tools_get_function_span + global _original_tools_get_meter global _original_agent_trace_invocation + global _original_get_tracer + global _original_get_meter if _applied: return @@ -152,6 +168,8 @@ def apply_util_genai_bridge() -> None: logger.warning("MAF util-genai bridge skipped: %s", exc) return + _original_get_tracer = getattr(observability, "get_tracer", None) + _original_get_meter = getattr(observability, "get_meter", None) _original_get_span = getattr(observability, "_get_span", None) _original_start_streaming_span = getattr( observability, "_start_streaming_span", None @@ -189,7 +207,21 @@ def apply_util_genai_bridge() -> None: if _original_create_mcp_client_span is not None else None ) + wrapped_get_tracer = ( + _wrap_provider_get_tracer(_original_get_tracer, tracer_provider) + if tracer_provider is not None and _original_get_tracer is not None + else None + ) + wrapped_get_meter = ( + _wrap_provider_get_meter(_original_get_meter, meter_provider) + if meter_provider is not None and _original_get_meter is not None + else None + ) + if wrapped_get_tracer is not None: + observability.get_tracer = wrapped_get_tracer # type: ignore[attr-defined] + if wrapped_get_meter is not None: + observability.get_meter = wrapped_get_meter # type: ignore[attr-defined] if wrapped_get_span is not None: observability._get_span = wrapped_get_span # type: ignore[attr-defined] if _original_start_streaming_span is not None: @@ -219,10 +251,17 @@ def apply_util_genai_bridge() -> None: _original_tools_get_function_span = getattr( tools_mod, "get_function_span", None ) + _original_tools_get_meter = getattr(tools_mod, "get_meter", None) if wrapped_get_function_span is not None: tools_mod.get_function_span = wrapped_get_function_span # type: ignore[attr-defined] + if ( + wrapped_get_meter is not None + and _original_tools_get_meter is not None + ): + tools_mod.get_meter = wrapped_get_meter # type: ignore[attr-defined] except ImportError: _original_tools_get_function_span = None + _original_tools_get_meter = None try: import agent_framework._mcp as mcp_mod # type: ignore @@ -249,13 +288,20 @@ def revert_util_genai_bridge() -> None: global _original_get_span global _original_start_streaming_span global _original_tools_get_function_span + global _original_tools_get_meter global _original_agent_trace_invocation + global _original_get_tracer + global _original_get_meter if not _applied: return try: import agent_framework.observability as observability # type: ignore + if _original_get_tracer is not None: + observability.get_tracer = _original_get_tracer # type: ignore[attr-defined] + if _original_get_meter is not None: + observability.get_meter = _original_get_meter # type: ignore[attr-defined] if _original_get_span is not None: observability._get_span = _original_get_span # type: ignore[attr-defined] if _original_start_streaming_span is not None: @@ -285,6 +331,8 @@ def revert_util_genai_bridge() -> None: if _original_tools_get_function_span is not None: tools_mod.get_function_span = _original_tools_get_function_span # type: ignore[attr-defined] + if _original_tools_get_meter is not None: + tools_mod.get_meter = _original_tools_get_meter # type: ignore[attr-defined] except ImportError: pass try: @@ -304,8 +352,11 @@ def revert_util_genai_bridge() -> None: _original_get_function_span = None _original_create_mcp_client_span = None _original_tools_get_function_span = None + _original_tools_get_meter = None _original_mcp_create_mcp_client_span = None _original_agent_trace_invocation = None + _original_get_tracer = None + _original_get_meter = None _restore_legacy_response_stream() @@ -602,6 +653,29 @@ def _activate_live_span(span: OtelSpan) -> Any: ) +@contextlib.contextmanager +def _activate_non_streaming_span( + span: OtelSpan, + attributes: Mapping[Any, Any], +) -> Generator[None, Any, Any]: + """Keep a MAF-owned non-streaming span current for the wrapped call.""" + context = otel_trace.set_span_in_context(span, otel_context.get_current()) + if ( + _SUPPRESS_LLM_SDK_KEY is not None + and _mapping_value(attributes, GEN_AI_SPAN_KIND) == GenAISpanKind.LLM + ): + context = otel_context.set_value( + _SUPPRESS_LLM_SDK_KEY, + True, + context, + ) + token = otel_context.attach(context) + try: + yield + finally: + otel_context.detach(token) + + def _wrap_get_span(original: Callable[..., Any]) -> Callable[..., Any]: @contextlib.contextmanager def _get_span( @@ -614,10 +688,11 @@ def _get_span( ) with span_cm as span: _mark_maf_live_span(span) - try: - yield span - finally: - _finalize_with_util_genai(span) + with _activate_non_streaming_span(span, bridge_attrs): + try: + yield span + finally: + _finalize_with_util_genai(span) return _get_span @@ -749,6 +824,60 @@ def _create_mcp_client_span( return _create_mcp_client_span +def _wrap_provider_get_tracer( + original: Callable[..., Any], tracer_provider: Any +) -> Callable[..., Any]: + """Route MAF's global-style tracer accessor to an explicit provider.""" + signature = inspect.signature(original) + + def _get_tracer(*args: Any, **kwargs: Any) -> Any: + bound = signature.bind_partial(*args, **kwargs) + bound.apply_defaults() + name = bound.arguments.get( + "instrumenting_module_name", "agent_framework" + ) + version = bound.arguments.get("instrumenting_library_version") + schema_url = bound.arguments.get("schema_url") + attributes = bound.arguments.get("attributes") + try: + return tracer_provider.get_tracer( + name, + version, + schema_url, + attributes, + ) + except TypeError: + return tracer_provider.get_tracer(name, version, schema_url) + + return _get_tracer + + +def _wrap_provider_get_meter( + original: Callable[..., Any], meter_provider: Any +) -> Callable[..., Any]: + """Route MAF's native metric instruments to an explicit provider.""" + signature = inspect.signature(original) + + def _get_meter(*args: Any, **kwargs: Any) -> Any: + bound = signature.bind_partial(*args, **kwargs) + bound.apply_defaults() + name = bound.arguments.get("name", "agent_framework") + version = bound.arguments.get("version") + schema_url = bound.arguments.get("schema_url") + attributes = bound.arguments.get("attributes") + try: + return meter_provider.get_meter( + name, + version, + schema_url, + attributes, + ) + except TypeError: + return meter_provider.get_meter(name, version, schema_url) + + return _get_meter + + def _current_span_context( original: Callable[..., Any], attributes: dict[str, Any], diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_processor.py b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_processor.py index 1b1822f12..c4bc2941c 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_processor.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_processor.py @@ -21,13 +21,16 @@ - Normalizes ``gen_ai.provider.name``. - Backfills ``gen_ai.response.time_to_first_token`` from streaming events. - Leaves successful spans with the SDK's default status, preserves ``ERROR`` on failure. -- Aggregates metrics counters. +- Leaves metrics to Microsoft Agent Framework's native instruments. """ from __future__ import annotations import json import time +from unittest.mock import MagicMock + +import pytest from opentelemetry.instrumentation.microsoft_agent_framework.semantic_conventions import ( GEN_AI_OPERATION_NAME, @@ -50,18 +53,14 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) -from opentelemetry.trace import SpanKind, Status, StatusCode +from opentelemetry.trace import SpanKind, StatusCode def _setup(): """Return ``(tracer_provider, tracer, exporter)`` with the MAF processor.""" tp = TracerProvider() exporter = InMemorySpanExporter() - processor = MAFSemanticProcessor( - meter_provider=None, - metrics_enabled=False, - capture_sensitive_data=False, - ) + processor = MAFSemanticProcessor(capture_sensitive_data=False) tp.add_span_processor(processor) tp.add_span_processor(SimpleSpanProcessor(exporter)) tracer = tp.get_tracer("test") @@ -72,11 +71,7 @@ def _setup_exporter_before_processor(): """Simulate exporters registered before the MAF semantic processor.""" tp = TracerProvider() exporter = InMemorySpanExporter() - processor = MAFSemanticProcessor( - meter_provider=None, - metrics_enabled=False, - capture_sensitive_data=False, - ) + processor = MAFSemanticProcessor(capture_sensitive_data=False) tp.add_span_processor(SimpleSpanProcessor(exporter)) tp.add_span_processor(processor) tracer = tp.get_tracer("test") @@ -88,6 +83,19 @@ def _flush(exporter): return exporter.get_finished_spans() +def test_processor_does_not_register_cumulative_gauges(): + meter_provider = MagicMock() + processor = MAFSemanticProcessor( + meter_provider=meter_provider, + slow_threshold_ms=250, + metrics_enabled=True, + capture_sensitive_data=True, + ) + + assert processor._capture_sensitive is True + meter_provider.get_meter.assert_not_called() + + def _mark_maf_span(span): setattr(span, MAF_LIVE_SPAN_MARKER, MAF_PROVIDER_NAME) @@ -462,66 +470,6 @@ def test_output_messages_finish_reason_tool_call_is_normalized(): assert messages[0]["finish_reason"] == "tool_calls" -def _setup_with_metrics(): - """Like ``_setup`` but with metrics aggregation enabled.""" - tp = TracerProvider() - exporter = InMemorySpanExporter() - processor = MAFSemanticProcessor( - meter_provider=None, metrics_enabled=True, capture_sensitive_data=False - ) - tp.add_span_processor(processor) - tp.add_span_processor(SimpleSpanProcessor(exporter)) - tracer = tp.get_tracer("test") - return tp, tracer, exporter, processor - - -def test_error_span_keeps_error_status_and_increments_error_counter(): - tp, tracer, exporter, processor = _setup_with_metrics() - try: - with tracer.start_as_current_span("chat gpt-4o") as span: - _mark_maf_span(span) - span.set_attribute(GEN_AI_OPERATION_NAME, GenAIOperation.CHAT) - span.set_attribute(GEN_AI_PROVIDER_NAME, "openai") - span.set_status(Status(StatusCode.ERROR, "boom")) - span.set_attribute("error.type", "RateLimitError") - raise RuntimeError("boom") - except RuntimeError: - pass - spans = _flush(exporter) - assert spans[0].status.status_code == StatusCode.ERROR - # Metric error counter incremented (model not set -> "unknown"). - found = False - for (m, k), count in processor._counters.calls_error_count.items(): - if k == GenAISpanKind.LLM and count == 1: - found = True - break - assert found, "error counter not incremented" - - -def test_metrics_counters_incremented_on_llm_span(): - tp, tracer, exporter, processor = _setup_with_metrics() - with tracer.start_as_current_span("chat gpt-4o") as span: - _mark_maf_span(span) - span.set_attribute(GEN_AI_OPERATION_NAME, GenAIOperation.CHAT) - span.set_attribute("gen_ai.request.model", "gpt-4o") - span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, 5) - span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, 7) - _ = _flush(exporter) - assert processor._counters.calls_count[("gpt-4o", GenAISpanKind.LLM)] == 1 - assert ( - processor._counters.llm_usage_input_tokens[ - ("gpt-4o", GenAISpanKind.LLM) - ] - == 5 - ) - assert ( - processor._counters.llm_usage_output_tokens[ - ("gpt-4o", GenAISpanKind.LLM) - ] - == 7 - ) - - def test_react_step_span_classification(): tp, tracer, exporter, _ = _setup() with tracer.start_as_current_span("react step") as span: @@ -568,19 +516,47 @@ def test_uninstrument_removes_registered_processor_from_provider(): assert procs is not None and processor not in procs +def test_instrument_rolls_back_bridge_when_processor_registration_fails( + monkeypatch, +): + import opentelemetry.instrumentation.microsoft_agent_framework as maf + + tracer_provider = MagicMock() + tracer_provider.add_span_processor.side_effect = RuntimeError( + "registration failed" + ) + apply_bridge = MagicMock() + revert_bridge = MagicMock() + monkeypatch.setattr(maf, "apply_util_genai_bridge", apply_bridge) + monkeypatch.setattr(maf, "revert_util_genai_bridge", revert_bridge) + + inst = maf.MicrosoftAgentFrameworkInstrumentor() + with pytest.raises(RuntimeError, match="registration failed"): + inst._instrument( + tracer_provider=tracer_provider, + react_step_enabled=False, + ) + + apply_bridge.assert_called_once_with( + tracer_provider=tracer_provider, + meter_provider=None, + ) + revert_bridge.assert_called_once_with() + assert inst._processor is None + + def test_non_maf_span_is_left_untouched(): - tp, tracer, exporter, processor = _setup_with_metrics() + tp, tracer, exporter, _ = _setup() with tracer.start_as_current_span("http request") as span: span.set_attribute("http.method", "GET") spans = _flush(exporter) s = spans[0] assert GEN_AI_SPAN_KIND not in s.attributes assert GEN_AI_OPERATION_NAME not in s.attributes - assert not processor._counters.calls_count def test_autogen_span_with_overlapping_agent_operation_is_left_untouched(): - tp, tracer, exporter, processor = _setup_with_metrics() + tp, tracer, exporter, _ = _setup() with tracer.start_as_current_span("invoke_agent assistant") as span: span.set_attribute(GEN_AI_OPERATION_NAME, GenAIOperation.INVOKE_AGENT) span.set_attribute(GEN_AI_PROVIDER_NAME, "autogen") @@ -588,11 +564,10 @@ def test_autogen_span_with_overlapping_agent_operation_is_left_untouched(): s = spans[0] assert s.attributes.get(GEN_AI_PROVIDER_NAME) == "autogen" assert GEN_AI_SPAN_KIND not in s.attributes - assert not processor._counters.calls_count def test_autogen_llm_span_with_private_marker_is_left_untouched(): - tp, tracer, exporter, processor = _setup_with_metrics() + tp, tracer, exporter, _ = _setup() with tracer.start_as_current_span("chat gpt-4o") as span: span.set_attribute(GEN_AI_OPERATION_NAME, GenAIOperation.CHAT) span.set_attribute(GEN_AI_PROVIDER_NAME, "openai") @@ -603,11 +578,10 @@ def test_autogen_llm_span_with_private_marker_is_left_untouched(): assert GEN_AI_SPAN_KIND not in s.attributes assert "_loongsuite_autogen_framework" not in s.attributes assert s.kind == SpanKind.INTERNAL - assert not processor._counters.calls_count def test_foreign_agent_spans_are_left_untouched(): - tp, tracer, exporter, processor = _setup_with_metrics() + tp, tracer, exporter, _ = _setup() for provider_name in ("langchain", "agentscope"): with tracer.start_as_current_span( f"invoke_agent {provider_name}-agent" @@ -622,7 +596,6 @@ def test_foreign_agent_spans_are_left_untouched(): assert len(spans) == 2 for span in spans: assert GEN_AI_SPAN_KIND not in span.attributes - assert not processor._counters.calls_count def test_dict_attribute_is_serialized_via_gen_ai_json_dumps(): @@ -806,11 +779,7 @@ def test_force_flush_sweeps_stale_live_spans(): class _StartedSpan: start_time = 0 - processor = MAFSemanticProcessor( - meter_provider=None, - metrics_enabled=False, - capture_sensitive_data=False, - ) + processor = MAFSemanticProcessor(capture_sensitive_data=False) processor._live_spans["deadbeef"] = _StartedSpan() processor._span_parents["deadbeef"] = None diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_util_genai_bridge.py b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_util_genai_bridge.py index 9891d4e96..ceaffa98a 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_util_genai_bridge.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-microsoft-agent-framework/tests/test_util_genai_bridge.py @@ -26,7 +26,9 @@ import json import sys import types +from unittest.mock import MagicMock +from opentelemetry import context as otel_context from opentelemetry import trace from opentelemetry.instrumentation.microsoft_agent_framework import ( util_genai_bridge, @@ -54,13 +56,12 @@ def _install_fake_observability(monkeypatch): tp = TracerProvider() exporter = InMemorySpanExporter() tp.add_span_processor(SimpleSpanProcessor(exporter)) - tracer = tp.get_tracer("fake-maf") @contextlib.contextmanager def _get_span(attributes, span_name_attribute): operation = attributes.get(GEN_AI_OPERATION_NAME, "operation") span_name = attributes.get(span_name_attribute, "unknown") - span = tracer.start_span(f"{operation} {span_name}") + span = obs_mod.get_tracer().start_span(f"{operation} {span_name}") span.set_attributes(attributes) with trace.use_span( span, @@ -73,7 +74,7 @@ def _get_span(attributes, span_name_attribute): def _start_streaming_span(attributes, span_name_attribute): operation = attributes.get(GEN_AI_OPERATION_NAME, "operation") span_name = attributes.get(span_name_attribute, "unknown") - span = tracer.start_span(f"{operation} {span_name}") + span = obs_mod.get_tracer().start_span(f"{operation} {span_name}") span.set_attributes(attributes) return span @@ -84,7 +85,7 @@ def _activate_span(span): @contextlib.contextmanager def get_function_span(attributes): - span = tracer.start_span( + span = obs_mod.get_tracer().start_span( f"{attributes[GEN_AI_OPERATION_NAME]} {attributes['gen_ai.tool.name']}" ) span.set_attributes(attributes) @@ -99,25 +100,50 @@ def get_function_span(attributes): @contextlib.contextmanager def create_mcp_client_span(method_name, target=None, attributes=None): span_name = f"{method_name} {target}" if target else method_name - span = tracer.start_span(span_name, kind=trace.SpanKind.CLIENT) + span = obs_mod.get_tracer().start_span( + span_name, kind=trace.SpanKind.CLIENT + ) span.set_attribute("mcp.method.name", method_name) if attributes: span.set_attributes(attributes) with trace.use_span(span, end_on_exit=True) as current_span: yield current_span + def get_tracer( + instrumenting_module_name="agent_framework", + instrumenting_library_version=None, + schema_url=None, + attributes=None, + ): + return tp.get_tracer( + instrumenting_module_name, + instrumenting_library_version, + schema_url, + attributes, + ) + + def get_meter( + name="agent_framework", + version=None, + schema_url=None, + attributes=None, + ): + return name, version, schema_url, attributes + obs_mod = types.ModuleType("agent_framework.observability") obs_mod._get_span = _get_span obs_mod._start_streaming_span = _start_streaming_span obs_mod._activate_span = _activate_span obs_mod.get_function_span = get_function_span obs_mod.create_mcp_client_span = create_mcp_client_span - obs_mod.get_tracer = lambda: tracer + obs_mod.get_tracer = get_tracer + obs_mod.get_meter = get_meter af_mod = types.ModuleType("agent_framework") af_mod.observability = obs_mod tools_mod = types.ModuleType("agent_framework._tools") tools_mod.get_function_span = get_function_span + tools_mod.get_meter = get_meter mcp_mod = types.ModuleType("agent_framework._mcp") mcp_mod.create_mcp_client_span = create_mcp_client_span monkeypatch.setitem(sys.modules, "agent_framework", af_mod) @@ -128,6 +154,80 @@ def create_mcp_client_span(method_name, target=None, attributes=None): return obs_mod, exporter +def test_non_streaming_span_stays_current_and_restores_context(monkeypatch): + obs_mod, exporter = _install_fake_observability(monkeypatch) + suppress_key = otel_context.create_key("test-suppress-llm-sdk") + monkeypatch.setattr( + util_genai_bridge, + "_SUPPRESS_LLM_SDK_KEY", + suppress_key, + ) + + util_genai_bridge.apply_util_genai_bridge() + tracer = obs_mod.get_tracer("test-parent") + try: + with tracer.start_as_current_span("entry") as outer: + with obs_mod._get_span( + { + GEN_AI_OPERATION_NAME: GenAIOperation.CHAT, + GEN_AI_REQUEST_MODEL: "qwen-plus", + }, + GEN_AI_REQUEST_MODEL, + ) as llm_span: + assert trace.get_current_span() is llm_span + assert otel_context.get_value(suppress_key) is True + assert trace.get_current_span() is outer + assert otel_context.get_value(suppress_key) is None + finally: + util_genai_bridge.revert_util_genai_bridge() + + spans = {span.name: span for span in exporter.get_finished_spans()} + assert ( + spans["chat qwen-plus"].parent.span_id + == spans["entry"].context.span_id + ) + + +def test_explicit_providers_route_maf_spans_and_native_metrics(monkeypatch): + obs_mod, default_exporter = _install_fake_observability(monkeypatch) + original_get_tracer = obs_mod.get_tracer + original_get_meter = obs_mod.get_meter + tools_mod = sys.modules["agent_framework._tools"] + original_tools_get_meter = tools_mod.get_meter + + explicit_provider = TracerProvider() + explicit_exporter = InMemorySpanExporter() + explicit_provider.add_span_processor( + SimpleSpanProcessor(explicit_exporter) + ) + meter_provider = MagicMock() + + util_genai_bridge.apply_util_genai_bridge( + tracer_provider=explicit_provider, + meter_provider=meter_provider, + ) + try: + with obs_mod._get_span( + { + GEN_AI_OPERATION_NAME: GenAIOperation.CHAT, + GEN_AI_REQUEST_MODEL: "qwen-plus", + }, + GEN_AI_REQUEST_MODEL, + ): + pass + obs_mod.get_meter("agent_framework.metrics") + tools_mod.get_meter("agent_framework.tools") + finally: + util_genai_bridge.revert_util_genai_bridge() + + assert not default_exporter.get_finished_spans() + assert len(explicit_exporter.get_finished_spans()) == 1 + assert meter_provider.get_meter.call_count == 2 + assert obs_mod.get_tracer is original_get_tracer + assert obs_mod.get_meter is original_get_meter + assert tools_mod.get_meter is original_tools_get_meter + + def test_llm_get_span_is_finalized_by_util_genai_before_export(monkeypatch): obs_mod, exporter = _install_fake_observability(monkeypatch) util_genai_bridge.apply_util_genai_bridge()