tests

sdk/monitor/azure-monitor-opentelemetry-exporter/samples/logs/test.py

@@ -0,0 +1,84 @@
+import os
+import logging
+
+from opentelemetry._logs import (
+    get_logger_provider,
+    set_logger_provider,
+)
+from opentelemetry.sdk._logs import (
+    LoggerProvider,
+    LoggingHandler,
+)
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+
+from azure.monitor.opentelemetry.exporter import AzureMonitorLogExporter
+
+from opentelemetry import trace
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
+
+class CustomFilter(logging.Filter):
+
+    def filter(self, record):
+        if hasattr(record, "ignore"):
+            return False
+        return True
+
+
+class LogHandler:
+    def __init__(
+        self,
+        app_name: str,
+        app_version: str,
+        **kwargs,
+    ):
+        """_summary_
+
+        Args:
+            app_name: app name
+            app_version: app version
+            kwargs Additional logging parameters
+        """
+        self.app_name = app_name
+        self.app_version = app_version
+        self.kwargs = kwargs
+
+        # Create the logging provider
+        self.logger_provider = LoggerProvider()
+        set_logger_provider(self.logger_provider)
+        exporter = AzureMonitorLogExporter(
+            connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"],
+        )
+        self.logger_provider.add_log_record_processor(BatchLogRecordProcessor(exporter))
+
+        # Create the logging handler
+        self.handler = LoggingHandler()
+
+        # Attach LoggingHandler to root logger
+        self._logger = logging.getLogger(self.app_name)
+        self._logger.addHandler(self.handler)
+        self._logger.addFilter(CustomFilter())
+        self._logger.setLevel(logging.INFO)
+
+log_handler = LogHandler(app_name='app-name', app_version=1.0)
+logger = log_handler._logger
+
+
+exporter = AzureMonitorTraceExporter.from_connection_string(
+    os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"]
+)
+
+tracer_provider = TracerProvider()
+trace.set_tracer_provider(tracer_provider)
+tracer = trace.get_tracer(__name__)
+span_processor = BatchSpanProcessor(exporter, schedule_delay_millis=60000)
+trace.get_tracer_provider().add_span_processor(span_processor)
+
+with tracer.start_as_current_span(name="my-application") as application_span:
+    logger.info("test logs message")
+    logger.info("test ignore message", extra={"ignore": "true"})
+    logger.exception("test exception message")
+
+input()

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/quickpulse/test_live_metrics.py

@@ -77,3 +77,44 @@ def test_init(self, generator_mock):
         self.assertEqual(qpm._reader._base_monitoring_data_point, qpm._base_monitoring_data_point)
         self.assertTrue(isinstance(qpm._meter_provider, MeterProvider))
         self.assertEqual(qpm._meter_provider._sdk_config.metric_readers, [qpm._reader])
+
+
+    def test_singleton(self):
+        resource = Resource.create(
+            {
+                ResourceAttributes.SERVICE_INSTANCE_ID: "test_instance",
+                ResourceAttributes.SERVICE_NAME: "test_service",
+            }
+        )
+        part_a_fields = _populate_part_a_fields(resource)
+        resource2 = Resource.create(
+            {
+                ResourceAttributes.SERVICE_INSTANCE_ID: "test_instance2",
+                ResourceAttributes.SERVICE_NAME: "test_service2",
+            }
+        )
+        qpm = _QuickpulseManager(
+            connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ac;LiveEndpoint=https://eastus.livediagnostics.monitor.azure.com/",
+            resource=resource,
+        )
+        qpm2 = _QuickpulseManager(
+            connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ac;LiveEndpoint=https://eastus.livediagnostics.monitor.azure.com/",
+            resource=resource2,
+        )
+        self.assertEqual(qpm, qpm2)
+        self.assertEqual(
+            qpm._base_monitoring_data_point.instance,
+            part_a_fields.get(ContextTagKeys.AI_CLOUD_ROLE_INSTANCE, "")
+        )
+        self.assertEqual(
+            qpm._base_monitoring_data_point.role_name,
+            part_a_fields.get(ContextTagKeys.AI_CLOUD_ROLE, "")
+        )
+        self.assertEqual(
+            qpm2._base_monitoring_data_point.instance,
+            part_a_fields.get(ContextTagKeys.AI_CLOUD_ROLE_INSTANCE, "")
+        )
+        self.assertEqual(
+            qpm2._base_monitoring_data_point.role_name,
+            part_a_fields.get(ContextTagKeys.AI_CLOUD_ROLE, "")
+        )

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

@@ -1,8 +1,10 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import datetime
 import os
 import platform
+import time
 import unittest
 
 from azure.monitor.opentelemetry.exporter import _utils
@@ -44,6 +46,21 @@ def test_nanoseconds_to_duration(self):
         self.assertEqual(ns_to_duration(3600 * 1000000000), "0.01:00:00.000")
         self.assertEqual(ns_to_duration(86400 * 1000000000), "1.00:00:00.000")
 
+
+    @patch("time.time")
+    def test_ticks_since_dot_net_epoch(self, time_mock):
+        current_time = time.time()
+        shift_time = int(
+            (
+                datetime.datetime(1970, 1, 1, 0, 0, 0) -
+                datetime.datetime(1, 1, 1, 0, 0, 0)).total_seconds()
+            ) * (10 ** 7)
+        time_mock.return_value = current_time
+        ticks = _utils._ticks_since_dot_net_epoch()
+        expected_ticks = int(current_time * (10**7)) + shift_time
+        self.assertEqual(ticks, expected_ticks)
+
+
     def test_populate_part_a_fields(self):
         resource = Resource(
             {"service.name": "testServiceName",