initialize event processor thread on demand (#636)

* initialize event processor thread on demand

Initialization is protected by a lock, as well as a PID check.

* refactor starting of background threads to be centrally owned by the agent instance

* add additional PID check when starting the transport thread

Author: beniwohli

CHANGELOG.asciidoc

@@ -43,6 +43,7 @@ This will be the last minor release to support the following versions:
 [float]
 ===== New Features
 
+ * Refactored spawning of background threads {pull}636[#636]
  * Added support for aiohttp client and server {pull}659[#659]
  * Added support for tornado web framework {pull}661[#661]
  * Added support for starlette/fastapi {pull}694[#694]

elasticapm/base.py

@@ -37,20 +37,20 @@
 import os
 import platform
 import sys
+import threading
 import time
 import warnings
 from copy import deepcopy
 
 import elasticapm
-from elasticapm.conf import Config, VersionedConfig, constants, update_config
+from elasticapm.conf import Config, VersionedConfig, constants
 from elasticapm.conf.constants import ERROR
 from elasticapm.metrics.base_metrics import MetricsRegistry
 from elasticapm.traces import Tracer, execution_context
 from elasticapm.utils import cgroup, compat, is_master_process, stacks, varmap
 from elasticapm.utils.encoding import enforce_label_format, keyword_field, shorten, transform
 from elasticapm.utils.logging import get_logger
 from elasticapm.utils.module_import import import_string
-from elasticapm.utils.threading import IntervalTimer
 
 __all__ = ("Client",)
 
@@ -92,6 +92,10 @@ def __init__(self, config=None, **inline):
         self.logger = get_logger("%s.%s" % (cls.__module__, cls.__name__))
         self.error_logger = get_logger("elasticapm.errors")
 
+        self._pid = None
+        self._thread_starter_lock = threading.Lock()
+        self._thread_managers = {}
+
         self.tracer = None
         self.processors = []
         self.filter_exception_types_dict = {}
@@ -141,6 +145,8 @@ def __init__(self, config=None, **inline):
         )
         transport_class = import_string(self.config.transport_class)
         self._transport = transport_class(self._api_endpoint_url, self, **transport_kwargs)
+        self.config.transport = self._transport
+        self._thread_managers["transport"] = self._transport
 
         for exc_to_filter in self.config.filter_exception_types or []:
             exc_to_filter_type = exc_to_filter.split(".")[-1]
@@ -188,15 +194,25 @@ def __init__(self, config=None, **inline):
             self._metrics.register(path)
         if self.config.breakdown_metrics:
             self._metrics.register("elasticapm.metrics.sets.breakdown.BreakdownMetricSet")
+        self._thread_managers["metrics"] = self._metrics
         compat.atexit_register(self.close)
         if self.config.central_config:
-            self._config_updater = IntervalTimer(
-                update_config, 1, "eapm conf updater", daemon=True, args=(self,), evaluate_function_interval=True
-            )
-            self._config_updater.start()
+            self._thread_managers["config"] = self.config
         else:
             self._config_updater = None
 
+        self.start_threads()
+
+    def start_threads(self):
+        with self._thread_starter_lock:
+            current_pid = os.getpid()
+            if self._pid != current_pid:
+                self.logger.debug("Detected PID change from %d to %d, starting threads", self._pid, current_pid)
+                for manager_type, manager in self._thread_managers.items():
+                    self.logger.debug("Starting %s thread", manager_type)
+                    manager.start_thread()
+                self._pid = current_pid
+
     def get_handler(self, name):
         return import_string(name)
 
@@ -243,6 +259,7 @@ def capture_exception(self, exc_info=None, handled=True, **kwargs):
     def queue(self, event_type, data, flush=False):
         if self.config.disable_send:
             return
+        self.start_threads()
         if flush and is_master_process():
             # don't flush in uWSGI master process to avoid ending up in an unpredictable threading state
             flush = False
@@ -272,11 +289,9 @@ def end_transaction(self, name=None, result="", duration=None):
         return transaction
 
     def close(self):
-        if self._metrics:
-            self._metrics._stop_collect_timer()
-        if self._config_updater:
-            self._config_updater.cancel()
-        self._transport.close()
+        with self._thread_starter_lock:
+            for _manager_type, manager in self._thread_managers.items():
+                manager.stop_thread()
 
     def get_service_info(self):
         if self._service_info:

elasticapm/conf/__init__.py

@@ -37,9 +37,11 @@
 
 from elasticapm.utils import compat, starmatch_to_regex
 from elasticapm.utils.logging import get_logger
+from elasticapm.utils.threading import IntervalTimer, ThreadManager
 
 __all__ = ("setup_logging", "Config")
 
+
 logger = get_logger("elasticapm.conf")
 
 
@@ -338,22 +340,24 @@ class Config(_ConfigBase):
     use_elastic_traceparent_header = _BoolConfigValue("USE_ELASTIC_TRACEPARENT_HEADER", default=True)
 
 
-class VersionedConfig(object):
+class VersionedConfig(ThreadManager):
     """
     A thin layer around Config that provides versioning
     """
 
-    __slots__ = ("_config", "_version", "_first_config", "_first_version", "_lock")
+    __slots__ = ("_config", "_version", "_first_config", "_first_version", "_lock", "transport", "_update_thread")
 
-    def __init__(self, config_object, version):
+    def __init__(self, config_object, version, transport=None):
         """
         Create a new VersionedConfig with an initial Config object
         :param config_object: the initial Config object
         :param version: a version identifier for the configuration
         """
         self._config = self._first_config = config_object
         self._version = self._first_version = version
+        self.transport = transport
         self._lock = threading.Lock()
+        self._update_thread = None
 
     def update(self, version, **config):
         """
@@ -399,32 +403,44 @@ def __setattr__(self, name, value):
     def config_version(self):
         return self._version
 
-
-def update_config(agent):
-    logger.debug("Checking for new config...")
-    transport = agent._transport
-    keys = {"service": {"name": agent.config.service_name}}
-    if agent.config.environment:
-        keys["service"]["environment"] = agent.config.environment
-    new_version, new_config, next_run = transport.get_config(agent.config.config_version, keys)
-    if new_version and new_config:
-        errors = agent.config.update(new_version, **new_config)
-        if errors:
-            logger.error("Error applying new configuration: %s", repr(errors))
-        else:
-            logger.info(
-                "Applied new configuration: %s",
-                "; ".join(
-                    "%s=%s" % (compat.text_type(k), compat.text_type(v)) for k, v in compat.iteritems(new_config)
-                ),
-            )
-    elif new_version == agent.config.config_version:
-        logger.debug("Remote config unchanged")
-    elif not new_config and agent.config.changed:
-        logger.debug("Remote config disappeared, resetting to original")
-        agent.config.reset()
-
-    return next_run
+    def update_config(self):
+        if not self.transport:
+            logger.warning("No transport set for config updates, skipping")
+            return
+        logger.debug("Checking for new config...")
+        keys = {"service": {"name": self.service_name}}
+        if self.environment:
+            keys["service"]["environment"] = self.environment
+        new_version, new_config, next_run = self.transport.get_config(self.config_version, keys)
+        if new_version and new_config:
+            errors = self.update(new_version, **new_config)
+            if errors:
+                logger.error("Error applying new configuration: %s", repr(errors))
+            else:
+                logger.info(
+                    "Applied new configuration: %s",
+                    "; ".join(
+                        "%s=%s" % (compat.text_type(k), compat.text_type(v)) for k, v in compat.iteritems(new_config)
+                    ),
+                )
+        elif new_version == self.config_version:
+            logger.debug("Remote config unchanged")
+        elif not new_config and self.changed:
+            logger.debug("Remote config disappeared, resetting to original")
+            self.reset()
+
+        return next_run
+
+    def start_thread(self):
+        self._update_thread = IntervalTimer(
+            self.update_config, 1, "eapm conf updater", daemon=True, evaluate_function_interval=True
+        )
+        self._update_thread.start()
+
+    def stop_thread(self):
+        if self._update_thread:
+            self._update_thread.cancel()
+            self._update_thread = None
 
 
 def setup_logging(handler, exclude=("gunicorn", "south", "elasticapm.errors")):

elasticapm/metrics/base_metrics.py

@@ -33,17 +33,17 @@
 from collections import defaultdict
 
 from elasticapm.conf import constants
-from elasticapm.utils import compat, is_master_process
+from elasticapm.utils import compat
 from elasticapm.utils.logging import get_logger
 from elasticapm.utils.module_import import import_string
-from elasticapm.utils.threading import IntervalTimer
+from elasticapm.utils.threading import IntervalTimer, ThreadManager
 
 logger = get_logger("elasticapm.metrics")
 
 DISTINCT_LABEL_LIMIT = 1000
 
 
-class MetricsRegistry(object):
+class MetricsRegistry(ThreadManager):
     def __init__(self, collect_interval, queue_func, tags=None, ignore_patterns=None):
         """
         Creates a new metric registry
@@ -58,13 +58,6 @@ def __init__(self, collect_interval, queue_func, tags=None, ignore_patterns=None
         self._tags = tags or {}
         self._collect_timer = None
         self._ignore_patterns = ignore_patterns or ()
-        if self._collect_interval:
-            # we only start the thread if we are not in a uwsgi master process
-            if not is_master_process():
-                self._start_collect_timer()
-            else:
-                # If we _are_ in a uwsgi master process, we use the postfork hook to start the thread after the fork
-                compat.postfork(lambda: self._start_collect_timer())
 
     def register(self, class_path):
         """
@@ -97,16 +90,19 @@ def collect(self):
             for data in metricset.collect():
                 self._queue_func(constants.METRICSET, data)
 
-    def _start_collect_timer(self, timeout=None):
-        timeout = timeout or self._collect_interval
-        self._collect_timer = IntervalTimer(self.collect, timeout, name="eapm metrics collect timer", daemon=True)
-        logger.debug("Starting metrics collect timer")
-        self._collect_timer.start()
-
-    def _stop_collect_timer(self):
-        if self._collect_timer:
+    def start_thread(self):
+        if self._collect_interval:
+            self._collect_timer = IntervalTimer(
+                self.collect, self._collect_interval, name="eapm metrics collect timer", daemon=True
+            )
+            logger.debug("Starting metrics collect timer")
+            self._collect_timer.start()
+
+    def stop_thread(self):
+        if self._collect_timer and self._collect_timer.is_alive():
             logger.debug("Cancelling collect timer")
             self._collect_timer.cancel()
+            self._collect_timer = None
 
 
 class MetricsSet(object):

elasticapm/transport/base.py

@@ -31,15 +31,17 @@
 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 import gzip
+import os
 import random
 import threading
 import time
 import timeit
 from collections import defaultdict
 
 from elasticapm.contrib.async_worker import AsyncWorker
-from elasticapm.utils import compat, is_master_process, json_encoder
+from elasticapm.utils import compat, json_encoder
 from elasticapm.utils.logging import get_logger
+from elasticapm.utils.threading import ThreadManager
 
 logger = get_logger("elasticapm.transport")
 
@@ -51,7 +53,7 @@ def __init__(self, message, data=None, print_trace=True):
         self.print_trace = print_trace
 
 
-class Transport(object):
+class Transport(ThreadManager):
     """
     All transport implementations need to subclass this class
 
@@ -93,18 +95,12 @@ def __init__(
         self._queued_data = None
         self._event_queue = self._init_event_queue(chill_until=queue_chill_count, max_chill_time=queue_chill_time)
         self._is_chilled_queue = isinstance(self._event_queue, ChilledQueue)
-        self._event_process_thread = None
+        self._thread = None
         self._last_flush = timeit.default_timer()
         self._counts = defaultdict(int)
         self._flushed = threading.Event()
         self._closed = False
         self._processors = processors if processors is not None else []
-        # only start the event processing thread if we are not in a uwsgi master process
-        if not is_master_process():
-            self._start_event_processor()
-        else:
-            # if we _are_ in a uwsgi master process, use the postfork mixup to start the thread after the fork
-            compat.postfork(lambda: self._start_event_processor())
 
     def queue(self, event_type, data, flush=False):
         try:
@@ -231,14 +227,14 @@ def _flush(self, buffer):
             except Exception as e:
                 self.handle_transport_fail(e)
 
-    def _start_event_processor(self):
-        if (not self._event_process_thread or not self._event_process_thread.is_alive()) and not self._closed:
+    def start_thread(self):
+        current_pid = os.getpid()
+        if (not self._thread or current_pid != self._thread.pid) and not self._closed:
             try:
-                self._event_process_thread = threading.Thread(
-                    target=self._process_queue, name="eapm event processor thread"
-                )
-                self._event_process_thread.daemon = True
-                self._event_process_thread.start()
+                self._thread = threading.Thread(target=self._process_queue, name="eapm event processor thread")
+                self._thread.daemon = True
+                self._thread.pid = current_pid
+                self._thread.start()
             except RuntimeError:
                 pass
 
@@ -254,13 +250,15 @@ def close(self):
         Cleans up resources and closes connection
         :return:
         """
-        if self._closed:
+        if self._closed or (not self._thread or self._thread.pid != os.getpid()):
             return
         self._closed = True
         self.queue("close", None)
         if not self._flushed.wait(timeout=self._max_flush_time):
             raise ValueError("close timed out")
 
+    stop_thread = close
+
     def flush(self):
         """
         Trigger a flush of the queue.

elasticapm/utils/threading.py

@@ -85,3 +85,11 @@ def run(self):
 
     def cancel(self):
         self._interval_done.set()
+
+
+class ThreadManager(object):
+    def start_thread(self):
+        raise NotImplementedError()
+
+    def stop_thread(self):
+        raise NotImplementedError()