feat: add open telemetry writer (#39)

README.md

@@ -439,6 +439,22 @@ writer:
   cert: cert.pem
 ```
 
+## Otel http exporter `Otel
+
+Exports/Sends the data via OTLP (OpenTelemetry Protocol) over HTTP to a collector.
+
+```yaml
+writer:
+  type: Otel
+```
+
+You can use the common otel environment variables to configure the exporter.
+
+```bash
+export OTEL_EXPORTER_OTLP_HEADERS='Authorization=Basic xxx=='
+export OTEL_EXPORTER_OTLP_METRICS_ENDPOINT='http://localhost:4318/v1/metrics'
+```
+
 # Multithreading
 
 By default, pollect executes all sources of a collection in parallel with 5 threads.

pollect/writers/OtelWriter.py

@@ -0,0 +1,57 @@
+
+from typing import Dict, List, Optional
+from pollect.core.ValueSet import ValueSet, Value
+from pollect.sources.Source import Source
+from pollect.writers.Writer import Writer
+
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.metrics import ObservableGauge
+
+
+class OtelWriter(Writer):
+
+    def __init__(self, config):
+        super().__init__(config)
+        self.exporter = OTLPMetricExporter()
+        self.reader = PeriodicExportingMetricReader(self.exporter)
+        self.provider = MeterProvider(metric_readers=[self.reader])
+
+        self.meter = self.provider.get_meter('pollect')
+
+        self.gauges = {}
+
+    def start(self):
+        pass
+
+    def stop(self):
+        pass
+
+    def get_or_create_gauge(self, name: str):
+        """
+        Gets or creates a gauge with the given meter.
+        There is no inbuilt way to obtain a gauge from the meter, so we store them in a dictionary
+        """
+        if name in self.gauges.keys():
+            return self.gauges[name]
+
+        gauge = self.meter.create_gauge(name=name)
+        self.gauges[name] = gauge
+        return gauge
+
+    def get_attributes_from_labels(self, labels: List[str], label_values: List[str]) -> Dict[str, str]:
+        """
+        Converts the labels used for prometheus to attributes used for opentelemetry
+        """
+        attributes = {}
+        for i in range(len(labels)):
+            attributes[labels[i]] = label_values[i]
+        return attributes
+
+    def write(self, data: List[ValueSet], source_ref: Optional[Source] = None):
+        for value_set in data:
+            for value_obj in value_set.values:
+                gauge = self.get_or_create_gauge(value_set.name)
+                attributes = self.get_attributes_from_labels(value_set.labels, value_obj.label_values)
+                gauge.set(value_obj.value, attributes=attributes)

setup.py

@@ -36,7 +36,9 @@ def get_version(rel_path):
         'schedule',
         'prometheus-client',
         'PyYAML',
-        'requests'
+        'requests',
+        'opentelemetry-sdk',
+        'opentelemetry-exporter-otlp',
     ],
     entry_points={
         'console_scripts': ['pollect=pollect.Pollect:main'],

tests/writers/test_OtelWriter.py

@@ -0,0 +1,32 @@
+from unittest import TestCase
+
+
+from pollect.core.ValueSet import ValueSet
+from pollect.writers.OtelWriter import OtelWriter
+
+
+class TestOtelWriter(TestCase):
+    writer: OtelWriter = None
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        # Singleton
+        if TestOtelWriter.writer is None:
+            TestOtelWriter.writer = OtelWriter({'port': 9123})
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        TestOtelWriter
+
+    def test_get_attributes_from_labels(self):
+        labels = ['a', 'b']
+        label_values = ['1', '2']
+        attributes = TestOtelWriter.writer.get_attributes_from_labels(labels, label_values)
+
+        assert attributes == {'a': '1', 'b': '2'}
+
+    def test_get_or_create_gauge(self):
+        # Execute twice to test if on first call the gauge is created and on the second the existing gauge is returned
+        gauge = TestOtelWriter.writer.get_or_create_gauge('test')
+        gauge = TestOtelWriter.writer.get_or_create_gauge('test')
+        assert self.writer.gauges['test'] == gauge