Attach rate metrics via Heartbeat for Azure exporter (#930)

Author: lzchen

contrib/opencensus-ext-azure/examples/traces/simple.py

@@ -23,3 +23,4 @@
 
 with tracer.span(name='foo'):
     print('Hello, World!')
+input(...)

contrib/opencensus-ext-azure/opencensus/ext/azure/common/exporter.py

@@ -61,7 +61,9 @@ def __init__(self, src, dst):
         self.src = src
         self.dst = dst
         self._stopping = False
-        super(Worker, self).__init__()
+        super(Worker, self).__init__(
+            name="AzureExporter Worker"
+        )
 
     def run(self):  # pragma: NO COVER
         # Indicate that this thread is an exporter thread.

contrib/opencensus-ext-azure/opencensus/ext/azure/common/storage.py

@@ -81,6 +81,7 @@ def __init__(
             maintenance_period=60,  # 1 minute
             retention_period=7*24*60*60,  # 7 days
             write_timeout=60,  # 1 minute
+            source=None,
     ):
         self.path = os.path.abspath(path)
         self.max_size = max_size
@@ -92,6 +93,7 @@ def __init__(
         self._maintenance_task = PeriodicTask(
             interval=self.maintenance_period,
             function=self._maintenance_routine,
+            name='{} Storage Worker'.format(source)
         )
         self._maintenance_task.daemon = True
         self._maintenance_task.start()

contrib/opencensus-ext-azure/opencensus/ext/azure/log_exporter/__init__.py

@@ -30,6 +30,7 @@
 )
 from opencensus.ext.azure.common.storage import LocalFileStorage
 from opencensus.ext.azure.common.transport import TransportMixin
+from opencensus.ext.azure.metrics_exporter import heartbeat_metrics
 from opencensus.trace import execution_context
 
 logger = logging.getLogger(__name__)
@@ -52,12 +53,15 @@ def __init__(self, **options):
             max_size=self.options.storage_max_size,
             maintenance_period=self.options.storage_maintenance_period,
             retention_period=self.options.storage_retention_period,
+            source=self.__class__.__name__,
         )
         self._telemetry_processors = []
         self.addFilter(SamplingFilter(self.options.logging_sampling_rate))
         self._queue = Queue(capacity=8192)  # TODO: make this configurable
         self._worker = Worker(self._queue, self)
         self._worker.start()
+        heartbeat_metrics.enable_heartbeat_metrics(
+            self.options.connection_string, self.options.instrumentation_key)
 
     def _export(self, batch, event=None):  # pragma: NO COVER
         try:

contrib/opencensus-ext-azure/opencensus/ext/azure/metrics_exporter/__init__.py

@@ -52,7 +52,9 @@ def __init__(self, **options):
             max_size=self.options.storage_max_size,
             maintenance_period=self.options.storage_maintenance_period,
             retention_period=self.options.storage_retention_period,
+            source=self.__class__.__name__,
         )
+        self._atexit_handler = atexit.register(self.shutdown)
         super(MetricsExporter, self).__init__()
 
     def export_metrics(self, metrics):
@@ -133,6 +135,13 @@ def _create_envelope(self, data_point, timestamp, properties):
         envelope.data = Data(baseData=data, baseType="MetricData")
         return envelope
 
+    def shutdown(self):
+        # flush metrics on exit
+        self.export_metrics(stats_module.stats.get_metrics())
+        if self._atexit_handler is not None:
+            atexit.unregister(self._atexit_handler)
+            self._atexit_handler = None
+
 
 def new_metrics_exporter(**options):
     exporter = MetricsExporter(**options)
@@ -142,5 +151,9 @@ def new_metrics_exporter(**options):
     transport.get_exporter_thread(producers,
                                   exporter,
                                   interval=exporter.options.export_interval)
-    atexit.register(exporter.export_metrics, stats_module.stats.get_metrics())
+    from opencensus.ext.azure.metrics_exporter import heartbeat_metrics
+    heartbeat_metrics.enable_heartbeat_metrics(
+        exporter.options.connection_string,
+        exporter.options.instrumentation_key
+    )
     return exporter

contrib/opencensus-ext-azure/opencensus/ext/azure/metrics_exporter/heartbeat_metrics/__init__.py

@@ -0,0 +1,62 @@
+# Copyright 2020, OpenCensus Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import threading
+
+from opencensus.ext.azure.metrics_exporter import MetricsExporter
+from opencensus.ext.azure.metrics_exporter.heartbeat_metrics.heartbeat import (
+    HeartbeatMetric,
+)
+from opencensus.metrics import transport
+from opencensus.metrics.export.gauge import Registry
+from opencensus.metrics.export.metric_producer import MetricProducer
+
+_HEARTBEAT_METRICS = None
+_HEARTBEAT_LOCK = threading.Lock()
+
+
+def enable_heartbeat_metrics(connection_string, ikey):
+    with _HEARTBEAT_LOCK:
+        # Only start heartbeat if did not exist before
+        global _HEARTBEAT_METRICS  # pylint: disable=global-statement
+        if _HEARTBEAT_METRICS is None:
+            exporter = MetricsExporter(
+                connection_string=connection_string,
+                instrumentation_key=ikey,
+                export_interval=900.0,  # Send every 15 minutes
+            )
+            producer = AzureHeartbeatMetricsProducer()
+            _HEARTBEAT_METRICS = producer
+            transport.get_exporter_thread([_HEARTBEAT_METRICS],
+                                          exporter,
+                                          exporter.options.export_interval)
+
+
+def register_metrics():
+    registry = Registry()
+    metric = HeartbeatMetric()
+    registry.add_gauge(metric())
+    return registry
+
+
+class AzureHeartbeatMetricsProducer(MetricProducer):
+    """Implementation of the producer of heartbeat metrics.
+
+    Includes Azure attach rate metrics, implemented using gauges.
+    """
+    def __init__(self):
+        self.registry = register_metrics()
+
+    def get_metrics(self):
+        return self.registry.get_metrics()

contrib/opencensus-ext-azure/opencensus/ext/azure/metrics_exporter/heartbeat_metrics/heartbeat.py

@@ -0,0 +1,64 @@
+# Copyright 2019, OpenCensus Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import platform
+from collections import OrderedDict
+
+from opencensus.common.version import __version__ as opencensus_version
+from opencensus.ext.azure.common.version import __version__ as ext_version
+from opencensus.metrics.export.gauge import LongGauge
+from opencensus.metrics.label_key import LabelKey
+from opencensus.metrics.label_value import LabelValue
+
+
+class HeartbeatMetric:
+    NAME = "Heartbeat"
+
+    def __init__(self):
+        self.properties = OrderedDict()
+        self.properties[LabelKey("sdk", '')] = LabelValue(
+            'py{}:oc{}:ext{}'.format(
+                platform.python_version(),
+                opencensus_version,
+                ext_version,
+            )
+        )
+        self.properties[LabelKey("osType", '')] = LabelValue(platform.system())
+        if os.environ.get("WEBSITE_SITE_NAME") is not None:
+            # Web apps
+            self.properties[LabelKey("appSrv_SiteName", '')] = \
+                LabelValue(os.environ.get("WEBSITE_SITE_NAME"))
+            self.properties[LabelKey("appSrv_wsStamp", '')] = \
+                LabelValue(os.environ.get("WEBSITE_HOME_STAMPNAME", ''))
+            self.properties[LabelKey("appSrv_wsHost", '')] = \
+                LabelValue(os.environ.get("WEBSITE_HOSTNAME", ''))
+        elif os.environ.get("FUNCTIONS_WORKER_RUNTIME") is not None:
+            # Function apps
+            self.properties[LabelKey("azfunction_appId", '')] = \
+                LabelValue(os.environ.get("WEBSITE_HOSTNAME"))
+
+    def __call__(self):
+        """ Returns a derived gauge for the heartbeat metric.
+
+        :rtype: :class:`opencensus.metrics.export.gauge.LongGauge`
+        :return: The gauge representing the heartbeat metric
+        """
+        gauge = LongGauge(
+            HeartbeatMetric.NAME,
+            'Heartbeat metric with custom dimensions',
+            'count',
+            list(self.properties.keys()))
+        gauge.get_or_create_time_series(list(self.properties.values()))
+        return gauge

contrib/opencensus-ext-azure/opencensus/ext/azure/trace_exporter/__init__.py

@@ -26,6 +26,7 @@
 )
 from opencensus.ext.azure.common.storage import LocalFileStorage
 from opencensus.ext.azure.common.transport import TransportMixin
+from opencensus.ext.azure.metrics_exporter import heartbeat_metrics
 from opencensus.trace.span import SpanKind
 
 try:
@@ -52,9 +53,12 @@ def __init__(self, **options):
             max_size=self.options.storage_max_size,
             maintenance_period=self.options.storage_maintenance_period,
             retention_period=self.options.storage_retention_period,
+            source=self.__class__.__name__,
         )
         self._telemetry_processors = []
         super(AzureExporter, self).__init__(**options)
+        heartbeat_metrics.enable_heartbeat_metrics(
+            self.options.connection_string, self.options.instrumentation_key)
 
     def span_data_to_envelope(self, sd):
         envelope = Envelope(