Skip to content

fix(ext/prometheus): support minmaxsumcount aggregator #945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Sep 24, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -67,22 +67,22 @@
import collections
import logging
import re
from typing import Sequence
from typing import Iterable, Optional, Sequence, Union

from prometheus_client import start_http_server
from prometheus_client.core import (
REGISTRY,
CollectorRegistry,
CounterMetricFamily,
SummaryMetricFamily,
UnknownMetricFamily,
)

from opentelemetry.metrics import Counter, Metric, ValueRecorder
from opentelemetry.metrics import Counter, ValueRecorder
from opentelemetry.sdk.metrics.export import (
MetricRecord,
MetricsExporter,
MetricsExportResult,
)
from opentelemetry.sdk.metrics.export.aggregate import MinMaxSumCountAggregator

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -110,8 +110,8 @@ def shutdown(self) -> None:


class CustomCollector:
""" CustomCollector represents the Prometheus Collector object
https://github.com/prometheus/client_python#custom-collectors
"""CustomCollector represents the Prometheus Collector object
https://github.com/prometheus/client_python#custom-collectors
"""

def __init__(self, prefix: str = ""):
Expand All @@ -121,7 +121,7 @@ def __init__(self, prefix: str = ""):
r"[^\w]", re.UNICODE | re.IGNORECASE
)

def add_metrics_data(self, metric_records: Sequence[MetricRecord]):
def add_metrics_data(self, metric_records: Sequence[MetricRecord]) -> None:
self._metrics_to_export.append(metric_records)

def collect(self):
Expand Down Expand Up @@ -152,34 +152,44 @@ def _translate_to_prometheus(self, metric_record: MetricRecord):
metric_name = self._prefix + "_"
metric_name += self._sanitize(metric_record.instrument.name)

description = getattr(metric_record.instrument, "description", "")
if isinstance(metric_record.instrument, Counter):
prometheus_metric = CounterMetricFamily(
name=metric_name,
documentation=metric_record.instrument.description,
labels=label_keys,
name=metric_name, documentation=description, labels=label_keys
)
prometheus_metric.add_metric(
labels=label_values, value=metric_record.aggregator.checkpoint
)
# TODO: Add support for histograms when supported in OT
elif isinstance(metric_record.instrument, ValueRecorder):
prometheus_metric = UnknownMetricFamily(
name=metric_name,
documentation=metric_record.instrument.description,
labels=label_keys,
)
prometheus_metric.add_metric(
labels=label_values, value=metric_record.aggregator.checkpoint
)
value = metric_record.aggregator.checkpoint
if isinstance(metric_record.aggregator, MinMaxSumCountAggregator):
prometheus_metric = SummaryMetricFamily(
name=metric_name,
documentation=description,
labels=label_keys,
)
prometheus_metric.add_metric(
labels=label_values,
count_value=value.count,
sum_value=value.sum,
)
else:
prometheus_metric = UnknownMetricFamily(
name=metric_name,
documentation=description,
labels=label_keys,
)
prometheus_metric.add_metric(labels=label_values, value=value)

else:
logger.warning(
"Unsupported metric type. %s", type(metric_record.instrument)
)
return prometheus_metric

def _sanitize(self, key):
""" sanitize the given metric name or label according to Prometheus rule.
def _sanitize(self, key: str) -> str:
"""sanitize the given metric name or label according to Prometheus rule.
Replace all characters other than [A-Za-z0-9_] with '_'.
"""
return self._non_letters_nor_digits_re.sub("_", key)
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import unittest
from unittest import mock

from prometheus_client import generate_latest
from prometheus_client.core import CounterMetricFamily

from opentelemetry.exporter.prometheus import (
Expand All @@ -24,7 +25,11 @@
from opentelemetry.metrics import get_meter_provider, set_meter_provider
from opentelemetry.sdk import metrics
from opentelemetry.sdk.metrics.export import MetricRecord, MetricsExportResult
from opentelemetry.sdk.metrics.export.aggregate import SumAggregator
from opentelemetry.sdk.metrics.export.aggregate import (
MinMaxSumCountAggregator,
SumAggregator,
)
from opentelemetry.sdk.util import get_dict_as_key


class TestPrometheusMetricExporter(unittest.TestCase):
Expand All @@ -35,7 +40,7 @@ def setUp(self):
"testname", "testdesc", "unit", int, metrics.Counter,
)
labels = {"environment": "staging"}
self._labels_key = metrics.get_dict_as_key(labels)
self._labels_key = get_dict_as_key(labels)

self._mock_registry_register = mock.Mock()
self._registry_register_patch = mock.patch(
Expand Down Expand Up @@ -70,13 +75,32 @@ def test_export(self):
self.assertEqual(len(exporter._collector._metrics_to_export), 1)
self.assertIs(result, MetricsExportResult.SUCCESS)

def test_min_max_sum_aggregator_to_prometheus(self):
meter = get_meter_provider().get_meter(__name__)
metric = meter.create_metric(
"test@name", "testdesc", "unit", int, metrics.ValueRecorder, []
)
labels = {}
key_labels = get_dict_as_key(labels)
aggregator = MinMaxSumCountAggregator()
aggregator.update(123)
aggregator.update(456)
aggregator.take_checkpoint()
record = MetricRecord(metric, key_labels, aggregator)
collector = CustomCollector("testprefix")
collector.add_metrics_data([record])
result_bytes = generate_latest(collector)
result = result_bytes.decode("utf-8")
self.assertIn("testprefix_test_name_count 2.0", result)
self.assertIn("testprefix_test_name_sum 579.0", result)

def test_counter_to_prometheus(self):
meter = get_meter_provider().get_meter(__name__)
metric = meter.create_metric(
"test@name", "testdesc", "unit", int, metrics.Counter,
)
labels = {"environment@": "staging", "os": "Windows"}
key_labels = metrics.get_dict_as_key(labels)
key_labels = get_dict_as_key(labels)
aggregator = SumAggregator()
aggregator.update(123)
aggregator.take_checkpoint()
Expand Down Expand Up @@ -107,7 +131,7 @@ def test_invalid_metric(self):
"tesname", "testdesc", "unit", int, StubMetric
)
labels = {"environment": "staging"}
key_labels = metrics.get_dict_as_key(labels)
key_labels = get_dict_as_key(labels)
record = MetricRecord(metric, key_labels, None)
collector = CustomCollector("testprefix")
collector.add_metrics_data([record])
Expand Down