Added support for breakdown metrics (#535)

closes #535
closes #479

Author: beniwohli

CHANGELOG.md

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

elasticapm/base.py

@@ -161,6 +161,7 @@ def __init__(self, config=None, **inline):
             max_spans=self.config.transaction_max_spans,
             span_frames_min_duration=self.config.span_frames_min_duration,
             ignore_patterns=self.config.transactions_ignore_patterns,
+            agent=self,
         )
         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
@@ -169,6 +170,8 @@ def __init__(self, config=None, **inline):
         )
         for path in self.config.metrics_sets:
             self._metrics.register(path)
+        if self.config.breakdown_metrics:
+            self._metrics.register("elasticapm.metrics.sets.breakdown.BreakdownMetricSet")
         compat.atexit_register(self.close)
 
     def get_handler(self, name):

elasticapm/conf/__init__.py

@@ -248,13 +248,20 @@ class Config(_ConfigBase):
             "elasticapm.processors.sanitize_http_request_body",
         ],
     )
-    metrics_sets = _ListConfigValue("METRICS_SETS", default=["elasticapm.metrics.sets.cpu.CPUMetricSet"])
+    metrics_sets = _ListConfigValue(
+        "METRICS_SETS",
+        default=[
+            "elasticapm.metrics.sets.cpu.CPUMetricSet",
+            "elasticapm.metrics.sets.transactions.TransactionsMetricSet",
+        ],
+    )
     metrics_interval = _ConfigValue(
         "METRICS_INTERVAL",
         type=int,
         validators=[duration_validator, ExcludeRangeValidator(1, 999, "{range_start} - {range_end} ms")],
         default=30000,
     )
+    breakdown_metrics = _BoolConfigValue("BREAKDOWN_METRICS", default=True)
     disable_metrics = _ListConfigValue("DISABLE_METRICS", type=starmatch_to_regex, default=[])
     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)

elasticapm/metrics/base_metrics.py

@@ -78,6 +78,12 @@ def register(self, class_path):
             except ImportError as e:
                 logger.warning("Could not register %s metricset: %s", class_path, compat.text_type(e))
 
+    def get_metricset(self, class_path):
+        try:
+            return self._metricsets[class_path]
+        except KeyError:
+            raise MetricSetNotFound(class_path)
+
     def collect(self):
         """
         Collect metrics from all registered metric sets and queues them for sending
@@ -104,48 +110,64 @@ def _stop_collect_timer(self):
 class MetricsSet(object):
     def __init__(self, registry):
         self._lock = threading.Lock()
-        self._counters = {}
-        self._gauges = {}
+        self._counters = defaultdict(dict)
+        self._gauges = defaultdict(dict)
+        self._timers = defaultdict(dict)
         self._registry = registry
 
-    def counter(self, name, **labels):
+    def counter(self, name, reset_on_collect=False, **labels):
         """
         Returns an existing or creates and returns a new counter
         :param name: name of the counter
+        :param reset_on_collect: indicate if the counter should be reset to 0 when collecting
         :param labels: a flat key/value map of labels
         :return: the counter object
         """
-        labels = self._labels_to_key(labels)
-        key = (name, labels)
-        with self._lock:
-            if key not in self._counters:
-                if self._registry._ignore_patterns and any(
-                    pattern.match(name) for pattern in self._registry._ignore_patterns
-                ):
-                    counter = noop_metric
-                else:
-                    counter = Counter(name)
-                self._counters[key] = counter
-            return self._counters[key]
+        return self._metric(self._counters, Counter, name, reset_on_collect, labels)
 
-    def gauge(self, name, **labels):
+    def gauge(self, name, reset_on_collect=False, **labels):
         """
         Returns an existing or creates and returns a new gauge
         :param name: name of the gauge
+        :param reset_on_collect: indicate if the gouge should be reset to 0 when collecting
+        :param labels: a flat key/value map of labels
         :return: the gauge object
         """
+        return self._metric(self._gauges, Gauge, name, reset_on_collect, labels)
+
+    def timer(self, name, reset_on_collect=False, **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 labels: a flat key/value map of labels
+        :return: the timer object
+        """
+        return self._metric(self._timers, Timer, name, reset_on_collect, labels)
+
+    def _metric(self, container, metric_class, name, reset_on_collect, labels):
+        """
+        Returns an existing or creates and returns a metric
+        :param container: the container for the metric
+        :param metric_class: the class of the metric
+        :param name: name of the metric
+        :param reset_on_collect: indicate if the metric should be reset to 0 when collecting
+        :param labels: a flat key/value map of labels
+        :return: the metric object
+        """
+
         labels = self._labels_to_key(labels)
         key = (name, labels)
         with self._lock:
-            if key not in self._gauges:
+            if key not in container:
                 if self._registry._ignore_patterns and any(
                     pattern.match(name) for pattern in self._registry._ignore_patterns
                 ):
-                    gauge = noop_metric
+                    metric = noop_metric
                 else:
-                    gauge = Gauge(name)
-                self._gauges[key] = gauge
-            return self._gauges[key]
+                    metric = metric_class(name, reset_on_collect=reset_on_collect)
+                container[key] = metric
+            return container[key]
 
     def collect(self):
         """
@@ -166,16 +188,28 @@ def collect(self):
             for (name, labels), c in compat.iteritems(self._counters):
                 if c is not noop_metric:
                     samples[labels].update({name: {"value": c.val}})
+                    if c.reset_on_collect:
+                        c.reset()
         if self._gauges:
             for (name, labels), g in compat.iteritems(self._gauges):
                 if g is not noop_metric:
                     samples[labels].update({name: {"value": g.val}})
+                    if g.reset_on_collect:
+                        g.reset()
+        if self._timers:
+            for (name, labels), t in compat.iteritems(self._timers):
+                if t is not noop_metric:
+                    val, count = t.val
+                    samples[labels].update({name + ".sum.us": {"value": int(val * 1000000)}})
+                    samples[labels].update({name + ".count": {"value": count}})
+                    if t.reset_on_collect:
+                        t.reset()
         if samples:
             for labels, sample in compat.iteritems(samples):
                 result = {"samples": sample, "timestamp": timestamp}
                 if labels:
                     result["tags"] = {k: v for k, v in labels}
-                yield result
+                yield self.before_yield(result)
 
     def before_collect(self):
         """
@@ -184,22 +218,39 @@ def before_collect(self):
         """
         pass
 
+    def before_yield(self, data):
+        return data
+
     def _labels_to_key(self, labels):
         return tuple((k, compat.text_type(v)) for k, v in sorted(compat.iteritems(labels)))
 
 
+class SpanBoundMetricSet(MetricsSet):
+    def before_yield(self, data):
+        tags = data.get("tags", None)
+        if tags:
+            span_type, span_subtype = tags.pop("span.type", None), tags.pop("span.subtype", "")
+            if span_type or span_subtype:
+                data["span"] = {"type": span_type, "subtype": span_subtype}
+            transaction_name, transaction_type = tags.pop("transaction.name", None), tags.pop("transaction.type", None)
+            if transaction_name or transaction_type:
+                data["transaction"] = {"name": transaction_name, "type": transaction_type}
+        return data
+
+
 class Counter(object):
-    __slots__ = ("label", "_lock", "_initial_value", "_val")
+    __slots__ = ("name", "_lock", "_initial_value", "_val", "reset_on_collect")
 
-    def __init__(self, label, initial_value=0):
+    def __init__(self, name, initial_value=0, reset_on_collect=False):
         """
         Creates a new counter
-        :param label: label of the counter
+        :param name: name of the counter
         :param initial_value: initial value of the counter, defaults to 0
         """
-        self.label = label
+        self.name = name
         self._lock = threading.Lock()
         self._val = self._initial_value = initial_value
+        self.reset_on_collect = reset_on_collect
 
     def inc(self, delta=1):
         """
@@ -237,15 +288,16 @@ def val(self):
 
 
 class Gauge(object):
-    __slots__ = ("label", "_val")
+    __slots__ = ("name", "_val", "reset_on_collect")
 
-    def __init__(self, label):
+    def __init__(self, name, reset_on_collect=False):
         """
         Creates a new gauge
-        :param label: label of the gauge
+        :param name: label of the gauge
         """
-        self.label = label
+        self.name = name
         self._val = None
+        self.reset_on_collect = reset_on_collect
 
     @property
     def val(self):
@@ -255,6 +307,35 @@ def val(self):
     def val(self, value):
         self._val = value
 
+    def reset(self):
+        self._val = 0
+
+
+class Timer(object):
+    __slots__ = ("name", "_val", "_count", "_lock", "reset_on_collect")
+
+    def __init__(self, name=None, reset_on_collect=False):
+        self.name = name
+        self._val = 0
+        self._count = 0
+        self._lock = threading.Lock()
+        self.reset_on_collect = reset_on_collect
+
+    def update(self, duration, count=1):
+        with self._lock:
+            self._val += duration
+            self._count += count
+
+    def reset(self):
+        with self._lock:
+            self._val = 0
+            self._count = 0
+
+    @property
+    def val(self):
+        with self._lock:
+            return self._val, self._count
+
 
 class NoopMetric(object):
     """
@@ -285,3 +366,8 @@ def reset(self):
 
 
 noop_metric = NoopMetric("noop")
+
+
+class MetricSetNotFound(LookupError):
+    def __init__(self, class_path):
+        super(MetricSetNotFound, self).__init__("%s metric set not found" % class_path)

elasticapm/metrics/sets/breakdown.py

@@ -0,0 +1,37 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2019, 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
+
+from elasticapm.metrics.base_metrics import SpanBoundMetricSet
+
+
+class BreakdownMetricSet(SpanBoundMetricSet):
+    pass

elasticapm/metrics/sets/transactions.py

@@ -0,0 +1,37 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2019, 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
+
+from elasticapm.metrics.base_metrics import SpanBoundMetricSet
+
+
+class TransactionsMetricSet(SpanBoundMetricSet):
+    pass