Merge pull request #88 from michi88/fix-mutating-metrics

Fix metrics being mutated during report

Author: keep94

.travis.yml

@@ -19,9 +19,7 @@ before_script:
   - python -m pydocstyle
 script:
   - python -m unittest discover
-  - if [[ "${TRAVIS_PULL_REQUEST}" != "false" ]]; then
-      test -n "$(git diff -G version= $TRAVIS_COMMIT_RANGE setup.py)";
-    fi
+
 deploy:
   - provider: pypi
     user: wavefront-cs

wavefront_sdk/client.py

@@ -92,8 +92,9 @@ def __init__(self, server, token, max_queue_size=50000, batch_size=10000,
         if enable_internal_metrics:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=self,
-                prefix='{}.core.sender.{}'.format(
-                    constants.SDK_METRIC_PREFIX, ingestion_type))
+                prefix=f'{constants.SDK_METRIC_PREFIX}'
+                       f'.core.sender.{ingestion_type}'
+            )
         else:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=None)
@@ -198,8 +199,8 @@ def _report(self, points, data_format, entity_prefix, report_errors):
                                          params=params, headers=self._headers,
                                          data=compressed_data)
 
-            self._sdk_metrics_registry.new_delta_counter('{}.report.{}'.format(
-                entity_prefix, response.status_code)).inc()
+            self._sdk_metrics_registry.new_delta_counter(
+                f'{entity_prefix}.report.{response.status_code}').inc()
             response.raise_for_status()
         except Exception as error:
             report_errors.inc()

wavefront_sdk/client_factory.py

@@ -43,12 +43,12 @@ def get_server_info_from_endpoint(self, url):
         base_url = urlparse(url)
         scheme = base_url.scheme
         if scheme == self.DIRECT_DATA_INGESTION_SCHEME:
-            server = '{}://{}'.format(self.DIRECT_DATA_INGESTION_SCHEME,
-                                      base_url.hostname)
+            server = f'{self.DIRECT_DATA_INGESTION_SCHEME}://' \
+                     f'{base_url.hostname}'
             token = base_url.username
         elif scheme in (self.PROXY_SCHEME, self.HTTP_PROXY_SCHEME):
-            server = '{}://{}:{}'.format(self.HTTP_PROXY_SCHEME,
-                                         base_url.hostname, base_url.port)
+            server = f'{self.HTTP_PROXY_SCHEME}://' \
+                     f'{base_url.hostname}:{base_url.port}'
             token = None
         else:
             raise RuntimeError("Unknown scheme specified while attempting to"

wavefront_sdk/common/constants.py

@@ -13,10 +13,10 @@
 WAVEFRONT_TRACING_SPAN_FORMAT = 'trace'
 
 # GREEK LETTER DELTA.
-DELTA_PREFIX = u'\u2206'
+DELTA_PREFIX = '\u2206'
 
 # GREEK CAPITAL LETTER DELTA.
-DELTA_PREFIX_2 = u'\u0394'
+DELTA_PREFIX_2 = '\u0394'
 
 # Heartbeat metric.
 HEART_BEAT_METRIC = '~component.heartbeat'

wavefront_sdk/common/metrics/registry.py

@@ -41,7 +41,9 @@ def _schedule_timer(self):
     # pylint: disable=broad-except
     def _report(self, timeout_secs=None):
         timestamp = time.time()
-        for key, val in self.metrics.items():
+
+        # Copying the dict prevents concurrent modification while iterating
+        for key, val in self.metrics.copy().items():
             if timeout_secs and time.time() - timestamp > timeout_secs:
                 break
             name = self.prefix + key

wavefront_sdk/direct.py

@@ -96,8 +96,7 @@ def __init__(self,
         if enable_internal_metrics:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=self,
-                prefix='{}.core.sender.direct'.format(
-                    constants.SDK_METRIC_PREFIX))
+                prefix=f'{constants.SDK_METRIC_PREFIX}.core.sender.direct')
         else:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=None)
@@ -205,8 +204,8 @@ def _report(self, points, data_format, entity_prefix, report_errors):
                                          headers=self._headers,
                                          data=compressed_data)
             status_code = response.status_code
-            self._sdk_metrics_registry.new_delta_counter('{}.report.{}'.format(
-                entity_prefix, status_code)).inc()
+            self._sdk_metrics_registry.new_delta_counter(
+                f'{entity_prefix}.report.{status_code}').inc()
         except requests.exceptions.RequestException:
             report_errors.inc()
         return status_code

wavefront_sdk/entities/histogram/histogram_impl.py

@@ -155,17 +155,16 @@ def bulk_update_dist_by_thread_id(self, thread_id, means, counts):
     def get_centroids(self):
         """Get list of centroids for dists of all threads in this minute."""
         centroids = []
-        for thread_id in self.per_thread_dist:
-            centroids.extend(
-                self.per_thread_dist[thread_id].centroids_to_list())
+        for _, digest in self.per_thread_dist.items():
+            centroids.extend(digest.centroids_to_list())
         return centroids
 
     def to_distribution(self):
         """Convert to Distribution."""
         distributions = []
-        for thread_id in self.per_thread_dist:
+        for _, digest in self.per_thread_dist.items():
             centroids = [(centroid['m'], int(centroid['c'])) for centroid in
-                         self.per_thread_dist[thread_id].centroids_to_list()]
+                         digest.centroids_to_list()]
             distributions.append(
                 Distribution(self.minute_millis, centroids))
         return distributions

wavefront_sdk/proxy.py

@@ -58,8 +58,7 @@ def __init__(self, host, metrics_port, distribution_port, tracing_port,
         if enable_internal_metrics:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=self,
-                prefix='{}.core.sender.proxy'.format(
-                    constants.SDK_METRIC_PREFIX))
+                prefix=f'{constants.SDK_METRIC_PREFIX}.core.sender.proxy')
         else:
             self._sdk_metrics_registry = registry.WavefrontSdkMetricsRegistry(
                 wf_metric_sender=None)