Added support for central config management (#511)

The agent will periodically poll the APM Server for configuration
changes, and apply them if a new configuration is detected. Currently,
only `transaction_sample_rate` is supported

closes #511

Author: beniwohli

CHANGELOG.md

@@ -10,6 +10,7 @@
 ## Other changes
 
  * Added support for recording breakdown metrics (#535)
+ * Added support for central config management (#511)
  * Added instrumentation for `urllib2` (Python 2) / `urllib.request` (Python 3) (#464)
  * Added `disable_metrics` setting (#399)
  * Updated elasticsearch instrumentation for 7.x (#482, #483)

docs/configuration.asciidoc

@@ -649,6 +649,22 @@ NOTE: this setting only disables the *sending* of the given metrics, not collect
 Enable/disable the tracking and collection of breakdown metrics.
 By setting this to `False`, tracking this metric is completely disabled, which can reduce the overhead of the agent. 
 
+NOTE: This feature requires APM Server and Kibana >= 7.3.
+
+[float]
+[[config-central_config]]
+==== `central_config`
+|============
+| Environment                       | Django/Flask        | Default
+| `ELASTIC_APM_CENTRAL_CONFIG`      | `CENTRAL_CONFIG` | `True`
+|============
+
+When enabled, the agent will make periodic requests to the APM Server to fetch updated configuration.
+
+See {kibana-ref}/agent-configuration.html#agent-configuration[APM Agent Configuration] for more documentation on central agent configuration.
+
+NOTE: This feature requires APM Server and Kibana >= 7.3.
+
 [float]
 [[config-django-specific]]
 === Django-specific configuration

elasticapm/base.py

@@ -42,13 +42,14 @@
 from copy import deepcopy
 
 import elasticapm
-from elasticapm.conf import Config, constants
+from elasticapm.conf import Config, VersionedConfig, constants, update_config
 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 keyword_field, shorten, transform
 from elasticapm.utils.module_import import import_string
+from elasticapm.utils.threading import IntervalTimer
 
 __all__ = ("Client",)
 
@@ -95,11 +96,12 @@ def __init__(self, config=None, **inline):
         self.filter_exception_types_dict = {}
         self._service_info = None
 
-        self.config = Config(config, inline_dict=inline)
-        if self.config.errors:
-            for msg in self.config.errors.values():
+        config = Config(config, inline_dict=inline)
+        if config.errors:
+            for msg in config.errors.values():
                 self.error_logger.error(msg)
-            self.config.disable_send = True
+            config.disable_send = True
+        self.config = VersionedConfig(config, version=None)
 
         headers = {
             "Content-Type": "application/x-ndjson",
@@ -157,10 +159,7 @@ def __init__(self, config=None, **inline):
                 ),
             ),
             queue_func=self.queue,
-            sample_rate=self.config.transaction_sample_rate,
-            max_spans=self.config.transaction_max_spans,
-            span_frames_min_duration=self.config.span_frames_min_duration,
-            ignore_patterns=self.config.transactions_ignore_patterns,
+            config=self.config,
             agent=self,
         )
         self.include_paths_re = stacks.get_path_regex(self.config.include_paths) if self.config.include_paths else None
@@ -173,6 +172,13 @@ def __init__(self, config=None, **inline):
         if self.config.breakdown_metrics:
             self._metrics.register("elasticapm.metrics.sets.breakdown.BreakdownMetricSet")
         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()
+        else:
+            self._config_updater = None
 
     def get_handler(self, name):
         return import_string(name)
@@ -241,6 +247,8 @@ def end_transaction(self, name=None, result=""):
     def close(self):
         if self._metrics:
             self._metrics._stop_collect_timer()
+        if self._config_updater:
+            self._config_updater.cancel()
         self._transport.close()
 
     def get_service_info(self):

elasticapm/conf/__init__.py

@@ -33,11 +33,14 @@
 import os
 import re
 import socket
+import threading
 
 from elasticapm.utils import compat, starmatch_to_regex
 
 __all__ = ("setup_logging", "Config")
 
+logger = logging.getLogger("elasticapm.conf")
+
 
 class ConfigurationError(ValueError):
     def __init__(self, msg, field_name):
@@ -64,8 +67,6 @@ def __get__(self, instance, owner):
 
     def __set__(self, instance, value):
         value = self._validate(instance, value)
-        if value is not None:
-            value = self.type(value)
         instance._values[self.dict_key] = value
 
     def _validate(self, instance, value):
@@ -76,6 +77,11 @@ def _validate(self, instance, value):
         if self.validators and value is not None:
             for validator in self.validators:
                 value = validator(value, self.dict_key)
+        if self.type and value is not None:
+            try:
+                value = self.type(value)
+            except ValueError as e:
+                raise ConfigurationError("{}: {}".format(self.dict_key, compat.text_type(e)), self.dict_key)
         instance._errors.pop(self.dict_key, None)
         return value
 
@@ -183,6 +189,9 @@ class _ConfigBase(object):
     def __init__(self, config_dict=None, env_dict=None, inline_dict=None):
         self._values = {}
         self._errors = {}
+        self.update(config_dict, env_dict, inline_dict)
+
+    def update(self, config_dict=None, env_dict=None, inline_dict=None):
         if config_dict is None:
             config_dict = {}
         if env_dict is None:
@@ -209,6 +218,14 @@ def __init__(self, config_dict=None, env_dict=None, inline_dict=None):
                 except ConfigurationError as e:
                     self._errors[e.field_name] = str(e)
 
+    @property
+    def values(self):
+        return self._values
+
+    @values.setter
+    def values(self, values):
+        self._values = values
+
     @property
     def errors(self):
         return self._errors
@@ -263,6 +280,7 @@ class Config(_ConfigBase):
     )
     breakdown_metrics = _BoolConfigValue("BREAKDOWN_METRICS", default=True)
     disable_metrics = _ListConfigValue("DISABLE_METRICS", type=starmatch_to_regex, default=[])
+    central_config = _BoolConfigValue("CENTRAL_CONFIG", default=True)
     api_request_size = _ConfigValue("API_REQUEST_SIZE", type=int, validators=[size_validator], default=750 * 1024)
     api_request_time = _ConfigValue("API_REQUEST_TIME", type=int, validators=[duration_validator], default=10 * 1000)
     transaction_sample_rate = _ConfigValue("TRANSACTION_SAMPLE_RATE", type=float, default=1.0)
@@ -296,6 +314,95 @@ class Config(_ConfigBase):
     django_transaction_name_from_route = _BoolConfigValue("DJANGO_TRANSACTION_NAME_FROM_ROUTE", default=False)
 
 
+class VersionedConfig(object):
+    """
+    A thin layer around Config that provides versioning
+    """
+
+    __slots__ = ("_config", "_version", "_first_config", "_first_version", "_lock")
+
+    def __init__(self, config_object, version):
+        """
+        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._lock = threading.Lock()
+
+    def update(self, version, **config):
+        """
+        Update the configuration version
+        :param version: version identifier for the new configuration
+        :param config: a key/value map of new configuration
+        :return: configuration errors, if any
+        """
+        new_config = Config()
+        new_config.values = self._config.values.copy()
+
+        # pass an empty env dict to ensure the environment doesn't get precedence
+        new_config.update(inline_dict=config, env_dict={})
+        if not new_config.errors:
+            with self._lock:
+                self._version = version
+                self._config = new_config
+        else:
+            return new_config.errors
+
+    def reset(self):
+        """
+        Reset state to the original configuration
+        """
+        with self._lock:
+            self._version = self._first_version
+            self._config = self._first_config
+
+    @property
+    def changed(self):
+        return self._config != self._first_config
+
+    def __getattr__(self, item):
+        return getattr(self._config, item)
+
+    def __setattr__(self, name, value):
+        if name not in self.__slots__:
+            setattr(self._config, name, value)
+        else:
+            super(VersionedConfig, self).__setattr__(name, value)
+
+    @property
+    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 setup_logging(handler, exclude=("gunicorn", "south", "elasticapm.errors")):
     """
     Configures logging to pipe to Elastic APM.

elasticapm/conf/constants.py

@@ -31,6 +31,7 @@
 import decimal
 
 EVENTS_API_PATH = "intake/v2/events"
+AGENT_CONFIG_PATH = "config/v1/agents"
 
 TRACE_CONTEXT_VERSION = 0
 TRACEPARENT_HEADER_NAME = "elastic-apm-traceparent"

elasticapm/traces.py

@@ -212,7 +212,7 @@ def _begin_span(
         tracer = self.tracer
         if parent_span and parent_span.leaf:
             span = DroppedSpan(parent_span, leaf=True)
-        elif tracer.max_spans and self._span_counter > tracer.max_spans - 1:
+        elif tracer.config.transaction_max_spans and self._span_counter > tracer.config.transaction_max_spans - 1:
             self.dropped_spans += 1
             span = DroppedSpan(parent_span)
             self._span_counter += 1
@@ -450,29 +450,18 @@ def child_ended(self, timestamp):
 
 
 class Tracer(object):
-    def __init__(
-        self,
-        frames_collector_func,
-        frames_processing_func,
-        queue_func,
-        sample_rate=1.0,
-        max_spans=0,
-        span_frames_min_duration=None,
-        ignore_patterns=None,
-        agent=None,
-    ):
-        self.max_spans = max_spans
+    def __init__(self, frames_collector_func, frames_processing_func, queue_func, config, agent):
+        self.config = config
         self.queue_func = queue_func
         self.frames_processing_func = frames_processing_func
         self.frames_collector_func = frames_collector_func
-        self._ignore_patterns = [re.compile(p) for p in ignore_patterns or []]
-        self._sample_rate = sample_rate
         self._agent = agent
-        if span_frames_min_duration in (-1, None):
+        self._ignore_patterns = [re.compile(p) for p in config.transactions_ignore_patterns or []]
+        if config.span_frames_min_duration in (-1, None):
             # both None and -1 mean "no minimum"
             self.span_frames_min_duration = None
         else:
-            self.span_frames_min_duration = span_frames_min_duration / 1000.0
+            self.span_frames_min_duration = config.span_frames_min_duration / 1000.0
 
     def begin_transaction(self, transaction_type, trace_parent=None):
         """
@@ -483,7 +472,9 @@ def begin_transaction(self, transaction_type, trace_parent=None):
         if trace_parent:
             is_sampled = bool(trace_parent.trace_options.recorded)
         else:
-            is_sampled = self._sample_rate == 1.0 or self._sample_rate > random.random()
+            is_sampled = (
+                self.config.transaction_sample_rate == 1.0 or self.config.transaction_sample_rate > random.random()
+            )
         transaction = Transaction(self, transaction_type, trace_parent=trace_parent, is_sampled=is_sampled)
         if trace_parent is None:
             transaction.trace_parent = TraceParent(

elasticapm/transport/http.py

@@ -33,6 +33,7 @@
 import hashlib
 import logging
 import os
+import re
 import ssl
 
 import certifi
@@ -41,7 +42,7 @@
 
 from elasticapm.transport.base import TransportException
 from elasticapm.transport.http_base import AsyncHTTPTransportBase, HTTPTransportBase
-from elasticapm.utils import compat, read_pem_file
+from elasticapm.utils import compat, json_encoder, read_pem_file
 
 logger = logging.getLogger("elasticapm.transport.http")
 
@@ -103,6 +104,35 @@ def send(self, data):
             if response:
                 response.close()
 
+    def get_config(self, current_version=None, keys=None):
+        url = self._config_url
+        data = json_encoder.dumps(keys).encode("utf-8")
+        headers = self._headers.copy()
+        max_age = 300
+        if current_version:
+            headers["If-None-Match"] = current_version
+        try:
+            response = self.http.urlopen(
+                "POST", url, body=data, headers=headers, timeout=self._timeout, preload_content=False
+            )
+        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:
+            logger.debug("HTTP error while fetching remote config: %s", compat.text_type(e))
+            return current_version, None, max_age
+        body = response.read()
+        if "Cache-Control" in response.headers:
+            try:
+                max_age = int(next(re.finditer(r"max-age=(\d+)", response.headers["Cache-Control"])).groups()[0])
+            except StopIteration:
+                logger.debug("Could not parse Cache-Control header: %s", response["Cache-Control"])
+        if response.status == 304:
+            # config is unchanged, return
+            logger.debug("Configuration unchanged")
+            return current_version, None, max_age
+        elif response.status >= 400:
+            return None, None, max_age
+
+        return response.headers.get("Etag"), json_encoder.loads(body.decode("utf-8")), max_age
+
     @property
     def cert_fingerprint(self):
         if self._server_cert: