Add an AIOHTTP exporter (#1139)

* Always run the asgi tests

Since the client now requires a minimum of Python 3.9, we don't need to
have this feature gate in place any more

Signed-off-by: Lexi Robinson <[email protected]>

* Add an AIOHTTP exporter

Unfortunately the AIOHTTP library doesn't support ASGI and apparently
has no plans to do so which makes the ASGI exporter not suitable for
anyone using it to run their python server.

Where possible this commit follows the existing ASGI implementation and
runs the same tests for consistency.

Signed-off-by: Lexi Robinson <[email protected]>

---------

Signed-off-by: Lexi Robinson <[email protected]>

Author: Lexicality

docs/content/exporting/http/aiohttp.md

@@ -0,0 +1,23 @@
+---
+title: AIOHTTP
+weight: 6
+---
+
+To use Prometheus with a [AIOHTTP server](https://docs.aiohttp.org/en/stable/web.html),
+there is `make_aiohttp_handler` which creates a handler.
+
+```python
+from aiohttp import web
+from prometheus_client.aiohttp import make_aiohttp_handler
+
+app = web.Application()
+app.router.add_get("/metrics", make_aiohttp_handler())
+```
+
+By default, this handler will instruct AIOHTTP to automatically compress the
+response if requested by the client. This behaviour can be disabled by passing
+`disable_compression=True` when creating the app, like this:
+
+```python
+app.router.add_get("/metrics", make_aiohttp_handler(disable_compression=True))
+```

prometheus_client/aiohttp/__init__.py

@@ -0,0 +1,5 @@
+from .exposition import make_aiohttp_handler
+
+__all__ = [
+    "make_aiohttp_handler",
+]

prometheus_client/aiohttp/exposition.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from aiohttp import hdrs, web
+from aiohttp.typedefs import Handler
+
+from ..exposition import _bake_output
+from ..registry import CollectorRegistry, REGISTRY
+
+
+def make_aiohttp_handler(
+    registry: CollectorRegistry = REGISTRY,
+    disable_compression: bool = False,
+) -> Handler:
+    """Create a aiohttp handler which serves the metrics from a registry."""
+
+    async def prometheus_handler(request: web.Request) -> web.Response:
+        # Prepare parameters
+        params = {key: request.query.getall(key) for key in request.query.keys()}
+        accept_header = ",".join(request.headers.getall(hdrs.ACCEPT, []))
+        accept_encoding_header = ""
+        # Bake output
+        status, headers, output = _bake_output(
+            registry,
+            accept_header,
+            accept_encoding_header,
+            params,
+            # use AIOHTTP's compression
+            disable_compression=True,
+        )
+        response = web.Response(
+            status=int(status.split(" ")[0]),
+            headers=headers,
+            body=output,
+        )
+        if not disable_compression:
+            response.enable_compression()
+        return response
+
+    return prometheus_handler

pyproject.toml

@@ -43,6 +43,9 @@ classifiers = [
 twisted = [
     "twisted",
 ]
+aiohttp = [
+	"aiohttp",
+]
 
 [project.urls]
 Homepage = "https://github.com/prometheus/client_python"

tests/test_aiohttp.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+import gzip
+from typing import TYPE_CHECKING
+from unittest import skipUnless
+
+from prometheus_client import CollectorRegistry, Counter
+from prometheus_client.exposition import CONTENT_TYPE_PLAIN_0_0_4
+
+try:
+    from aiohttp import ClientResponse, hdrs, web
+    from aiohttp.test_utils import AioHTTPTestCase
+
+    from prometheus_client.aiohttp import make_aiohttp_handler
+
+    AIOHTTP_INSTALLED = True
+except ImportError:
+    if TYPE_CHECKING:
+        assert False
+
+    from unittest import IsolatedAsyncioTestCase as AioHTTPTestCase
+
+    AIOHTTP_INSTALLED = False
+
+
+class AioHTTPTest(AioHTTPTestCase):
+    @skipUnless(AIOHTTP_INSTALLED, "AIOHTTP is not installed")
+    def setUp(self) -> None:
+        self.registry = CollectorRegistry()
+
+    async def get_application(self) -> web.Application:
+        app = web.Application()
+        # The AioHTTPTestCase requires that applications be static, so we need
+        # both versions to be available so the test can choose between them
+        app.router.add_get("/metrics", make_aiohttp_handler(self.registry))
+        app.router.add_get(
+            "/metrics_uncompressed",
+            make_aiohttp_handler(self.registry, disable_compression=True),
+        )
+        return app
+
+    def increment_metrics(
+        self,
+        metric_name: str,
+        help_text: str,
+        increments: int,
+    ) -> None:
+        c = Counter(metric_name, help_text, registry=self.registry)
+        for _ in range(increments):
+            c.inc()
+
+    def assert_metrics(
+        self,
+        output: str,
+        metric_name: str,
+        help_text: str,
+        increments: int,
+    ) -> None:
+        self.assertIn("# HELP " + metric_name + "_total " + help_text + "\n", output)
+        self.assertIn("# TYPE " + metric_name + "_total counter\n", output)
+        self.assertIn(metric_name + "_total " + str(increments) + ".0\n", output)
+
+    def assert_not_metrics(
+        self,
+        output: str,
+        metric_name: str,
+        help_text: str,
+        increments: int,
+    ) -> None:
+        self.assertNotIn("# HELP " + metric_name + "_total " + help_text + "\n", output)
+        self.assertNotIn("# TYPE " + metric_name + "_total counter\n", output)
+        self.assertNotIn(metric_name + "_total " + str(increments) + ".0\n", output)
+
+    async def assert_outputs(
+        self,
+        response: ClientResponse,
+        metric_name: str,
+        help_text: str,
+        increments: int,
+    ) -> None:
+        self.assertIn(
+            CONTENT_TYPE_PLAIN_0_0_4,
+            response.headers.getall(hdrs.CONTENT_TYPE),
+        )
+        output = await response.text()
+        self.assert_metrics(output, metric_name, help_text, increments)
+
+    async def validate_metrics(
+        self,
+        metric_name: str,
+        help_text: str,
+        increments: int,
+    ) -> None:
+        """
+        AIOHTTP handler serves the metrics from the provided registry.
+        """
+        self.increment_metrics(metric_name, help_text, increments)
+        async with self.client.get("/metrics") as response:
+            response.raise_for_status()
+            await self.assert_outputs(response, metric_name, help_text, increments)
+
+    async def test_report_metrics_1(self):
+        await self.validate_metrics("counter", "A counter", 2)
+
+    async def test_report_metrics_2(self):
+        await self.validate_metrics("counter", "Another counter", 3)
+
+    async def test_report_metrics_3(self):
+        await self.validate_metrics("requests", "Number of requests", 5)
+
+    async def test_report_metrics_4(self):
+        await self.validate_metrics("failed_requests", "Number of failed requests", 7)
+
+    async def test_gzip(self):
+        # Increment a metric.
+        metric_name = "counter"
+        help_text = "A counter"
+        increments = 2
+        self.increment_metrics(metric_name, help_text, increments)
+
+        async with self.client.get(
+            "/metrics",
+            auto_decompress=False,
+            headers={hdrs.ACCEPT_ENCODING: "gzip"},
+        ) as response:
+            response.raise_for_status()
+            self.assertIn(hdrs.CONTENT_ENCODING, response.headers)
+            self.assertIn("gzip", response.headers.getall(hdrs.CONTENT_ENCODING))
+            body = await response.read()
+            output = gzip.decompress(body).decode("utf8")
+            self.assert_metrics(output, metric_name, help_text, increments)
+
+    async def test_gzip_disabled(self):
+        # Increment a metric.
+        metric_name = "counter"
+        help_text = "A counter"
+        increments = 2
+        self.increment_metrics(metric_name, help_text, increments)
+
+        async with self.client.get(
+            "/metrics_uncompressed",
+            auto_decompress=False,
+            headers={hdrs.ACCEPT_ENCODING: "gzip"},
+        ) as response:
+            response.raise_for_status()
+            self.assertNotIn(hdrs.CONTENT_ENCODING, response.headers)
+            output = await response.text()
+            self.assert_metrics(output, metric_name, help_text, increments)
+
+    async def test_openmetrics_encoding(self):
+        """Response content type is application/openmetrics-text when appropriate Accept header is in request"""
+        async with self.client.get(
+            "/metrics",
+            auto_decompress=False,
+            headers={hdrs.ACCEPT: "application/openmetrics-text; version=1.0.0"},
+        ) as response:
+            response.raise_for_status()
+            self.assertEqual(
+                response.headers.getone(hdrs.CONTENT_TYPE).split(";", maxsplit=1)[0],
+                "application/openmetrics-text",
+            )
+
+    async def test_plaintext_encoding(self):
+        """Response content type is text/plain when Accept header is missing in request"""
+        async with self.client.get("/metrics") as response:
+            response.raise_for_status()
+            self.assertEqual(
+                response.headers.getone(hdrs.CONTENT_TYPE).split(";", maxsplit=1)[0],
+                "text/plain",
+            )
+
+    async def test_qs_parsing(self):
+        """Only metrics that match the 'name[]' query string param appear"""
+
+        metrics = [("asdf", "first test metric", 1), ("bsdf", "second test metric", 2)]
+
+        for m in metrics:
+            self.increment_metrics(*m)
+
+        for i_1 in range(len(metrics)):
+            async with self.client.get(
+                "/metrics",
+                params={"name[]": f"{metrics[i_1][0]}_total"},
+            ) as response:
+                output = await response.text()
+                self.assert_metrics(output, *metrics[i_1])
+
+                for i_2 in range(len(metrics)):
+                    if i_1 == i_2:
+                        continue
+
+                    self.assert_not_metrics(output, *metrics[i_2])

tests/test_asgi.py

@@ -1,19 +1,11 @@
+import asyncio
 import gzip
-from unittest import skipUnless, TestCase
+from unittest import TestCase
 
-from prometheus_client import CollectorRegistry, Counter
-from prometheus_client.exposition import CONTENT_TYPE_PLAIN_0_0_4
-
-try:
-    # Python >3.5 only
-    import asyncio
+from asgiref.testing import ApplicationCommunicator
 
-    from asgiref.testing import ApplicationCommunicator
-
-    from prometheus_client import make_asgi_app
-    HAVE_ASYNCIO_AND_ASGI = True
-except ImportError:
-    HAVE_ASYNCIO_AND_ASGI = False
+from prometheus_client import CollectorRegistry, Counter, make_asgi_app
+from prometheus_client.exposition import CONTENT_TYPE_PLAIN_0_0_4
 
 
 def setup_testing_defaults(scope):
@@ -33,7 +25,6 @@ def setup_testing_defaults(scope):
 
 
 class ASGITest(TestCase):
-    @skipUnless(HAVE_ASYNCIO_AND_ASGI, "Don't have asyncio/asgi installed.")
     def setUp(self):
         self.registry = CollectorRegistry()
         self.captured_status = None

tox.ini

@@ -3,11 +3,13 @@ envlist = coverage-clean,py{3.9,3.10,3.11,3.12,3.13,py3.9,3.9-nooptionals},cover
 
 [testenv]
 deps =
+    asgiref
     coverage
     pytest
     pytest-benchmark
     attrs
     {py3.9,pypy3.9}: twisted
+    {py3.9,pypy3.9}: aiohttp
 commands = coverage run --parallel -m pytest {posargs}
 
 [testenv:py3.9-nooptionals]
@@ -44,6 +46,7 @@ commands =
 [testenv:mypy]
 deps =
     pytest
+    aiohttp
     asgiref
     mypy==0.991
 skip_install = true