Fix runtime workers not flushing to Dogstatsd (#939)

* Force update to dogstatsd constant tags

* Less confusing log message

* Move service update to after runtime worker

* Fix run_periodic for runtime worker

* Fix test for runtime metrics worker

* Force float division

Author: majorgreys

ddtrace/internal/runtime/runtime_metrics.py

@@ -60,7 +60,8 @@ class RuntimeWorker(_worker.PeriodicWorkerThread):
 
     FLUSH_INTERVAL = 10
 
-    def __init__(self, statsd_client, flush_interval=FLUSH_INTERVAL):
+    def __init__(self, statsd_client, flush_interval=None):
+        flush_interval = self.FLUSH_INTERVAL if flush_interval is None else flush_interval
         super(RuntimeWorker, self).__init__(interval=flush_interval,
                                             name=self.__class__.__name__)
         self._statsd_client = statsd_client
@@ -78,7 +79,7 @@ def flush(self):
         for key, value in self._runtime_metrics:
             self._write_metric(key, value)
 
-    on_periodic = flush
+    run_periodic = flush
     on_shutdown = flush
 
     def reset(self):

ddtrace/tracer.py

@@ -69,6 +69,8 @@ def __init__(self):
         self._runtime_id = generate_runtime_id()
         self._runtime_worker = None
         self._dogstatsd_client = None
+        self._dogstatsd_host = self.DEFAULT_HOSTNAME
+        self._dogstatsd_port = self.DEFAULT_DOGSTATSD_PORT
 
     def get_call_context(self, *args, **kwargs):
         """
@@ -154,12 +156,11 @@ def configure(self, enabled=None, hostname=None, port=None, dogstatsd_host=None,
             self._wrap_executor = wrap_executor
 
         if collect_metrics and self._runtime_worker is None:
+            self._dogstatsd_host = dogstatsd_host or self._dogstatsd_host
+            self._dogstatsd_port = dogstatsd_port or self._dogstatsd_port
             # start dogstatsd client if not already running
             if not self._dogstatsd_client:
-                self._start_dogstatsd_client(
-                    dogstatsd_host or self.DEFAULT_HOSTNAME,
-                    dogstatsd_port or self.DEFAULT_DOGSTATSD_PORT,
-                )
+                self._start_dogstatsd_client()
 
             self._start_runtime_worker()
 
@@ -271,18 +272,18 @@ def start_span(self, name, child_of=None, service=None, resource=None, span_type
         # add it to the current context
         context.add_span(span)
 
+        # check for new process if runtime metrics worker has already been started
+        if self._runtime_worker:
+            self._check_new_process()
+
         # update set of services handled by tracer
-        if service:
+        if service and service not in self._services:
             self._services.add(service)
 
             # The constant tags for the dogstatsd client needs to updated with any new
             # service(s) that may have been added.
             self._update_dogstatsd_constant_tags()
 
-        # check for new process if runtime metrics worker has already been started
-        if self._runtime_worker:
-            self._check_new_process()
-
         return span
 
     def _update_dogstatsd_constant_tags(self):
@@ -299,12 +300,15 @@ def _update_dogstatsd_constant_tags(self):
         log.debug('Updating constant tags {}'.format(tags))
         self._dogstatsd_client.constant_tags = tags
 
-    def _start_dogstatsd_client(self, host, port):
+    def _start_dogstatsd_client(self):
         # start dogstatsd as client with constant tags
-        log.debug('Starting DogStatsd on {}:{}'.format(host, port))
+        log.debug('Connecting to DogStatsd on {}:{}'.format(
+            self._dogstatsd_host,
+            self._dogstatsd_port
+        ))
         self._dogstatsd_client = DogStatsd(
-            host=host,
-            port=port,
+            host=self._dogstatsd_host,
+            port=self._dogstatsd_port,
         )
 
     def _start_runtime_worker(self):
@@ -330,6 +334,10 @@ def _check_new_process(self):
 
         self._start_runtime_worker()
 
+        # force an immediate update constant tags since we have reset services
+        # and generated a new runtime id
+        self._update_dogstatsd_constant_tags()
+
     def trace(self, name, service=None, resource=None, span_type=None):
         """
         Return a span that will trace an operation called `name`. The context that created

tests/internal/runtime/test_runtime_metrics.py

@@ -1,3 +1,5 @@
+import time
+
 from ddtrace.internal.runtime.runtime_metrics import (
     RuntimeTags,
     RuntimeMetrics,
@@ -43,21 +45,30 @@ def test_one_metric(self):
 
 
 class TestRuntimeWorker(BaseTracerTestCase):
-    def test_worker_metrics(self):
-        self.tracer.configure(collect_metrics=True)
+    def test_tracer_metrics(self):
+        # mock dogstatsd client before configuring tracer for runtime metrics
+        self.tracer._dogstatsd_client = DogStatsd()
+        self.tracer._dogstatsd_client.socket = FakeSocket()
+
+        default_flush_interval = RuntimeWorker.FLUSH_INTERVAL
+        try:
+            # lower flush interval
+            RuntimeWorker.FLUSH_INTERVAL = 1./4
+
+            # configure tracer for runtime metrics
+            self.tracer.configure(collect_metrics=True)
+        finally:
+            # reset flush interval
+            RuntimeWorker.FLUSH_INTERVAL = default_flush_interval
 
         with self.override_global_tracer(self.tracer):
-            self.tracer._dogstatsd_client = DogStatsd()
-            self.tracer._dogstatsd_client.socket = FakeSocket()
-
             root = self.start_span('parent', service='parent')
             context = root.context
             self.start_span('child', service='child', child_of=context)
 
-            self.worker = RuntimeWorker(self.tracer._dogstatsd_client, 0)
-            self.worker.start()
-            self.worker.stop()
-            self.worker.join()
+            time.sleep(self.tracer._runtime_worker.interval * 2)
+            self.tracer._runtime_worker.stop()
+            self.tracer._runtime_worker.join()
 
             # get all received metrics
             received = []
@@ -69,7 +80,8 @@ def test_worker_metrics(self):
                 received.append(new)
 
             # expect received all default metrics
-            self.assertEqual(len(received), len(DEFAULT_RUNTIME_METRICS))
+            # we expect more than one flush since it is also called on shutdown
+            assert len(received) / len(DEFAULT_RUNTIME_METRICS) > 1
 
             # expect all metrics in default set are received
             # DEV: dogstatsd gauges in form "{metric_name}:{metric_value}|g#t{tag_name}:{tag_value},..."
@@ -78,7 +90,8 @@ def test_worker_metrics(self):
                 DEFAULT_RUNTIME_METRICS
             )
 
-            for gauge in received:
+            # check to last set of metrics returned to confirm tags were set
+            for gauge in received[-len(DEFAULT_RUNTIME_METRICS):]:
                 self.assertRegexpMatches(gauge, 'runtime-id:')
                 self.assertRegexpMatches(gauge, 'service:parent')
                 self.assertRegexpMatches(gauge, 'service:child')