add log exporter and metric reader params to configure azure monitor (#44367)

* add log exporter and metric reader params

* consistent ordering

* added changelog

* add some test configs

* test fixes

* fix pylint errors

* Add more tests and rename

* Remove duplicate metric reader

* Add samples

* Fix black

* Fix format using tox

* update test

---------

Co-authored-by: Radhika Gupta <[email protected]>

Author: eavanvalkenburg

sdk/monitor/azure-monitor-opentelemetry/CHANGELOG.md

@@ -3,6 +3,8 @@
 ## 1.8.4 (Unreleased)
 
 ### Features Added
+- Added ability to add additional Log Record Processors and Metric Readers via configure_azure_monitor
+  ([#44367](https://github.com/Azure/azure-sdk-for-python/pull/44367))
 
 ### Breaking Changes
 

sdk/monitor/azure-monitor-opentelemetry/README.md

@@ -68,6 +68,8 @@ You can use `configure_azure_monitor` to set up instrumentation for your app to
 | `resource` | Specifies the OpenTelemetry [Resource][ot_spec_resource] associated with your application. Passed in [Resource Attributes][ot_spec_resource_attributes] take priority over default attributes and those from [Resource Detectors][ot_python_resource_detectors]. | [OTEL_SERVICE_NAME][ot_spec_service_name], [OTEL_RESOURCE_ATTRIBUTES][ot_spec_resource_attributes], [OTEL_EXPERIMENTAL_RESOURCE_DETECTORS][ot_python_resource_detectors] |
 | `span_processors` | A list of [span processors][ot_span_processor] that will perform processing on each of your spans before they are exported. Useful for filtering/modifying telemetry. | `N/A` |
 | `views` | A list of [views][ot_view] that will be used to customize metrics exported by the SDK. | `N/A` |
+| `log_record_processors` | A list of [log record processors][ot_log_record_processor] that will process log records before they are exported. | `N/A` |
+| `metric_readers` | A list of [metric reader][ot_metric_reader] that will process metric readers before they are exported | `N/A` |
 | `traces_per_second` | Configures the Rate Limited sampler by specifying the maximum number of traces to sample per second. When set, this automatically enables the rate-limited sampler. Alternatively, you can configure sampling using the `OTEL_TRACES_SAMPLER` and `OTEL_TRACES_SAMPLER_ARG` environment variables as described in the table below. Please note that the sampling configuration via environment variables will have precedence over the sampling exporter/distro options. | `N/A`
 
 You can configure further with [OpenTelemetry environment variables][ot_env_vars].
@@ -231,6 +233,7 @@ contact [[email protected]](mailto:[email protected]) with any additio
 [ot_sdk_python]: https://github.com/open-telemetry/opentelemetry-python
 [ot_sdk_python_metric_reader]: https://opentelemetry-python.readthedocs.io/en/latest/sdk/metrics.export.html#opentelemetry.sdk.metrics.export.MetricReader
 [ot_span_processor]: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/sdk.md#span-processor
+[ot_log_record_processor]: https://github.com/open-telemetry/opentelemetry-specification/tree/main/specification/logs/sdk.md#log-record-processor
 [ot_view]: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#view
 [ot_sdk_python_view_examples]: https://github.com/open-telemetry/opentelemetry-python/tree/main/docs/examples/metrics/views
 [ot_instrumentation_django]: https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-django

sdk/monitor/azure-monitor-opentelemetry/azure/monitor/opentelemetry/_configure.py

@@ -12,7 +12,7 @@
 )
 from opentelemetry.metrics import set_meter_provider
 from opentelemetry.sdk.metrics import MeterProvider
-from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, MetricReader
 from opentelemetry.sdk.metrics.view import View
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
@@ -38,6 +38,8 @@
     SAMPLING_RATIO_ARG,
     SAMPLING_TRACES_PER_SECOND_ARG,
     SPAN_PROCESSORS_ARG,
+    LOG_RECORD_PROCESSORS_ARG,
+    METRIC_READERS_ARG,
     VIEWS_ARG,
     ENABLE_TRACE_BASED_SAMPLING_ARG,
 )
@@ -102,6 +104,10 @@ def configure_azure_monitor(**kwargs) -> None:  # pylint: disable=C4758
      Attributes take priority over default attributes and those from Resource Detectors.
     :keyword list[~opentelemetry.sdk.trace.SpanProcessor] span_processors: List of `SpanProcessor` objects
      to process every span prior to exporting. Will be run sequentially.
+    :keyword list[~opentelemetry.sdk._logs.LogRecordProcessor] log_record_processors: List of `LogRecordProcessor`
+     objects to process every log record prior to exporting. Will be run sequentially.
+    :keyword list[~opentelemetry.sdk.metrics.MetricReader] metric_readers: List of MetricReader objects to read and
+     export metrics. Each reader can have its own exporter and collection interval.
     :keyword bool enable_live_metrics: Boolean value to determine whether to enable live metrics feature.
      Defaults to `False`.
     :keyword bool enable_performance_counters: Boolean value to determine whether to enable performance counters.
@@ -212,6 +218,8 @@ def _setup_logging(configurations: Dict[str, ConfigurationValue]):
         enable_performance_counters_config = configurations[ENABLE_PERFORMANCE_COUNTERS_ARG]
         logger_provider = LoggerProvider(resource=resource)
         enable_trace_based_sampling_for_logs = configurations[ENABLE_TRACE_BASED_SAMPLING_ARG]
+        for custom_log_record_processor in configurations[LOG_RECORD_PROCESSORS_ARG]:  # type: ignore
+            logger_provider.add_log_record_processor(custom_log_record_processor)  # type: ignore
         if configurations.get(ENABLE_LIVE_METRICS_ARG):
             qlp = _QuickpulseLogRecordProcessor()
             logger_provider.add_log_record_processor(qlp)
@@ -270,11 +278,12 @@ def _setup_logging(configurations: Dict[str, ConfigurationValue]):
 def _setup_metrics(configurations: Dict[str, ConfigurationValue]):
     resource: Resource = configurations[RESOURCE_ARG]  # type: ignore
     views: List[View] = configurations[VIEWS_ARG]  # type: ignore
+    readers: list[MetricReader] = configurations[METRIC_READERS_ARG]  # type: ignore
     enable_performance_counters_config = configurations[ENABLE_PERFORMANCE_COUNTERS_ARG]
     metric_exporter = AzureMonitorMetricExporter(**configurations)
-    reader = PeriodicExportingMetricReader(metric_exporter)
+    readers.append(PeriodicExportingMetricReader(metric_exporter))
     meter_provider = MeterProvider(
-        metric_readers=[reader],
+        metric_readers=readers,
         resource=resource,
         views=views,
     )

sdk/monitor/azure-monitor-opentelemetry/azure/monitor/opentelemetry/_constants.py

@@ -24,6 +24,8 @@
 RESOURCE_ARG = "resource"
 SAMPLING_RATIO_ARG = "sampling_ratio"
 SPAN_PROCESSORS_ARG = "span_processors"
+LOG_RECORD_PROCESSORS_ARG = "log_record_processors"
+METRIC_READERS_ARG = "metric_readers"
 VIEWS_ARG = "views"
 RATE_LIMITED_SAMPLER = "microsoft.rate_limited"
 FIXED_PERCENTAGE_SAMPLER = "microsoft.fixed.percentage"

sdk/monitor/azure-monitor-opentelemetry/azure/monitor/opentelemetry/_utils/configurations.py

@@ -43,6 +43,8 @@
     SAMPLING_RATIO_ARG,
     SAMPLING_TRACES_PER_SECOND_ARG,
     SPAN_PROCESSORS_ARG,
+    LOG_RECORD_PROCESSORS_ARG,
+    METRIC_READERS_ARG,
     VIEWS_ARG,
     RATE_LIMITED_SAMPLER,
     FIXED_PERCENTAGE_SAMPLER,
@@ -78,6 +80,8 @@ def _get_configurations(**kwargs) -> Dict[str, ConfigurationValue]:
     _default_sampling_ratio(configurations)
     _default_instrumentation_options(configurations)
     _default_span_processors(configurations)
+    _default_log_record_processors(configurations)
+    _default_metric_readers(configurations)
     _default_enable_live_metrics(configurations)
     _default_enable_performance_counters(configurations)
     _default_views(configurations)
@@ -225,6 +229,14 @@ def _default_span_processors(configurations):
     configurations.setdefault(SPAN_PROCESSORS_ARG, [])
 
 
+def _default_log_record_processors(configurations):
+    configurations.setdefault(LOG_RECORD_PROCESSORS_ARG, [])
+
+
+def _default_metric_readers(configurations):
+    configurations.setdefault(METRIC_READERS_ARG, [])
+
+
 def _default_enable_live_metrics(configurations):
     configurations.setdefault(ENABLE_LIVE_METRICS_ARG, False)
 

sdk/monitor/azure-monitor-opentelemetry/samples/logging/modify_logs.py

@@ -0,0 +1,46 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+import logging
+from logging import getLogger
+from azure.monitor.opentelemetry import configure_azure_monitor
+from azure.monitor.opentelemetry.exporter._generated.models import ContextTagKeys
+from opentelemetry import trace
+from opentelemetry.sdk._logs import LogRecordProcessor, ReadableLogRecord
+
+logger = getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+class LogRecordEnrichingProcessor(LogRecordProcessor):
+    """Enriches log records with operation name from the current span context."""
+
+    def on_emit(self, readable_log_record: ReadableLogRecord) -> None:
+        current_span = trace.get_current_span()
+        if current_span and getattr(current_span, "name", None):
+            if readable_log_record.log_record.attributes is None:
+                readable_log_record.log_record.attributes = {}
+            readable_log_record.log_record.attributes[ContextTagKeys.AI_OPERATION_NAME] = current_span.name
+
+    def shutdown(self) -> None:
+        pass
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        return True
+
+
+# Create the log record enriching processor
+log_enriching_processor = LogRecordEnrichingProcessor()
+
+# Configure Azure Monitor with the custom log record processor
+configure_azure_monitor(log_record_processors=[log_enriching_processor])
+
+tracer = trace.get_tracer(__name__)
+
+with tracer.start_as_current_span("span-name-here"):
+    logger.info("This log will be enriched with operation name")
+
+input()

sdk/monitor/azure-monitor-opentelemetry/samples/metrics/modify_metrics.py

@@ -0,0 +1,50 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+from time import sleep
+
+from azure.monitor.opentelemetry import configure_azure_monitor
+from opentelemetry import metrics
+from opentelemetry.sdk.metrics.export import (
+    MetricExportResult,
+    MetricExporter,
+    MetricsData,
+    PeriodicExportingMetricReader,
+)
+
+
+class PrintMetricExporter(MetricExporter):
+    """Minimal exporter that prints metric data."""
+
+    def export(self, metrics_data: MetricsData, **kwargs) -> MetricExportResult:  # type: ignore[override]
+        # In a real exporter, send metrics_data to your backend
+        print(f"exported metrics: {metrics_data}")
+        return MetricExportResult.SUCCESS
+
+    def shutdown(self, timeout_millis: float = 30000, **kwargs) -> None:  # type: ignore[override]
+        return None
+
+    def force_flush(self, timeout_millis: float = 30000, **kwargs) -> bool:  # type: ignore[override]
+        return True
+
+
+# Add a custom reader; the SDK will append its own Azure Monitor reader
+custom_reader = PeriodicExportingMetricReader(
+    PrintMetricExporter(),
+    export_interval_millis=5000,
+)
+
+configure_azure_monitor(
+    enable_performance_counters=False,
+    metric_readers=[custom_reader],
+)
+
+meter = metrics.get_meter_provider().get_meter("metric-readers-sample")
+counter = meter.create_counter("example.counter")
+
+for _ in range(3):
+    counter.add(1)
+    sleep(1)

sdk/monitor/azure-monitor-opentelemetry/tests/test_configure.py

@@ -536,6 +536,7 @@ def test_setup_logging(self, get_logger_mock, pclp_mock):
         logging_handler_mock.return_value = logging_handler_init_mock
         logger_mock = Mock()
         logger_mock.handlers = []
+        custom_lrp = Mock()
         get_logger_mock.return_value = logger_mock
         formatter_init_mock = Mock()
         elp_init_mock = Mock()
@@ -547,6 +548,7 @@ def test_setup_logging(self, get_logger_mock, pclp_mock):
             "enable_performance_counters": True,
             "logger_name": "test",
             "resource": TEST_RESOURCE,
+            "log_record_processors": [custom_lrp],
             "logging_formatter": formatter_init_mock,
             "enable_trace_based_sampling_for_logs": False,
         }
@@ -572,7 +574,11 @@ def test_setup_logging(self, get_logger_mock, pclp_mock):
         set_logger_provider_mock.assert_called_once_with(lp_init_mock)
         log_exporter_mock.assert_called_once_with(**configurations)
         blrp_mock.assert_called_once_with(log_exp_init_mock, {"enable_trace_based_sampling_for_logs": False})
-        self.assertEqual(lp_init_mock.add_log_record_processor.call_count, 2)
+        self.assertEqual(lp_init_mock.add_log_record_processor.call_count, 3)
+        lp_init_mock.add_log_record_processor.assert_has_calls(
+            [call(custom_lrp), call(pclp_init_mock), call(blrp_init_mock)]
+        )
+        self.assertEqual(lp_init_mock.add_log_record_processor.call_count, 3)
         lp_init_mock.add_log_record_processor.assert_has_calls([call(pclp_init_mock), call(blrp_init_mock)])
         logging_handler_mock.assert_called_once_with(logger_provider=lp_init_mock)
         logging_handler_init_mock.setFormatter.assert_called_once_with(formatter_init_mock)
@@ -620,6 +626,7 @@ def test_setup_logging_duplicate_logger(self, get_logger_mock, instance_mock, pc
             "enable_performance_counters": True,
             "logger_name": "test",
             "resource": TEST_RESOURCE,
+            "log_record_processors": [],
             "logging_formatter": None,
             "enable_trace_based_sampling_for_logs": True,
         }
@@ -686,6 +693,7 @@ def test_setup_logging_disable_performance_counters(self, get_logger_mock, pclp_
             "enable_performance_counters": False,
             "logger_name": "test",
             "resource": TEST_RESOURCE,
+            "log_record_processors": [],
             "logging_formatter": formatter_init_mock,
             "enable_trace_based_sampling_for_logs": False,
         }
@@ -745,15 +753,20 @@ def test_setup_metrics(
         reader_init_mock = Mock()
         reader_mock.return_value = reader_init_mock
 
+        # Custom metric readers provided by user
+        custom_reader_1 = Mock()
+        custom_reader_2 = Mock()
+
         configurations = {
             "connection_string": "test_cs",
             "enable_performance_counters": True,
             "resource": TEST_RESOURCE,
+            "metric_readers": [custom_reader_1, custom_reader_2],
             "views": [],
         }
         _setup_metrics(configurations)
         mp_mock.assert_called_once_with(
-            metric_readers=[reader_init_mock],
+            metric_readers=[custom_reader_1, custom_reader_2, reader_init_mock],
             resource=TEST_RESOURCE,
             views=[],
         )
@@ -793,6 +806,7 @@ def test_setup_metrics_views(
             "connection_string": "test_cs",
             "enable_performance_counters": False,
             "resource": TEST_RESOURCE,
+            "metric_readers": [],
             "views": [view_mock],
         }
         _setup_metrics(configurations)
@@ -836,6 +850,7 @@ def test_setup_metrics_perf_counters_disabled(
             "connection_string": "test_cs",
             "enable_performance_counters": False,
             "resource": TEST_RESOURCE,
+            "metric_readers": [],
             "views": [],
         }
         _setup_metrics(configurations)

sdk/monitor/azure-monitor-opentelemetry/tests/utils/test_configurations.py

@@ -33,6 +33,7 @@
     RATE_LIMITED_SAMPLER,
     FIXED_PERCENTAGE_SAMPLER,
     ENABLE_TRACE_BASED_SAMPLING_ARG,
+    METRIC_READERS_ARG,
 )
 from opentelemetry.environment_variables import (
     OTEL_LOGS_EXPORTER,
@@ -76,6 +77,8 @@ def test_get_configurations(self, resource_create_mock):
             views=["test_view"],
             logger_name="test_logger",
             span_processors=["test_processor"],
+            log_record_processors=["test_log_record_processor"],
+            metric_readers=["test_metric_reader"],
             enable_trace_based_sampling_for_logs=True,
         )
 
@@ -110,6 +113,8 @@ def test_get_configurations(self, resource_create_mock):
         self.assertEqual(configurations["views"], ["test_view"])
         self.assertEqual(configurations["logger_name"], "test_logger")
         self.assertEqual(configurations["span_processors"], ["test_processor"])
+        self.assertEqual(configurations["log_record_processors"], ["test_log_record_processor"])
+        self.assertEqual(configurations[METRIC_READERS_ARG], ["test_metric_reader"])
         self.assertEqual(configurations[ENABLE_TRACE_BASED_SAMPLING_ARG], True)
 
     @patch.dict("os.environ", {}, clear=True)
@@ -144,6 +149,8 @@ def test_get_configurations_defaults(self, resource_create_mock):
         self.assertEqual(configurations["enable_performance_counters"], True)
         self.assertEqual(configurations["logger_name"], "")
         self.assertEqual(configurations["span_processors"], [])
+        self.assertEqual(configurations["log_record_processors"], [])
+        self.assertEqual(configurations["metric_readers"], [])
         self.assertEqual(configurations["views"], [])
         self.assertEqual(configurations[ENABLE_TRACE_BASED_SAMPLING_ARG], False)