introduce IntervalTimer and use it instead of threading.Timer (#453)

`threading.Timer` only runs the function once, and then stops the thread.
To get an interval timer, we used to start a new thread when the original
thread expired. This is wasteful, and in the worst case can lead to
runoff creation of timer threads (#450).

Instead, the `IntervalTimer` will run in an endless loop, executing the
given function every n seconds.

closes #453

Author: beniwohli

CHANGELOG.md

@@ -4,6 +4,7 @@
 [Check the diff](https://github.com/elastic/apm-agent-python/compare/v4.2.1...master)
  * fixed an issue with Celery and the prefork worker pool (#444)
  * fixed an issue when running uwsgi without a master process (#446)
+ * introduced `IntervalTimer` and use it instead of `threading.Timer` (#452)
 
 ## v4.2.1 
 [Check the diff](https://github.com/elastic/apm-agent-python/compare/v4.2.0...v4.2.1)

elasticapm/metrics/base_metrics.py

@@ -34,6 +34,7 @@
 
 from elasticapm.utils import compat, is_master_process
 from elasticapm.utils.module_import import import_string
+from elasticapm.utils.threading import IntervalTimer
 
 logger = logging.getLogger("elasticapm.metrics")
 
@@ -74,15 +75,11 @@ def register(self, class_path):
             except ImportError as e:
                 logger.warning("Could not register %s metricset: %s", class_path, compat.text_type(e))
 
-    def collect(self, start_timer=True):
+    def collect(self):
         """
         Collect metrics from all registered metric sets
-        :param start_timer: if True, restarts the collect timer after collection
         :return:
         """
-        if start_timer:
-            self._start_collect_timer()
-
         logger.debug("Collecting metrics")
 
         for name, metricset in compat.iteritems(self._metricsets):
@@ -92,9 +89,7 @@ def collect(self, start_timer=True):
 
     def _start_collect_timer(self, timeout=None):
         timeout = timeout or self._collect_interval
-        self._collect_timer = threading.Timer(timeout, self.collect, kwargs={"start_timer": True})
-        self._collect_timer.name = "elasticapm metrics collect timer"
-        self._collect_timer.daemon = True
+        self._collect_timer = IntervalTimer(self.collect, timeout, name="eapm metrics collect timer", daemon=True)
         logger.debug("Starting metrics collect timer")
         self._collect_timer.start()
 

elasticapm/utils/threading.py

@@ -0,0 +1,68 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2019, Elasticsearch BV
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions are met:
+#
+#  * Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+#
+#  * Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+#  * Neither the name of the copyright holder nor the names of its
+#    contributors may be used to endorse or promote products derived from
+#    this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+
+
+from __future__ import absolute_import
+
+import threading
+from timeit import default_timer
+
+
+class IntervalTimer(threading.Thread):
+    """
+    A timer that runs a function repeatedly. In contrast to threading.Timer,
+    IntervalTimer runs the given function in perpetuity or until it is cancelled.
+    When run, it will wait `interval` seconds until the first execution.
+    """
+
+    def __init__(self, function, interval, name=None, args=(), kwargs=None, daemon=None):
+        super(IntervalTimer, self).__init__(name=name)
+        self.daemon = daemon
+        self._args = args
+        self._kwargs = kwargs if kwargs is not None else {}
+        self._function = function
+        self._interval = interval
+        self._interval_done = threading.Event()
+
+    def run(self):
+        execution_time = 0
+        while True:
+            real_interval = self._interval - execution_time
+            interval_completed = True
+            if real_interval:
+                interval_completed = not self._interval_done.wait(real_interval)
+            if not interval_completed:
+                # we've been cancelled, time to go home
+                return
+            start = default_timer()
+            self._function(*self._args, **self._kwargs)
+            execution_time = default_timer() - start
+
+    def cancel(self):
+        self._interval_done.set()

tests/utils/threading_tests.py

@@ -0,0 +1,49 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2019, Elasticsearch BV
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions are met:
+#
+#  * Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+#
+#  * Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+#  * Neither the name of the copyright holder nor the names of its
+#    contributors may be used to endorse or promote products derived from
+#    this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+
+import time
+
+import mock
+
+from elasticapm.utils.threading import IntervalTimer
+
+
+def test_interval_timer():
+    func = mock.Mock()
+    timer = IntervalTimer(function=func, interval=0.1, args=(1,), kwargs={"a": "b"})
+    timer.start()
+    time.sleep(0.25)
+    try:
+        assert func.call_count == 2
+        for call in func.call_args_list:
+            assert call == ((1,), {"a": "b"})
+    finally:
+        timer.cancel()
+    time.sleep(0.05)
+    assert not timer.is_alive()