system metrics (CPU/memory)

.appveyor.yml

@@ -41,6 +41,7 @@ install:
   # We need wheel installed to build wheels
   - ".\\tests\\appveyor\\build.cmd %PYTHON%\\python.exe -m pip install -U wheel pip setuptools"
   - ".\\tests\\appveyor\\build.cmd %PYTHON%\\python.exe -m pip install -r tests\\requirements\\requirements-%WEBFRAMEWORK%.txt"
+  - ".\\tests\\appveyor\\build.cmd %PYTHON%\\python.exe -m pip install psutil"
 
 build: off
 

conftest.py

@@ -93,6 +93,7 @@ def pytest_configure(config):
                     },
                 }
             ],
+            ELASTIC_APM={"METRICS_INTERVAL": "0ms"},  # avoid autostarting the metrics collector thread
         )
         settings_dict.update(
             **middleware_setting(

docs/index.asciidoc

@@ -17,6 +17,8 @@ include::./django.asciidoc[Django support]
 
 include::./flask.asciidoc[Flask support]
 
+include::./metrics.asciidoc[Metrics]
+
 include::./advanced-topics.asciidoc[Advanced Topics]
 
 include::./api.asciidoc[API documentation]

docs/metrics.asciidoc

@@ -0,0 +1,27 @@
+[[metrics]]
+== Metrics
+
+With Elastic APM, you can capture system and process metrics.
+These metrics will be sent regularly to the APM Server and from there to Elasticsearch
+
+[[metric-sets]]
+=== Metric Sets
+
+[[cpu-memory-metricset]]
+==== CPU/Memory Metric Set
+
+`elasticapm.metrics.sets.cpu.CPUMetricSet`
+
+This metric set collects various system metrics and metrics of the current process.
+
+|============
+| Metric                    | Type | Format      | Description
+| `system.cpu.total.norm.pct`| scaled_float | percent| The percentage of CPU time in states other than Idle and IOWait, normalised by the number of cores.
+| `system.memory.total`| long | bytes | The total memory of the system in bytes
+| `system.memory.actual.free`| long | bytes | Actual free memory in bytes
+| `system.process.cpu.total.norm.pct`| scaled_float | percent | The percentage of CPU time spent by the process since the last event. This value is normalized by the number of CPU cores and it ranges from 0 to 100%.
+| `system.process.memory.size`| long | bytes | The total virtual memory the process has.
+| `system.process.memory.rss.bytes`| long | bytes | The Resident Set Size. The amount of memory the process occupied in main memory (RAM).
+|============
+
+NOTE: if you do *not* use Linux, you need to install https://pypi.org/project/psutil/[`psutil`] for this metric set.

elasticapm/base.py

@@ -24,6 +24,7 @@
 import elasticapm
 from elasticapm.conf import Config, constants
 from elasticapm.conf.constants import ERROR
+from elasticapm.metrics.base_metrics import MetricsRegistry
 from elasticapm.traces import Tracer, get_transaction
 from elasticapm.utils import compat, is_master_process, stacks, varmap
 from elasticapm.utils.encoding import keyword_field, shorten, transform
@@ -140,6 +141,9 @@ def __init__(self, config=None, **inline):
         )
         self.include_paths_re = stacks.get_path_regex(self.config.include_paths) if self.config.include_paths else None
         self.exclude_paths_re = stacks.get_path_regex(self.config.exclude_paths) if self.config.exclude_paths else None
+        self._metrics = MetricsRegistry(self.config.metrics_interval / 1000.0, self.queue)
+        for path in self.config.metrics_sets:
+            self._metrics.register(path)
         compat.atexit_register(self.close)
 
     def get_handler(self, name):

elasticapm/conf/__init__.py

@@ -122,12 +122,29 @@ def __call__(self, value, field_name):
         return val
 
 
-duration_validator = UnitValidator("^((?:-)?\d+)(ms|s|m)$", "\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000})
+duration_validator = UnitValidator(r"^((?:-)?\d+)(ms|s|m)$", r"\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000})
 size_validator = UnitValidator(
-    "^(\d+)(b|kb|mb|gb)$", "\d+(b|KB|MB|GB)", {"b": 1, "kb": 1024, "mb": 1024 * 1024, "gb": 1024 * 1024 * 1024}
+    r"^(\d+)(b|kb|mb|gb)$", r"\d+(b|KB|MB|GB)", {"b": 1, "kb": 1024, "mb": 1024 * 1024, "gb": 1024 * 1024 * 1024}
 )
 
 
+class ExcludeRangeValidator(object):
+    def __init__(self, range_start, range_end, range_desc):
+        self.range_start = range_start
+        self.range_end = range_end
+        self.range_desc = range_desc
+
+    def __call__(self, value, field_name):
+        if self.range_start <= value <= self.range_end:
+            raise ConfigurationError(
+                "{} cannot be in range: {}".format(
+                    value, self.range_desc.format(**{"range_start": self.range_start, "range_end": self.range_end})
+                ),
+                field_name,
+            )
+        return value
+
+
 class _ConfigBase(object):
     _NO_VALUE = object()  # sentinel object
 
@@ -178,7 +195,9 @@ class Config(_ConfigBase):
     server_timeout = _ConfigValue(
         "SERVER_TIMEOUT",
         type=float,
-        validators=[UnitValidator("^((?:-)?\d+)(ms|s|m)?$", "\d+(ms|s|m)", {"ms": 0.001, "s": 1, "m": 60, None: 1000})],
+        validators=[
+            UnitValidator(r"^((?:-)?\d+)(ms|s|m)?$", r"\d+(ms|s|m)", {"ms": 0.001, "s": 1, "m": 60, None: 1000})
+        ],
         default=5,
     )
     hostname = _ConfigValue("HOSTNAME", default=socket.gethostname())
@@ -196,14 +215,23 @@ class Config(_ConfigBase):
             "elasticapm.processors.sanitize_http_request_body",
         ],
     )
+    metrics_sets = _ListConfigValue("METRICS_SETS", default=["elasticapm.metrics.sets.cpu.CPUMetricSet"])
+    metrics_interval = _ConfigValue(
+        "METRICS_INTERVAL",
+        type=int,
+        validators=[duration_validator, ExcludeRangeValidator(1, 999, "{range_start} - {range_end} ms")],
+        default=30000,
+    )
     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)
     transaction_max_spans = _ConfigValue("TRANSACTION_MAX_SPANS", type=int, default=500)
     span_frames_min_duration = _ConfigValue(
         "SPAN_FRAMES_MIN_DURATION",
         default=5,
-        validators=[UnitValidator("^((?:-)?\d+)(ms|s|m)?$", "\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000, None: 1})],
+        validators=[
+            UnitValidator(r"^((?:-)?\d+)(ms|s|m)?$", r"\d+(ms|s|m)", {"ms": 1, "s": 1000, "m": 60000, None: 1})
+        ],
         type=int,
     )
     collect_local_variables = _ConfigValue("COLLECT_LOCAL_VARIABLES", default="errors")

elasticapm/contrib/django/management/commands/elasticapm.py

@@ -130,7 +130,7 @@ def handle_test(self, command, **options):
     def handle_check(self, command, **options):
         """Check your settings for common misconfigurations"""
         passed = True
-        client = DjangoClient()
+        client = DjangoClient(metrics_interval="0ms")
 
         def is_set(x):
             return x and x != "None"

elasticapm/contrib/pylons/__init__.py

@@ -10,6 +10,7 @@
 """
 from elasticapm.base import Client
 from elasticapm.middleware import ElasticAPM as Middleware
+from elasticapm.utils import compat
 
 
 def list_from_setting(config, setting):
@@ -21,13 +22,6 @@ def list_from_setting(config, setting):
 
 class ElasticAPM(Middleware):
     def __init__(self, app, config, client_cls=Client):
-        client = client_cls(
-            server_url=config.get("elasticapm.server_url"),
-            server_timeout=config.get("elasticapm.server_timeout"),
-            name=config.get("elasticapm.name"),
-            service_name=config.get("elasticapm.service_name"),
-            secret_token=config.get("elasticapm.secret_token"),
-            include_paths=list_from_setting(config, "elasticapm.include_paths"),
-            exclude_paths=list_from_setting(config, "elasticapm.exclude_paths"),
-        )
+        client_config = {key[11:]: val for key, val in compat.iteritems(config) if key.startswith("elasticapm.")}
+        client = client_cls(**client_config)
         super(ElasticAPM, self).__init__(app, client)

elasticapm/metrics/base_metrics.py

@@ -0,0 +1,180 @@
+import logging
+import threading
+import time
+
+from elasticapm.utils import compat
+from elasticapm.utils.module_import import import_string
+
+logger = logging.getLogger("elasticapm.metrics")
+
+
+class MetricsRegistry(object):
+    def __init__(self, collect_interval, queue_func, tags=None):
+        """
+        Creates a new metric registry
+
+        :param collect_interval: the interval to collect metrics from registered metric sets
+        :param queue_func: the function to call with the collected metrics
+        :param tags:
+        """
+        self._collect_interval = collect_interval
+        self._queue_func = queue_func
+        self._metricsets = {}
+        self._tags = tags or {}
+        self._collect_timer = None
+        if self._collect_interval:
+            self._start_collect_timer()
+
+    def register(self, class_path):
+        """
+        Register a new metric set
+        :param class_path: a string with the import path of the metricset class
+        """
+        if class_path in self._metricsets:
+            return
+        else:
+            try:
+                class_obj = import_string(class_path)
+                self._metricsets[class_path] = class_obj()
+            except ImportError as e:
+                logger.warning("Could not register %s metricset: %s", class_path, compat.text_type(e))
+
+    def collect(self, start_timer=True):
+        """
+        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):
+            data = metricset.collect()
+            if data:
+                self._queue_func("metricset", data)
+
+    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
+        logger.debug("Starting metrics collect timer")
+        self._collect_timer.start()
+
+    def _stop_collect_timer(self):
+        if self._collect_timer:
+            logger.debug("Cancelling collect timer")
+            self._collect_timer.cancel()
+
+
+class MetricsSet(object):
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._counters = {}
+        self._gauges = {}
+
+    def counter(self, name):
+        """
+        Returns an existing or creates and returns a new counter
+        :param name: name of the counter
+        :return: the counter object
+        """
+        with self._lock:
+            if name not in self._counters:
+                self._counters[name] = Counter(name)
+            return self._counters[name]
+
+    def gauge(self, name):
+        """
+        Returns an existing or creates and returns a new gauge
+        :param name: name of the gauge
+        :return: the gauge object
+        """
+        with self._lock:
+            if name not in self._gauges:
+                self._gauges[name] = Gauge(name)
+            return self._gauges[name]
+
+    def collect(self):
+        """
+        Collects all metrics attached to this metricset, and returns it as a list, together with a timestamp
+        in microsecond precision.
+
+        The format of the return value should be
+
+            {
+                "samples": {"metric.name": {"value": some_float}, ...},
+                "timestamp": unix epoch in microsecond precision
+            }
+        """
+        samples = {}
+        if self._counters:
+            samples.update({label: {"value": c.val} for label, c in compat.iteritems(self._counters)})
+        if self._gauges:
+            samples.update({label: {"value": g.val} for label, g in compat.iteritems(self._gauges)})
+        if samples:
+            return {"samples": samples, "timestamp": int(time.time() * 1000000)}
+
+
+class Counter(object):
+    __slots__ = ("label", "_lock", "_initial_value", "_val")
+
+    def __init__(self, label, initial_value=0):
+        """
+        Creates a new counter
+        :param label: label of the counter
+        :param initial_value: initial value of the counter, defaults to 0
+        """
+        self.label = label
+        self._lock = threading.Lock()
+        self._val = self._initial_value = initial_value
+
+    def inc(self, delta=1):
+        """
+        Increments the counter. If no delta is provided, it is incremented by one
+        :param delta: the amount to increment the counter by
+        """
+        with self._lock:
+            self._val += delta
+
+    def dec(self, delta=1):
+        """
+        Decrements the counter. If no delta is provided, it is decremented by one
+        :param delta: the amount to decrement the counter by
+        """
+        with self._lock:
+            self._val -= delta
+
+    def reset(self):
+        """
+        Reset the counter to the initial value
+        """
+        with self._lock:
+            self._val = self._initial_value
+
+    @property
+    def val(self):
+        """Returns the current value of the counter"""
+        return self._val
+
+
+class Gauge(object):
+    __slots__ = ("label", "_val")
+
+    def __init__(self, label):
+        """
+        Creates a new gauge
+        :param label: label of the gauge
+        """
+        self.label = label
+        self._val = None
+
+    @property
+    def val(self):
+        return self._val
+
+    @val.setter
+    def val(self, value):
+        self._val = value

elasticapm/metrics/sets/cpu.py

@@ -0,0 +1,7 @@
+import os
+import platform
+
+if platform.system() == "Linux" and "ELASTIC_APM_FORCE_PSUTIL_METRICS" not in os.environ:
+    from elasticapm.metrics.sets.cpu_linux import CPUMetricSet  # noqa: F401
+else:
+    from elasticapm.metrics.sets.cpu_psutil import CPUMetricSet  # noqa: F401