basic udp+influx+victoria tests

Author: gistart

.gitignore

@@ -12,6 +12,8 @@
 *.egg-info
 __pycache__
 .vscode*
+.coverage
+coverage.xml
 
 !.gitkeep
 !/.gitignore

README.md

@@ -1,6 +1,7 @@
 # prometheus-push-client
 
 ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/gistart/prometheus-push-client/test-all)
+![Codecov](https://img.shields.io/codecov/c/github/gistart/prometheus-push-client)
 
 ## Default labelvalues
 

prometheus_push_client/clients/bg.py

@@ -2,6 +2,8 @@
 import threading
 import time
 
+from prometheus_push_client import compat
+
 
 class BackgroundClient:
 
@@ -56,7 +58,7 @@ class AsyncioBGClient(BackgroundClient):
     async def start(self):
         self.stop_event = asyncio.Event()
         await self.transport.start()
-        self._runner = asyncio.create_task(self.run())
+        self._runner = compat.create_task(self.run())
 
     async def stop(self):
         self.stop_event.set()

prometheus_push_client/compat.py

@@ -2,5 +2,12 @@
 
 if hasattr(asyncio, "get_running_loop"):
     get_running_loop = asyncio.get_running_loop
-else:  # < 3.7, may be unsafe outside coroutines
+else:  # < 3.7, may be unsafe outside coroutines (running loop)
     get_running_loop = asyncio.get_event_loop
+
+
+if hasattr(asyncio, "create_task"):
+    create_task = asyncio.create_task
+else: # < 3.7, totally unsafe outside loop
+    def create_task(coro):
+        return asyncio.get_event_loop().create_task(coro)

test/test_offline/test_influx.py

@@ -2,7 +2,7 @@
 
 import prometheus_push_client as ppc
 
-from testutils import make_metric_fixture, collect_metrics
+from testutils import make_metric_fixture, collect_formatter
 
 
 NS = "test_offline"
@@ -13,7 +13,12 @@
 def sum1(request):
     return make_metric_fixture(
         request,
-        ppc.Summary(name="s1", namespace=NS, subsystem=SUB)
+        ppc.Summary(
+            name="s1",
+            namespace=NS,
+            subsystem=SUB,
+            labelnames=["l1", "l2"]
+        )
     )
 
 
@@ -29,13 +34,46 @@ def counter1(request):
 def histogram1(request):
     return make_metric_fixture(
         request,
-        ppc.Histogram(name="h1", namespace=NS, subsystem=SUB, buckets=[1,2,float("inf")])
+        ppc.Histogram(
+            name="h1",
+            namespace=NS,
+            subsystem=SUB,
+            buckets=[1, 2, float("inf")],
+            labelnames=["l1"],
+        )
     )
 
 
-
-def test_influx_formatter(sum1, counter1, histogram1):
-    pass
+@pytest.fixture
+def influx_format_expected():
+    return (""
+"""
+test_offline_influx_s1,l1=1,l2=2 count=2.0
+test_offline_influx_s1,l1=1,l2=2 sum=10.0
+test_offline_influx_s1,l1=10,l2=20 count=1.0
+test_offline_influx_s1,l1=10,l2=20 sum=10.0
+test_offline_influx_c1 total=10.0
+test_offline_influx_h1,l1=1,le=1.0 bucket=1.0
+test_offline_influx_h1,l1=1,le=2.0 bucket=2.0
+test_offline_influx_h1,l1=1,le=+Inf bucket=3.0
+test_offline_influx_h1,l1=1 count=3.0
+test_offline_influx_h1,l1=1 sum=4.5
+""").strip()
+
+
+def test_influx_formatter(sum1, counter1, histogram1, influx_format_expected):
+    sum1.labels(1, 2).observe(5)
+    sum1.labels(l1=1, l2=2).observe(5)
+    sum1.labels(l1=10, l2=20).observe(10)
+
+    counter1.inc(10)
+
+    histogram1.labels(l1=1).observe(0.5)
+    histogram1.labels(l1=1).observe(1.5)
+    histogram1.labels(l1=1).observe(2.5)
+
+    data = collect_formatter(ppc.InfluxFormat)
+    assert data == influx_format_expected
 
 
 @pytest.fixture

test/test_online/test_victoria.py

@@ -1,10 +1,112 @@
 import pytest
+import json
+import time
+import math
+import asyncio
 import requests
+from collections import Counter
+
+import prometheus_push_client as ppc
+
+from testutils import make_metric_fixture
+
+
+NS = "test_vic"
+
+
+def export(cfg, pattern=NS):
+    response = requests.get(
+        cfg.vm_export_url,
+        params={'match[]': r'{__name__=~"%s.*"}' % pattern}
+    )
+
+    found = []
+    for line in response.iter_lines():
+        metric = json.loads(line)
+        found.append(metric)
+    return found
+
+
+def count_samples(found, pattern):
+    n_samples = Counter()
+    for metric in found:
+        mname = metric["metric"]["__name__"]
+        if mname.startswith(pattern):
+            n_samples[mname] += len(metric["values"])
+    return n_samples
+
 
 
 def test_ping(cfg):
     response = requests.get(cfg.vm_ping_url)
     content = response.text
 
     assert response.status_code == 200, content
-    assert content == "OK", content
+    assert content == "OK", content
+
+
+@pytest.fixture
+def counter1(request):
+    return make_metric_fixture(
+        request,
+        ppc.Counter("c1", namespace=NS, subsystem="influx_udp")
+    )
+
+
+def test_vic_influx_udp_thread(cfg, counter1):
+
+    found_before = export(cfg)
+    count_before = count_samples(found_before, counter1._name)
+
+    period = 0.4
+    sleeps = 2
+    sleep_sec = 1.0
+    samples_expected = math.ceil(sleeps / period)  # + graceful one
+
+    @ppc.influx_udp_thread(cfg.vm_host, cfg.vm_influx_port, period=period)
+    def _test():
+        for _ in range(sleeps):
+            counter1.inc()
+            time.sleep(sleep_sec)
+        return 1 / 0   # testing graceful stop
+
+    with pytest.raises(ZeroDivisionError):
+        _test()
+
+    time.sleep(2.0)  # let them sync
+
+    found_after = export(cfg)
+    count_after = count_samples(found_after, counter1._name)
+
+    for k in count_after.keys():
+        assert samples_expected == count_after[k] - count_before[k]
+
+
+@pytest.mark.asyncio
+async def test_vic_influx_udp_async(cfg, counter1):
+
+    found_before = export(cfg)
+    count_before = count_samples(found_before, counter1._name)
+
+    period = 0.4
+    sleeps = 2
+    sleep_sec = 1.0
+    samples_expected = math.ceil(sleeps / period)  # + graceful one
+
+    @ppc.influx_udp_async(cfg.vm_host, cfg.vm_influx_port, period=period)
+    async def _test():
+        for _ in range(sleeps):
+            counter1.inc()
+            await asyncio.sleep(sleep_sec)
+        return 1 / 0   # testing graceful stop
+
+    with pytest.raises(ZeroDivisionError):
+        await _test()
+
+    await asyncio.sleep(2.0)  # let them sync
+
+    found_after = export(cfg)
+    count_after = count_samples(found_after, counter1._name)
+
+    for k in count_after.keys():
+        assert samples_expected == count_after[k] - count_before[k]

test/testutils.py

@@ -29,3 +29,22 @@ def only_interesting(line):
 
     interesting_lines = filter(only_interesting, all_data.split("\n"))
     return "\n".join(interesting_lines)
+
+
+def collect_formatter(formatter_cls, *metric_names):
+    fmt = formatter_cls()
+
+    collected = []
+    for line in fmt.iter_samples():
+        line = line.decode()
+
+        if metric_names:
+            for mnane in metric_names:
+                if line.startswith(mname):
+                    collected.append(line)
+                    break
+        else:
+            collected.append(line)
+
+    collected = filter(lambda line: " created=" not in line, collected)
+    return "\n".join(collected)