diff --git a/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md b/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md index ba3a1ab6b47b..d6d3e9392d62 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md +++ b/sdk/agentserver/azure-ai-agentserver-core/CHANGELOG.md @@ -1,5 +1,12 @@ # Release History +## 2.0.0b8 (Unreleased) + +### Other Changes + +- Added local OTLP HTTP/protobuf and gRPC protocol regression tests covering traces, metrics, and logs without requiring external collector infrastructure. +- Added Agent Server-managed OTLP/gRPC export when `OTEL_EXPORTER_OTLP_PROTOCOL=grpc` is configured. + ## 2.0.0b7 (2026-06-28) ### Features Added diff --git a/sdk/agentserver/azure-ai-agentserver-core/README.md b/sdk/agentserver/azure-ai-agentserver-core/README.md index 2a9d56c42346..a994213077c8 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/README.md +++ b/sdk/agentserver/azure-ai-agentserver-core/README.md @@ -143,6 +143,16 @@ export APPLICATIONINSIGHTS_CONNECTION_STRING="InstrumentationKey=..." python my_agent.py ``` +OTLP export is enabled when `OTEL_EXPORTER_OTLP_ENDPOINT` is set. HTTP/protobuf +is the default protocol; set `OTEL_EXPORTER_OTLP_PROTOCOL=grpc` to use an +OTLP/gRPC collector: + +```bash +export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317" +export OTEL_EXPORTER_OTLP_PROTOCOL="grpc" +python my_agent.py +``` + ## Troubleshooting ### Logging diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_tracing.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_tracing.py index b0ed26bbeda1..ca7325d2781c 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_tracing.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_tracing.py @@ -33,9 +33,10 @@ OpenTelemetry is a required dependency — these functions always create real spans. Azure Monitor export is optional (auto-configured by the distro). """ +from collections.abc import AsyncIterable, AsyncIterator # pylint: disable=import-error +from contextlib import contextmanager, nullcontext import logging import os -from collections.abc import AsyncIterable, AsyncIterator # pylint: disable=import-error from typing import Any, Optional, Union from opentelemetry import baggage as _otel_baggage, context as _otel_context, trace @@ -76,6 +77,21 @@ logger = logging.getLogger("azure.ai.agentserver") +_OTLP_HTTP_PROTOBUF = "http/protobuf" +_OTLP_GRPC = "grpc" +_OTLP_ENDPOINT = "OTEL_EXPORTER_OTLP_ENDPOINT" +_OTLP_PROTOCOL = "OTEL_EXPORTER_OTLP_PROTOCOL" +_OTLP_TRACES_ENDPOINT = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT" +_OTLP_METRICS_ENDPOINT = "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" +_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT" +_OTLP_ENV_VARS = ( + _OTLP_ENDPOINT, + _OTLP_PROTOCOL, + _OTLP_TRACES_ENDPOINT, + _OTLP_METRICS_ENDPOINT, + _OTLP_LOGS_ENDPOINT, +) + # ====================================================================== # Public API: observability setup @@ -180,7 +196,7 @@ def _configure_tracing( agent_blueprint_id = _config.resolve_agent_blueprint_id() or None agent_tenant_id = _config.resolve_agent_tenant_id() or None - span_processors = [ + resolved_span_processors = [ _FoundryEnrichmentSpanProcessor( agent_name=agent_name, agent_version=agent_version, agent_id=agent_id, project_id=project_id, @@ -195,26 +211,36 @@ def _configure_tracing( session_id=_config.resolve_session_id() or None, ), ] + metric_readers: list[Any] = [] + suppress_distro_otlp = _append_grpc_otlp_components( + resolved_span_processors, + metric_readers, + log_record_processors, + ) try: - _setup_distro_export( - resource=resource, - span_processors=span_processors, - log_record_processors=log_record_processors, - connection_string=connection_string, - enable_sensitive_data=enable_sensitive_data, - ) + context = _without_otlp_env() if suppress_distro_otlp else nullcontext() + with context: + _setup_distro_export( + resource=resource, + span_processors=resolved_span_processors, + metric_readers=metric_readers, + log_record_processors=log_record_processors, + connection_string=connection_string, + enable_sensitive_data=enable_sensitive_data, + ) logger.info("Tracing configured successfully via microsoft-opentelemetry distro.") except ImportError: logger.warning("microsoft-opentelemetry is not installed — tracing export disabled.") # Still set up TracerProvider with enrichment processor so spans are created - _ensure_trace_provider(resource, span_processors) + _ensure_trace_provider(resource, resolved_span_processors) def _setup_distro_export( *, resource: Any, span_processors: list[Any], + metric_readers: list[Any], log_record_processors: list[Any], connection_string: Optional[str] = None, enable_sensitive_data: bool = False, @@ -226,6 +252,7 @@ def _setup_distro_export( :keyword resource: OTel resource describing this service. :keyword span_processors: Span processors to register. + :keyword metric_readers: Metric readers to register. :keyword log_record_processors: Log record processors to register. :keyword connection_string: Application Insights connection string. :keyword enable_sensitive_data: Enable sensitive data recording for @@ -236,6 +263,7 @@ def _setup_distro_export( kwargs: dict[str, Any] = { "resource": resource, "span_processors": span_processors, + "metric_readers": metric_readers, "log_record_processors": log_record_processors, "enable_sensitive_data": enable_sensitive_data, } @@ -259,6 +287,87 @@ def _setup_distro_export( use_microsoft_opentelemetry(**kwargs) +def _append_grpc_otlp_components( + span_processors: list[Any], + metric_readers: list[Any], + log_record_processors: list[Any], +) -> bool: + """Append SDK-managed OTLP/gRPC exporters when requested by env vars. + + The Microsoft OpenTelemetry distro currently owns the normal OTLP path but + only bundles the HTTP/protobuf exporter. Agent Server handles the gRPC + protocol here so customers only need to set OTLP environment variables. + Returns True when OTLP environment variables should be hidden from the + distro call so it does not also create HTTP/protobuf exporters. + + :param span_processors: Span processors to append gRPC trace export to. + :type span_processors: list[~typing.Any] + :param metric_readers: Metric readers to append gRPC metric export to. + :type metric_readers: list[~typing.Any] + :param log_record_processors: Log record processors to append gRPC log export to. + :type log_record_processors: list[~typing.Any] + :return: Whether OTLP environment variables should be hidden from the distro call. + :rtype: bool + """ + if not _is_otlp_enabled() or _resolve_otlp_protocol() != _OTLP_GRPC: + return False + + from opentelemetry.exporter.otlp.proto.grpc._log_exporter import ( + OTLPLogExporter as GrpcLogExporter, + ) + from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( + OTLPMetricExporter as GrpcMetricExporter, + ) + from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter as GrpcSpanExporter, + ) + from opentelemetry.sdk._logs.export import BatchLogRecordProcessor + from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader + from opentelemetry.sdk.trace.export import BatchSpanProcessor + + span_processors.append(BatchSpanProcessor(GrpcSpanExporter())) + metric_readers.append(PeriodicExportingMetricReader(GrpcMetricExporter())) + log_record_processors.append(BatchLogRecordProcessor(GrpcLogExporter())) + + return True + + +def _is_otlp_enabled() -> bool: + return any( + os.environ.get(env_var) + for env_var in ( + _OTLP_ENDPOINT, + _OTLP_TRACES_ENDPOINT, + _OTLP_METRICS_ENDPOINT, + _OTLP_LOGS_ENDPOINT, + ) + ) + + +def _resolve_otlp_protocol() -> str: + protocol = os.environ.get(_OTLP_PROTOCOL) or _OTLP_HTTP_PROTOBUF + normalized = protocol.strip().lower() + if normalized not in (_OTLP_HTTP_PROTOBUF, _OTLP_GRPC): + raise ValueError( + f"Unsupported OTLP protocol {protocol!r}. Use " + f"{_OTLP_HTTP_PROTOBUF!r} or {_OTLP_GRPC!r}." + ) + return normalized + + +@contextmanager +def _without_otlp_env() -> Any: + saved = { + env_var: os.environ.pop(env_var) + for env_var in _OTLP_ENV_VARS + if env_var in os.environ + } + try: + yield + finally: + os.environ.update(saved) + + # ====================================================================== # Public API: span operations # ====================================================================== @@ -502,21 +611,22 @@ def _on_ending(self, span: Any) -> None: if attrs is None: return try: + target = getattr(attrs, "_dict", attrs) if self.agent_name: - attrs[_ATTR_GEN_AI_AGENT_NAME] = self.agent_name + target[_ATTR_GEN_AI_AGENT_NAME] = self.agent_name if self.agent_version: - attrs[_ATTR_GEN_AI_AGENT_VERSION] = self.agent_version + target[_ATTR_GEN_AI_AGENT_VERSION] = self.agent_version if self.agent_id: - attrs[_ATTR_GEN_AI_AGENT_ID] = self.agent_id + target[_ATTR_GEN_AI_AGENT_ID] = self.agent_id if self.agent_blueprint_id: - attrs[_ATTR_GEN_AI_AGENT_BLUEPRINT_ID] = self.agent_blueprint_id + target[_ATTR_GEN_AI_AGENT_BLUEPRINT_ID] = self.agent_blueprint_id if self.agent_tenant_id: - attrs[_ATTR_GEN_AI_AGENT_TENANT_ID] = self.agent_tenant_id + target[_ATTR_GEN_AI_AGENT_TENANT_ID] = self.agent_tenant_id except Exception: # pylint: disable=broad-exception-caught logger.debug("Failed to enrich span attributes in _on_ending", exc_info=True) - def on_end(self, span: Any) -> None: # pylint: disable=unused-argument - pass + def on_end(self, span: Any) -> None: + self._on_ending(span) def shutdown(self) -> None: pass diff --git a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_version.py b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_version.py index 369f0dcc3bea..213b705c79c6 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_version.py +++ b/sdk/agentserver/azure-ai-agentserver-core/azure/ai/agentserver/core/_version.py @@ -2,4 +2,4 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # --------------------------------------------------------- -VERSION = "2.0.0b7" +VERSION = "2.0.0b8" diff --git a/sdk/agentserver/azure-ai-agentserver-core/dev_requirements.txt b/sdk/agentserver/azure-ai-agentserver-core/dev_requirements.txt index 2e6ad8b7bf67..0c1b5a8d678a 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/dev_requirements.txt +++ b/sdk/agentserver/azure-ai-agentserver-core/dev_requirements.txt @@ -4,5 +4,11 @@ pytest httpx pytest-asyncio -opentelemetry-api>=1.40.0 -opentelemetry-sdk>=1.40.0 \ No newline at end of file +starlette>=0.45.0 +hypercorn>=0.17.0 +opentelemetry-api~=1.43.0 +opentelemetry-sdk~=1.43.0 +opentelemetry-exporter-otlp-proto-grpc~=1.43.0 +microsoft-opentelemetry~=1.3.5 +grpcio +opentelemetry-proto diff --git a/sdk/agentserver/azure-ai-agentserver-core/pyproject.toml b/sdk/agentserver/azure-ai-agentserver-core/pyproject.toml index 5e19c7a03b89..de80a9010de8 100644 --- a/sdk/agentserver/azure-ai-agentserver-core/pyproject.toml +++ b/sdk/agentserver/azure-ai-agentserver-core/pyproject.toml @@ -23,9 +23,10 @@ keywords = ["azure", "azure sdk", "agent", "agentserver", "core"] dependencies = [ "starlette>=0.45.0", "hypercorn>=0.17.0", - "opentelemetry-api>=1.40.0", - "opentelemetry-sdk>=1.40.0", - "microsoft-opentelemetry>=1.0.0", + "opentelemetry-api~=1.43.0", + "opentelemetry-sdk~=1.43.0", + "opentelemetry-exporter-otlp-proto-grpc~=1.43.0", + "microsoft-opentelemetry~=1.3.5", ] [build-system] diff --git a/sdk/agentserver/azure-ai-agentserver-core/tests/test_otlp_protocol.py b/sdk/agentserver/azure-ai-agentserver-core/tests/test_otlp_protocol.py new file mode 100644 index 000000000000..0601fdaa9aa4 --- /dev/null +++ b/sdk/agentserver/azure-ai-agentserver-core/tests/test_otlp_protocol.py @@ -0,0 +1,247 @@ +# --------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# --------------------------------------------------------- +"""End-to-end OTLP protocol tests using in-process loopback receivers.""" + +import os +from concurrent.futures import ThreadPoolExecutor +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import subprocess +import sys +from threading import Lock, Thread + +import grpc +import pytest +from opentelemetry.proto.collector.logs.v1 import ( + logs_service_pb2, + logs_service_pb2_grpc, +) +from opentelemetry.proto.collector.metrics.v1 import ( + metrics_service_pb2, + metrics_service_pb2_grpc, +) +from opentelemetry.proto.collector.trace.v1 import ( + trace_service_pb2, + trace_service_pb2_grpc, +) + +_SIGNALS = {"traces", "metrics", "logs"} +_PACKAGE_ROOT = Path(__file__).resolve().parents[1] +_OTLP_ENV_VARS = ( + "APPLICATIONINSIGHTS_CONNECTION_STRING", + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_PROTOCOL", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL", +) + +_EMIT_ALL_SIGNALS = """ +from azure.ai.agentserver.core import configure_observability +from opentelemetry import metrics, trace +from opentelemetry._logs import LogRecord, SeverityNumber, get_logger_provider + +configure_observability() + +with trace.get_tracer("agentserver.otlp.test").start_as_current_span("otlp-test-span"): + pass + +metrics.get_meter("agentserver.otlp.test").create_counter("otlp.test").add(1) + +get_logger_provider().get_logger("agentserver.otlp.test").emit( + LogRecord( + severity_text="INFO", + severity_number=SeverityNumber.INFO, + body="otlp-test-log", + ) +) + +trace.get_tracer_provider().force_flush() +metrics.get_meter_provider().force_flush() +get_logger_provider().force_flush() +""" + + +class _SignalReceiver: + def __init__(self) -> None: + self._signals: set[str] = set() + self._lock = Lock() + + @property + def signals(self) -> set[str]: + with self._lock: + return set(self._signals) + + @property + def endpoint(self) -> str: + raise NotImplementedError + + def record(self, signal: str) -> None: + with self._lock: + self._signals.add(signal) + + def __enter__(self) -> "_SignalReceiver": + raise NotImplementedError + + def __exit__(self, *_args: object) -> None: + raise NotImplementedError + + +class _HttpOtlpReceiver(_SignalReceiver): + def __init__(self) -> None: + super().__init__() + receiver = self + services = { + "/v1/traces": ( + "traces", + trace_service_pb2.ExportTraceServiceRequest, + trace_service_pb2.ExportTraceServiceResponse, + ), + "/v1/metrics": ( + "metrics", + metrics_service_pb2.ExportMetricsServiceRequest, + metrics_service_pb2.ExportMetricsServiceResponse, + ), + "/v1/logs": ( + "logs", + logs_service_pb2.ExportLogsServiceRequest, + logs_service_pb2.ExportLogsServiceResponse, + ), + } + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self) -> None: # pylint: disable=invalid-name + service = services.get(self.path) + if service is None: + self.send_error(404) + return + + signal, request_type, response_type = service + body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) + request_type.FromString(body) + receiver.record(signal) + + response = response_type().SerializeToString() + self.send_response(200) + self.send_header("Content-Type", "application/x-protobuf") + self.send_header("Content-Length", str(len(response))) + self.end_headers() + self.wfile.write(response) + + def log_message( # pylint: disable=redefined-builtin + self, format: str, *args: object + ) -> None: + del format, args + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self._thread = Thread(target=self._server.serve_forever, daemon=True) + + @property + def endpoint(self) -> str: + return f"http://127.0.0.1:{self._server.server_port}" + + def __enter__(self) -> "_HttpOtlpReceiver": + self._thread.start() + return self + + def __exit__(self, *_args: object) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + + +class _GrpcOtlpReceiver(_SignalReceiver): + def __init__(self) -> None: + super().__init__() + receiver = self + + class TraceService(trace_service_pb2_grpc.TraceServiceServicer): + def Export( + self, request, context + ): # pylint: disable=invalid-name,unused-argument + receiver.record("traces") + return trace_service_pb2.ExportTraceServiceResponse() + + class MetricsService(metrics_service_pb2_grpc.MetricsServiceServicer): + def Export( + self, request, context + ): # pylint: disable=invalid-name,unused-argument + receiver.record("metrics") + return metrics_service_pb2.ExportMetricsServiceResponse() + + class LogsService(logs_service_pb2_grpc.LogsServiceServicer): + def Export( + self, request, context + ): # pylint: disable=invalid-name,unused-argument + receiver.record("logs") + return logs_service_pb2.ExportLogsServiceResponse() + + self._server = grpc.server(ThreadPoolExecutor(max_workers=3)) + trace_service_pb2_grpc.add_TraceServiceServicer_to_server( + TraceService(), self._server + ) + metrics_service_pb2_grpc.add_MetricsServiceServicer_to_server( + MetricsService(), self._server + ) + logs_service_pb2_grpc.add_LogsServiceServicer_to_server( + LogsService(), self._server + ) + self._port = self._server.add_insecure_port("127.0.0.1:0") + + @property + def endpoint(self) -> str: + return f"http://127.0.0.1:{self._port}" + + def __enter__(self) -> "_GrpcOtlpReceiver": + self._server.start() + return self + + def __exit__(self, *_args: object) -> None: + self._server.stop(grace=0).wait(timeout=5) + + +@pytest.mark.parametrize( + ("protocol", "receiver_type"), + [ + pytest.param("http/protobuf", _HttpOtlpReceiver, id="http-protobuf"), + pytest.param("grpc", _GrpcOtlpReceiver, id="grpc"), + ], +) +def test_otlp_protocol_exports_all_signals( + protocol: str, + receiver_type: type[_SignalReceiver], +) -> None: + """Agent Server must honor the configured OTLP protocol for every signal.""" + with receiver_type() as receiver: + env = os.environ.copy() + for variable in _OTLP_ENV_VARS: + env.pop(variable, None) + env.update( + { + "OTEL_EXPORTER_OTLP_ENDPOINT": receiver.endpoint, + "OTEL_EXPORTER_OTLP_PROTOCOL": protocol, + "OTEL_EXPORTER_OTLP_TIMEOUT": "1", + } + ) + + result = subprocess.run( + [sys.executable, "-c", _EMIT_ALL_SIGNALS], + cwd=_PACKAGE_ROOT, + env=env, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert receiver.signals == _SIGNALS, ( + f"{protocol} receiver got {sorted(receiver.signals)} instead of " + f"{sorted(_SIGNALS)}.\nsubprocess stderr:\n{result.stderr}" + )