Ensure that metrics are flushed before shutting down (#1139)

* Ensure that metrics are flushed before shutting down

To ensure that the flushed metrics are actually sent, we need
to make sure that the transport thread is shut down last.

* use a variable name that doesn't clash with the thread priority concept

Author: beniwohli

elasticapm/base.py

@@ -213,11 +213,13 @@ def __init__(self, config=None, **inline):
         set_client(self)
 
     def start_threads(self):
-        with self._thread_starter_lock:
-            current_pid = os.getpid()
-            if self._pid != current_pid:
+        current_pid = os.getpid()
+        if self._pid != current_pid:
+            with self._thread_starter_lock:
                 self.logger.debug("Detected PID change from %r to %r, starting threads", self._pid, current_pid)
-                for manager_type, manager in self._thread_managers.items():
+                for manager_type, manager in sorted(
+                    self._thread_managers.items(), key=lambda item: item[1].start_stop_order
+                ):
                     self.logger.debug("Starting %s thread", manager_type)
                     manager.start_thread(pid=current_pid)
                 self._pid = current_pid
@@ -303,7 +305,7 @@ def end_transaction(self, name=None, result="", duration=None):
     def close(self):
         if self.config.enabled:
             with self._thread_starter_lock:
-                for _, manager in self._thread_managers.items():
+                for _, manager in sorted(self._thread_managers.items(), key=lambda item: item[1].start_stop_order):
                     manager.stop_thread()
         global CLIENT_SINGLETON
         CLIENT_SINGLETON = None

elasticapm/metrics/base_metrics.py

@@ -103,6 +103,8 @@ def stop_thread(self):
             logger.debug("Cancelling collect timer")
             self._collect_timer.cancel()
             self._collect_timer = None
+            # collect one last time
+            self.collect()
 
     @property
     def collect_interval(self):

elasticapm/transport/base.py

@@ -33,6 +33,7 @@
 import gzip
 import os
 import random
+import sys
 import threading
 import time
 import timeit
@@ -86,6 +87,7 @@ def __init__(
         self._closed = False
         self._processors = processors if processors is not None else []
         super(Transport, self).__init__()
+        self.start_stop_order = sys.maxsize  # ensure that the transport thread is always started/stopped last
 
     @property
     def _max_flush_time(self):

elasticapm/utils/threading.py

@@ -91,6 +91,7 @@ def cancel(self):
 class ThreadManager(object):
     def __init__(self):
         self.pid = None
+        self.start_stop_order = 100
 
     def start_thread(self, pid=None):
         if not pid:

tests/metrics/base_tests.py

@@ -125,6 +125,24 @@ def target():
     assert metricset.counter("x").val == expected
 
 
+@pytest.mark.parametrize("sending_elasticapm_client", [{"metrics_interval": "30s"}], indirect=True)
+def test_metrics_flushed_on_shutdown(sending_elasticapm_client):
+    # this is ugly, we need an API for this at some point...
+    metricset = MetricsSet(sending_elasticapm_client._metrics)
+    sending_elasticapm_client._metrics._metricsets["foo"] = metricset
+    metricset.counter("x").inc()
+    sending_elasticapm_client.close()
+    assert sending_elasticapm_client.httpserver.payloads
+    for item in sending_elasticapm_client.httpserver.payloads[0]:
+        try:
+            assert item["metricset"]["samples"]["x"]["value"] == 1
+            break
+        except KeyError:
+            pass
+    else:
+        assert False, "no item found with matching dict path metricset.samples.x.value"
+
+
 @mock.patch("elasticapm.metrics.base_metrics.DISTINCT_LABEL_LIMIT", 3)
 def test_metric_limit(caplog, elasticapm_client):
     m = MetricsSet(MetricsRegistry(elasticapm_client))