[core] Collect run-time metrics (#819)

* [metrics] initial implementation

- add data structures with tests
- add collectors for psutil and platform modules
- send metrics to agent statsd

* [metrics] add gc generation metrics

* [metrics] clean-up

* [metrics] add thread worker, additional metrics

* [metrics] linting

* [metrics] code organization

* [metrics] add runtime_id to tracer

* [metrics] resolve rebase conflicts

* [metrics] linting

* [metrics] add runtime-id tag

* [metrics] linting

* [metrics] linting

* Add environment variable for enabling runtime metrics

* Environment configuration for dogstatsd

* apply brettlinter

- thanks brett :)

Co-Authored-By: Kyle-Verhoog <[email protected]>

* [metrics] remove unnecessary LazyValues

* [metrics] in-line psutil method calls

* [metrics] use internal logger

* [metrics] add reset method, gather services

* [metrics] support multiple services properly

* [metrics] use base test case

* [metrics] handle process forking

* [metrics] add runtime metrics tags to spans

* Remove LazyValue

* Add dependencies for runtime metrics to library

* Refactor metrics collectors and add tests

* Begin major refactoring of api

* Decouple dogstatsd from runtime metrics

* Fix constant

* Fix flake8

* Separate host/port for trace agent and dogstatsd

* Update ddtrace_run tests

* Fix integration test

* Vendor datadogpy to fix issues with gevent+requests

* Revert change to on import

* Add license for dogstatsd

* Move runtime metrics into internal

* Fixes for ddtrace.internal.runtime

* Wrap worker flush in try-except to log errors

* Flush calls gauge which is a UDP so no need to catch errors

* Remove unused datadog and metrics tests

* Rename class in repr

* Remove collect_fn argument from ValueCollector

* Fix flake8

* Remove tags not called for in RFC

* Better metric names for cpu

* Use 0-1-2 for gc collections

* Comments

* Fix daemon for threading

* Add test on metrics received by dogstatsd

* Remove datadog dependency since we have it vendored

* Fix cpu metrics

* Fix cumulative metrics

* Fix reset

* Flag check unnecessary

* Fix runtime tag names

Co-Authored-By: majorgreys <[email protected]>

* Only tag root span with runtime info

* Use common namespace for gc metric names

* Remove unnecessary set check

* Wait for tests of metrics received

* Fix for constant tags and services

* Fix broken config

* Fix flake8

* Fix ddtrace-run test for runtime metrics enabled

* Update ddtrace/bootstrap/sitecustomize.py

Co-Authored-By: majorgreys <[email protected]>

Author: Kyle-Verhoog

ddtrace/bootstrap/sitecustomize.py

@@ -98,6 +98,8 @@ def add_global_tags(tracer):
     if priority_sampling:
         opts["priority_sampling"] = asbool(priority_sampling)
 
+    opts['collect_metrics'] = asbool(get_env('runtime_metrics', 'enabled'))
+
     if opts:
         tracer.configure(**opts)
 

ddtrace/internal/runtime/__init__.py

@@ -0,0 +1,12 @@
+from .runtime_metrics import (
+    RuntimeTags,
+    RuntimeMetrics,
+    RuntimeWorker,
+)
+
+
+__all__ = [
+    'RuntimeTags',
+    'RuntimeMetrics',
+    'RuntimeWorker',
+]

ddtrace/internal/runtime/collector.py

@@ -0,0 +1,85 @@
+import importlib
+
+from ..logger import get_logger
+
+log = get_logger(__name__)
+
+
+class ValueCollector(object):
+    """A basic state machine useful for collecting, caching and updating data
+    obtained from different Python modules.
+
+    The two primary use-cases are
+    1) data loaded once (like tagging information)
+    2) periodically updating data sources (like thread count)
+
+    Functionality is provided for requiring and importing modules which may or
+    may not be installed.
+    """
+    enabled = True
+    periodic = False
+    required_modules = []
+    value = None
+    value_loaded = False
+
+    def __init__(self, enabled=None, periodic=None, required_modules=None):
+        self.enabled = self.enabled if enabled is None else enabled
+        self.periodic = self.periodic if periodic is None else periodic
+        self.required_modules = self.required_modules if required_modules is None else required_modules
+
+        self._modules_successfully_loaded = False
+        self.modules = self._load_modules()
+        if self._modules_successfully_loaded:
+            self._on_modules_load()
+
+    def _on_modules_load(self):
+        """Hook triggered after all required_modules have been successfully loaded.
+        """
+
+    def _load_modules(self):
+        modules = {}
+        try:
+            for module in self.required_modules:
+                modules[module] = importlib.import_module(module)
+            self._modules_successfully_loaded = True
+        except ImportError:
+            # DEV: disable collector if we cannot load any of the required modules
+            self.enabled = False
+            log.warn('Could not import module "{}" for {}. Disabling collector.'.format(module, self))
+            return None
+        return modules
+
+    def collect(self, keys=None):
+        """Returns metrics as collected by `collect_fn`.
+
+        :param keys: The keys of the metrics to collect.
+        """
+        if not self.enabled:
+            return self.value
+
+        keys = keys or set()
+
+        if not self.periodic and self.value_loaded:
+            return self.value
+
+        # call underlying collect function and filter out keys not requested
+        self.value = self.collect_fn(keys)
+
+        # filter values for keys
+        if len(keys) > 0 and isinstance(self.value, list):
+            self.value = [
+                (k, v)
+                for (k, v) in self.value
+                if k in keys
+            ]
+
+        self.value_loaded = True
+        return self.value
+
+    def __repr__(self):
+        return '<{}(enabled={},periodic={},required_modules={})>'.format(
+            self.__class__.__name__,
+            self.enabled,
+            self.periodic,
+            self.required_modules,
+        )

ddtrace/internal/runtime/constants.py

@@ -0,0 +1,46 @@
+GC_COUNT_GEN0 = 'runtime.python.gc.count.gen0'
+GC_COUNT_GEN1 = 'runtime.python.gc.count.gen1'
+GC_COUNT_GEN2 = 'runtime.python.gc.count.gen2'
+
+THREAD_COUNT = 'runtime.python.thread_count'
+MEM_RSS = 'runtime.python.mem.rss'
+CPU_TIME_SYS = 'runtime.python.cpu.time.sys'
+CPU_TIME_USER = 'runtime.python.cpu.time.user'
+CPU_PERCENT = 'runtime.python.cpu.percent'
+CTX_SWITCH_VOLUNTARY = 'runtime.python.cpu.ctx_switch.voluntary'
+CTX_SWITCH_INVOLUNTARY = 'runtime.python.cpu.ctx_switch.involuntary'
+
+GC_RUNTIME_METRICS = set([
+    GC_COUNT_GEN0,
+    GC_COUNT_GEN1,
+    GC_COUNT_GEN2,
+])
+
+PSUTIL_RUNTIME_METRICS = set([
+    THREAD_COUNT,
+    MEM_RSS,
+    CTX_SWITCH_VOLUNTARY,
+    CTX_SWITCH_INVOLUNTARY,
+    CPU_TIME_SYS,
+    CPU_TIME_USER,
+    CPU_PERCENT,
+])
+
+DEFAULT_RUNTIME_METRICS = GC_RUNTIME_METRICS | PSUTIL_RUNTIME_METRICS
+
+RUNTIME_ID = 'runtime-id'
+SERVICE = 'service'
+LANG_INTERPRETER = 'lang_interpreter'
+LANG_VERSION = 'lang_version'
+
+TRACER_TAGS = set([
+    RUNTIME_ID,
+    SERVICE,
+])
+
+PLATFORM_TAGS = set([
+    LANG_INTERPRETER,
+    LANG_VERSION
+])
+
+DEFAULT_RUNTIME_TAGS = TRACER_TAGS

ddtrace/internal/runtime/metric_collectors.py

@@ -0,0 +1,92 @@
+import os
+
+from .collector import ValueCollector
+from .constants import (
+    GC_COUNT_GEN0,
+    GC_COUNT_GEN1,
+    GC_COUNT_GEN2,
+    THREAD_COUNT,
+    MEM_RSS,
+    CTX_SWITCH_VOLUNTARY,
+    CTX_SWITCH_INVOLUNTARY,
+    CPU_TIME_SYS,
+    CPU_TIME_USER,
+    CPU_PERCENT,
+)
+
+
+class RuntimeMetricCollector(ValueCollector):
+    value = []
+    periodic = True
+
+
+class GCRuntimeMetricCollector(RuntimeMetricCollector):
+    """ Collector for garbage collection generational counts
+
+    More information at https://docs.python.org/3/library/gc.html
+    """
+    required_modules = ['gc']
+
+    def collect_fn(self, keys):
+        gc = self.modules.get('gc')
+
+        counts = gc.get_count()
+        metrics = [
+            (GC_COUNT_GEN0, counts[0]),
+            (GC_COUNT_GEN1, counts[1]),
+            (GC_COUNT_GEN2, counts[2]),
+        ]
+
+        return metrics
+
+
+class PSUtilRuntimeMetricCollector(RuntimeMetricCollector):
+    """Collector for psutil metrics.
+
+    Performs batched operations via proc.oneshot() to optimize the calls.
+    See https://psutil.readthedocs.io/en/latest/#psutil.Process.oneshot
+    for more information.
+    """
+    required_modules = ['psutil']
+    stored_value = dict(
+        CPU_TIME_SYS_TOTAL=0,
+        CPU_TIME_USER_TOTAL=0,
+        CTX_SWITCH_VOLUNTARY_TOTAL=0,
+        CTX_SWITCH_INVOLUNTARY_TOTAL=0,
+    )
+
+    def _on_modules_load(self):
+        self.proc = self.modules['psutil'].Process(os.getpid())
+
+    def collect_fn(self, keys):
+        with self.proc.oneshot():
+            # only return time deltas
+            # TODO[tahir]: better abstraction for metrics based on last value
+            cpu_time_sys_total = self.proc.cpu_times().system
+            cpu_time_user_total = self.proc.cpu_times().user
+            cpu_time_sys = cpu_time_sys_total - self.stored_value['CPU_TIME_SYS_TOTAL']
+            cpu_time_user = cpu_time_user_total - self.stored_value['CPU_TIME_USER_TOTAL']
+
+            ctx_switch_voluntary_total = self.proc.num_ctx_switches().voluntary
+            ctx_switch_involuntary_total = self.proc.num_ctx_switches().involuntary
+            ctx_switch_voluntary = ctx_switch_voluntary_total - self.stored_value['CTX_SWITCH_VOLUNTARY_TOTAL']
+            ctx_switch_involuntary = ctx_switch_involuntary_total - self.stored_value['CTX_SWITCH_INVOLUNTARY_TOTAL']
+
+            self.stored_value = dict(
+                CPU_TIME_SYS_TOTAL=cpu_time_sys_total,
+                CPU_TIME_USER_TOTAL=cpu_time_user_total,
+                CTX_SWITCH_VOLUNTARY_TOTAL=ctx_switch_voluntary_total,
+                CTX_SWITCH_INVOLUNTARY_TOTAL=ctx_switch_involuntary_total,
+            )
+
+            metrics = [
+                (THREAD_COUNT, self.proc.num_threads()),
+                (MEM_RSS, self.proc.memory_info().rss),
+                (CTX_SWITCH_VOLUNTARY, ctx_switch_voluntary),
+                (CTX_SWITCH_INVOLUNTARY, ctx_switch_involuntary),
+                (CPU_TIME_SYS, cpu_time_sys),
+                (CPU_TIME_USER, cpu_time_user),
+                (CPU_PERCENT, self.proc.cpu_percent()),
+            ]
+
+            return metrics

ddtrace/internal/runtime/runtime_metrics.py

@@ -0,0 +1,107 @@
+import threading
+import time
+import itertools
+
+from ..logger import get_logger
+from .constants import (
+    DEFAULT_RUNTIME_METRICS,
+    DEFAULT_RUNTIME_TAGS,
+)
+from .metric_collectors import (
+    GCRuntimeMetricCollector,
+    PSUtilRuntimeMetricCollector,
+)
+from .tag_collectors import (
+    TracerTagCollector,
+)
+
+log = get_logger(__name__)
+
+
+class RuntimeCollectorsIterable(object):
+    def __init__(self, enabled=None):
+        self._enabled = enabled or self.ENABLED
+        # Initialize the collectors.
+        self._collectors = [c() for c in self.COLLECTORS]
+
+    def __iter__(self):
+        collected = (
+            collector.collect(self._enabled)
+            for collector in self._collectors
+        )
+        return itertools.chain.from_iterable(collected)
+
+    def __repr__(self):
+        return '{}(enabled={})'.format(
+            self.__class__.__name__,
+            self._enabled,
+        )
+
+
+class RuntimeTags(RuntimeCollectorsIterable):
+    ENABLED = DEFAULT_RUNTIME_TAGS
+    COLLECTORS = [
+        TracerTagCollector,
+    ]
+
+
+class RuntimeMetrics(RuntimeCollectorsIterable):
+    ENABLED = DEFAULT_RUNTIME_METRICS
+    COLLECTORS = [
+        GCRuntimeMetricCollector,
+        PSUtilRuntimeMetricCollector,
+    ]
+
+
+class RuntimeWorker(object):
+    """ Worker thread for collecting and writing runtime metrics to a DogStatsd
+        client.
+    """
+
+    FLUSH_INTERVAL = 10
+
+    def __init__(self, statsd_client, flush_interval=None):
+        self._stay_alive = None
+        self._thread = None
+        self._flush_interval = flush_interval or self.FLUSH_INTERVAL
+        self._statsd_client = statsd_client
+        self._runtime_metrics = RuntimeMetrics()
+
+    def _target(self):
+        while self._stay_alive:
+            self.flush()
+            time.sleep(self._flush_interval)
+
+    def start(self):
+        if not self._thread:
+            log.debug("Starting {}".format(self))
+            self._stay_alive = True
+            self._thread = threading.Thread(target=self._target)
+            self._thread.setDaemon(True)
+            self._thread.start()
+
+    def stop(self):
+        if self._thread and self._stay_alive:
+            log.debug("Stopping {}".format(self))
+            self._stay_alive = False
+
+    def _write_metric(self, key, value):
+        log.debug('Writing metric {}:{}'.format(key, value))
+        self._statsd_client.gauge(key, value)
+
+    def flush(self):
+        if not self._statsd_client:
+            log.warn('Attempted flush with uninitialized or failed statsd client')
+            return
+
+        for key, value in self._runtime_metrics:
+            self._write_metric(key, value)
+
+    def reset(self):
+        self._runtime_metrics = RuntimeMetrics()
+
+    def __repr__(self):
+        return '{}(runtime_metrics={})'.format(
+            self.__class__.__name__,
+            self._runtime_metrics,
+        )