statsd format first try, statsd udp tests, http transport, vic+http basic tests

Author: gistart

.coveragerc

@@ -0,0 +1,12 @@
+[run]
+branch = True
+include =
+    prometheus_push_client/**
+omit =
+    test/**
+
+[report]
+exclude_lines =
+    pragma: no cover
+    raise NotImplementedError
+    if __name__ == .__main__.:

.github/workflows/tests.yml

@@ -12,21 +12,25 @@ jobs:
     name: run all tests
     runs-on: ubuntu-20.04
     strategy:
+      fail-fast: false
       matrix:
         python-version: [3.6, 3.7, 3.8, 3.9, 'pypy3']
     services:
       victoria:
         image: victoriametrics/victoria-metrics:v1.59.0
-        #options: >-
-        #  --entrypoint "/victoria-metrics-prod", "--envflag.enable"]
         env:
           httpListenAddr: 8429
           influxListenAddr: 8491
-          graphiteListenAddr: 8492
         ports:
         - 8428:8428
         - 8491:8491
-        - 8492:8492
+        - 8491:8491/udp
+      statsd-exporter:
+        image: prom/statsd-exporter:v0.20.2
+        ports:
+        - 9102:9102
+        - 9125:9125
+        - 9125:9125/udp
     steps:
     - name: checkout
       uses: actions/checkout@v2
@@ -37,14 +41,15 @@ jobs:
     - name: pip-requirements
       run: pip install --upgrade -e .[test]
     - name: run-test
-      run: pytest -svvv --continue-on-collection-errors --cov=./prometheus_push_client --cov-report=xml:coverage.xml
+      run: pytest
       env:
         VM_HOST: localhost
         VM_API_PORT: 8428
         VM_INFLUX_PORT: 8491
-        VM_GRAPHITE_PORT: 8492
+        STATSD_HOST: localhost
+        STATSD_API_PORT: 9102
+        STATSD_UDP_PORT: 9125
     - name: upload-cov
       uses: codecov/codecov-action@v1
       with:
         files: coverage.xml
-        verbose: true

README.md

@@ -1,7 +1,7 @@
-# prometheus-push-client
+[![GitHub Workflow Status](https://img.shields.io/github/workflow/status/gistart/prometheus-push-client/test-all)](https://github.com/gistart/prometheus-push-client/actions)
+[![Codecov](https://img.shields.io/codecov/c/github/gistart/prometheus-push-client)](https://codecov.io/gh/gistart/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)
+# prometheus-push-client
 
 ## Default labelvalues
 

prometheus_push_client/__init__.py

@@ -2,14 +2,17 @@
 # TODO: clients.fg
 from .decorators import (
     influx_udp_async,
-    graphite_udp_async,
+    statsd_udp_async,
     influx_udp_thread,
-    graphite_udp_thread,
+    statsd_udp_thread,
+    influx_http_async,
+    influx_http_thread,
 )
 from .formats.influx import InfluxFormat
-from .formats.graphite import GraphiteFormat
-from .formats.prometheus import PrometheusFormat
+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
-# TODO: transports.http
+from .version import __version__

prometheus_push_client/clients/bg.py

@@ -33,8 +33,10 @@ def start(self):
 
     def stop(self):
         self.stop_event.set()
-        self.join()
-        self.transport.stop()
+        try:
+            self.join()
+        finally:
+            self.transport.stop()
 
     def run(self):
         period = self.period
@@ -62,8 +64,10 @@ async def start(self):
 
     async def stop(self):
         self.stop_event.set()
-        await asyncio.wait([self._runner])  # TODO timeout, catch errors
-        self.transport.stop()
+        try:
+            await asyncio.wait([self._runner])
+        finally:
+            await self.transport.stop()
 
     async def run(self):
         period = self.period

prometheus_push_client/decorators.py

@@ -47,18 +47,36 @@ def influx_udp_thread(host, port, period=15.0):
     return _thread_bg(client)
 
 
-def graphite_udp_async(host, port, period=15.0):
+def influx_http_async(url, period=15.0):
     client = ppc.AsyncioBGClient(
-        format=ppc.GraphiteFormat(),
+        format=ppc.InfluxFormat(),
+        transport=ppc.AioHttpTransport(url),
+        period=period,
+    )
+    return _async_bg(client)
+
+
+def influx_http_thread(url, period=15.0):
+    client = ppc.ThreadBGClient(
+        format=ppc.InfluxFormat(),
+        transport=ppc.SyncHttpTransport(url),
+        period=period,
+    )
+    return _thread_bg(client)
+
+
+def statsd_udp_async(host, port, period=15.0):
+    client = ppc.AsyncioBGClient(
+        format=ppc.StatsdFormat(),
         transport=ppc.AioUdpTransport(host, port),
         period=period,
     )
     return _async_bg(client)
 
 
-def graphite_udp_thread(host, port, period=15.0):
+def statsd_udp_thread(host, port, period=15.0):
     client = ppc.ThreadBGClient(
-        format=ppc.GraphiteFormat(),
+        format=ppc.StatsdFormat(),
         transport=ppc.SyncUdpTransport(host, port),
         period=period,
     )

prometheus_push_client/formats/graphite.py

@@ -25,18 +25,18 @@ def validate_metric_names(self, registry):
     def format_sample(self, sample, *_):
         sample_name, measurement_name = sample.name.rsplit("_", 1)
 
-        tags = ""
+        tag_set = ""
         if sample.labels:
-            tags = ",".join(f"{k}={v}" for k,v in sample.labels.items())
-            tags = f",{tags}"
+            tags = ["", *(f"{k}={v}" for k, v in sample.labels.items())]
+            tag_set = ",".join(tags)
 
         ts = ""
         if sample.timestamp:
             ts = f" {int(sample.timestamp * 1e9)}"  # to nanoseconds
 
         return self.FMT_SAMPLE.format(
             sample_name=sample_name,
-            tag_set=tags,
+            tag_set=tag_set,
             measurement_name=measurement_name,
             value=sample.value,
             timestamp=ts,

prometheus_push_client/formats/influx.py

@@ -0,0 +1,37 @@
+import prometheus_client as pc
+
+from prometheus_push_client.formats.base import BaseFormat
+
+
+class OpenMetricsFormat(BaseFormat):
+    """
+    As described at:
+    https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md
+    """
+
+    FMT_DATAPOINT = "%(measurement)s%(tag_set)s %(value)s%(timestamp)s"
+
+    ESCAPING = str.maketrans({
+        '"': r'\"',
+        '\\': r'\\',
+        '\n': r'\n',
+    })
+
+
+    def format_sample(self, sample, metric):
+        tag_set = ""
+        if sample.labels:
+            tags = (f'{k}="{v.translate(self.ESCAPING)}"' for k, v in sample.labels.items())
+            tag_set = "{%s}" % ",".join(tags)
+
+        ts = ""
+        if sample.timestamp:
+            ts = " %s" % sample.timestamp
+
+        # TODO: TYPE string?
+        return self.FMT_DATAPOINT % dict(
+            measurement=sample.name,
+            tag_set=tag_set,
+            value=sample.value,
+            timestamp=ts,
+        )