Merge pull request #2 from gistart/feat/pushgw

http fixes, pushgateway client

Author: gistart

.github/workflows/tests.yml

@@ -31,6 +31,10 @@ jobs:
         - 9102:9102
         - 9125:9125
         - 9125:9125/udp
+      pushgateway:
+        image: prom/pushgateway:v1.4.0
+        ports:
+        - 9091:9091
     steps:
     - name: checkout
       uses: actions/checkout@v2
@@ -49,6 +53,8 @@ jobs:
         STATSD_HOST: localhost
         STATSD_API_PORT: 9102
         STATSD_UDP_PORT: 9125
+        PUSHGW_HOST: localhost
+        PUSHGW_PORT: 9091
     - name: upload-cov
       uses: codecov/codecov-action@v1
       with:

README.md

@@ -3,11 +3,13 @@
 
 # prometheus-push-client
 
-Push metrics from your regular and/or long-running jobs to existing Prometheus/VictoriaMetrics monitoring system.
+Push metrics from your periodic long-running jobs to existing Prometheus/VictoriaMetrics monitoring system.
 
-Currently supports pushes directly to VictoriaMetrics via UDP and HTTP using InfluxDB line protocol as [described here](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html?highlight=telegraf#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf).
+Currently supports pushes directly to `VictoriaMetrics` via UDP and HTTP using InfluxDB line protocol as [described here](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html?highlight=telegraf#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf).
 
-For pure Prometheus setups, pushes into StatsD/statsd-exporter in StatsD format via UDP are supported ([see exporter docs](https://github.com/prometheus/statsd_exporter#with-statsd)). Prometheus and StatsD metric types are not fully compatible, so currenly all metrics become StatsD gauges, but `rate`, `increase`, `histogram_quantile` and other PromQL functions produce same results as if types never changed.
+For `pure Prometheus` setups, several options are supported:
+- to [StatsD](https://github.com/statsd/statsd) or [statsd-exporter](https://github.com/prometheus/statsd_exporter#with-statsd) in StatsD format via UDP. Prometheus and StatsD metric types are not fully compatible, so currenly all metrics become StatsD gauges, but `rate`, `increase`, `histogram_quantile` and other PromQL functions produce same results as if types never changed.
+- to [pushgateway](https://github.com/prometheus/pushgateway) or [prom-aggregation-gateway](https://github.com/weaveworks/prom-aggregation-gateway) in OpenMetrics format via HTTP. Please read corresponding docs about appropriate use cases and limitations.
 
 Install it via pip:
 
@@ -74,10 +76,12 @@ Best way to use them is via decorators / context managers. These clients are int
 ``` python
 def influx_udp_async(host, port, period=15.0):
 def influx_udp_thread(host, port, period=15.0):
-def influx_http_async(url, period=15.0):
-def influx_http_thread(url, period=15.0):
 def statsd_udp_async(host, port, period=15.0):
 def statsd_udp_thread(host, port, period=15.0):
+def influx_http_async(url, verb="POST", period=15.0):
+def influx_http_thread(url, verb="POST", period=15.0):
+def openmetrics_http_async(url, verb="POST", period=15.0):
+def openmetrics_http_thread(url, verb="POST", period=15.0):
 ```
 
 Usage example:
@@ -108,6 +112,8 @@ async def main(urls):
         req_hist.labels(gethostname(url)).observe(response.elapsed)
 ```
 
+Please read about mandatory `job` tag within url while [using pushgateway](https://github.com/prometheus/pushgateway#url).
+
 
 ### Streaming clients
 

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):  # pragma: no cover
+        # as in generate_latest 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,27 @@
 
 
 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)
+        if data.endswith(b"\n\n"):  # pragma: no cover
+            return data
+        to_pad = 1 if data[-1] == ord(b"\n") else 2
+        return data.ljust(len(data) + to_pad, b"\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 +42,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)
+        self.session.request(self.verb, self.url, data=data)
 
 
 class AioHttpTransport(BaseHttpTransport):
@@ -52,6 +58,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"

setup.py

@@ -12,7 +12,7 @@
 readme_lines = []
 with open('README.md') as fd:
     readme_lines = filter(None, fd.read().splitlines())
-readme_lines = list(readme_lines)[:10]
+readme_lines = list(readme_lines)[:12]
 readme_lines.append('Read more at [github page](%s).' % github_url)
 readme = '\n\n'.join(readme_lines)
 

test/test_online/conftest.py

@@ -14,10 +14,15 @@ class Cfg:
         statsd_api_port = os.getenv("STATSD_API_PORT", "9102")
         statsd_udp_port = os.getenv("STATSD_UDP_PORT", "9125")
 
+        pushgw_host = os.getenv("PUSHGW_HOST", "localhost")
+        pushgw_port = os.getenv("PUSHGW_PORT", "9091")
+
         vm_ping_url = f"http://{vm_host}:{vm_api_port}/health"
         vm_write_url = f"http://{vm_host}:{vm_api_port}/write"
         vm_export_url = f"http://{vm_host}:{vm_api_port}/api/v1/export"
 
         statsd_url = f"http://{statsd_host}:{statsd_api_port}/metrics"
 
+        pushgw_url = f"http://{pushgw_host}:{pushgw_port}/metrics"
+
     return Cfg()