Standard Metrics - Outgoing Requests Per Second (#729)

Author: lzchen

contrib/opencensus-ext-azure/README.rst

@@ -153,6 +153,7 @@ Below is a list of standard metrics that are currently available:
 
 - Available Memory (bytes)
 - CPU Processor Time (percentage)
+- Outgoing Request Rate (per second)
 - Process CPU Usage (percentage)
 - Process Private Bytes (bytes)
 

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

@@ -19,6 +19,7 @@
 from opencensus.common.schedule import Queue
 from opencensus.common.schedule import QueueEvent
 from opencensus.ext.azure.common import Options
+from opencensus.trace import execution_context
 
 
 class BaseExporter(object):
@@ -64,6 +65,8 @@ def __init__(self, src, dst):
         super(Worker, self).__init__()
 
     def run(self):  # pragma: NO COVER
+        # Indicate that this thread is an exporter thread.
+        execution_context.set_is_exporter(True)
         src = self.src
         dst = self.dst
         while True:

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

@@ -16,8 +16,6 @@
 import logging
 import requests
 
-from opencensus.trace import execution_context
-
 logger = logging.getLogger(__name__)
 
 
@@ -42,14 +40,6 @@ def _transmit(self, envelopes):
         Return the next retry time in seconds for retryable failure.
         This function should never throw exception.
         """
-        # TODO: prevent requests being tracked
-        blacklist_hostnames = execution_context.get_opencensus_attr(
-            'blacklist_hostnames',
-        )
-        execution_context.set_opencensus_attr(
-            'blacklist_hostnames',
-            ['dc.services.visualstudio.com'],
-        )
         try:
             response = requests.post(
                 url=self.options.endpoint,
@@ -64,11 +54,7 @@ def _transmit(self, envelopes):
             logger.warning('Transient client side error %s.', ex)
             # client side error (retryable)
             return self.options.minimum_retry_interval
-        finally:
-            execution_context.set_opencensus_attr(
-                'blacklist_hostnames',
-                blacklist_hostnames,
-            )
+
         text = 'N/A'
         data = None
         try:

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

@@ -28,6 +28,7 @@
 from opencensus.ext.azure.common.protocol import Message
 from opencensus.ext.azure.common.storage import LocalFileStorage
 from opencensus.ext.azure.common.transport import TransportMixin
+from opencensus.trace import execution_context
 
 logger = logging.getLogger(__name__)
 
@@ -76,6 +77,8 @@ def __init__(self, src, dst):
         )
 
     def run(self):
+        # Indicate that this thread is an exporter thread.
+        execution_context.set_is_exporter(True)
         src = self._src
         dst = self._dst
         while True:

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

@@ -27,7 +27,6 @@
 from opencensus.metrics import transport
 from opencensus.metrics.export.metric_descriptor import MetricDescriptorType
 from opencensus.stats import stats as stats_module
-from opencensus.trace import execution_context
 
 __all__ = ['MetricsExporter', 'new_metrics_exporter']
 
@@ -127,13 +126,6 @@ def _transmit_without_retry(self, envelopes):
         non-retryable failure, simply outputs result to logs.
         This function should never throw exception.
         """
-        blacklist_hostnames = execution_context.get_opencensus_attr(
-            'blacklist_hostnames',
-        )
-        execution_context.set_opencensus_attr(
-            'blacklist_hostnames',
-            ['dc.services.visualstudio.com'],
-        )
         try:
             response = requests.post(
                 url=self.options.endpoint,
@@ -148,11 +140,6 @@ def _transmit_without_retry(self, envelopes):
             # No retry policy, log output
             logger.warning('Transient client side error %s.', ex)
             return
-        finally:
-            execution_context.set_opencensus_attr(
-                'blacklist_hostnames',
-                blacklist_hostnames,
-            )
 
         text = 'N/A'
         data = None

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

@@ -16,6 +16,8 @@
 from opencensus.metrics.export.metric_producer import MetricProducer
 from opencensus.ext.azure.metrics_exporter.standard_metrics.cpu \
     import ProcessorTimeMetric
+from opencensus.ext.azure.metrics_exporter.standard_metrics.dependency \
+    import DependencyRateMetric
 from opencensus.ext.azure.metrics_exporter.standard_metrics.memory \
     import AvailableMemoryMetric
 from opencensus.ext.azure.metrics_exporter.standard_metrics.process \
@@ -24,8 +26,11 @@
     import ProcessMemoryMetric
 
 # List of standard metrics to track
-STANDARD_METRICS = [AvailableMemoryMetric, ProcessCPUMetric,
-                    ProcessMemoryMetric, ProcessorTimeMetric]
+STANDARD_METRICS = [AvailableMemoryMetric,
+                    DependencyRateMetric,
+                    ProcessCPUMetric,
+                    ProcessMemoryMetric,
+                    ProcessorTimeMetric]
 
 
 def register_metrics():

contrib/opencensus-ext-azure/opencensus/ext/azure/metrics_exporter/standard_metrics/dependency.py

@@ -0,0 +1,87 @@
+# 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 requests
+import time
+
+from opencensus.metrics.export.gauge import DerivedDoubleGauge
+from opencensus.trace import execution_context
+
+dependency_map = dict()
+ORIGINAL_REQUEST = requests.Session.request
+
+
+def dependency_patch(*args, **kwargs):
+    result = ORIGINAL_REQUEST(*args, **kwargs)
+    # Only collect request metric if sent from non-exporter thread
+    if not execution_context.is_exporter():
+        count = dependency_map.get('count', 0)
+        dependency_map['count'] = count + 1
+    return result
+
+
+def setup():
+    # Patch the requests library functions to track dependency information
+    requests.Session.request = dependency_patch
+
+
+class DependencyRateMetric(object):
+    # Dependency call metrics can be found under custom metrics
+    NAME = "\\ApplicationInsights\\Dependency Calls/Sec"
+
+    def __init__(self):
+        setup()
+
+    @staticmethod
+    def get_value():
+        current_count = dependency_map.get('count', 0)
+        current_time = time.time()
+        last_count = dependency_map.get('last_count', 0)
+        last_time = dependency_map.get('last_time')
+        last_result = dependency_map.get('last_result', 0)
+
+        try:
+            # last_time is None the very first time this function is called
+            if last_time is not None:
+                elapsed_seconds = current_time - last_time
+                interval_count = current_count - last_count
+                result = interval_count / elapsed_seconds
+            else:
+                result = 0
+            dependency_map['last_time'] = current_time
+            dependency_map['last_count'] = current_count
+            dependency_map['last_result'] = result
+            return result
+        except ZeroDivisionError:
+            # If elapsed_seconds is 0, exporter call made too close to previous
+            # Return the previous result if this is the case
+            return last_result
+
+    def __call__(self):
+        """ Returns a derived gauge for outgoing requests per second
+
+        Calculated by obtaining by getting the number of outgoing requests made
+        using the requests library within an elapsed time and dividing that
+        value over the elapsed time.
+
+        :rtype: :class:`opencensus.metrics.export.gauge.DerivedLongGauge`
+        :return: The gauge representing the available memory metric
+        """
+        gauge = DerivedDoubleGauge(
+            DependencyRateMetric.NAME,
+            'Outgoing Requests per second',
+            'rps',
+            [])
+        gauge.create_default_time_series(DependencyRateMetric.get_value)
+        return gauge

contrib/opencensus-ext-azure/tests/test_azure_standard_metrics.py

@@ -14,12 +14,21 @@
 
 import collections
 import mock
+import requests
 import unittest
 
 from opencensus.ext.azure.metrics_exporter import standard_metrics
+from opencensus.trace import execution_context
+
+ORIGINAL_FUNCTION = requests.Session.request
 
 
 class TestStandardMetrics(unittest.TestCase):
+    def setUp(self):
+        standard_metrics.dependency.dependency_map.clear()
+        requests.Session.request = ORIGINAL_FUNCTION
+        standard_metrics.dependency.ORIGINAL_REQUEST = ORIGINAL_FUNCTION
+
     @mock.patch('opencensus.ext.azure.metrics_exporter'
                 '.standard_metrics.register_metrics')
     def test_producer_ctor(self, avail_mock):
@@ -31,12 +40,12 @@ def test_producer_get_metrics(self):
         producer = standard_metrics.AzureStandardMetricsProducer()
         metrics = producer.get_metrics()
 
-        self.assertEqual(len(metrics), 4)
+        self.assertEqual(len(metrics), 5)
 
     def test_register_metrics(self):
         registry = standard_metrics.register_metrics()
 
-        self.assertEqual(len(registry.get_metrics()), 4)
+        self.assertEqual(len(registry.get_metrics()), 5)
 
     def test_get_available_memory_metric(self):
         metric = standard_metrics.AvailableMemoryMetric()
@@ -124,3 +133,55 @@ def test_get_process_cpu_usage_exception(self, logger_mock):
             standard_metrics.ProcessCPUMetric.get_value()
 
             logger_mock.exception.assert_called()
+
+    def test_dependency_patch(self):
+        map = standard_metrics.dependency.dependency_map
+        standard_metrics.dependency.ORIGINAL_REQUEST = lambda x: None
+        session = requests.Session()
+        execution_context.set_is_exporter(False)
+        result = standard_metrics.dependency.dependency_patch(session)
+
+        self.assertEqual(map['count'], 1)
+        self.assertIsNone(result)
+
+    def test_dependency_patch_exporter_thread(self):
+        map = standard_metrics.dependency.dependency_map
+        standard_metrics.dependency.ORIGINAL_REQUEST = lambda x: None
+        session = mock.Mock()
+        execution_context.set_is_exporter(True)
+        result = standard_metrics.dependency.dependency_patch(session)
+
+        self.assertIsNone(map.get('count'))
+        self.assertIsNone(result)
+
+    def test_get_dependency_rate_metric(self):
+        metric = standard_metrics.DependencyRateMetric()
+        gauge = metric()
+
+        self.assertEqual(gauge.descriptor.name,
+                         '\\ApplicationInsights\\Dependency Calls/Sec')
+
+    def test_get_dependency_rate_first_time(self):
+        rate = standard_metrics.DependencyRateMetric.get_value()
+
+        self.assertEqual(rate, 0)
+
+    @mock.patch('opencensus.ext.azure.metrics_exporter'
+                '.standard_metrics.dependency.time')
+    def test_get_dependency_rate(self, time_mock):
+        time_mock.time.return_value = 100
+        standard_metrics.dependency.dependency_map['last_time'] = 98
+        standard_metrics.dependency.dependency_map['count'] = 4
+        rate = standard_metrics.DependencyRateMetric.get_value()
+
+        self.assertEqual(rate, 2)
+
+    @mock.patch('opencensus.ext.azure.metrics_exporter'
+                '.standard_metrics.dependency.time')
+    def test_get_dependency_rate_error(self, time_mock):
+        time_mock.time.return_value = 100
+        standard_metrics.dependency.dependency_map['last_result'] = 5
+        standard_metrics.dependency.dependency_map['last_time'] = 100
+        result = standard_metrics.DependencyRateMetric.get_value()
+
+        self.assertEqual(result, 5)

contrib/opencensus-ext-httplib/opencensus/ext/httplib/trace.py

@@ -61,6 +61,10 @@ def wrap_httplib_request(request_func):
     """
 
     def call(self, method, url, body, headers, *args, **kwargs):
+        # Check if request was sent from an exporter. If so, do not wrap.
+        if execution_context.is_exporter():
+            return request_func(self, method, url, body,
+                                headers, *args, **kwargs)
         _tracer = execution_context.get_opencensus_tracer()
         blacklist_hostnames = execution_context.get_opencensus_attr(
             'blacklist_hostnames')
@@ -100,6 +104,9 @@ def wrap_httplib_response(response_func):
     """
 
     def call(self, *args, **kwargs):
+        # Check if request was sent from an exporter. If so, do not wrap.
+        if execution_context.is_exporter():
+            return response_func(self, *args, **kwargs)
         _tracer = execution_context.get_opencensus_tracer()
         current_span_id = execution_context.get_opencensus_attr(
             'httplib/current_span_id')