Otel metrics

CHANGELOG.md

@@ -5,8 +5,9 @@
 - **[FEATURE]**: feat(plugins): change plugins to allow better customisation [#22](https://github.com/intergral/deep/pull/22) [@Umaaz](https://github.com/Umaaz)
 - **[FEATURE]**: feat(metrics): initial implementation of metric points [#21](https://github.com/intergral/deep/pull/21) [@Umaaz](https://github.com/Umaaz)
 - **[FEATURE]**: feat(spans): initial implementation of span points [#25](https://github.com/intergral/deep/pull/25) [@Umaaz](https://github.com/Umaaz)
+- **[FEATURE]**: feat(api): add api function to register tracepoint directly [#8](https://github.com/intergral/deep/pull/8) [@Umaaz](https://github.com/Umaaz)
 - **[ENHANCEMENT]**: enhancement(trigger): change tracepoint handling to use triggers [#16](https://github.com/intergral/deep/pull/16) [@Umaaz](https://github.com/Umaaz)
-- **[BUGFIX]**: feat(api): add api function to register tracepoint directly [#8](https://github.com/intergral/deep/pull/8) [@Umaaz](https://github.com/Umaaz)
+- **[BUGFIX]**: fix(metrics): correct handling of metrics in otel [#30](https://github.com/intergral/deep/pull/30) [@Umaaz](https://github.com/Umaaz)
 
 # 1.0.1 (22/06/2023)
 

src/deep/api/plugin/__init__.py

@@ -33,6 +33,7 @@
     'deep.api.plugin.otel.OTelPlugin',
     'deep.api.plugin.python.PythonPlugin',
     'deep.api.plugin.metric.prometheus_metrics.PrometheusPlugin',
+    'deep.api.plugin.metric.otel_metrics.OTelMetrics',
 ]
 """System provided default plugins."""
 

src/deep/api/plugin/metric/__init__.py

@@ -21,8 +21,10 @@
 import abc
 from typing import Dict
 
+from deep.api.plugin import Plugin
 
-class MetricProcessor(abc.ABC):
+
+class MetricProcessor(Plugin, abc.ABC):
     """Metric processor connects Deep to a metric provider."""
 
     @abc.abstractmethod

src/deep/api/plugin/metric/otel_metrics.py

@@ -0,0 +1,133 @@
+#       Copyright (C) 2024  Intergral GmbH
+#
+#      This program is free software: you can redistribute it and/or modify
+#      it under the terms of the GNU Affero General Public License as published by
+#      the Free Software Foundation, either version 3 of the License, or
+#      (at your option) any later version.
+#
+#      This program is distributed in the hope that it will be useful,
+#      but WITHOUT ANY WARRANTY; without even the implied warranty of
+#      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#      GNU Affero General Public License for more details.
+#
+#      You should have received a copy of the GNU Affero General Public License
+#      along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""Handling for OTEL metric support."""
+import threading
+from typing import Dict
+
+import deep.logging
+from deep.api.plugin import DidNotEnable
+from deep.api.plugin.metric import MetricProcessor
+
+try:
+    from opentelemetry.metrics import get_meter, Counter, Histogram, UpDownCounter
+except ImportError as e:
+    raise DidNotEnable("opentelemetry.metrics is not installed", e)
+
+
+class OTelMetrics(MetricProcessor):
+    """
+    Metric processor for otel.
+
+    Separate from OTEL Plugin for spans, as you can have one without the other.
+    """
+
+    def __init__(self, config):
+        """Create new plugin."""
+        super().__init__("OTelMetrics", config)
+        self.__cache = {}
+        self.__lock = threading.Lock()
+
+    def __check_cache(self, name, type_name, from_default):
+        cache_key = f'{name}_{type_name}'
+        if cache_key in self.__cache:
+            return self.__cache[cache_key]
+        default = from_default()
+        self.__cache[cache_key] = default
+        return default
+
+    def counter(self, name: str, labels: Dict[str, str], namespace: str, help_string: str, unit: str, value: float):
+        """
+        Create a counter type value in the provider.
+
+        :param name: the metric name
+        :param labels: the metric labels
+        :param namespace: the metric namespace
+        :param help_string: the metric help string
+        :param unit: the metric unit
+        :param value: the metric value
+        """
+        try:
+            with self.__lock:
+                counter: Counter
+                counter = self.__check_cache(name, 'counter',
+                                             lambda: get_meter('deep').create_counter(f'{namespace}_{name}', unit,
+                                                                                      help_string))
+                counter.add(value, labels)
+        except Exception:
+            deep.logging.exception(f"Error registering metric counter {namespace}_{name}")
+
+    def gauge(self, name: str, labels: Dict[str, str], namespace: str, help_string: str, unit: str, value: float):
+        """
+        Create a gauge type value in the provider.
+
+        :param name: the metric name
+        :param labels: the metric labels
+        :param namespace: the metric namespace
+        :param help_string: the metric help string
+        :param unit: the metric unit
+        :param value: the metric value
+        """
+        try:
+            with self.__lock:
+                gauge: UpDownCounter
+                gauge = self.__check_cache(name, 'gauge',
+                                           lambda: get_meter('deep').create_up_down_counter(f'{namespace}_{name}',
+                                                                                            unit, help_string))
+                gauge.add(value, labels)
+        except Exception:
+            deep.logging.exception(f"Error registering metric histogram {namespace}_{name}")
+
+    def histogram(self, name: str, labels: Dict[str, str], namespace: str, help_string: str, unit: str, value: float):
+        """
+        Create a histogram type value in the provider.
+
+        :param name: the metric name
+        :param labels: the metric labels
+        :param namespace: the metric namespace
+        :param help_string: the metric help string
+        :param unit: the metric unit
+        :param value: the metric value
+        """
+        try:
+            with self.__lock:
+                histogram: Histogram
+                histogram = self.__check_cache(name, 'histogram',
+                                               lambda: get_meter('deep').create_histogram(f'{namespace}_{name}', unit,
+                                                                                          help_string))
+                histogram.record(value, labels)
+        except Exception:
+            deep.logging.exception(f"Error registering metric histogram {namespace}_{name}")
+
+    def summary(self, name: str, labels: Dict[str, str], namespace: str, help_string: str, unit: str, value: float):
+        """
+        Create a summary type value in the provider.
+
+        :param name: the metric name
+        :param labels: the metric labels
+        :param namespace: the metric namespace
+        :param help_string: the metric help string
+        :param unit: the metric unit
+        :param value: the metric value
+        """
+        try:
+            with self.__lock:
+                histogram: Histogram
+                histogram = self.__check_cache(name, 'summary',
+                                               lambda: get_meter('deep').create_histogram(f'{namespace}_{name}', unit,
+                                                                                          help_string))
+                histogram.record(value, labels)
+        except Exception:
+            deep.logging.exception(f"Error registering metric summary {namespace}_{name}")

src/deep/api/plugin/metric/prometheus_metrics.py

@@ -19,16 +19,16 @@
 from typing import Dict
 
 import deep.logging
-from deep.api.plugin import DidNotEnable, Plugin
+from deep.api.plugin import DidNotEnable
 from deep.api.plugin.metric import MetricProcessor
 
 try:
     from prometheus_client import Summary, Counter, REGISTRY, Histogram, Gauge
 except ImportError as e:
-    raise DidNotEnable("opentelemetry is not installed", e)
+    raise DidNotEnable("prometheus_client is not installed", e)
 
 
-class PrometheusPlugin(Plugin, MetricProcessor):
+class PrometheusPlugin(MetricProcessor):
     """Connect Deep to prometheus."""
 
     def __init__(self, config):
@@ -68,7 +68,6 @@ def counter(self, name: str, labels: Dict[str, str], namespace: str, help_string
                 counter.inc(value)
         except Exception:
             deep.logging.exception(f"Error registering metric counter {namespace}_{name}")
-            pass
 
     def gauge(self, name: str, labels: Dict[str, str], namespace: str, help_string: str, unit: str, value: float):
         """
@@ -145,6 +144,10 @@ def summary(self, name: str, labels: Dict[str, str], namespace: str, help_string
             deep.logging.exception(f"Error registering metric summary {namespace}_{name}")
             pass
 
+    @property
+    def _cache(self):
+        return self.__cache
+
     def shutdown(self):
         """Clean up and shutdown the plugin."""
         self.clear()

src/deep/api/plugin/otel.py

@@ -29,7 +29,7 @@
     # noinspection PyProtectedMember
     from opentelemetry.sdk.trace import _Span, TracerProvider
 except ImportError as e:
-    raise DidNotEnable("opentelemetry is not installed", e)
+    raise DidNotEnable("opentelemetry.sdk is not installed", e)
 
 
 class _OtelSpan(Span):

tests/unit_tests/api/plugin/metrics/test_otel_metrics.py

@@ -0,0 +1,117 @@
+#       Copyright (C) 2024  Intergral GmbH
+#
+#      This program is free software: you can redistribute it and/or modify
+#      it under the terms of the GNU Affero General Public License as published by
+#      the Free Software Foundation, either version 3 of the License, or
+#      (at your option) any later version.
+#
+#      This program is distributed in the hope that it will be useful,
+#      but WITHOUT ANY WARRANTY; without even the implied warranty of
+#      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#      GNU Affero General Public License for more details.
+#
+#      You should have received a copy of the GNU Affero General Public License
+#      along with this program.  If not, see <https://www.gnu.org/licenses/>.
+import unittest
+
+from mockito import when, mock, verify
+from opentelemetry.metrics import set_meter_provider, MeterProvider
+
+import deep.logging
+from deep.api.plugin.metric.otel_metrics import OTelMetrics
+
+
+class TestOtelMetrics(unittest.TestCase):
+    mock_provider = mock(MeterProvider)
+
+    @classmethod
+    def setUpClass(cls):
+        deep.logging.init()
+
+        set_meter_provider(cls.mock_provider)
+
+    def setUp(self):
+        self.deep_provider = mock()
+        when(self.mock_provider).get_meter('deep', '').thenReturn(self.deep_provider)
+        when(self.mock_provider).get_meter('deep', '', None).thenReturn(self.deep_provider)
+
+    def test_post_counter_twice(self):
+        metric = mock()
+        when(self.deep_provider).create_counter('test_counter', "unit", "help").thenReturn(metric)
+        metrics = OTelMetrics(None)
+        metrics.counter("counter", {}, "test", "help", "unit", 1)
+        metrics.counter("counter", {}, "test", "help", "unit", 1)
+
+        verify(self.deep_provider, 1).create_counter('test_counter', "unit", "help")
+        verify(metric, 2).add(1, {})
+
+    def test_post_counter(self):
+        metric = mock()
+        when(self.deep_provider).create_counter('test_counter', "unit", "help").thenReturn(metric)
+        metrics = OTelMetrics(None)
+        metrics.counter("counter", {}, "test", "help", "unit", 1)
+
+        verify(self.deep_provider, 1).create_counter('test_counter', "unit", "help")
+
+        verify(metric).add(1, {})
+
+    def test_post_histogram(self):
+        metric = mock()
+        when(self.deep_provider).create_histogram('test_histogram', "unit", "help").thenReturn(metric)
+        metrics = OTelMetrics(None)
+        metrics.histogram("histogram", {}, "test", "help", "unit", 1)
+
+        verify(metric).record(1, {})
+
+    def test_post_gauge(self):
+        metric = mock()
+        when(self.deep_provider).create_up_down_counter('test_gauge', "unit", "help").thenReturn(metric)
+        metrics = OTelMetrics(None)
+        metrics.gauge("gauge", {}, "test", "help", "unit", 1)
+
+        verify(metric).add(1, {})
+
+    def test_post_summary(self):
+        metric = mock()
+        when(self.deep_provider).create_histogram('test_summary', "unit", "help").thenReturn(metric)
+        metrics = OTelMetrics(None)
+        metrics.summary("summary", {}, "test", "help", "unit", 1)
+
+        verify(metric).record(1, {})
+
+    def test_post_counter_error(self):
+        metric = mock()
+        when(self.deep_provider).create_counter('test_counter', "unit", "help").thenRaise(Exception("test otel error"))
+        metrics = OTelMetrics(None)
+        metrics.counter("counter", {}, "test", "help", "unit", 1)
+
+        verify(self.deep_provider, 1).create_counter('test_counter', "unit", "help")
+
+        verify(metric, 0).add(1, {})
+
+    def test_post_histogram_error(self):
+        metric = mock()
+        when(self.deep_provider).create_histogram('test_histogram', "unit", "help").thenRaise(
+            Exception("test otel error"))
+        metrics = OTelMetrics(None)
+        metrics.histogram("histogram", {}, "test", "help", "unit", 1)
+
+        verify(metric, 0).record(1, {})
+
+    def test_post_gauge_error(self):
+        metric = mock()
+        when(self.deep_provider).create_up_down_counter('test_gauge', "unit", "help").thenRaise(
+            Exception("test otel error"))
+        metrics = OTelMetrics(None)
+        metrics.gauge("gauge", {}, "test", "help", "unit", 1)
+
+        verify(metric, 0).add(1, {})
+
+    def test_post_summary_error(self):
+        metric = mock()
+        when(self.deep_provider).create_histogram('test_summary', "unit", "help").thenRaise(
+            Exception("test otel error"))
+        metrics = OTelMetrics(None)
+        metrics.summary("summary", {}, "test", "help", "unit", 1)
+
+        verify(metric, 0).record(1, {})

tests/unit_tests/api/plugin/metrics/test_prometheus_metrics.py

@@ -29,10 +29,52 @@ def tearDown(self):
 
     def test_counter(self):
         self.plugin.counter("test", {}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
 
     def test_counter_with_labels(self):
         self.plugin.counter("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
 
-    def test_duplicate_registration(self):
+    def test_duplicate_counter_registration(self):
         self.plugin.counter("test_other", {}, "deep", "", "", 123)
         self.plugin.counter("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_gauge(self):
+        self.plugin.gauge("test", {}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_gauge_with_labels(self):
+        self.plugin.gauge("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_duplicate_gauge_registration(self):
+        self.plugin.gauge("test_other", {}, "deep", "", "", 123)
+        self.plugin.gauge("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_histogram(self):
+        self.plugin.histogram("test", {}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_histogram_with_labels(self):
+        self.plugin.histogram("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_duplicate_histogram_registration(self):
+        self.plugin.histogram("test_other", {}, "deep", "", "", 123)
+        self.plugin.histogram("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_summary(self):
+        self.plugin.summary("test", {}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_summary_with_labels(self):
+        self.plugin.summary("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))
+
+    def test_duplicate_summary_registration(self):
+        self.plugin.summary("test_other", {}, "deep", "", "", 123)
+        self.plugin.summary("test_other", {'value': 'name'}, "deep", "", "", 123)
+        self.assertEqual(1, len(self.plugin._cache))