basic test for openmetrics_http + pushgateway

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/formats/openmetrics.py

@@ -17,8 +17,8 @@ class OpenMetricsFormat(BaseFormat):
     })
 
 
-    def format_header(self, metric):
-        # as in generate_lastest implemented at:
+    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

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()

test/test_online/test_pushgateway.py

@@ -0,0 +1,130 @@
+import asyncio
+from collections import namedtuple
+import time
+import pytest
+import requests
+
+import re
+
+import prometheus_push_client as ppc
+
+from testutils import make_metric_fixture, collect_metrics
+
+
+NS = "test_pushgw"
+JOB = f"job/{NS}"
+
+RX_JOB_TAG = re.compile(r"(job=\".*?\",?)")
+RX_INSTANCE_TAG = re.compile(r"(instance=\".*?\",?)")
+
+
+def get_metrics(cfg):
+    response = requests.get(cfg.pushgw_url)
+    content = response.text
+
+    assert response.status_code == 200, content
+    assert len(content), content
+
+    return content
+
+
+def test_ping(cfg):
+    assert get_metrics(cfg)
+
+
+
+@pytest.fixture
+def summ1(request):
+    return make_metric_fixture(request, ppc.Summary(
+        name="summ1",
+        namespace=NS,
+        unit="seconds",
+        labelnames=["l"],
+    ))
+
+
+@pytest.fixture
+def gauge1(request):
+    return make_metric_fixture(request, ppc.Gauge(
+        name="gauge1",
+        namespace=NS,
+        unit="meters",
+        labelnames=["l"],
+    ))
+
+
+def test_pushgateway_http_sync(cfg, summ1, gauge1):
+    period = 0.4
+    sleeps = 2
+    sleep_sec = 1.0
+
+    @ppc.openmetrics_http_thread(f"{cfg.pushgw_url}/{JOB}", period=period)
+    def _test():
+        for _ in range(sleeps):
+            with gauge1.labels("L1").track_inprogress():
+                with summ1.labels("L2").time():
+                    time.sleep(sleep_sec)
+        return 1 / 0   # testing graceful stop
+
+    with pytest.raises(ZeroDivisionError):
+        _test()
+
+    time.sleep(1.0)  # let them sync
+
+    local_res = collect_metrics(summ1._name, gauge1._name)
+    local_res_lines = sorted(local_res.split("\n"))
+
+    remote_all = get_metrics(cfg)
+    remote_res = collect_metrics(summ1._name, gauge1._name, data=remote_all)
+
+    remote_res_lines = []
+    for rem_line in sorted(remote_res.split("\n")):
+        rem_line = RX_INSTANCE_TAG.sub("", rem_line)
+        rem_line = RX_JOB_TAG.sub("", rem_line)
+        remote_res_lines.append(rem_line)
+
+    assert len(local_res_lines) == len(remote_res_lines), remote_res
+    for loc, rem in zip(local_res_lines, remote_res_lines):
+        assert (
+            loc == rem or
+            loc.startswith(f"{rem}.")  # pushgw-side float -> int
+        )
+
+
+@pytest.mark.asyncio
+async def test_pushgateway_http_sync(cfg, summ1, gauge1):
+    period = 0.4
+    sleeps = 2
+    sleep_sec = 1.0
+
+    async def _test():
+        async with ppc.openmetrics_http_async(f"{cfg.pushgw_url}/{JOB}", period=period):
+            for _ in range(sleeps):
+                with gauge1.labels("L1").track_inprogress():
+                    with summ1.labels("L2").time():
+                        await asyncio.sleep(sleep_sec)
+        return 1 / 0   # testing graceful stop
+
+    with pytest.raises(ZeroDivisionError):
+        await _test()
+
+    await asyncio.sleep(1.0)  # let them sync
+
+    local_res = collect_metrics(summ1._name, gauge1._name)
+    local_res_lines = sorted(local_res.split("\n"))
+
+    remote_all = get_metrics(cfg)
+    remote_res = collect_metrics(summ1._name, gauge1._name, data=remote_all)
+
+    remote_res_lines = []
+    for rem_line in sorted(remote_res.split("\n")):
+        rem_line = RX_INSTANCE_TAG.sub("", rem_line)
+        rem_line = RX_JOB_TAG.sub("", rem_line)
+        remote_res_lines.append(rem_line)
+
+    assert len(local_res_lines) == len(remote_res_lines), remote_res
+    for loc, rem in zip(local_res_lines, remote_res_lines):
+        assert (
+            loc == rem or
+            loc.startswith(f"{rem}.")  # pushgw-side float -> int
+        )