Prometheus metrics (#1083)

* implement a Prometheus bridge

The bridge reads metrics from the Prometheus registry, and converts
them to our own metrics. This is currently supported for Counters,
Gauges, and Summaries. Histograms are not supported.

Author: beniwohli

.ci/.jenkins_framework.yml

@@ -43,3 +43,4 @@ FRAMEWORK:
   - graphene-2
   - httpx-newest
   - httplib2-newest
+  - prometheus_client-newest

docs/configuration.asciidoc

@@ -954,6 +954,38 @@ By setting this to `False`, tracking this metric is completely disabled, which c
 
 NOTE: This feature requires APM Server and Kibana >= 7.3.
 
+[float]
+[[config-prometheus_metrics]]
+==== `prometheus_metrics` (Beta)
+
+[options="header"]
+|============
+| Environment                        | Django/Flask         | Default
+| `ELASTIC_APM_PROMETHEUS_METRICS`   | `PROMETHEUS_METRICS` | `False`
+|============
+
+Enable/disable the tracking and collection of metrics from `prometheus_client`.
+
+See <<prometheus-metricset>> for more information.
+
+NOTE: This feature is currently in beta status.
+
+[float]
+[[config-prometheus_metrics_prefix]]
+==== `prometheus_metrics_prefix` (Beta)
+
+[options="header"]
+|============
+| Environment                               | Django/Flask        | Default
+| `ELASTIC_APM_PROMETHEUS_METRICS_PREFIX`   | `PROMETHEUS_METRICS_PREFIX` | `prometheus.metrics.`
+|============
+
+A prefix to prepend to Prometheus metrics names.
+
+See <<prometheus-metricset>> for more information.
+
+NOTE: This feature is currently in beta status.
+
 [float]
 [[config-central_config]]
 ==== `central_config`

docs/metrics.asciidoc

@@ -11,6 +11,7 @@ These metrics will be sent regularly to the APM Server and from there to Elastic
 * <<cpu-memory-metricset>>
 * <<transactions-metricset>>
 * <<breakdown-metricset>>
+* <<prometheus-metricset>>
 
 [float]
 [[cpu-memory-metricset]]
@@ -178,3 +179,28 @@ You can filter and group by these dimensions:
 * `span.subtype`: The sub-type of the span, for example `mysql` (optional)
 
 --
+[float]
+[[prometheus-metricset]]
+==== Prometheus metric set (beta)
+
+beta[]
+
+If you use https://github.com/prometheus/client_python[`prometheus_client`] to collect metrics, the agent can
+collect them as well and make them available in Elasticsearch.
+
+The following types of metrics are supported:
+
+ * Counters
+ * Gauges
+ * Summaries
+
+To use the Prometheus metric set, you have to enable it with the <<config-prometheus_metrics, `prometheus_metrics`>> configuration option.
+
+All metrics collected from `prometheus_client` are prefixed with `"prometheus.metrics."`. This can be changed using the <<config-prometheus_metrics_prefix, `prometheus_metrics_prefix`>> configuration option.
+
+[float]
+[[prometheus-metricset-beta]]
+===== Beta limitations
+ * The metrics format may change without backwards compatibility in future releases.
+ * Histograms are currently not supported.
+ 

elasticapm/base.py

@@ -195,6 +195,8 @@ def __init__(self, config=None, **inline):
             self._metrics.register(path)
         if self.config.breakdown_metrics:
             self._metrics.register("elasticapm.metrics.sets.breakdown.BreakdownMetricSet")
+        if self.config.prometheus_metrics:
+            self._metrics.register("elasticapm.metrics.sets.prometheus.PrometheusMetrics")
         self._thread_managers["metrics"] = self._metrics
         compat.atexit_register(self.close)
         if self.config.central_config:

elasticapm/conf/__init__.py

@@ -559,6 +559,8 @@ class Config(_ConfigBase):
         default=30000,
     )
     breakdown_metrics = _BoolConfigValue("BREAKDOWN_METRICS", default=True)
+    prometheus_metrics = _BoolConfigValue("PROMETHEUS_METRICS", default=False)
+    prometheus_metrics_prefix = _ConfigValue("PROMETHEUS_METRICS_PREFIX", default="prometheus.metrics.")
     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=768 * 1024)

elasticapm/metrics/base_metrics.py

@@ -142,17 +142,18 @@ def gauge(self, name, reset_on_collect=False, **labels):
         """
         return self._metric(self._gauges, Gauge, name, reset_on_collect, labels)
 
-    def timer(self, name, reset_on_collect=False, **labels):
+    def timer(self, name, reset_on_collect=False, unit=None, **labels):
         """
         Returns an existing or creates and returns a new timer
         :param name: name of the timer
         :param reset_on_collect: indicate if the timer should be reset to 0 when collecting
+        :param unit: Unit of the observed metric
         :param labels: a flat key/value map of labels
         :return: the timer object
         """
-        return self._metric(self._timers, Timer, name, reset_on_collect, labels)
+        return self._metric(self._timers, Timer, name, reset_on_collect, labels, unit)
 
-    def _metric(self, container, metric_class, name, reset_on_collect, labels):
+    def _metric(self, container, metric_class, name, reset_on_collect, labels, unit=None):
         """
         Returns an existing or creates and returns a metric
         :param container: the container for the metric
@@ -178,7 +179,7 @@ def _metric(self, container, metric_class, name, reset_on_collect, labels):
                         )
                     metric = noop_metric
                 else:
-                    metric = metric_class(name, reset_on_collect=reset_on_collect)
+                    metric = metric_class(name, reset_on_collect=reset_on_collect, unit=unit)
                 container[key] = metric
             return container[key]
 
@@ -219,7 +220,10 @@ def collect(self):
                 if t is not noop_metric:
                     val, count = t.val
                     if val or not t.reset_on_collect:
-                        samples[labels].update({name + ".sum.us": {"value": int(val * 1000000)}})
+                        sum_name = ".sum"
+                        if t._unit:
+                            sum_name += "." + t._unit
+                        samples[labels].update({name + sum_name: {"value": val}})
                         samples[labels].update({name + ".count": {"value": count}})
                     if t.reset_on_collect:
                         t.reset()
@@ -260,11 +264,12 @@ def before_yield(self, data):
 class Counter(object):
     __slots__ = ("name", "_lock", "_initial_value", "_val", "reset_on_collect")
 
-    def __init__(self, name, initial_value=0, reset_on_collect=False):
+    def __init__(self, name, initial_value=0, reset_on_collect=False, unit=None):
         """
         Creates a new counter
         :param name: name of the counter
         :param initial_value: initial value of the counter, defaults to 0
+        :param unit: unit of the observed counter. Unused for counters
         """
         self.name = name
         self._lock = threading.Lock()
@@ -305,14 +310,20 @@ def val(self):
         """Returns the current value of the counter"""
         return self._val
 
+    @val.setter
+    def val(self, value):
+        with self._lock:
+            self._val = value
+
 
 class Gauge(object):
     __slots__ = ("name", "_val", "reset_on_collect")
 
-    def __init__(self, name, reset_on_collect=False):
+    def __init__(self, name, reset_on_collect=False, unit=None):
         """
         Creates a new gauge
         :param name: label of the gauge
+        :param unit of the observed gauge. Unused for gauges
         """
         self.name = name
         self._val = None
@@ -331,12 +342,13 @@ def reset(self):
 
 
 class Timer(object):
-    __slots__ = ("name", "_val", "_count", "_lock", "reset_on_collect")
+    __slots__ = ("name", "_val", "_count", "_lock", "_unit", "reset_on_collect")
 
-    def __init__(self, name=None, reset_on_collect=False):
+    def __init__(self, name=None, reset_on_collect=False, unit=None):
         self.name = name
         self._val = 0
         self._count = 0
+        self._unit = unit
         self._lock = threading.Lock()
         self.reset_on_collect = reset_on_collect
 
@@ -355,6 +367,11 @@ def val(self):
         with self._lock:
             return self._val, self._count
 
+    @val.setter
+    def val(self, value):
+        with self._lock:
+            self._val, self._count = value
+
 
 class NoopMetric(object):
     """

elasticapm/metrics/sets/prometheus.py

@@ -0,0 +1,89 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2020, Elasticsearch BV
+#  All rights reserved.
+#
+#  Redistribution and use in source and binary forms, with or without
+#  modification, are permitted provided that the following conditions are met:
+#
+#  * Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+#
+#  * Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+#  * Neither the name of the copyright holder nor the names of its
+#    contributors may be used to endorse or promote products derived from
+#    this software without specific prior written permission.
+#
+#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+from __future__ import absolute_import
+
+import itertools
+
+import prometheus_client
+
+from elasticapm.metrics.base_metrics import MetricsSet
+
+
+class PrometheusMetrics(MetricsSet):
+    def __init__(self, registry):
+        super(PrometheusMetrics, self).__init__(registry)
+        self._prometheus_registry = prometheus_client.REGISTRY
+
+    def before_collect(self):
+        for metric in self._prometheus_registry.collect():
+            metric_type = self.METRIC_MAP.get(metric.type, None)
+            if not metric_type:
+                continue
+            metric_type(self, metric.name, metric.samples)
+
+    def _prom_counter_handler(self, name, samples):
+        # Counters can be converted 1:1 from Prometheus to our
+        # format. Each pair of samples represents a distinct labelset for a
+        # given name. The pair consists of the value, and a "created" timestamp.
+        # We only use the former.
+        for total_sample, _ in grouper(samples, 2):
+            self.counter(
+                self._registry.client.config.prometheus_metrics_prefix + name, **total_sample.labels
+            ).val = total_sample.value
+
+    def _prom_gauge_handler(self, name, samples):
+        # Counters can be converted 1:1 from Prometheus to our
+        # format. Each sample represents a distinct labelset for a
+        # given name
+        for sample in samples:
+            self.gauge(
+                self._registry.client.config.prometheus_metrics_prefix + name, **sample.labels
+            ).val = sample.value
+
+    def _prom_summary_handler(self, name, samples):
+        # Prometheus Summaries are analogous to our Timers, having
+        # a count and a sum. A prometheus summary is represented by
+        # three values. The list of samples for a given name can be
+        # grouped into 3-pairs of (count, sum, creation_timestamp).
+        # Each 3-pair represents a labelset.
+        for count_sample, sum_sample, _ in grouper(samples, 3):
+            self.timer(self._registry.client.config.prometheus_metrics_prefix + name, **count_sample.labels).val = (
+                sum_sample.value,
+                count_sample.value,
+            )
+
+    METRIC_MAP = {"counter": _prom_counter_handler, "gauge": _prom_gauge_handler, "summary": _prom_summary_handler}
+
+
+def grouper(iterable, n, fillvalue=None):
+    """Collect data into fixed-length chunks or blocks"""
+    # grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx"
+    args = [iter(iterable)] * n
+    return itertools.zip_longest(*args, fillvalue=fillvalue)

elasticapm/traces.py

@@ -165,8 +165,9 @@ def end(self, skip_frames=0, duration=None):
             self._transaction_metrics.timer(
                 "transaction.duration",
                 reset_on_collect=True,
+                unit="us",
                 **{"transaction.name": self.name, "transaction.type": self.transaction_type}
-            ).update(self.duration)
+            ).update(int(self.duration * 1000000))
         if self._breakdown:
             for (span_type, span_subtype), timer in compat.iteritems(self._span_timers):
                 labels = {
@@ -176,15 +177,19 @@ def end(self, skip_frames=0, duration=None):
                 }
                 if span_subtype:
                     labels["span.subtype"] = span_subtype
-                self._breakdown.timer("span.self_time", reset_on_collect=True, **labels).update(*timer.val)
+                val = timer.val
+                self._breakdown.timer("span.self_time", reset_on_collect=True, unit="us", **labels).update(
+                    int(val[0] * 1000000), val[1]
+                )
             labels = {"transaction.name": self.name, "transaction.type": self.transaction_type}
             if self.is_sampled:
                 self._breakdown.counter("transaction.breakdown.count", reset_on_collect=True, **labels).inc()
                 self._breakdown.timer(
                     "span.self_time",
                     reset_on_collect=True,
+                    unit="us",
                     **{"span.type": "app", "transaction.name": self.name, "transaction.type": self.transaction_type}
-                ).update(self.duration - self._child_durations.duration)
+                ).update(int((self.duration - self._child_durations.duration) * 1000000))
 
     def _begin_span(
         self,

setup.cfg

@@ -148,6 +148,7 @@ markers =
     starlette
     graphene
     httpx
+    prometheus_client
 
 [isort]
 line_length=120

tests/metrics/base_tests.py

@@ -150,8 +150,8 @@ def test_metrics_not_collected_if_zero_and_reset(elasticapm_client):
     resetting_counter = m.counter("resetting_counter", reset_on_collect=True)
     gauge = m.gauge("gauge", reset_on_collect=False)
     resetting_gauge = m.gauge("resetting_gauge", reset_on_collect=True)
-    timer = m.timer("timer", reset_on_collect=False)
-    resetting_timer = m.timer("resetting_timer", reset_on_collect=True)
+    timer = m.timer("timer", reset_on_collect=False, unit="us")
+    resetting_timer = m.timer("resetting_timer", reset_on_collect=True, unit="us")
 
     counter.inc(), resetting_counter.inc()
     gauge.val = 5