http fixes, pushgateway client

Author: gistart

prometheus_push_client/__init__.py

@@ -13,6 +13,8 @@
     statsd_udp_aiostream,
     influx_udp_stream,
     statsd_udp_stream,
+    openmetrics_http_async,
+    openmetrics_http_thread,
 )
 from .formats.influx import InfluxFormat
 from .formats.statsd import StatsdFormat

prometheus_push_client/decorators.py

@@ -21,19 +21,19 @@ def influx_udp_thread(host, port, period=15.0, registry=ppc.PUSH_REGISTRY):
     return _sync_wrap(client)
 
 
-def influx_http_async(url, period=15.0, registry=ppc.PUSH_REGISTRY):
+def influx_http_async(url, verb="POST", period=15.0, registry=ppc.PUSH_REGISTRY):
     client = ppc.AsyncBatchClient(
         format=ppc.InfluxFormat(registry=registry),
-        transport=ppc.AioHttpTransport(url),
+        transport=ppc.AioHttpTransport(url, verb),
         period=period,
     )
     return _async_wrap(client)
 
 
-def influx_http_thread(url, period=15.0, registry=ppc.PUSH_REGISTRY):
+def influx_http_thread(url, verb="POST", period=15.0, registry=ppc.PUSH_REGISTRY):
     client = ppc.ThreadBatchClient(
         format=ppc.InfluxFormat(registry=registry),
-        transport=ppc.SyncHttpTransport(url),
+        transport=ppc.SyncHttpTransport(url, verb),
         period=period,
     )
     return _sync_wrap(client)
@@ -89,6 +89,24 @@ def statsd_udp_stream(host, port, registry=ppc.PUSH_REGISTRY):
     return _sync_wrap(client)
 
 
+def openmetrics_http_async(url, verb="POST", period=15.0, registry=ppc.PUSH_REGISTRY):
+    client = ppc.AsyncBatchClient(
+        format=ppc.OpenMetricsFormat(registry=registry),
+        transport=ppc.AioHttpTransport(url, verb),
+        period=period,
+    )
+    return _async_wrap(client)
+
+
+def openmetrics_http_thread(url, verb="POST", period=15.0, registry=ppc.PUSH_REGISTRY):
+    client = ppc.ThreadBatchClient(
+        format=ppc.OpenMetricsFormat(registry=registry),
+        transport=ppc.SyncHttpTransport(url, verb),
+        period=period,
+    )
+    return _sync_wrap(client)
+
+
 #
 # decorator AND context manager
 #

prometheus_push_client/formats/base.py

@@ -16,11 +16,22 @@ def iter_samples(self):
 
     def format_metrics(self, *metrics):
         for metric in metrics:
+            for line in self.format_header(metric):
+                yield _enc(line)
+
             for sample in metric.samples:
                 line = self.format_sample(sample, metric)
-                if isinstance(line, str):
-                    line = line.encode("utf-8")
-                yield line
+                yield _enc(line)
+
+    def format_header(self, metric):
+        return
+        yield  # "empty" generator by default
 
     def format_sample(self, metric, sample):
-        raise NotImplementedError()
+        raise NotImplementedError()
+
+
+def _enc(line):
+    if isinstance(line, str):
+        return line.encode("utf-8")
+    return line

prometheus_push_client/formats/openmetrics.py

@@ -7,6 +7,7 @@ class OpenMetricsFormat(BaseFormat):
     https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md
     """
 
+    FMT_DATATYPE = "# TYPE %(measurement)s %(dtype)s"
     FMT_DATAPOINT = "%(measurement)s%(tag_set)s %(value)s%(timestamp)s"
 
     ESCAPING = str.maketrans({
@@ -16,7 +17,25 @@ class OpenMetricsFormat(BaseFormat):
     })
 
 
-    def format_sample(self, sample, metric):
+    def format_header(self, metric):
+        # as in generate_lastest implemented at:
+        # https://github.com/prometheus/client_python/blob/master/prometheus_client/exposition.py
+
+        mname, mtype = metric.name, metric.type
+        if metric.type == "counter":
+            mname = f"{mname}_total"
+        elif metric.type in ("info", "stateset"):
+            mtype = "gauge"
+        elif metric.type == "gaugehistogram":
+            mtype = "histogram"
+
+        yield self.FMT_DATATYPE % dict(measurement=mname, dtype=mtype)
+        yield self.FMT_DATATYPE % dict(
+            measurement=f"{metric.name}_created",
+            dtype="gauge"
+        )
+
+    def format_sample(self, sample, _):
         tag_set = ""
         if sample.labels:
             tags = (f'{k}="{v.translate(self.ESCAPING)}"' for k, v in sample.labels.items())
@@ -26,10 +45,9 @@ def format_sample(self, sample, metric):
         if sample.timestamp:  # pragma: no cover
             ts = " %s" % sample.timestamp
 
-        # TODO: TYPE string?
         return self.FMT_DATAPOINT % dict(
             measurement=sample.name,
             tag_set=tag_set,
             value=sample.value,
             timestamp=ts,
-        )
+        )

prometheus_push_client/transports/http.py

@@ -9,21 +9,25 @@
 
 
 class BaseHttpTransport:
-    def __init__(self, url):
+    def __init__(self, url, verb="POST"):
         self._validate()
         self.url = url
+        self.verb = verb
         self.session = None
+        #TODO: basic auth, custom headers, tls ?
 
     def _validate(self):  # pragma: no cover
         pass
 
+    def prepare_data(self, iterable):
+        data = b"\n".join(iterable)
+        to_pad = sum(ord(b"\n") != char for char in data[-2:])
+        return data.ljust(len(data) + to_pad, b"\n")  # ensure ends with "\n\n"
+
     def push_all_sync(self):
         raise NotImplementedError("brave proposal")
 
 
-# TODO: configurable push formats ?
-
-
 class SyncHttpTransport(BaseHttpTransport):
     def _validate(self):
         if requests is None:
@@ -36,8 +40,8 @@ def stop(self):
         self.session.close()
 
     def push_all(self, iterable):
-        data = b"\n".join(iterable)
-        self.session.post(self.url, data=data)
+        data = self.prepare_data(iterable)
+        resp = self.session.request(self.verb, self.url, data=data)
 
 
 class AioHttpTransport(BaseHttpTransport):
@@ -52,6 +56,6 @@ async def stop(self):
         await self.session.close()
 
     async def push_all(self, iterable):
-        data = b"\n".join(iterable)
-        async with self.session.post(self.url, data=data) as _:
+        data = self.prepare_data(iterable)
+        async with self.session.request(self.verb, self.url, data=data) as _:
             pass

prometheus_push_client/version.py

@@ -1 +1 @@
-__version__ = "0.0.7"
+__version__ = "0.0.8"