diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py index 5e266e94adee..28be6d5a718d 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py @@ -11,6 +11,7 @@ import tempfile import json import time +from threading import Lock from typing import Any, Callable, Dict, Iterable, Iterator, List, Literal, Optional, Set, Tuple, TypedDict, Union, cast from openai import OpenAI, AzureOpenAI @@ -18,6 +19,7 @@ from azure.ai.evaluation._legacy._adapters.entities import Run import pandas as pd +from azure.core.credentials import AccessToken, TokenCredential from azure.ai.evaluation._common.math import list_mean_nan_safe, apply_transform_nan_safe from azure.ai.evaluation._common.utils import validate_azure_ai_project, is_onedp_project from azure.ai.evaluation._evaluators._common._base_eval import EvaluatorBase @@ -64,6 +66,7 @@ ) LOGGER = logging.getLogger(__name__) +AZURE_MONITOR_SCOPE = "https://monitor.azure.com/.default" # For metrics (aggregates) whose metric names intentionally differ from their # originating column name, usually because the aggregation of the original value @@ -1383,19 +1386,22 @@ def emit_eval_result_events_to_app_insights( :type results: List[Dict] """ + if not results: + LOGGER.debug("No results to log to App Insights") + return + + exporter_options = _get_app_insights_exporter_options(app_insights_config) + use_entra_authentication = app_insights_config.get("credential_type") == "ProjectManagedIdentity" + from opentelemetry import _logs from opentelemetry.sdk._logs import LoggerProvider - from opentelemetry.sdk._logs.export import BatchLogRecordProcessor + from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogExportResult from opentelemetry.sdk.resources import Resource from opentelemetry.semconv.resource import ResourceAttributes from azure.monitor.opentelemetry.exporter import AzureMonitorLogExporter from opentelemetry._events import get_event_logger from opentelemetry.sdk._events import EventLoggerProvider - if not results: - LOGGER.debug("No results to log to App Insights") - return - logger_provider = None try: # Configure OpenTelemetry logging with anonymized Resource attributes @@ -1414,13 +1420,16 @@ def emit_eval_result_events_to_app_insights( _logs.set_logger_provider(logger_provider) # Create Azure Monitor log exporter - azure_log_exporter = AzureMonitorLogExporter(connection_string=app_insights_config["connection_string"]) + azure_log_exporter = AzureMonitorLogExporter(**exporter_options) + export_result_tracker: Optional["_ExportResultTrackingLogExporter"] = None + log_exporter: Any = azure_log_exporter + if use_entra_authentication: + export_result_tracker = _ExportResultTrackingLogExporter(azure_log_exporter, LogExportResult.FAILURE) + log_exporter = export_result_tracker # Add the Azure Monitor exporter to the logger provider # Set export_timeout_millis to prevent individual batch exports from hanging - logger_provider.add_log_record_processor( - BatchLogRecordProcessor(azure_log_exporter, export_timeout_millis=60000) - ) + logger_provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter, export_timeout_millis=60000)) # Create event logger event_provider = EventLoggerProvider(logger_provider) @@ -1454,16 +1463,23 @@ def emit_eval_result_events_to_app_insights( # Force flush to ensure events are sent, with a timeout to prevent hanging flush_timeout_millis = 60000 # 60 seconds flush_success = logger_provider.force_flush(timeout_millis=flush_timeout_millis) - if flush_success: - LOGGER.info(f"Successfully logged {len(results)} evaluation results to App Insights") - else: - LOGGER.warning( + if export_result_tracker is not None and export_result_tracker.export_failed: + raise RuntimeError("Failed to export evaluation results to App Insights.") + if not flush_success: + timeout_message = ( f"App Insights force_flush timed out after {flush_timeout_millis}ms. " "Some evaluation events may not have been sent." ) + if use_entra_authentication: + raise TimeoutError(timeout_message) + LOGGER.warning(timeout_message) + else: + LOGGER.info(f"Successfully logged {len(results)} evaluation results to App Insights") - except Exception as e: - LOGGER.error(f"Failed to emit evaluation results to App Insights: {e}") + except Exception as ex: + if use_entra_authentication: + raise + LOGGER.error("Failed to emit evaluation results to App Insights: %s", ex) finally: # Shut down the logger provider to stop background threads (e.g. OneSettings # configuration poller) that would otherwise keep the process alive indefinitely. @@ -1474,6 +1490,77 @@ def emit_eval_result_events_to_app_insights( pass +class _AzureMonitorScopedCredential: # pylint: disable=too-few-public-methods + """Delegate token acquisition with Azure Monitor's required scope.""" + + def __init__(self, credential: TokenCredential) -> None: + self._credential = credential + + def get_token(self, *_scopes: str, **kwargs: Any) -> AccessToken: + """Request a fresh Azure Monitor token from the configured credential.""" + return self._credential.get_token(AZURE_MONITOR_SCOPE, **kwargs) + + +class _ExportResultTrackingLogExporter: # pylint: disable=too-few-public-methods + """Track failures returned by a log exporter running on a worker thread.""" + + def __init__(self, exporter: Any, failure_result: Any) -> None: + self._exporter = exporter + self._failure_result = failure_result + self._export_failed = False + self._lock = Lock() + + @property + def export_failed(self) -> bool: + """Return whether any batch export failed.""" + with self._lock: + return self._export_failed + + def export(self, batch: Any) -> Any: + """Delegate a batch export and retain its failure status.""" + try: + result = self._exporter.export(batch) + except Exception: + with self._lock: + self._export_failed = True + raise + + if result == self._failure_result: + with self._lock: + self._export_failed = True + return result + + def shutdown(self) -> None: + """Shut down the wrapped exporter.""" + self._exporter.shutdown() + + +def _get_app_insights_exporter_options( + app_insights_config: AppInsightsConfig, +) -> Dict[str, Any]: + exporter_options: Dict[str, Any] = {"connection_string": app_insights_config["connection_string"]} + credential_type = app_insights_config.get("credential_type") + if credential_type is None: + return exporter_options + if credential_type == "ApiKey": + exporter_options["credential"] = None + return exporter_options + if credential_type != "ProjectManagedIdentity": + raise EvaluationException( + message=f"Unsupported App Insights credential type: {credential_type}.", + target=ErrorTarget.EVALUATE, + category=ErrorCategory.INVALID_VALUE, + blame=ErrorBlame.SYSTEM_ERROR, + ) + + credential = app_insights_config.get("credential") + if credential is None: + raise ValueError("App Insights ProjectManagedIdentity authentication requires a TokenCredential.") + exporter_options["credential"] = _AzureMonitorScopedCredential(credential) + exporter_options["credential_scopes"] = [AZURE_MONITOR_SCOPE] + return exporter_options + + def _preprocess_data( data: Union[str, os.PathLike], evaluators_and_graders: Dict[str, Union[Callable, AzureOpenAIGrader]], diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_model_configurations.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_model_configurations.py index d922d597fb93..32eb70130d93 100644 --- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_model_configurations.py +++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_model_configurations.py @@ -5,6 +5,7 @@ from typing import Any, Dict, List, Literal, TypedDict, Union from typing_extensions import NotRequired +from azure.core.credentials import TokenCredential from ._evaluator_definition import EvaluatorDefinition from typing import Dict, List, Optional, Any @@ -175,6 +176,8 @@ class EvaluationResult(TypedDict): class AppInsightsConfig(TypedDict): connection_string: str + credential_type: NotRequired[Literal["ApiKey", "ProjectManagedIdentity"]] + credential: NotRequired[TokenCredential] project_id: NotRequired[str] run_type: NotRequired[str] schedule_type: NotRequired[str] diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py index 2ac7d282f237..edb93f9176f4 100644 --- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py +++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py @@ -5,8 +5,12 @@ import os import pathlib import numpy as np -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, call, patch +from azure.core.credentials import AccessToken, TokenCredential +from azure.core.pipeline import PipelineContext, PipelineRequest +from azure.core.pipeline.policies import BearerTokenCredentialPolicy +from azure.core.rest import HttpRequest import pandas as pd import pytest from pandas.testing import assert_frame_equal @@ -53,11 +57,12 @@ _is_inverse_metric, _create_result_object, emit_eval_result_events_to_app_insights, + _get_app_insights_exporter_options, ) from azure.ai.evaluation._evaluate._utils import _convert_name_map_into_property_entries from azure.ai.evaluation._evaluate._utils import _apply_column_mapping, _trace_destination_from_project_scope from azure.ai.evaluation._evaluators._eci._eci import ECIEvaluator -from azure.ai.evaluation._exceptions import EvaluationException +from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException from azure.ai.evaluation._legacy._adapters._check import MISSING_LEGACY_SDK @@ -3613,3 +3618,203 @@ def test_no_shutdown_when_results_empty(self, mock_lp_cls): emit_eval_result_events_to_app_insights(config, []) mock_lp_cls.assert_not_called() + + +@pytest.mark.skipif(MISSING_OPENTELEMETRY, reason="This test requires the opentelemetry package") +class TestAppInsightsAuthentication: + """Tests for Application Insights exporter authentication configuration.""" + + _RESULTS = [ + { + "results": [{"metric": "coherence", "score": 4.5}], + "datasource_item": {}, + } + ] + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_project_managed_identity_credential_and_scope_are_passed_to_exporter(self, mock_lp_cls): + mock_lp_cls.return_value.force_flush.return_value = True + credential = MagicMock(spec=TokenCredential) + exporter_module = MagicMock() + config = { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + "credential": credential, + } + + with patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}): + emit_eval_result_events_to_app_insights(config, self._RESULTS) + + exporter_options = exporter_module.AzureMonitorLogExporter.call_args.kwargs + assert exporter_options["connection_string"] == "InstrumentationKey=fake-key" + assert exporter_options["credential_scopes"] == ["https://monitor.azure.com/.default"] + assert exporter_options["credential"] is not credential + assert "token" not in config + + def test_exporter_credential_refresh_uses_monitor_scope(self): + credential = MagicMock(spec=TokenCredential) + credential.get_token.side_effect = [ + AccessToken("token-1", 0), + AccessToken("token-2", 0), + ] + exporter_options = _get_app_insights_exporter_options( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + "credential": credential, + } + ) + policy = BearerTokenCredentialPolicy( + exporter_options["credential"], + "https://monitor.azure.com//.default", + ) + request = PipelineRequest( + HttpRequest("POST", "https://example.test"), + PipelineContext(None), + ) + + policy.on_request(request) + policy.on_request(request) + + assert credential.get_token.call_args_list == [ + call("https://monitor.azure.com/.default"), + call("https://monitor.azure.com/.default"), + ] + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_project_managed_identity_exporter_failure_is_surfaced(self, mock_lp_cls): + credential = MagicMock(spec=TokenCredential) + exporter_module = MagicMock() + exporter_module.AzureMonitorLogExporter.side_effect = RuntimeError("authentication failed") + + with patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}): + with pytest.raises(RuntimeError, match="authentication failed"): + emit_eval_result_events_to_app_insights( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + "credential": credential, + }, + self._RESULTS, + ) + + mock_lp_cls.return_value.shutdown.assert_called_once() + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_project_managed_identity_batch_export_failure_is_surfaced(self, mock_lp_cls): + from opentelemetry.sdk._logs.export import LogExportResult + + mock_lp_cls.return_value.force_flush.return_value = True + credential = MagicMock(spec=TokenCredential) + exporter = MagicMock() + exporter.export.return_value = LogExportResult.FAILURE + exporter_module = MagicMock() + exporter_module.AzureMonitorLogExporter.return_value = exporter + + def create_processor(tracked_exporter, **_kwargs): + tracked_exporter.export([]) + return MagicMock() + + with patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}), patch( + "opentelemetry.sdk._logs.export.BatchLogRecordProcessor", + side_effect=create_processor, + ): + with pytest.raises(RuntimeError, match="Failed to export evaluation results"): + emit_eval_result_events_to_app_insights( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + "credential": credential, + }, + self._RESULTS, + ) + + exporter.export.assert_called_once_with([]) + mock_lp_cls.return_value.force_flush.assert_called_once() + mock_lp_cls.return_value.shutdown.assert_called_once() + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_project_managed_identity_flush_timeout_is_surfaced(self, mock_lp_cls): + mock_lp_cls.return_value.force_flush.return_value = False + credential = MagicMock(spec=TokenCredential) + exporter_module = MagicMock() + + with patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}): + with pytest.raises(TimeoutError, match="force_flush timed out"): + emit_eval_result_events_to_app_insights( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + "credential": credential, + }, + self._RESULTS, + ) + + mock_lp_cls.return_value.shutdown.assert_called_once() + + @patch("azure.identity.DefaultAzureCredential") + def test_entra_authentication_requires_credential_without_fallback(self, mock_default_credential): + environment = { + "APPLICATIONINSIGHTS_CONNECTION_STRING": "InstrumentationKey=environment-key", + "AZURE_CLIENT_ID": "environment-client-id", + } + with patch.dict(os.environ, environment, clear=False): + original_environment = os.environ.copy() + with pytest.raises( + ValueError, + match="ProjectManagedIdentity authentication requires a TokenCredential", + ): + emit_eval_result_events_to_app_insights( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ProjectManagedIdentity", + }, + self._RESULTS, + ) + assert os.environ == original_environment + mock_default_credential.assert_not_called() + + def test_unknown_credential_type_does_not_fall_back_to_api_key(self): + with pytest.raises(EvaluationException, match="Unsupported App Insights credential type") as exc_info: + emit_eval_result_events_to_app_insights( + { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "FutureCredential", + }, + self._RESULTS, + ) + assert exc_info.value.target == ErrorTarget.EVALUATE + assert exc_info.value.category == ErrorCategory.INVALID_VALUE + assert exc_info.value.blame == ErrorBlame.SYSTEM_ERROR + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_missing_credential_type_configuration_remains_compatible(self, mock_lp_cls): + mock_lp_cls.return_value.force_flush.return_value = True + exporter_module = MagicMock() + config = {"connection_string": "InstrumentationKey=fake-key"} + + with patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}): + emit_eval_result_events_to_app_insights(config, self._RESULTS) + + exporter_module.AzureMonitorLogExporter.assert_called_once_with(connection_string="InstrumentationKey=fake-key") + + @patch("opentelemetry.sdk._logs.LoggerProvider") + def test_api_key_configuration_disables_environment_fallback(self, mock_lp_cls): + mock_lp_cls.return_value.force_flush.return_value = True + exporter_module = MagicMock() + config = { + "connection_string": "InstrumentationKey=fake-key", + "credential_type": "ApiKey", + } + + with patch.dict( + os.environ, + {"APPLICATIONINSIGHTS_AUTHENTICATION_STRING": "Authorization=AAD"}, + clear=False, + ), patch.dict("sys.modules", {"azure.monitor.opentelemetry.exporter": exporter_module}): + emit_eval_result_events_to_app_insights(config, self._RESULTS) + + exporter_module.AzureMonitorLogExporter.assert_called_once_with( + connection_string="InstrumentationKey=fake-key", + credential=None, + )