Implement live metrics filtering for docs (Azure#38925)

Author: lzchen

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

@@ -4,6 +4,13 @@
 
 ### Features Added
 
+- Implement live metrics filtering for metrics
+  ([#37998](https://github.com/Azure/azure-sdk-for-python/pull/37998))
+- Add applying filter/validating filter logic to live metrics filtering
+  ([#38451](https://github.com/Azure/azure-sdk-for-python/pull/38451))
+- Implement live metrics filtering for docs
+  ([#38925](https://github.com/Azure/azure-sdk-for-python/pull/38925))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_filter.py

@@ -7,6 +7,7 @@
 
 from azure.monitor.opentelemetry.exporter._quickpulse._generated.models import (
     DerivedMetricInfo,
+    DocumentStreamInfo,
     FilterConjunctionGroupInfo,
     FilterInfo,
     PredicateType,
@@ -18,14 +19,18 @@
 from azure.monitor.opentelemetry.exporter._quickpulse._state import (
     _clear_quickpulse_projection_map,
     _set_quickpulse_derived_metric_infos,
+    _set_quickpulse_doc_stream_infos,
     _set_quickpulse_etag,
 )
 from azure.monitor.opentelemetry.exporter._quickpulse._types import (
     _DATA_FIELD_NAMES,
     _TelemetryData,
 )
 from azure.monitor.opentelemetry.exporter._quickpulse._utils import _filter_time_stamp_to_ms
-from azure.monitor.opentelemetry.exporter._quickpulse._validate import _validate_derived_metric_info
+from azure.monitor.opentelemetry.exporter._quickpulse._validate import (
+    _validate_derived_metric_info,
+    _validate_document_filter_group_info,
+)
 
 
 # Apply filter configuration based off response
@@ -38,7 +43,7 @@ def _update_filter_configuration(etag: str, config_bytes: bytes):
     # Process metric filter configuration
     _parse_metric_filter_configuration(config)
     # # Process document filter configuration
-    # _parse_document_filter_configuration(config)
+    _parse_document_filter_configuration(config)
     # Update new etag
     _set_quickpulse_etag(etag)
 
@@ -67,21 +72,23 @@ def _parse_metric_filter_configuration(config: Dict[str, Any]) -> None:
     _set_quickpulse_derived_metric_infos(metric_infos)
 
 
-# def _parse_document_filter_configuration(config: Dict[str, Any]) -> None:
-#     # Process document filter configuration
-#     doc_infos: Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]] = {}
-#     for doc_stream_dict in config.get("document_streams", []):
-#         doc_stream = DocumentStreamInfo.from_dict(doc_stream_dict)
-#         for doc_filter_group in doc_stream.document_filter_groups:
-#             if not _validate_document_filter_group_info(doc_filter_group):
-#                 continue
-#             # TODO: Rename exception fields
-#             telemetry_type: TelemetryType = TelemetryType(doc_filter_group.telemetry_type)
-#             if telemetry_type not in doc_infos:
-#                 doc_infos[telemetry_type] = {}
-#             if doc_stream.id not in doc_infos[telemetry_type]:
-#                 doc_infos[telemetry_type][doc_stream.id] = []
-#             doc_infos[telemetry_type][doc_stream.id].append(doc_filter_group.filters)
+def _parse_document_filter_configuration(config: Dict[str, Any]) -> None:
+    # Process document filter configuration
+    doc_infos: Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]] = {}
+    for doc_stream_dict in config.get("DocumentStreams", []):
+        doc_stream = DocumentStreamInfo.from_dict(doc_stream_dict)
+        for doc_filter_group in doc_stream.document_filter_groups:
+            if not _validate_document_filter_group_info(doc_filter_group):
+                continue
+            # Rename exception fields by parsing out "Exception." portion
+            _rename_exception_fields_for_filtering(doc_filter_group.filters)
+            telemetry_type: TelemetryType = TelemetryType(doc_filter_group.telemetry_type)
+            if telemetry_type not in doc_infos:
+                doc_infos[telemetry_type] = {}
+            if doc_stream.id not in doc_infos[telemetry_type]:
+                doc_infos[telemetry_type][doc_stream.id] = []
+            doc_infos[telemetry_type][doc_stream.id].append(doc_filter_group.filters)
+    _set_quickpulse_doc_stream_infos(doc_infos)
 
 
 def _rename_exception_fields_for_filtering(filter_groups: FilterConjunctionGroupInfo):

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_live_metrics.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 # cSpell:disable
-from typing import Any
+from typing import Any, Dict, List, Optional
 
 import logging
 import platform
@@ -39,9 +39,12 @@
     _QuickpulseMetricReader,
 )
 from azure.monitor.opentelemetry.exporter._quickpulse._filter import (
+    _check_filters,
     _check_metric_filters,
 )
 from azure.monitor.opentelemetry.exporter._quickpulse._generated.models import (
+    DerivedMetricInfo,
+    FilterConjunctionGroupInfo,
     MonitoringDataPoint,
     TelemetryType,
 )
@@ -53,6 +56,7 @@
     _is_post_state,
     _append_quickpulse_document,
     _get_quickpulse_derived_metric_infos,
+    _get_quickpulse_doc_stream_infos,
     _set_global_quickpulse_state,
 )
 from azure.monitor.opentelemetry.exporter._quickpulse._types import (
@@ -174,8 +178,6 @@ def _record_span(self, span: ReadableSpan) -> None:
         # Only record if in post state
         if _is_post_state():
             try:
-                document = _get_span_document(span)
-                _append_quickpulse_document(document)
                 duration_ms = 0
                 if span.end_time and span.start_time:
                     duration_ms = (span.end_time - span.start_time) / 1e9  # type: ignore
@@ -195,13 +197,14 @@ def _record_span(self, span: ReadableSpan) -> None:
                         self._dependency_failure_rate_counter.add(1)
                     self._dependency_duration.record(duration_ms)
 
-                metric_infos_dict = _get_quickpulse_derived_metric_infos()
-                # check if filtering is enabled
-                if metric_infos_dict:
-                    # Derive metrics for quickpulse filtering
-                    data = _TelemetryData._from_span(span)
-                    _derive_metrics_from_telemetry_data(data)
-                    # TODO: derive exception metrics from span events
+                # Derive metrics for quickpulse filtering
+                data = _TelemetryData._from_span(span)
+                _derive_metrics_from_telemetry_data(data)
+
+                # Process docs for quickpulse filtering
+                _apply_document_filters_from_telemetry_data(data)
+
+                # TODO: derive exception metrics from span events
             except Exception:  # pylint: disable=broad-except
                 _logger.exception("Exception occurred while recording span.")
 
@@ -210,31 +213,33 @@ def _record_log_record(self, log_data: LogData) -> None:
         if _is_post_state():
             try:
                 if log_data.log_record:
+                    exc_type = None
                     log_record = log_data.log_record
                     if log_record.attributes:
-                        document = _get_log_record_document(log_data)
-                        _append_quickpulse_document(document)
                         exc_type = log_record.attributes.get(SpanAttributes.EXCEPTION_TYPE)
                         exc_message = log_record.attributes.get(SpanAttributes.EXCEPTION_MESSAGE)
                         if exc_type is not None or exc_message is not None:
                             self._exception_rate_counter.add(1)
 
-                    metric_infos_dict = _get_quickpulse_derived_metric_infos()
-                    # check if filtering is enabled
-                    if metric_infos_dict:
-                        # Derive metrics for quickpulse filtering
-                        data = _TelemetryData._from_log_record(log_record)
-                        _derive_metrics_from_telemetry_data(data)
+                    # Derive metrics for quickpulse filtering
+                    data = _TelemetryData._from_log_record(log_record)
+                    _derive_metrics_from_telemetry_data(data)
+
+                    # Process docs for quickpulse filtering
+                    _apply_document_filters_from_telemetry_data(data, exc_type)  # type: ignore
             except Exception:  # pylint: disable=broad-except
                 _logger.exception("Exception occurred while recording log record.")
 
 
 # Filtering
 
-# Called by record_span/record_log when processing a span/log_record
+# Called by record_span/record_log when processing a span/log_record for metrics filtering
 # Derives metrics from projections if applicable to current filters in config
 def _derive_metrics_from_telemetry_data(data: _TelemetryData):
-    metric_infos_dict = _get_quickpulse_derived_metric_infos()
+    metric_infos_dict: Dict[TelemetryType, List[DerivedMetricInfo]] = _get_quickpulse_derived_metric_infos()
+    # if empty, filtering was not configured
+    if not metric_infos_dict:
+        return
     metric_infos = []  # type: ignore
     if isinstance(data, _RequestData):
         metric_infos = metric_infos_dict.get(TelemetryType.REQUEST)  # type: ignore
@@ -249,4 +254,44 @@ def _derive_metrics_from_telemetry_data(data: _TelemetryData):
         # generate filtered metrics
         _create_projections(metric_infos, data)
 
+
+# Called by record_span/record_log when processing a span/log_record for docs filtering
+# Finds doc stream Ids and their doc filter configurations
+def _apply_document_filters_from_telemetry_data(data: _TelemetryData, exc_type: Optional[str] = None):
+    doc_config_dict: Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]] = _get_quickpulse_doc_stream_infos()  # pylint: disable=C0301
+    stream_ids = set()
+    doc_config = {}  # type: ignore
+    if isinstance(data, _RequestData):
+        doc_config = doc_config_dict.get(TelemetryType.REQUEST, {})  # type: ignore
+    elif isinstance(data, _DependencyData):
+        doc_config = doc_config_dict.get(TelemetryType.DEPENDENCY, {})  # type: ignore
+    elif isinstance(data, _ExceptionData):
+        doc_config = doc_config_dict.get(TelemetryType.EXCEPTION, {})  # type: ignore
+    elif isinstance(data, _TraceData):
+        doc_config = doc_config_dict.get(TelemetryType.TRACE, {})  # type: ignore
+    for stream_id, filter_groups in doc_config.items():
+        for filter_group in filter_groups:
+            if _check_filters(filter_group.filters, data):
+                stream_ids.add(stream_id)
+                break
+
+    # We only append and send the document if either:
+    # 1. The document matched the filtering for a specific streamId
+    # 2. Filtering was not enabled for this telemetry type (empty doc_config)
+    if len(stream_ids) > 0 or not doc_config:
+        if type(data) in (_DependencyData, _RequestData):
+            document = _get_span_document(data)  # type: ignore
+        else:
+            document = _get_log_record_document(data, exc_type)  # type: ignore
+        # A stream (with a unique streamId) is relevant if there are multiple sources sending to the same
+        # ApplicationInsights instace with live metrics enabled
+        # Modify the document's streamIds to determine which stream to send to in post
+        # Note that the default case is that the list of document_stream_ids is empty, in which
+        # case no filtering is done for the telemetry type and it is sent to all streams
+        if stream_ids:
+            document.document_stream_ids = list(stream_ids)
+
+        # Add the generated document to be sent to quickpulse
+        _append_quickpulse_document(document)
+
 # cSpell:enable

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_state.py

@@ -15,6 +15,7 @@
     AggregationType,
     DerivedMetricInfo,
     DocumentIngress,
+    FilterConjunctionGroupInfo,
     TelemetryType,
 )
 
@@ -39,6 +40,7 @@ class _QuickpulseState(Enum):
 _QUICKPULSE_ETAG = ""
 _QUICKPULSE_DERIVED_METRIC_INFOS: Dict[TelemetryType, List[DerivedMetricInfo]] = {}
 _QUICKPULSE_PROJECTION_MAP: Dict[str, Tuple[AggregationType, float, int]] = {}
+_QUICKPULSE_DOC_STREAM_INFOS: Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]] = {}
 
 
 def _set_global_quickpulse_state(state: _QuickpulseState) -> None:
@@ -127,8 +129,8 @@ def _get_quickpulse_etag() -> str:
     return _QUICKPULSE_ETAG
 
 
-# Used for updating filter configuration when etag has changed
-# Contains filter and projection to apply for each telemetry type if exists
+# Used for updating metric filter configuration when etag has changed
+# Contains filter and projection of metrics to apply for each telemetry type if exists
 def _set_quickpulse_derived_metric_infos(filters: Dict[TelemetryType, List[DerivedMetricInfo]]) -> None:
     # pylint: disable=global-statement
     global _QUICKPULSE_DERIVED_METRIC_INFOS
@@ -175,3 +177,16 @@ def _clear_quickpulse_projection_map():
     # pylint: disable=global-variable-not-assigned
     global _QUICKPULSE_PROJECTION_MAP
     _QUICKPULSE_PROJECTION_MAP.clear()
+
+
+# Used for updating doc filter configuration when etag has changed
+# Contains filter and projection of docs to apply for each telemetry type if exists
+# Format is Dict[TelemetryType, Dict[stream.id, List[FilterConjunctionGroupInfo]]]
+def _set_quickpulse_doc_stream_infos(filters: Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]]) -> None:
+    # pylint: disable=global-statement
+    global _QUICKPULSE_DOC_STREAM_INFOS
+    _QUICKPULSE_DOC_STREAM_INFOS = filters
+
+
+def _get_quickpulse_doc_stream_infos() -> Dict[TelemetryType, Dict[str, List[FilterConjunctionGroupInfo]]]:
+    return _QUICKPULSE_DOC_STREAM_INFOS

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_types.py

@@ -94,6 +94,8 @@ def _from_span(span: ReadableSpan):
         attributes = {}
         dependency_type = "InProc"
         data = ""
+        if span.end_time and span.start_time:
+            duration_ms = (span.end_time - span.start_time) / 1e9
         if span.attributes:
             attributes = span.attributes
             target = trace_utils._get_target_for_dependency_from_peer(attributes)