first stab at implementing metrics [WIP]

Author: beniwohli

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.cpu_psutil.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/metrics/__init__.py

@@ -0,0 +1,118 @@
+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):
+        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):
+        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):
+        logger.info("Starting metrics collect timer")
+
+        for name, metricset in compat.iteritems(self._metricsets):
+            data = metricset.collect()
+            if data:
+                self._queue_func("metricset", data)
+
+        if start_timer:
+            self._start_collect_timer()
+
+    def _start_collect_timer(self, timeout=None):
+        timeout = timeout or self._collect_interval
+        self._collect_timer = threading.Timer(timeout, self.collect)
+        self._collect_timer.name = "elasticapm metrics collect timer"
+        self._collect_timer.daemon = True
+        logger.info("Starting metrics collect timer")
+        self._collect_timer.start()
+
+    def _stop_collect_timer(self):
+        if self._collect_timer:
+            logger.info("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):
+        with self._lock:
+            if name not in self._counters:
+                self._counters[name] = Counter(name)
+            return self._counters[name]
+
+    def gauge(self, name):
+        with self._lock:
+            if name not in self._gauges:
+                self._gauges[name] = Gauge(name)
+            return self._gauges[name]
+
+    def collect(self):
+        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):
+    def __init__(self, label):
+        self.label = label
+        self._lock = threading.Lock()
+        self._val = 0
+
+    def inc(self, delta=1):
+        with self._lock:
+            self._val += delta
+
+    def dec(self, delta=1):
+        with self._lock:
+            self._val += delta
+
+    def reset(self):
+        with self._lock:
+            self._val = 0
+
+    @property
+    def val(self):
+        return self._val
+
+
+class Gauge(object):
+    def __init__(self, label):
+        self.label = label
+        self._val = None
+
+    @property
+    def val(self):
+        return self._val
+
+    @val.setter
+    def val(self, value):
+        self._val = value

elasticapm/metrics/base_metrics.py

@@ -0,0 +1,26 @@
+from elasticapm.metrics.base_metrics import MetricsSet
+
+try:
+    import psutil
+except ImportError:
+    raise ImportError("psutil not found. Install it to get system and process metrics")
+
+
+class CPUMetricSet(MetricsSet):
+    def __init__(self):
+        psutil.cpu_percent(interval=None)
+        super(CPUMetricSet, self).__init__()
+
+    def collect(self):
+        cpu_count = psutil.cpu_count()
+        self.gauge("system.cpu.total.norm.pct").val = psutil.cpu_percent(interval=None) / cpu_count
+        self.gauge("system.memory.actual.free").val = psutil.virtual_memory().available
+        self.gauge("system.memory.total").val = psutil.virtual_memory().total
+
+        p = psutil.Process()
+        with p.oneshot():
+            memory_info = p.memory_full_info()
+            self.gauge("system.process.cpu.total.norm.pct").val = p.cpu_percent() / cpu_count
+            self.gauge("system.process.memory.size").val = memory_info.vms
+            self.gauge("system.process.memory.rss.bytes").val = memory_info.rss
+        return super(CPUMetricSet, self).collect()

elasticapm/metrics/cpu_psutil.py

@@ -3,6 +3,7 @@
 import logging
 import threading
 import timeit
+from collections import defaultdict
 
 from elasticapm.contrib.async_worker import AsyncWorker
 from elasticapm.utils import json_encoder
@@ -56,11 +57,13 @@ def __init__(
         self._queue_lock = threading.Lock()
         self._last_flush = timeit.default_timer()
         self._flush_timer = None
+        self._counts = defaultdict(int)
 
     def queue(self, event_type, data, flush=False):
         with self._queue_lock:
             queued_data = self.queued_data
             queued_data.write((self._json_serializer({event_type: data}) + "\n").encode("utf-8"))
+            self._counts[event_type] += 1
             since_last_flush = timeit.default_timer() - self._last_flush
             queue_size = 0 if queued_data.fileobj is None else queued_data.fileobj.tell()
         if flush: