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))