[Monitor OpenTelemetry Exporter] Add Check for Synthetic Source (Azure#41733)

* Add check for user-agent and synthetic source.

* Update CHANGELOG.md

* Update CHANGELOG.md

* Fix typing issue.

* Update _utils.py

* Update CHANGELOG.md

* Update CHANGELOG.md

---------

Co-authored-by: Leighton Chen <[email protected]>

Author: JacksonWeber

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

@@ -4,6 +4,9 @@
 
 ### Features Added
 
+- Detect synthetically created telemetry based on the user-agent header
+  ([#41733](https://github.com/Azure/azure-sdk-for-python/pull/41733))
+
 ### Breaking Changes
 
 ### Bugs Fixed

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

@@ -9,7 +9,7 @@
 import threading
 import time
 import warnings
-from typing import Callable, Dict, Any
+from typing import Callable, Dict, Any, Optional
 
 from opentelemetry.semconv.resource import ResourceAttributes
 from opentelemetry.sdk.resources import Resource
@@ -292,12 +292,50 @@ def _get_cloud_role_instance(resource: Resource) -> str:
     return platform.node()  # hostname default
 
 
-def _is_synthetic_source(properties: Attributes) -> bool:
+def _is_synthetic_source(properties: Optional[Any]) -> bool:
     # TODO: Use semconv symbol when released in upstream
+    if not properties:
+        return False
     synthetic_type = properties.get("user_agent.synthetic.type")  # type: ignore
     return synthetic_type in ("bot", "test")
 
 
+def _is_synthetic_load(properties: Optional[Any]) -> bool:
+    """
+    Check if the request is from a synthetic load test by examining the HTTP user agent.
+
+    :param properties: The attributes/properties to check for user agent information
+    :type properties: Optional[Any]
+    :return: True if the user agent contains "AlwaysOn", False otherwise
+    :rtype: bool
+    """
+    if not properties:
+        return False
+
+    # Check both old and new semantic convention attributes for HTTP user agent
+    user_agent = (
+        properties.get("user_agent.original") or  # type: ignore  # New semantic convention
+        properties.get("http.user_agent")  # type: ignore  # Legacy semantic convention
+    )
+
+    if user_agent and isinstance(user_agent, str):
+        return "AlwaysOn" in user_agent
+
+    return False
+
+
+def _is_any_synthetic_source(properties: Optional[Any]) -> bool:
+    """
+    Check if the telemetry should be marked as synthetic from any source.
+
+    :param properties: The attributes/properties to check
+    :type properties: Optional[Any]
+    :return: True if any synthetic source is detected, False otherwise
+    :rtype: bool
+    """
+    return _is_synthetic_source(properties) or _is_synthetic_load(properties)
+
+
 # pylint: disable=W0622
 def _filter_custom_properties(properties: Attributes, filter=None) -> Dict[str, str]:
     truncated_properties: Dict[str, str] = {}

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/logs/_exporter.py

@@ -126,6 +126,8 @@ def _convert_log_to_envelope(log_data: LogData) -> TelemetryItem:
     envelope.tags[ContextTagKeys.AI_OPERATION_PARENT_ID] = "{:016x}".format(  # type: ignore
         log_record.span_id or _DEFAULT_SPAN_ID
     )
+    if _utils._is_any_synthetic_source(log_record.attributes):
+        envelope.tags[ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE] = "True"  # type: ignore
     # Special use case: Customers want to be able to set location ip on log records
     location_ip = trace_utils._get_location_ip(log_record.attributes)
     if location_ip:

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/metrics/_exporter.py

@@ -217,7 +217,7 @@ def _convert_point_to_envelope(
     envelope = _utils._create_telemetry_item(point.time_unix_nano)
     envelope.name = _METRIC_ENVELOPE_NAME
     envelope.tags.update(_utils._populate_part_a_fields(resource))  # type: ignore
-    if _utils._is_synthetic_source(point.attributes):
+    if _utils._is_any_synthetic_source(point.attributes):
         envelope.tags[ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE] = "True"  # type: ignore
     namespace = None
     if scope is not None and _is_metric_namespace_opted_in():

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/trace/_exporter.py

@@ -221,7 +221,7 @@ def _convert_span_to_envelope(span: ReadableSpan) -> TelemetryItem:
     envelope.tags[ContextTagKeys.AI_OPERATION_ID] = "{:032x}".format(span.context.trace_id)
     if SpanAttributes.ENDUSER_ID in span.attributes:
         envelope.tags[ContextTagKeys.AI_USER_ID] = span.attributes[SpanAttributes.ENDUSER_ID]
-    if _utils._is_synthetic_source(span.attributes):
+    if _utils._is_any_synthetic_source(span.attributes):
         envelope.tags[ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE] = "True"
     if span.parent and span.parent.span_id:
         envelope.tags[ContextTagKeys.AI_OPERATION_PARENT_ID] = "{:016x}".format(span.parent.span_id)

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/logs/test_logs.py

@@ -552,6 +552,70 @@ def test_log_to_envelope_timestamp(self):
         self.assertEqual(envelope.time, ns_to_iso_str(record.observed_timestamp))
         self._log_data.log_record = old_record
 
+    def test_log_to_envelope_synthetic_source(self):
+        exporter = self._exporter
+        resource = Resource.create(
+            {
+                "service.name": "testServiceName",
+                "service.namespace": "testServiceNamespace",
+                "service.instance.id": "testServiceInstanceId",
+            }
+        )
+        log_data = _logs.LogData(
+            _logs.LogRecord(
+                timestamp=1646865018558419456,
+                trace_id=125960616039069540489478540494783893221,
+                span_id=2909973987304607650,
+                severity_text="WARNING",
+                trace_flags=None,
+                severity_number=SeverityNumber.WARN,
+                body="Test message",
+                resource=resource,
+                attributes={
+                    "test": "attribute",
+                    "user_agent.synthetic.type": "bot",
+                },
+            ),
+            InstrumentationScope("test_name"),
+        )
+        envelope = exporter._log_to_envelope(log_data)
+
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE), "True")
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_CLOUD_ROLE), "testServiceNamespace.testServiceName")
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_CLOUD_ROLE_INSTANCE), "testServiceInstanceId")
+
+    def test_log_to_envelope_synthetic_load_always_on(self):
+        exporter = self._exporter
+        resource = Resource.create(
+            {
+                "service.name": "testServiceName",
+                "service.namespace": "testServiceNamespace",
+                "service.instance.id": "testServiceInstanceId",
+            }
+        )
+        log_data = _logs.LogData(
+            _logs.LogRecord(
+                timestamp=1646865018558419456,
+                trace_id=125960616039069540489478540494783893221,
+                span_id=2909973987304607650,
+                severity_text="WARNING",
+                trace_flags=None,
+                severity_number=SeverityNumber.WARN,
+                body="Test message",
+                resource=resource,
+                attributes={
+                    "test": "attribute",
+                    "http.user_agent": "Azure-Load-Testing/1.0 AlwaysOn",
+                },
+            ),
+            InstrumentationScope("test_name"),
+        )
+        envelope = exporter._log_to_envelope(log_data)
+
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_OPERATION_SYNTHETIC_SOURCE), "True")
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_CLOUD_ROLE), "testServiceNamespace.testServiceName")
+        self.assertEqual(envelope.tags.get(ContextTagKeys.AI_CLOUD_ROLE_INSTANCE), "testServiceInstanceId")
+
 
 class TestAzureLogExporterWithDisabledStorage(TestAzureLogExporter):
     _exporter_class = partial(AzureMonitorLogExporter, disable_offline_storage=True)

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_utils.py

@@ -586,3 +586,42 @@ def test_is_synthetic_source_none(self):
     def test_is_synthetic_source_other(self):
         properties = {"user_agent.synthetic.type": "user"}
         self.assertFalse(_utils._is_synthetic_source(properties))
+
+    def test_is_synthetic_load_always_on_legacy(self):
+        properties = {"http.user_agent": "Mozilla/5.0 AlwaysOn"}
+        self.assertTrue(_utils._is_synthetic_load(properties))
+
+    def test_is_synthetic_load_always_on_new_convention(self):
+        properties = {"user_agent.original": "Azure-Load-Testing/1.0 AlwaysOn"}
+        self.assertTrue(_utils._is_synthetic_load(properties))
+
+    def test_is_synthetic_load_always_on_case_sensitive(self):
+        properties = {"http.user_agent": "Mozilla/5.0 alwayson"}
+        self.assertFalse(_utils._is_synthetic_load(properties))
+
+    def test_is_synthetic_load_no_user_agent(self):
+        properties = {}
+        self.assertFalse(_utils._is_synthetic_load(properties))
+
+    def test_is_synthetic_load_normal_user_agent(self):
+        properties = {"http.user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
+        self.assertFalse(_utils._is_synthetic_load(properties))
+
+    def test_is_any_synthetic_source_bot(self):
+        properties = {"user_agent.synthetic.type": "bot"}
+        self.assertTrue(_utils._is_any_synthetic_source(properties))
+
+    def test_is_any_synthetic_source_always_on(self):
+        properties = {"http.user_agent": "Azure-Load-Testing/1.0 AlwaysOn"}
+        self.assertTrue(_utils._is_any_synthetic_source(properties))
+
+    def test_is_any_synthetic_source_both(self):
+        properties = {
+            "user_agent.synthetic.type": "bot",
+            "http.user_agent": "Azure-Load-Testing/1.0 AlwaysOn"
+        }
+        self.assertTrue(_utils._is_any_synthetic_source(properties))
+
+    def test_is_any_synthetic_source_none(self):
+        properties = {"http.user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
+        self.assertFalse(_utils._is_any_synthetic_source(properties))