Fix statsbeat bugs (#1113)

Author: lzchen

contrib/opencensus-ext-azure/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Statsbeat bug fixes
+([#1113](https://github.com/census-instrumentation/opencensus-python/pull/1113))
+
 ## 1.1.3
 Released 2022-03-03
 

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

@@ -21,3 +21,5 @@
 # and place it in the APPLICATIONINSIGHTS_CONNECTION_STRING
 # environment variable.
 logger.addHandler(AzureLogHandler())
+
+logger.warning("Hello World!")

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

@@ -94,9 +94,6 @@ def _transmit(self, envelopes):
         except requests.RequestException as ex:
             logger.warning(
                 'Retrying due to transient client side error %s.', ex)
-            if self._check_stats_collection():
-                with _requests_lock:
-                    _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
             # client side error (retryable)
             exception = self.options.minimum_retry_interval
         except CredentialUnavailableError as ex:
@@ -108,9 +105,6 @@ def _transmit(self, envelopes):
         except Exception as ex:
             logger.warning(
                 'Error when sending request %s. Dropping telemetry.', ex)
-            if self._check_stats_collection():
-                with _requests_lock:
-                    _requests_map['exception'] = _requests_map.get('exception', 0) + 1  # noqa: E501
             # Extraneous error (non-retryable)
             exception = -1
         finally:
@@ -120,6 +114,13 @@ def _transmit(self, envelopes):
                     duration = _requests_map.get('duration', 0)
                     _requests_map['duration'] = duration + (end_time - start_time)  # noqa: E501
             if exception is not None:
+                if self._check_stats_collection():
+                    with _requests_lock:
+                        if exception >= 0:
+                            _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
+                        else:
+                            _requests_map['exception'] = _requests_map.get('exception', 0) + 1  # noqa: E501
+
                 return exception
 
         text = 'N/A'
@@ -139,10 +140,11 @@ def _transmit(self, envelopes):
                 with _requests_lock:
                     _requests_map['success'] = _requests_map.get('success', 0) + 1  # noqa: E501
             return 0
-        # Status code not 200 counts as failure
+        # Status code not 200, 439 or 402 counts as failures
         if self._check_stats_collection():
-            with _requests_lock:
-                _requests_map['failure'] = _requests_map.get('failure', 0) + 1  # noqa: E501
+            if response.status_code != 439 and response.status_code != 402:
+                with _requests_lock:
+                    _requests_map['failure'] = _requests_map.get('failure', 0) + 1  # noqa: E501
         if response.status_code == 206:  # Partial Content
             if data:
                 try:
@@ -162,6 +164,9 @@ def _transmit(self, envelopes):
                                 envelopes[error['index']],
                             )
                     if resend_envelopes:
+                        if self._check_stats_collection():
+                            with _requests_lock:
+                                _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
                         self.storage.put(resend_envelopes)
                 except Exception as ex:
                     logger.error(
@@ -170,9 +175,6 @@ def _transmit(self, envelopes):
                         text,
                         ex,
                     )
-                if self._check_stats_collection():
-                    with _requests_lock:
-                        _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
                 return -response.status_code
             # cannot parse response body, fallback to retry
         if response.status_code in (
@@ -189,11 +191,7 @@ def _transmit(self, envelopes):
             # server side error (retryable)
             if self._check_stats_collection():
                 with _requests_lock:
-                    # 429 counts as throttle instead of retry
-                    if response.status_code == 429:
-                        _requests_map['throttle'] = _requests_map.get('throttle', 0) + 1  # noqa: E501
-                    else:
-                        _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
+                    _requests_map['retry'] = _requests_map.get('retry', 0) + 1  # noqa: E501
             return self.options.minimum_retry_interval
         # Authentication error
         if response.status_code == 401:
@@ -235,14 +233,25 @@ def _transmit(self, envelopes):
                 logger.error(
                     "Error parsing redirect information."
                 )
-            logger.error(
-                "Error sending telemetry because of circular redirects."
-                " Please check the integrity of your connection string."
-            )
+            else:
+                logger.error(
+                    "Error sending telemetry because of circular redirects."
+                    " Please check the integrity of your connection string."
+                )
+            # If redirect but did not return, exception occured
+            if self._check_stats_collection():
+                with _requests_lock:
+                    _requests_map['exception'] = _requests_map.get('exception', 0) + 1  # noqa: E501
+        # Other, server side error (non-retryable)
         logger.error(
             'Non-retryable server side error %s: %s.',
             response.status_code,
             text,
         )
-        # server side error (non-retryable)
+        if self._check_stats_collection():
+            if response.status_code == 402 or response.status_code == 439:
+                # 402: Monthly Quota Exceeded (new SDK)
+                # 439: Monthly Quota Exceeded (old SDK) <- Currently OC SDK
+                with _requests_lock:
+                    _requests_map['throttle'] = _requests_map.get('throttle', 0) + 1  # noqa: E501
         return -response.status_code

contrib/opencensus-ext-azure/opencensus/ext/azure/metrics_exporter/statsbeat_metrics/statsbeat.py

@@ -150,27 +150,24 @@ def _get_feature_properties():
 
 def _get_success_count_value():
     with _requests_lock:
-        interval_count = _requests_map.get('success', 0) \
-                    - _requests_map.get('last_success', 0)
-        _requests_map['last_success'] = _requests_map.get('success', 0)
+        interval_count = _requests_map.get('success', 0)
+        _requests_map['success'] = 0
         return interval_count
 
 
 def _get_failure_count_value():
     with _requests_lock:
-        interval_count = _requests_map.get('failure', 0) \
-                    - _requests_map.get('last_failure', 0)
-        _requests_map['last_failure'] = _requests_map.get('failure', 0)
+        interval_count = _requests_map.get('failure', 0)
+        _requests_map['failure'] = 0
         return interval_count
 
 
 def _get_average_duration_value():
     with _requests_lock:
-        interval_duration = _requests_map.get('duration', 0) \
-            - _requests_map.get('last_duration', 0)
-        interval_count = _requests_map.get('count', 0) \
-            - _requests_map.get('last_count', 0)
-        _requests_map['last_duration'] = _requests_map.get('duration', 0)
+        interval_duration = _requests_map.get('duration', 0)
+        interval_count = _requests_map.get('count', 0)
+        _requests_map['duration'] = 0
+        _requests_map['count'] = 0
         if interval_duration > 0 and interval_count > 0:
             result = interval_duration / interval_count
             # Convert to milliseconds
@@ -180,25 +177,22 @@ def _get_average_duration_value():
 
 def _get_retry_count_value():
     with _requests_lock:
-        interval_count = _requests_map.get('retry', 0) \
-                    - _requests_map.get('last_retry', 0)
-        _requests_map['last_retry'] = _requests_map.get('retry', 0)
+        interval_count = _requests_map.get('retry', 0)
+        _requests_map['retry'] = 0
         return interval_count
 
 
 def _get_throttle_count_value():
     with _requests_lock:
-        interval_count = _requests_map.get('throttle', 0) \
-                    - _requests_map.get('last_throttle', 0)
-        _requests_map['last_throttle'] = _requests_map.get('throttle', 0)
+        interval_count = _requests_map.get('throttle', 0)
+        _requests_map['throttle'] = 0
         return interval_count
 
 
 def _get_exception_count_value():
     with _requests_lock:
-        interval_count = _requests_map.get('exception', 0) \
-                    - _requests_map.get('last_exception', 0)
-        _requests_map['last_exception'] = _requests_map.get('exception', 0)
+        interval_count = _requests_map.get('exception', 0)
+        _requests_map['exception'] = 0
         return interval_count
 
 

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

@@ -337,10 +337,46 @@ def test_get_common_properties(self):
 
     def test_get_success_count_value(self):
         _requests_map.clear()
-        _requests_map['last_success'] = 5
         _requests_map['success'] = 10
-        self.assertEqual(_get_success_count_value(), 5)
-        self.assertEqual(_requests_map['last_success'], 10)
+        self.assertEqual(_get_success_count_value(), 10)
+        self.assertEqual(_requests_map['success'], 0)
+        _requests_map.clear()
+
+    def test_get_failure_count_value(self):
+        _requests_map.clear()
+        _requests_map['failure'] = 10
+        self.assertEqual(_get_failure_count_value(), 10)
+        self.assertEqual(_requests_map['failure'], 0)
+        _requests_map.clear()
+
+    def test_get_average_duration_value(self):
+        _requests_map.clear()
+        _requests_map['duration'] = 10
+        _requests_map['count'] = 2
+        self.assertEqual(_get_average_duration_value(), 5000.0)
+        self.assertEqual(_requests_map['duration'], 0)
+        self.assertEqual(_requests_map['count'], 0)
+        _requests_map.clear()
+
+    def test_get_retry_count_value(self):
+        _requests_map.clear()
+        _requests_map['retry'] = 10
+        self.assertEqual(_get_retry_count_value(), 10)
+        self.assertEqual(_requests_map['retry'], 0)
+        _requests_map.clear()
+
+    def test_get_throttle_count_value(self):
+        _requests_map.clear()
+        _requests_map['throttle'] = 10
+        self.assertEqual(_get_throttle_count_value(), 10)
+        self.assertEqual(_requests_map['throttle'], 0)
+        _requests_map.clear()
+
+    def test_get_exception_count_value(self):
+        _requests_map.clear()
+        _requests_map['exception'] = 10
+        self.assertEqual(_get_exception_count_value(), 10)
+        self.assertEqual(_requests_map['exception'], 0)
         _requests_map.clear()
 
     def test_statsbeat_metric_get_initial_metrics(self):