From e5107dff3335f81523f9b04a47eb27e46804e902 Mon Sep 17 00:00:00 2001
From: Benjamin Wohlwend <beni@elastic.co>
Date: Wed, 2 Jun 2021 10:55:51 +0200
Subject: [PATCH 1/5] catch and log exceptions in the interval timer function
Instead of letting the thread die, we log the exception with
a stack trace to ease debugging.
---
elasticapm/utils/threading.py | 17 ++++++++++++-----
tests/utils/threading_tests.py | 14 ++++++++++++++
2 files changed, 26 insertions(+), 5 deletions(-)
diff --git a/elasticapm/utils/threading.py b/elasticapm/utils/threading.py
index a36b91fda..b5d6983b8 100644
--- a/elasticapm/utils/threading.py
+++ b/elasticapm/utils/threading.py
@@ -30,10 +30,13 @@
from __future__ import absolute_import
+import logging
import os
import threading
from timeit import default_timer
+logger = logging.getLogger("elasticapm.utils.threading")
+
class IntervalTimer(threading.Thread):
"""
@@ -77,11 +80,15 @@ def run(self):
# we've been cancelled, time to go home
return
start = default_timer()
- rval = self._function(*self._args, **self._kwargs)
- if self._evaluate_function_interval and isinstance(rval, (int, float)):
- interval_override = rval
- else:
- interval_override = None
+ try:
+ rval = self._function(*self._args, **self._kwargs)
+ if self._evaluate_function_interval and isinstance(rval, (int, float)):
+ interval_override = rval
+ else:
+ interval_override = None
+ except Exception:
+ logger.error("Exception in interval timer function", exc_info=True)
+
execution_time = default_timer() - start
def cancel(self):
diff --git a/tests/utils/threading_tests.py b/tests/utils/threading_tests.py
index dc80ce5bd..451bccf6f 100644
--- a/tests/utils/threading_tests.py
+++ b/tests/utils/threading_tests.py
@@ -32,6 +32,7 @@
import mock
from elasticapm.utils.threading import IntervalTimer
+from tests.utils import assert_any_record_contains
def test_interval_timer():
@@ -75,3 +76,16 @@ def test_interval_timer_interval_override_non_number():
timer.cancel()
time.sleep(0.05)
assert not timer.is_alive()
+
+
+def test_interval_timer_exception(caplog):
+ def my_func():
+ return 1 / 0
+
+ with caplog.at_level("ERROR", "elasticapm.utils.threading"):
+ timer = IntervalTimer(function=my_func, interval=0.1)
+ timer.start()
+ time.sleep(0.25)
+ assert timer.is_alive()
+ timer.cancel()
+ assert_any_record_contains(caplog.records, "Exception in interval timer function")
From 3bc05d530f5b1629ec369ea2c6f49679e40d9373 Mon Sep 17 00:00:00 2001
From: Benjamin Wohlwend <beni@elastic.co>
Date: Mon, 14 Jun 2021 14:11:37 +0200
Subject: [PATCH 2/5] implement histogram metrics and a prometheus histogram
bridge
The spec for histograms has been merged in
https://github.com/elastic/apm-server/pull/5360
---
.../contrib/django/middleware/__init__.py | 2 +-
elasticapm/metrics/base_metrics.py | 135 +++++++++++++-----
elasticapm/metrics/sets/prometheus.py | 43 +++++-
tests/metrics/base_tests.py | 20 +++
tests/metrics/prometheus_tests.py | 31 ++++
5 files changed, 188 insertions(+), 43 deletions(-)
diff --git a/elasticapm/contrib/django/middleware/__init__.py b/elasticapm/contrib/django/middleware/__init__.py
index e270caa8d..55a2b8b3d 100644
--- a/elasticapm/contrib/django/middleware/__init__.py
+++ b/elasticapm/contrib/django/middleware/__init__.py
@@ -67,7 +67,7 @@ class ElasticAPMClientMiddlewareMixin(object):
@property
def client(self):
try:
- app = apps.get_app_config("elasticapm.contrib.django")
+ app = apps.get_app_config("elasticapm")
return app.client
except LookupError:
return get_client()
diff --git a/elasticapm/metrics/base_metrics.py b/elasticapm/metrics/base_metrics.py
index 393d4c8be..eff12f9d3 100644
--- a/elasticapm/metrics/base_metrics.py
+++ b/elasticapm/metrics/base_metrics.py
@@ -121,6 +121,7 @@ def __init__(self, registry):
self._counters = {}
self._gauges = {}
self._timers = {}
+ self._histograms = {}
self._registry = registry
self._label_limit_logged = False
@@ -155,7 +156,10 @@ def timer(self, name, reset_on_collect=False, unit=None, **labels):
"""
return self._metric(self._timers, Timer, name, reset_on_collect, labels, unit)
- def _metric(self, container, metric_class, name, reset_on_collect, labels, unit=None):
+ def histogram(self, name, reset_on_collect=False, unit=None, buckets=None, **labels):
+ return self._metric(self._histograms, Histogram, name, reset_on_collect, labels, unit, buckets=buckets)
+
+ def _metric(self, container, metric_class, name, reset_on_collect, labels, unit=None, **kwargs):
"""
Returns an existing or creates and returns a metric
:param container: the container for the metric
@@ -172,7 +176,10 @@ def _metric(self, container, metric_class, name, reset_on_collect, labels, unit=
if key not in container:
if any(pattern.match(name) for pattern in self._registry.ignore_patterns):
metric = noop_metric
- elif len(self._gauges) + len(self._counters) + len(self._timers) >= DISTINCT_LABEL_LIMIT:
+ elif (
+ len(self._gauges) + len(self._counters) + len(self._timers) + len(self._histograms)
+ >= DISTINCT_LABEL_LIMIT
+ ):
if not self._label_limit_logged:
self._label_limit_logged = True
logger.warning(
@@ -181,7 +188,7 @@ def _metric(self, container, metric_class, name, reset_on_collect, labels, unit=
)
metric = noop_metric
else:
- metric = metric_class(name, reset_on_collect=reset_on_collect, unit=unit)
+ metric = metric_class(name, reset_on_collect=reset_on_collect, unit=unit, **kwargs)
container[key] = metric
return container[key]
@@ -202,33 +209,42 @@ def collect(self):
samples = defaultdict(dict)
if self._counters:
# iterate over a copy of the dict to avoid threading issues, see #717
- for (name, labels), c in compat.iteritems(self._counters.copy()):
- if c is not noop_metric:
- val = c.val
- if val or not c.reset_on_collect:
- samples[labels].update({name: {"value": val}})
- if c.reset_on_collect:
- c.reset()
+ for (name, labels), counter in compat.iteritems(self._counters.copy()):
+ if counter is not noop_metric:
+ val = counter.val
+ if val or not counter.reset_on_collect:
+ samples[labels].update({name: {"value": val, "type": "counter"}})
+ if counter.reset_on_collect:
+ counter.reset()
if self._gauges:
- for (name, labels), g in compat.iteritems(self._gauges.copy()):
- if g is not noop_metric:
- val = g.val
- if val or not g.reset_on_collect:
- samples[labels].update({name: {"value": val}})
- if g.reset_on_collect:
- g.reset()
+ for (name, labels), gauge in compat.iteritems(self._gauges.copy()):
+ if gauge is not noop_metric:
+ val = gauge.val
+ if val or not gauge.reset_on_collect:
+ samples[labels].update({name: {"value": val, "type": "gauge"}})
+ if gauge.reset_on_collect:
+ gauge.reset()
if self._timers:
- for (name, labels), t in compat.iteritems(self._timers.copy()):
- if t is not noop_metric:
- val, count = t.val
- if val or not t.reset_on_collect:
+ for (name, labels), timer in compat.iteritems(self._timers.copy()):
+ if timer is not noop_metric:
+ val, count = timer.val
+ if val or not timer.reset_on_collect:
sum_name = ".sum"
- if t._unit:
- sum_name += "." + t._unit
+ if timer._unit:
+ sum_name += "." + timer._unit
samples[labels].update({name + sum_name: {"value": val}})
samples[labels].update({name + ".count": {"value": count}})
- if t.reset_on_collect:
- t.reset()
+ if timer.reset_on_collect:
+ timer.reset()
+ if self._histograms:
+ for (name, labels), histo in compat.iteritems(self._histograms.copy()):
+ if histo is not noop_metric:
+ counts = histo.val
+ if counts or not histo.reset_on_collect:
+ samples[labels].update({name: {"counts": counts, "values": histo.buckets, "type": "histogram"}})
+ if histo.reset_on_collect:
+ histo.reset()
+
if samples:
for labels, sample in compat.iteritems(samples):
result = {"samples": sample, "timestamp": timestamp}
@@ -263,8 +279,16 @@ def before_yield(self, data):
return data
-class Counter(object):
- __slots__ = ("name", "_lock", "_initial_value", "_val", "reset_on_collect")
+class BaseMetric(object):
+ __slots__ = ("name", "reset_on_collect")
+
+ def __init__(self, name, reset_on_collect=False, **kwargs):
+ self.name = name
+ self.reset_on_collect = reset_on_collect
+
+
+class Counter(BaseMetric):
+ __slots__ = BaseMetric.__slots__ + ("_lock", "_initial_value", "_val")
def __init__(self, name, initial_value=0, reset_on_collect=False, unit=None):
"""
@@ -273,10 +297,9 @@ def __init__(self, name, initial_value=0, reset_on_collect=False, unit=None):
: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()
self._val = self._initial_value = initial_value
- self.reset_on_collect = reset_on_collect
+ super(Counter, self).__init__(name, reset_on_collect=reset_on_collect)
def inc(self, delta=1):
"""
@@ -318,8 +341,8 @@ def val(self, value):
self._val = value
-class Gauge(object):
- __slots__ = ("name", "_val", "reset_on_collect")
+class Gauge(BaseMetric):
+ __slots__ = BaseMetric.__slots__ + ("_val",)
def __init__(self, name, reset_on_collect=False, unit=None):
"""
@@ -327,9 +350,8 @@ def __init__(self, name, reset_on_collect=False, unit=None):
:param name: label of the gauge
:param unit of the observed gauge. Unused for gauges
"""
- self.name = name
self._val = None
- self.reset_on_collect = reset_on_collect
+ super(Gauge, self).__init__(name, reset_on_collect=reset_on_collect)
@property
def val(self):
@@ -343,16 +365,15 @@ def reset(self):
self._val = 0
-class Timer(object):
- __slots__ = ("name", "_val", "_count", "_lock", "_unit", "reset_on_collect")
+class Timer(BaseMetric):
+ __slots__ = BaseMetric.__slots__ + ("_val", "_count", "_lock", "_unit")
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
+ super(Timer, self).__init__(name, reset_on_collect=reset_on_collect)
def update(self, duration, count=1):
with self._lock:
@@ -375,6 +396,46 @@ def val(self, value):
self._val, self._count = value
+class Histogram(BaseMetric):
+ DEFAULT_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, float("inf"))
+
+ __slots__ = BaseMetric.__slots__ + ("_lock", "_buckets", "_counts", "_lock", "_unit")
+
+ def __init__(self, name=None, reset_on_collect=False, unit=None, buckets=None):
+ self._lock = threading.Lock()
+ self._buckets = buckets or Histogram.DEFAULT_BUCKETS
+ if self._buckets[-1] < float("inf"):
+ self._buckets.append(float("inf"))
+ self._counts = [0] * len(self._buckets)
+ self._unit = unit
+ super(Histogram, self).__init__(name, reset_on_collect=reset_on_collect)
+
+ def update(self, value, count=1):
+ pos = 0
+ while value > self._buckets[pos]:
+ pos += 1
+ with self._lock:
+ self._counts[pos] += count
+
+ @property
+ def val(self):
+ with self._lock:
+ return self._counts
+
+ @val.setter
+ def val(self, value):
+ with self._lock:
+ self._counts = value
+
+ @property
+ def buckets(self):
+ return self._buckets
+
+ def reset(self):
+ with self._lock:
+ self._counts = [0] * len(self._buckets)
+
+
class NoopMetric(object):
"""
A no-op metric that implements the "interface" of both Counter and Gauge.
diff --git a/elasticapm/metrics/sets/prometheus.py b/elasticapm/metrics/sets/prometheus.py
index 894de159c..2dffce812 100644
--- a/elasticapm/metrics/sets/prometheus.py
+++ b/elasticapm/metrics/sets/prometheus.py
@@ -46,9 +46,9 @@ def before_collect(self):
metric_type = self.METRIC_MAP.get(metric.type, None)
if not metric_type:
continue
- metric_type(self, metric.name, metric.samples)
+ metric_type(self, metric.name, metric.samples, metric.unit)
- def _prom_counter_handler(self, name, samples):
+ def _prom_counter_handler(self, name, samples, unit):
# 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.
@@ -58,7 +58,7 @@ def _prom_counter_handler(self, name, samples):
self._registry.client.config.prometheus_metrics_prefix + name, **total_sample.labels
).val = total_sample.value
- def _prom_gauge_handler(self, name, samples):
+ def _prom_gauge_handler(self, name, samples, unit):
# Counters can be converted 1:1 from Prometheus to our
# format. Each sample represents a distinct labelset for a
# given name
@@ -67,7 +67,7 @@ def _prom_gauge_handler(self, name, samples):
self._registry.client.config.prometheus_metrics_prefix + name, **sample.labels
).val = sample.value
- def _prom_summary_handler(self, name, samples):
+ def _prom_summary_handler(self, name, samples, unit):
# 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
@@ -79,7 +79,40 @@ def _prom_summary_handler(self, name, samples):
count_sample.value,
)
- METRIC_MAP = {"counter": _prom_counter_handler, "gauge": _prom_gauge_handler, "summary": _prom_summary_handler}
+ def _prom_histogram_handler(self, name, samples, unit):
+ # Prometheus histograms are structured as a series of counts
+ # with an "le" label. The count of each label signifies all
+ # observations with a lower-or-equal value with respect to
+ # the "le" label value.
+ # After the le-samples, 3 more samples follow, with an overall
+ # count, overall sum, and creation timestamp.
+ sample_pos = 0
+ prev_val = 0
+ counts = []
+ values = []
+ name = self._registry.client.config.prometheus_metrics_prefix + name
+ while sample_pos < len(samples):
+ sample = samples[sample_pos]
+ if "le" in sample.labels:
+ values.append(float(sample.labels["le"]))
+ counts.append(sample.value - prev_val)
+ prev_val = sample.value
+ sample_pos += 1
+
+ else:
+ # we reached the end of one set of buckets/values, this is the "count" sample
+ self.histogram(name, unit=unit, buckets=values, **sample.labels).val = counts
+ prev_val = 0
+ counts = []
+ values = []
+ sample_pos += 3 # skip sum/created samples
+
+ METRIC_MAP = {
+ "counter": _prom_counter_handler,
+ "gauge": _prom_gauge_handler,
+ "summary": _prom_summary_handler,
+ "histogram": _prom_histogram_handler,
+ }
def grouper(iterable, n, fillvalue=None):
diff --git a/tests/metrics/base_tests.py b/tests/metrics/base_tests.py
index 36e4b3223..00ee3d19a 100644
--- a/tests/metrics/base_tests.py
+++ b/tests/metrics/base_tests.py
@@ -86,6 +86,26 @@ def test_metrics_counter(elasticapm_client):
assert data["samples"]["x"]["value"] == 0
+def test_metrics_histogram(elasticapm_client):
+ metricset = MetricsSet(MetricsRegistry(elasticapm_client))
+ hist = metricset.histogram("x", buckets=[1, 10, 100])
+ assert len(hist.buckets) == 4
+ assert hist.buckets[3] == float("inf")
+
+ hist.update(0.3)
+ hist.update(1)
+ hist.update(5)
+ hist.update(20)
+ hist.update(100)
+ hist.update(1000)
+
+ data = list(metricset.collect())
+ assert len(data) == 1
+ d = data[0]
+ assert d["samples"]["x"]["counts"] == [2, 1, 2, 1]
+ assert d["samples"]["x"]["values"] == [1, 10, 100, float("inf")]
+
+
def test_metrics_labels(elasticapm_client):
metricset = MetricsSet(MetricsRegistry(elasticapm_client))
metricset.counter("x", mylabel="a").inc()
diff --git a/tests/metrics/prometheus_tests.py b/tests/metrics/prometheus_tests.py
index 756465e21..67a2912b2 100644
--- a/tests/metrics/prometheus_tests.py
+++ b/tests/metrics/prometheus_tests.py
@@ -104,3 +104,34 @@ def test_summary(elasticapm_client):
assert data[2]["samples"]["prometheus.metrics.summary_with_labels.count"]["value"] == 1.0
assert data[2]["samples"]["prometheus.metrics.summary_with_labels.sum"]["value"] == 11.0
assert data[2]["tags"] == {"alabel": "bar", "anotherlabel": "bazzinga"}
+
+
+def test_histogram(elasticapm_client):
+ metricset = PrometheusMetrics(MetricsRegistry(elasticapm_client))
+ histo = prometheus_client.Histogram("test", "test histogram", buckets=[1, 10, 100, float("inf")])
+ histo_with_labels = prometheus_client.Histogram(
+ "testwithlabel", "test histogram with labels", ["alabel", "anotherlabel"], buckets=[1, 10, 100, float("inf")]
+ )
+ histo.observe(0.5)
+ histo.observe(0.6)
+ histo.observe(1.5)
+ histo.observe(26)
+ histo.observe(42)
+ histo.observe(12)
+ histo.observe(105)
+ histo_with_labels.labels(alabel="foo", anotherlabel="baz").observe(1)
+ histo_with_labels.labels(alabel="foo", anotherlabel="baz").observe(10)
+ histo_with_labels.labels(alabel="foo", anotherlabel="baz").observe(100)
+ histo_with_labels.labels(alabel="foo", anotherlabel="bazzinga").observe(1000)
+ data = list(metricset.collect())
+ assert len(data) == 3
+ assert data[0]["samples"]["prometheus.metrics.test"]["values"] == [1, 10, 100, float("inf")]
+ assert data[0]["samples"]["prometheus.metrics.test"]["counts"] == [2, 1, 3, 1]
+
+ assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [1, 10, 100, float("inf")]
+ assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [1, 1, 1, 0]
+ assert data[1]["tags"] == {"alabel": "foo", "anotherlabel": "baz"}
+
+ assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [1, 10, 100, float("inf")]
+ assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [0, 0, 0, 1]
+ assert data[2]["tags"] == {"alabel": "foo", "anotherlabel": "bazzinga"}
From 7626a9606c4898a63e02a3da6b01ccb96b7e77ae Mon Sep 17 00:00:00 2001
From: Benjamin Wohlwend <beni@elastic.co>
Date: Mon, 14 Jun 2021 17:12:11 +0200
Subject: [PATCH 3/5] trying to debug failure that only happens on CI...
---
tests/metrics/prometheus_tests.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/tests/metrics/prometheus_tests.py b/tests/metrics/prometheus_tests.py
index 67a2912b2..5f8b9a5c6 100644
--- a/tests/metrics/prometheus_tests.py
+++ b/tests/metrics/prometheus_tests.py
@@ -124,7 +124,7 @@ def test_histogram(elasticapm_client):
histo_with_labels.labels(alabel="foo", anotherlabel="baz").observe(100)
histo_with_labels.labels(alabel="foo", anotherlabel="bazzinga").observe(1000)
data = list(metricset.collect())
- assert len(data) == 3
+ assert len(data) == 3, data
assert data[0]["samples"]["prometheus.metrics.test"]["values"] == [1, 10, 100, float("inf")]
assert data[0]["samples"]["prometheus.metrics.test"]["counts"] == [2, 1, 3, 1]
From 99ffc0d6339179809fe49a0ecb126c1f52bfe5b4 Mon Sep 17 00:00:00 2001
From: Benjamin Wohlwend <beni@elastic.co>
Date: Wed, 7 Jul 2021 13:15:56 +0200
Subject: [PATCH 4/5] adapt prometheus histograms to use centroids instead of
upper limit buckets
---
elasticapm/metrics/base_metrics.py | 4 +---
elasticapm/metrics/sets/prometheus.py | 27 ++++++++++++++++++---------
tests/metrics/prometheus_tests.py | 6 +++---
3 files changed, 22 insertions(+), 15 deletions(-)
diff --git a/elasticapm/metrics/base_metrics.py b/elasticapm/metrics/base_metrics.py
index eff12f9d3..2847d9a1c 100644
--- a/elasticapm/metrics/base_metrics.py
+++ b/elasticapm/metrics/base_metrics.py
@@ -213,7 +213,7 @@ def collect(self):
if counter is not noop_metric:
val = counter.val
if val or not counter.reset_on_collect:
- samples[labels].update({name: {"value": val, "type": "counter"}})
+ samples[labels].update({name: {"value": val}})
if counter.reset_on_collect:
counter.reset()
if self._gauges:
@@ -404,8 +404,6 @@ class Histogram(BaseMetric):
def __init__(self, name=None, reset_on_collect=False, unit=None, buckets=None):
self._lock = threading.Lock()
self._buckets = buckets or Histogram.DEFAULT_BUCKETS
- if self._buckets[-1] < float("inf"):
- self._buckets.append(float("inf"))
self._counts = [0] * len(self._buckets)
self._unit = unit
super(Histogram, self).__init__(name, reset_on_collect=reset_on_collect)
diff --git a/elasticapm/metrics/sets/prometheus.py b/elasticapm/metrics/sets/prometheus.py
index 2dffce812..a75abb127 100644
--- a/elasticapm/metrics/sets/prometheus.py
+++ b/elasticapm/metrics/sets/prometheus.py
@@ -87,24 +87,33 @@ def _prom_histogram_handler(self, name, samples, unit):
# After the le-samples, 3 more samples follow, with an overall
# count, overall sum, and creation timestamp.
sample_pos = 0
- prev_val = 0
- counts = []
- values = []
name = self._registry.client.config.prometheus_metrics_prefix + name
+ hist_samples = []
while sample_pos < len(samples):
sample = samples[sample_pos]
if "le" in sample.labels:
- values.append(float(sample.labels["le"]))
- counts.append(sample.value - prev_val)
- prev_val = sample.value
+ hist_samples.append((float(sample.labels["le"]), sample))
sample_pos += 1
-
else:
# we reached the end of one set of buckets/values, this is the "count" sample
- self.histogram(name, unit=unit, buckets=values, **sample.labels).val = counts
- prev_val = 0
counts = []
values = []
+ prev_sample = None
+ prev_le = 0
+ for i, (le, hist_sample) in enumerate(hist_samples):
+ if i == 0:
+ val = le if le < 0 else le / 2.0
+ elif le == float("+Inf"):
+ val = prev_le
+ else:
+ val = prev_le + (le - prev_le) / 2.0
+ values.append(val)
+ counts.append(int(hist_sample.value - (prev_sample.value if prev_sample else 0)))
+ prev_le = le
+ prev_sample = hist_sample
+ self.histogram(name, unit=unit, buckets=values, **sample.labels).val = counts
+
+ hist_samples = []
sample_pos += 3 # skip sum/created samples
METRIC_MAP = {
diff --git a/tests/metrics/prometheus_tests.py b/tests/metrics/prometheus_tests.py
index 5f8b9a5c6..7ac00cbf2 100644
--- a/tests/metrics/prometheus_tests.py
+++ b/tests/metrics/prometheus_tests.py
@@ -125,13 +125,13 @@ def test_histogram(elasticapm_client):
histo_with_labels.labels(alabel="foo", anotherlabel="bazzinga").observe(1000)
data = list(metricset.collect())
assert len(data) == 3, data
- assert data[0]["samples"]["prometheus.metrics.test"]["values"] == [1, 10, 100, float("inf")]
+ assert data[0]["samples"]["prometheus.metrics.test"]["values"] == [0.5, 5.5, 55.0, 100.0]
assert data[0]["samples"]["prometheus.metrics.test"]["counts"] == [2, 1, 3, 1]
- assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [1, 10, 100, float("inf")]
+ assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [1, 1, 1, 0]
assert data[1]["tags"] == {"alabel": "foo", "anotherlabel": "baz"}
- assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [1, 10, 100, float("inf")]
+ assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [0, 0, 0, 1]
assert data[2]["tags"] == {"alabel": "foo", "anotherlabel": "bazzinga"}
From ff607b156477dd9c3e0ab4c61d685e79cd3be26a Mon Sep 17 00:00:00 2001
From: Benjamin Wohlwend <beni@elastic.co>
Date: Mon, 19 Jul 2021 11:54:49 +0200
Subject: [PATCH 5/5] move midpoint calculation into base metrics
---
elasticapm/metrics/base_metrics.py | 27 +++++++++++++++++++++-
elasticapm/metrics/sets/prometheus.py | 27 ++++++++--------------
tests/metrics/base_tests.py | 3 +--
tests/metrics/prometheus_tests.py | 32 ++++++++++++++++-----------
4 files changed, 55 insertions(+), 34 deletions(-)
diff --git a/elasticapm/metrics/base_metrics.py b/elasticapm/metrics/base_metrics.py
index 2847d9a1c..c3320bd8a 100644
--- a/elasticapm/metrics/base_metrics.py
+++ b/elasticapm/metrics/base_metrics.py
@@ -241,7 +241,30 @@ def collect(self):
if histo is not noop_metric:
counts = histo.val
if counts or not histo.reset_on_collect:
- samples[labels].update({name: {"counts": counts, "values": histo.buckets, "type": "histogram"}})
+ # For the bucket values, we follow the approach described by Prometheus's
+ # histogram_quantile function
+ # (https://prometheus.io/docs/prometheus/latest/querying/functions/#histogram_quantile)
+ # to achieve consistent percentile aggregation results:
+ #
+ # "The histogram_quantile() function interpolates quantile values by assuming a linear
+ # distribution within a bucket. (...) If a quantile is located in the highest bucket,
+ # the upper bound of the second highest bucket is returned. A lower limit of the lowest
+ # bucket is assumed to be 0 if the upper bound of that bucket is greater than 0. In that
+ # case, the usual linear interpolation is applied within that bucket. Otherwise, the upper
+ # bound of the lowest bucket is returned for quantiles located in the lowest bucket."
+ bucket_midpoints = []
+ for i, bucket_le in enumerate(histo.buckets):
+ if i == 0:
+ if bucket_le > 0:
+ bucket_le /= 2.0
+ elif i == len(histo.buckets) - 1:
+ bucket_le = histo.buckets[i - 1]
+ else:
+ bucket_le = histo.buckets[i - 1] + (bucket_le - histo.buckets[i - 1]) / 2.0
+ bucket_midpoints.append(bucket_le)
+ samples[labels].update(
+ {name: {"counts": counts, "values": bucket_midpoints, "type": "histogram"}}
+ )
if histo.reset_on_collect:
histo.reset()
@@ -404,6 +427,8 @@ class Histogram(BaseMetric):
def __init__(self, name=None, reset_on_collect=False, unit=None, buckets=None):
self._lock = threading.Lock()
self._buckets = buckets or Histogram.DEFAULT_BUCKETS
+ if self._buckets[-1] != float("inf"):
+ self._buckets.append(float("inf"))
self._counts = [0] * len(self._buckets)
self._unit = unit
super(Histogram, self).__init__(name, reset_on_collect=reset_on_collect)
diff --git a/elasticapm/metrics/sets/prometheus.py b/elasticapm/metrics/sets/prometheus.py
index a75abb127..2dffce812 100644
--- a/elasticapm/metrics/sets/prometheus.py
+++ b/elasticapm/metrics/sets/prometheus.py
@@ -87,33 +87,24 @@ def _prom_histogram_handler(self, name, samples, unit):
# After the le-samples, 3 more samples follow, with an overall
# count, overall sum, and creation timestamp.
sample_pos = 0
+ prev_val = 0
+ counts = []
+ values = []
name = self._registry.client.config.prometheus_metrics_prefix + name
- hist_samples = []
while sample_pos < len(samples):
sample = samples[sample_pos]
if "le" in sample.labels:
- hist_samples.append((float(sample.labels["le"]), sample))
+ values.append(float(sample.labels["le"]))
+ counts.append(sample.value - prev_val)
+ prev_val = sample.value
sample_pos += 1
+
else:
# we reached the end of one set of buckets/values, this is the "count" sample
+ self.histogram(name, unit=unit, buckets=values, **sample.labels).val = counts
+ prev_val = 0
counts = []
values = []
- prev_sample = None
- prev_le = 0
- for i, (le, hist_sample) in enumerate(hist_samples):
- if i == 0:
- val = le if le < 0 else le / 2.0
- elif le == float("+Inf"):
- val = prev_le
- else:
- val = prev_le + (le - prev_le) / 2.0
- values.append(val)
- counts.append(int(hist_sample.value - (prev_sample.value if prev_sample else 0)))
- prev_le = le
- prev_sample = hist_sample
- self.histogram(name, unit=unit, buckets=values, **sample.labels).val = counts
-
- hist_samples = []
sample_pos += 3 # skip sum/created samples
METRIC_MAP = {
diff --git a/tests/metrics/base_tests.py b/tests/metrics/base_tests.py
index 00ee3d19a..36c925bec 100644
--- a/tests/metrics/base_tests.py
+++ b/tests/metrics/base_tests.py
@@ -90,7 +90,6 @@ def test_metrics_histogram(elasticapm_client):
metricset = MetricsSet(MetricsRegistry(elasticapm_client))
hist = metricset.histogram("x", buckets=[1, 10, 100])
assert len(hist.buckets) == 4
- assert hist.buckets[3] == float("inf")
hist.update(0.3)
hist.update(1)
@@ -103,7 +102,7 @@ def test_metrics_histogram(elasticapm_client):
assert len(data) == 1
d = data[0]
assert d["samples"]["x"]["counts"] == [2, 1, 2, 1]
- assert d["samples"]["x"]["values"] == [1, 10, 100, float("inf")]
+ assert d["samples"]["x"]["values"] == [0.5, 5.5, 55.0, 100]
def test_metrics_labels(elasticapm_client):
diff --git a/tests/metrics/prometheus_tests.py b/tests/metrics/prometheus_tests.py
index 7ac00cbf2..0308c2cf9 100644
--- a/tests/metrics/prometheus_tests.py
+++ b/tests/metrics/prometheus_tests.py
@@ -45,7 +45,14 @@
prometheus_client.REGISTRY.unregister(prometheus_client.GC_COLLECTOR)
-def test_counter(elasticapm_client):
+@pytest.fixture()
+def prometheus():
+ # reset registry
+ prometheus_client.REGISTRY._collector_to_names = {}
+ prometheus_client.REGISTRY._names_to_collectors = {}
+
+
+def test_counter(elasticapm_client, prometheus):
metricset = PrometheusMetrics(MetricsRegistry(elasticapm_client))
counter = prometheus_client.Counter("a_bare_counter", "Bare counter")
counter_with_labels = prometheus_client.Counter(
@@ -64,7 +71,7 @@ def test_counter(elasticapm_client):
assert data[2]["tags"] == {"alabel": "bar", "anotherlabel": "bazzinga"}
-def test_gauge(elasticapm_client):
+def test_gauge(elasticapm_client, prometheus):
metricset = PrometheusMetrics(MetricsRegistry(elasticapm_client))
gauge = prometheus_client.Gauge("a_bare_gauge", "Bare gauge")
gauge_with_labels = prometheus_client.Gauge("gauge_with_labels", "Gauge with labels", ["alabel", "anotherlabel"])
@@ -81,7 +88,7 @@ def test_gauge(elasticapm_client):
assert data[2]["tags"] == {"alabel": "bar", "anotherlabel": "bazzinga"}
-def test_summary(elasticapm_client):
+def test_summary(elasticapm_client, prometheus):
metricset = PrometheusMetrics(MetricsRegistry(elasticapm_client))
summary = prometheus_client.Summary("a_bare_summary", "Bare summary")
summary_with_labels = prometheus_client.Summary(
@@ -106,11 +113,11 @@ def test_summary(elasticapm_client):
assert data[2]["tags"] == {"alabel": "bar", "anotherlabel": "bazzinga"}
-def test_histogram(elasticapm_client):
+def test_histogram(elasticapm_client, prometheus):
metricset = PrometheusMetrics(MetricsRegistry(elasticapm_client))
- histo = prometheus_client.Histogram("test", "test histogram", buckets=[1, 10, 100, float("inf")])
+ histo = prometheus_client.Histogram("histo", "test histogram", buckets=[1, 10, 100, float("inf")])
histo_with_labels = prometheus_client.Histogram(
- "testwithlabel", "test histogram with labels", ["alabel", "anotherlabel"], buckets=[1, 10, 100, float("inf")]
+ "histowithlabel", "test histogram with labels", ["alabel", "anotherlabel"], buckets=[1, 10, 100, float("inf")]
)
histo.observe(0.5)
histo.observe(0.6)
@@ -124,14 +131,13 @@ def test_histogram(elasticapm_client):
histo_with_labels.labels(alabel="foo", anotherlabel="baz").observe(100)
histo_with_labels.labels(alabel="foo", anotherlabel="bazzinga").observe(1000)
data = list(metricset.collect())
- assert len(data) == 3, data
- assert data[0]["samples"]["prometheus.metrics.test"]["values"] == [0.5, 5.5, 55.0, 100.0]
- assert data[0]["samples"]["prometheus.metrics.test"]["counts"] == [2, 1, 3, 1]
+ assert data[0]["samples"]["prometheus.metrics.histo"]["values"] == [0.5, 5.5, 55.0, 100.0]
+ assert data[0]["samples"]["prometheus.metrics.histo"]["counts"] == [2, 1, 3, 1]
- assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
- assert data[1]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [1, 1, 1, 0]
+ assert data[1]["samples"]["prometheus.metrics.histowithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
+ assert data[1]["samples"]["prometheus.metrics.histowithlabel"]["counts"] == [1, 1, 1, 0]
assert data[1]["tags"] == {"alabel": "foo", "anotherlabel": "baz"}
- assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
- assert data[2]["samples"]["prometheus.metrics.testwithlabel"]["counts"] == [0, 0, 0, 1]
+ assert data[2]["samples"]["prometheus.metrics.histowithlabel"]["values"] == [0.5, 5.5, 55.0, 100.0]
+ assert data[2]["samples"]["prometheus.metrics.histowithlabel"]["counts"] == [0, 0, 0, 1]
assert data[2]["tags"] == {"alabel": "foo", "anotherlabel": "bazzinga"}