tests for streaming clients and decorator/context wrappers

gistart · gistart · commit e55a78e8d828 · 2021-05-10T16:19:45.000+03:00
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -17,10 +17,10 @@ jobs:
         python-version: [3.6, 3.7, 3.8, 3.9, 'pypy3']
     services:
       victoria:
-        image: victoriametrics/victoria-metrics:v1.59.0
+        image: gistart/victoria-metrics:v1.59.0
         env:
-          httpListenAddr: 8429
-          influxListenAddr: 8491
+          httpListenAddr: :8428
+          influxListenAddr: :8491
         ports:
         - 8428:8428
         - 8491:8491
diff --git a/.gitignore b/.gitignore
@@ -15,6 +15,8 @@ __pycache__
 .coverage
 coverage.xml
 test/poke.py
+build/*
+dist/*
 compare/*
 
 !.gitkeep
diff --git a/README.md b/README.md
@@ -17,15 +17,15 @@ pip install prometheus-push-client
 
 ## Metrics
 
-This library uses `prometheus-client` metric implementation, but adds with some minor tweaks.
+This library uses `prometheus-client` metric implementation, but adds some minor tweaks.
 
 ### Separate registry
 
 New metric constructors use separate `PUSH_REGISTRY` as a default, not to interfere with other metrics already defined and monitored in existing projects.
 
 ### Default labelvalues
 
-With regular prometheus_client, defaults may be defined for either _none_ or _all_ the labels (with `labelvalues`), but that's not enough (and sometimes doesn't work?).
+With regular prometheus_client, defaults may be defined for either _none_ or _all_ the labels (with `labelvalues`), but that's not enough. Moreover `labelvalues` sometimes doesn't work as expected.
 
 We probably want to define _some_ defaults, like `hostname`, or more importantly, **if we use VictoriaMetrics cluster**, we always need to push label `VictoriaMetrics_AccountID=<int>` (usually 1) or else our metrics will be ignored.
 
@@ -63,11 +63,11 @@ To avoid that we'll have to properly isolate each task's metrics, which can be i
 
 Batch clients spawn synchronization jobs "in background" (meaning in a thread or asyncio task) to periodically send all metrics from `ppc.PUSH_REGISTRY` to the destination.
 
-Clients will attempt to stop gracefully, synchronizing registry "one last time" after job exits or crashes. Sometimes this _may_ mess up Grafana sampling, but the worst picture I could atrifically create looks like this:
+Clients will attempt to stop gracefully, synchronizing registry "one last time" after job exits or crashes. Sometimes this _may_ mess up Grafana sampling, but the worst picture I could artifically create looks like this:
 
 ![graceful push effect](./docs/img/graceful_stop_effect01.png)
 
-Best way to use them is via decorators. These clients are intended to be used with long running, but finite tasks, which could be spawned anywhere, therefor not easily accessible by the scraper. If that's not the case -- just use "passive mode" w/ the scraper instead.
+Best way to use them is via decorators / context managers. These clients are intended to be used with long running, but finite tasks, which could be spawned anywhere, therefor not easily accessible by the scraper. If that's not the case -- just use "passive mode" w/ the scraper instead.
 
 ``` python
 def influx_udp_async(host, port, period=15.0):
@@ -97,10 +97,27 @@ req_hist = ppc.Histogram(
 async def main(urls):
     # the job ...
     req_hist.labels(gethostname(url)).observe(response.elapsed)
-    # ...
+
+# OR
+
+async def main(urls):
+    async with ppc.influx_udp_async("victoria.acme.inc.net", 9876, period=15):
+        # the job ...
+        req_hist.labels(gethostname(url)).observe(response.elapsed)
 ```
 
 
-## Streming clients
+## Streaming clients
+
+If for some reason every metric change needs to be synced, UDP streaming clients are implemented in this library.
+
+```python
+def influx_udp_aiostream(host, port):
+def influx_udp_stream(host, port):
+def statsd_udp_aiostream(host, port):
+def statsd_udp_stream(host, port):
+```
+
+Usage is completely identical to batch clients' decorators / context managers.
 
-:warning: histogram.obsrever doesn't work, fix later
+:warning: Histogram and Summary `.time() decorator` doesn't work in this mode atm, because it can't be easily monkey-patched.
diff --git a/prometheus_push_client/__init__.py b/prometheus_push_client/__init__.py
@@ -17,7 +17,7 @@
 from .formats.influx import InfluxFormat
 from .formats.statsd import StatsdFormat
 from .formats.openmetrics import OpenMetricsFormat
-from .metrics import Counter, Gauge, Summary, Histogram, Info, Enum
+from .metrics import Counter, Gauge, Summary, Histogram
 from .transports.http import SyncHttpTransport, AioHttpTransport
 from .transports.udp import SyncUdpTransport, AioUdpTransport
 from .version import __version__
diff --git a/prometheus_push_client/clients/batch.py b/prometheus_push_client/clients/batch.py
@@ -16,7 +16,7 @@ def __init__(self, format, transport, period=15.0, *args, **kwargs):
         self.period = float(period)
         self.stop_event = None
 
-        self._period_step = 0.5  # check event every 0.5 seconds
+        self._period_step = 0.25  # check event every 0.25 seconds
 
         super().__init__(*args, **kwargs)
 
diff --git a/prometheus_push_client/decorators.py b/prometheus_push_client/decorators.py
@@ -93,6 +93,7 @@ def statsd_udp_stream(host, port, registry=ppc.PUSH_REGISTRY):
 # decorator AND context manager
 #
 
+
 class _sync_wrap:
     def __init__(self, client):
         self.client = client
diff --git a/prometheus_push_client/formats/influx.py b/prometheus_push_client/formats/influx.py
@@ -35,7 +35,7 @@ def format_sample(self, sample, *_):
             tag_set = ",".join(tags)
 
         ts = ""
-        if sample.timestamp:
+        if sample.timestamp:  # pragma: no cover
             ts = f" {int(sample.timestamp * 1e9)}"  # to nanoseconds
 
         return self.FMT_SAMPLE.format(
diff --git a/prometheus_push_client/formats/openmetrics.py b/prometheus_push_client/formats/openmetrics.py
@@ -25,7 +25,7 @@ def format_sample(self, sample, metric):
             tag_set = "{%s}" % ",".join(tags)
 
         ts = ""
-        if sample.timestamp:
+        if sample.timestamp:  # pragma: no cover
             ts = " %s" % sample.timestamp
 
         # TODO: TYPE string?
diff --git a/prometheus_push_client/metrics.py b/prometheus_push_client/metrics.py
@@ -30,6 +30,4 @@ class Gauge(_PushMixin, pc.Gauge): pass
 class Summary(_PushMixin, pc.Summary): pass
 class Histogram(_PushMixin, pc.Histogram): pass
 
-# TODO: TEST!!1
-class Info(_PushMixin, pc.Info): pass
-class Enum(_PushMixin, pc.Enum): pass
+# TODO: Info, Enum ?
diff --git a/prometheus_push_client/registry.py b/prometheus_push_client/registry.py
@@ -4,7 +4,7 @@
 PUSH_REGISTRY = pc.CollectorRegistry(auto_describe=False, target_info=None)
 
 
-def transfer_metrics(src_registry, dst_registry):
+def transfer_metrics(src_registry, dst_registry):  # pragma: no cover
     """
     From ppc.PUSH_REGISTRY to pc.REGISTRY or vice versa
     """
diff --git a/prometheus_push_client/transports/http.py b/prometheus_push_client/transports/http.py
@@ -1,11 +1,11 @@
 import asyncio
 try:
     import requests
-except ImportError:
+except ImportError:  # pragma: no cover
     requests = None
 try:
     import aiohttp
-except ImportError:
+except ImportError:  # pragma: no cover
     aiohttp = None
 
 
@@ -15,7 +15,7 @@ def __init__(self, url):
         self.url = url
         self.session = None
 
-    def _validate(self):
+    def _validate(self):  # pragma: no cover
         pass
 
     def push_all_sync(self):
@@ -54,5 +54,5 @@ async def stop(self):
 
     async def push_all(self, iterable):
         data = b"\n".join(iterable)
-        async with self.session.post(self.url, data=data) as response:
+        async with self.session.post(self.url, data=data) as _:
             pass
diff --git a/setup.py b/setup.py
@@ -1,4 +1,4 @@
-from distutils.core import setup
+import setuptools
 
 
 # version
@@ -32,13 +32,15 @@
 ]
 extras_require["dev"] = [
     "ipython",
+    "setuptools",
+    "wheel",
     "twine",
     *extras_require["test"]
 ]
 
 
 # setup
-setup(
+setuptools.setup(
     name="prometheus_push_client",
     version=__version__,
     author="gistart",
@@ -48,15 +50,15 @@
     long_description_content_type="text/markdown",
     url=github_url,
     license="Apache License 2.0",
-    classifiers=(
+    classifiers=[
         "Programming Language :: Python :: 3",
         "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
-        "License :: OSI Approved :: Apache License 2.0",
+        "License :: OSI Approved :: Apache Software License",
         "Operating System :: OS Independent",
-    ),
+    ],
     python_requires=">=3.6",
     packages=[
         "prometheus_push_client"
diff --git a/test/test_online/test_statsd_exporter.py b/test/test_online/test_statsd_exporter.py
@@ -39,34 +39,43 @@ def hist1(request):
     ))
 
 
+@pytest.fixture(scope="module")  #! important
+def gauge1(request):
+    return make_metric_fixture(request, ppc.Gauge(
+        name="g1",
+        namespace=NS,
+        subsystem="statsd_udp",
+        # no labels!
+    ))
 
-def test_statsd_udp_thread(cfg, hist1):
+
+def test_statsd_udp_thread(cfg, hist1, gauge1):
 
     for i in range(1, 251):
         hist1.labels(l1=1).observe(i / 100)
 
     period = 0.3
     sleeps = 2
     sleep_sec = 1.1
-    samples_expected = math.ceil(sleeps / period)  # + graceful
 
     @ppc.statsd_udp_thread(cfg.statsd_host, cfg.statsd_udp_port, period)
     def _test():
         for i in range(1, sleeps + 1):
             hist1.labels(l1=2).observe(i / 10)
             time.sleep(sleep_sec)
+            gauge1.set(i)
         return 1 / 0
 
     with pytest.raises(ZeroDivisionError):
         _test()
 
     time.sleep(2.0)  # wait remote sync
 
-    local_res = collect_metrics(hist1._name)
+    local_res = collect_metrics(hist1._name, gauge1._name)
     local_res_lines = sorted(local_res.split("\n"))
 
     remote_all = get_metrics(cfg)
-    remote_res = collect_metrics(hist1._name, data=remote_all)
+    remote_res = collect_metrics(hist1._name, gauge1._name, data=remote_all)
     remote_res_lines = sorted(remote_res.split("\n"))
 
     assert len(local_res_lines) == len(remote_res_lines), remote_res
@@ -78,32 +87,95 @@ def _test():
 
 
 @pytest.mark.asyncio
-async def test_statsd_udp_asyncio(cfg, hist1):
+async def test_statsd_udp_asyncio(cfg, hist1, gauge1):
     for i in range(1, 251):
         hist1.labels(l1=1).observe(i / 100)
 
     period = 0.3
     sleeps = 2
     sleep_sec = 1.1
-    samples_expected = math.ceil(sleeps / period)  # + graceful
 
-    @ppc.statsd_udp_async(cfg.statsd_host, cfg.statsd_udp_port, period)
     async def _test():
         for i in range(1, sleeps + 1):
             hist1.labels(l1=2).observe(i / 10)
             await asyncio.sleep(sleep_sec)
+            gauge1.set(i)
+        return 1 / 0
+
+    with pytest.raises(ZeroDivisionError):
+        async with ppc.statsd_udp_async(cfg.statsd_host, cfg.statsd_udp_port, period):
+            await _test()
+
+    await asyncio.sleep(2.0)  # wait remote sync
+
+    local_res = collect_metrics(hist1._name, gauge1._name)
+    local_res_lines = sorted(local_res.split("\n"))
+
+    remote_all = get_metrics(cfg)
+    remote_res = collect_metrics(hist1._name, gauge1._name, data=remote_all)
+    remote_res_lines = sorted(remote_res.split("\n"))
+
+    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}.")  # statsd-side float -> int
+        )
+
+
+def test_statsd_udp_stream(cfg, hist1, gauge1):
+    for i in range(1, 51):
+        hist1.labels(l1=1).observe(i / 100)
+
+    def _test():
+        for i in range(1, 31):
+            hist1.labels(l1=2).observe(i / 10)
+            gauge1.set(i)
+        return 1 / 0
+
+    with pytest.raises(ZeroDivisionError):
+        with ppc.statsd_udp_stream(cfg.statsd_host, cfg.statsd_udp_port):
+            _test()
+
+    time.sleep(2.0)  # wait remote sync
+
+    local_res = collect_metrics(hist1._name, gauge1._name)
+    local_res_lines = sorted(local_res.split("\n"))
+
+    remote_all = get_metrics(cfg)
+    remote_res = collect_metrics(hist1._name, gauge1._name, data=remote_all)
+    remote_res_lines = sorted(remote_res.split("\n"))
+
+    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}.")  # statsd-side float -> int
+        )
+
+
+@pytest.mark.asyncio
+async def test_statsd_udp_aiostream(cfg, hist1, gauge1):
+    for i in range(1, 51):
+        hist1.labels(l1=1).observe(i / 100)
+
+    @ppc.statsd_udp_aiostream(cfg.statsd_host, cfg.statsd_udp_port)
+    async def _test():
+        for i in range(1, 31):
+            hist1.labels(l1=2).observe(i / 10)
+            gauge1.set(i)
         return 1 / 0
 
     with pytest.raises(ZeroDivisionError):
         await _test()
 
     await asyncio.sleep(2.0)  # wait remote sync
 
-    local_res = collect_metrics(hist1._name)
+    local_res = collect_metrics(hist1._name, gauge1._name)
     local_res_lines = sorted(local_res.split("\n"))
 
     remote_all = get_metrics(cfg)
-    remote_res = collect_metrics(hist1._name, data=remote_all)
+    remote_res = collect_metrics(hist1._name, gauge1._name, data=remote_all)
     remote_res_lines = sorted(remote_res.split("\n"))
 
     assert len(local_res_lines) == len(remote_res_lines), remote_res
diff --git a/test/test_online/test_victoria.py b/test/test_online/test_victoria.py

Original file line number	Diff line number	Diff line change
`@@ -93,6 +93,7 @@ def statsd_udp_stream(host, port, registry=ppc.PUSH_REGISTRY):`
`93`	`93`	`# decorator AND context manager`
`94`	`94`	`#`
`95`	`95`
	`96`	`+`
`96`	`97`	`class _sync_wrap:`
`97`	`98`	`def __init__(self, client):`
`98`	`99`	`self.client = client`