streaming clients; decorators + context managers; more tests

Author: gistart

README.md

@@ -25,7 +25,7 @@ New metric constructors use separate `PUSH_REGISTRY` as a default, not to interf
 
 ### Default labelvalues
 
-With regular prometheus_client, defaults may be defined for either _none_ or _all_ the labels (with `labelvalues`), but that's not enough.
+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?).
 
 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.
 
@@ -55,13 +55,13 @@ counter1.labels(host="non-default", event_type="login").inc()
 counter1.labels("non-default", "login").inc()
 ```
 
-Metrics with no labels are initialized at creation time. This can have unpleasant side-effect: if we initialize lots of metrics not used in currently running job, background clients will have to push their non-changing values in every synchronization session.
+Metrics with no labels are initialized at creation time. This can have unpleasant side-effect: if we initialize lots of metrics not used in currently running job, batch clients will have to push their non-changing values in every synchronization session.
 
 To avoid that we'll have to properly isolate each task's metrics, which can be impossible or rather tricky, or we can create metrics with default, non-changing labels (like `hostname`). Such metrics will be initialized on fisrt use (inc), and we'll be pushing only those we actually used.
 
-## Background clients
+## Batch clients
 
-Background 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.
+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:
 
@@ -99,3 +99,8 @@ async def main(urls):
     req_hist.labels(gethostname(url)).observe(response.elapsed)
     # ...
 ```
+
+
+## Streming clients
+
+:warning: histogram.obsrever doesn't work, fix later

prometheus_push_client/__init__.py

@@ -1,18 +1,23 @@
-from .clients.bg import ThreadBGClient, AsyncioBGClient
-# TODO: clients.fg
+from .registry import PUSH_REGISTRY
+
+from .clients.batch import ThreadBatchClient, AsyncBatchClient
+from .clients.streaming import SyncStreamingClient, AsyncStreamingClient
 from .decorators import (
     influx_udp_async,
     statsd_udp_async,
     influx_udp_thread,
     statsd_udp_thread,
     influx_http_async,
     influx_http_thread,
+    influx_udp_aiostream,
+    statsd_udp_aiostream,
+    influx_udp_stream,
+    statsd_udp_stream,
 )
 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 .registry import PUSH_REGISTRY
 from .transports.http import SyncHttpTransport, AioHttpTransport
 from .transports.udp import SyncUdpTransport, AioUdpTransport
 from .version import __version__

prometheus_push_client/clients/base.py

@@ -0,0 +1,19 @@
+"""
+For documentation purposes
+"""
+
+
+class BaseSyncClient:  # pragma: no cover
+    def start(self):
+        raise NotImplementedError()
+
+    def stop(self):
+        raise NotImplementedError()
+
+
+class BaseAsyncClient:  # pragma: no cover
+    async def start(self):
+        raise NotImplementedError()
+
+    async def stop(self):
+        raise NotImplementedError()

prometheus_push_client/clients/bg.py renamed to prometheus_push_client/clients/batch.py

@@ -6,11 +6,10 @@
 from prometheus_push_client import compat
 
 
-log = logging.getLogger("prometheus.bg")
+log = logging.getLogger("prometheus.batch")
 
 
-class BackgroundClient:
-
+class BaseBatchClient:
     def __init__(self, format, transport, period=15.0, *args, **kwargs):
         self.format = format
         self.transport = transport
@@ -29,7 +28,7 @@ def sleep_steps_iter(self, period):
             yield last_step
 
 
-class ThreadBGClient(BackgroundClient, threading.Thread):
+class ThreadBatchClient(BaseBatchClient, threading.Thread):
     def start(self):
         self.stop_event = threading.Event()
         self.transport.start()
@@ -52,15 +51,15 @@ def run(self):
                     break
 
             ts_start = time.time()
-            samples_iter = self.format.iter_samples()
+            data_gen = self.format.iter_samples()
             try:
-                self.transport.push_all(samples_iter)
+                self.transport.push_all(data_gen)
             except Exception:
                 log.error("push crashed", exc_info=True)
             period = self.period - (time.time() - ts_start)
 
 
-class AsyncioBGClient(BackgroundClient):
+class AsyncBatchClient(BaseBatchClient):
     async def start(self):
         self.stop_event = asyncio.Event()
         await self.transport.start()
@@ -86,9 +85,9 @@ async def run(self):
                     break
 
             ts_start = time.time()
-            samples_iter = self.format.iter_samples()
+            data_gen = self.format.iter_samples()
             try:
-                await self.transport.push_all(samples_iter)
+                await self.transport.push_all(data_gen)
             except Exception:
                 log.error("push crashed", exc_info=True)
             period = self.period - (time.time() - ts_start)

prometheus_push_client/clients/streaming.py

@@ -0,0 +1,148 @@
+import logging
+from types import MethodType
+
+
+log = logging.getLogger("prometheus.streaming")
+
+METRIC_METHODS = ("inc", "dec", "set", "observe")
+
+
+class BaseStreamingClient:
+    def __init__(self, format, transport):
+        self.format = format
+        self.transport = transport
+
+    def _monkey_patch_all(self):
+        registry = self.format.registry
+        with registry._lock:
+            patch_registry(registry, self.format, self.transport)
+            for collector in registry._collector_to_names.keys():
+                patch_collector(collector, self.format, self.transport)
+
+    def _restore_all(self):
+        registry = self.format.registry
+        with registry._lock:
+            restore_registry(registry)
+            for collector in registry._collector_to_names.keys():
+                restore_collector(collector)
+
+
+class SyncStreamingClient(BaseStreamingClient):
+    def start(self):
+        self._monkey_patch_all()
+        self.transport.start()
+
+    def stop(self):
+        self._restore_all()
+        self.transport.stop()
+
+
+class AsyncStreamingClient(BaseStreamingClient):
+    async def start(self):
+        self._monkey_patch_all()
+        await self.transport.start()
+
+    async def stop(self):
+        self._restore_all()
+        await self.transport.stop()
+
+
+def push_all(collector, format, transport):
+    try:
+        metrics = collector.collect()
+        data_gen = format.format_metrics(*metrics)
+        transport.push_all_sync(data_gen)
+    except Exception:
+        log.error("push crashed", exc_info=True)
+
+
+#
+## Utils for monkey-patching
+#
+
+
+def patch_registry(registry, format, transport):
+    register_orig = registry.register
+
+    def register_new(self, collector, *args, **kwargs):
+        patch_collector(collector, format, transport)
+        register_orig(collector, *args, **kwargs)
+
+    setattr(registry, "register", MethodType(register_new, registry))
+
+
+def patch_collector(collector, format, transport, parent=None):
+    # own methods
+    for method_name in METRIC_METHODS:
+        if not hasattr(collector, method_name):
+            continue
+        patch_collector_method(collector, format, transport, method_name, parent)
+
+    # child _metrics
+    if hasattr(collector, "_metrics"):
+        with collector._lock:
+            patch_collector_submetrics(collector, format, transport)
+
+
+def patch_collector_method(collector, format, transport, method_name, parent=None):
+    method_orig = getattr(collector, method_name)
+
+    def method_new(self, *args, **kwargs):
+        method_orig(*args, **kwargs)
+
+        push_from = self if parent is None else parent
+        push_all(push_from, format, transport)
+
+    setattr(collector, method_name, MethodType(method_new, collector))
+
+
+def patch_collector_submetrics(collector, format, transport):
+    new_container = SelfPatchingMetrics(collector, format, transport)
+
+    for orig_key in list(collector._metrics.keys()):
+        orig_collector = collector._metrics.pop(orig_key)
+        new_container[orig_key] = orig_collector
+
+    collector._metrics = new_container
+
+
+class SelfPatchingMetrics(dict):
+    def __init__(self, parent, format, transport):
+        self.parent = parent
+        self.format = format
+        self.transport = transport
+        dict.__init__(self)
+
+    def __setitem__(self, key, submetric):
+        patch_collector(submetric, self.format, self.transport, self.parent)
+        dict.__setitem__(self, key, submetric)
+
+
+def restore_registry(registry):
+    setattr(
+        registry,
+        "register",
+        MethodType(
+            registry.__class__.register,
+            registry
+        )
+    )
+
+
+def restore_collector(collector):
+    for method_name in METRIC_METHODS:
+        if not hasattr(collector, method_name):
+            continue
+
+        setattr(
+            collector,
+            method_name,
+            MethodType(
+                getattr(collector.__class__, method_name),
+                collector,
+            )
+        )
+
+    if hasattr(collector, "_metrics"):
+        with collector._lock:
+            collector._metrics = {**collector._metrics}

prometheus_push_client/compat.py

@@ -1,5 +1,7 @@
+import contextlib
 import asyncio
 
+
 if hasattr(asyncio, "get_running_loop"):
     get_running_loop = asyncio.get_running_loop
 else:  # < 3.7, may be unsafe outside coroutines (running loop)
@@ -10,4 +12,4 @@
     create_task = asyncio.create_task
 else: # < 3.7, totally unsafe outside loop
     def create_task(coro):  # pragma: no cover
-        return asyncio.get_event_loop().create_task(coro)
+        return asyncio.get_event_loop().create_task(coro)