added metrics executor

Author: igorcoding

extapi/_meta.py

@@ -1,3 +1,4 @@
 import importlib.util
 
 has_open_telemetry = importlib.util.find_spec("opentelemetry") is not None
+has_prometheus = importlib.util.find_spec("prometheus_client") is not None

extapi/http/abc.py

@@ -2,6 +2,7 @@
 from typing import Any, Generic, Protocol, Self, TypeVar, runtime_checkable
 
 from multidict import CIMultiDict
+from yarl import URL
 
 from .types import RequestData, Response, StrOrURL
 
@@ -48,7 +49,7 @@ async def get(
         return await self.execute(
             RequestData(
                 method="GET",
-                url=url,
+                url=URL(url) if isinstance(url, str) else url,
                 params=params,
                 json=json,
                 data=data,
@@ -72,7 +73,7 @@ async def post(
         return await self.execute(
             RequestData(
                 method="POST",
-                url=url,
+                url=URL(url) if isinstance(url, str) else url,
                 params=params,
                 json=json,
                 data=data,
@@ -96,7 +97,7 @@ async def delete(
         return await self.execute(
             RequestData(
                 method="DELETE",
-                url=url,
+                url=URL(url) if isinstance(url, str) else url,
                 params=params,
                 json=json,
                 data=data,
@@ -120,7 +121,7 @@ async def put(
         return await self.execute(
             RequestData(
                 method="PUT",
-                url=url,
+                url=URL(url) if isinstance(url, str) else url,
                 params=params,
                 json=json,
                 data=data,
@@ -144,7 +145,7 @@ async def patch(
         return await self.execute(
             RequestData(
                 method="PATCH",
-                url=url,
+                url=URL(url) if isinstance(url, str) else url,
                 params=params,
                 json=json,
                 data=data,

extapi/http/addons/log.py

@@ -16,12 +16,10 @@ def __init__(self, *, log_params: bool = True):
         self._log_params = log_params
 
     def _get_url(self, request: RequestData) -> URL:
-        url_ = URL(request.url) if not isinstance(request.url, URL) else request.url
-
         if self._log_params and request.params:
-            return url_.with_query(request.params)
+            return request.url.with_query(request.params)
 
-        return url_
+        return request.url
 
     async def before_request(self, request: RequestData) -> None:
         url = self._get_url(request)

extapi/http/backends/aiohttp.py

@@ -37,6 +37,28 @@ async def json(
         return await self._original.json(encoding=encoding, loads=loads)
 
 
+_aiohttp_extra_kwargs = [
+    "cookies",
+    "skip_auto_headers",
+    "auth",
+    "allow_redirects",
+    "max_redirects",
+    "compress",
+    "chunked",
+    "expect100",
+    "raise_for_status",
+    "read_until_eof",
+    "proxy",
+    "proxy_auth",
+    "server_hostname",
+    "trace_request_ctx",
+    "read_bufsize",
+    "auto_decompress",
+    "max_line_size",
+    "max_field_size",
+]
+
+
 class AiohttpExecutor(AbstractExecutor[aiohttp.ClientResponse]):
     __slots__ = (
         "_ssl",
@@ -49,15 +71,27 @@ def __init__(
     ):
         super().__init__()
         self._ssl = ssl
-        self._session = aiohttp.ClientSession(*args, **kwargs)
+        self._session = self._make_session(*args, **kwargs)
         self._default_timeout = default_timeout
 
+    def _make_session(self, *args, **kwargs) -> aiohttp.ClientSession:
+        return aiohttp.ClientSession(*args, **kwargs)
+
     async def close(self):
         await self._session.close()
 
     async def execute(self, request: RequestData) -> Response[aiohttp.ClientResponse]:
         timeout = request.timeout or self._default_timeout
 
+        # aiohttp-specific kwargs
+        # we need to pull them individually because
+        # we may have our own custom kwargs
+        aiohttp_kwargs = {
+            key: request.kwargs[key]
+            for key in _aiohttp_extra_kwargs
+            if key in request.kwargs
+        }
+
         response = await self._session.request(
             method=request.method,
             url=request.url,
@@ -67,7 +101,7 @@ async def execute(self, request: RequestData) -> Response[aiohttp.ClientResponse
             headers=request.headers,
             timeout=timeout,  # type: ignore[arg-type]
             ssl=self._ssl,
-            **request.kwargs,
+            **aiohttp_kwargs,
         )
 
         return Response[aiohttp.ClientResponse](

extapi/http/backends/httpx.py

@@ -2,7 +2,6 @@
 
 import httpx
 from multidict import CIMultiDict
-from yarl import URL
 
 from extapi.http.abc import AbstractExecutor
 from extapi.http.types import (
@@ -28,6 +27,16 @@ async def read(self) -> bytes:
         return await self._original.aread()
 
 
+_httpx_extra_kwargs = [
+    "content",
+    "files",
+    "cookies",
+    "auth",
+    "follow_redirects",
+    "extensions",
+]
+
+
 class HttpxExecutor(AbstractExecutor[httpx.Response], metaclass=abc.ABCMeta):
     __slots__ = (
         "_client",
@@ -46,26 +55,35 @@ def __init__(
         verify = kwargs.pop("verify", None)
         if verify is None:
             verify = check_ssl
-        self._client = httpx.AsyncClient(
+        self._client = self._make_client(
             verify=verify, follow_redirects=follow_redirects, **kwargs
         )
         self._default_timeout = default_timeout
 
+    def _make_client(self, *args, **kwargs) -> httpx.AsyncClient:
+        return httpx.AsyncClient(*args, **kwargs)
+
     async def close(self):
         await self._client.aclose()
 
     async def execute(self, request: RequestData) -> Response[httpx.Response]:
         timeout = request.timeout or self._default_timeout
-        url = request.url
-
-        if isinstance(url, URL):
-            url = str(url)
+        url = str(request.url)
 
         if request.headers is None:
             httpx_headers = []
         else:
             httpx_headers = [(k, str(v)) for k, v in request.headers.items()]
 
+        # httpx-specific kwargs
+        # we need to pull them individually because
+        # we may have our own custom kwargs
+        httpx_kwargs = {
+            key: request.kwargs[key]
+            for key in _httpx_extra_kwargs
+            if key in request.kwargs
+        }
+
         response = await self._client.stream(
             method=request.method,
             url=url,
@@ -74,11 +92,11 @@ async def execute(self, request: RequestData) -> Response[httpx.Response]:
             data=request.data,
             headers=httpx_headers,
             timeout=timeout,
-            **request.kwargs,
+            **httpx_kwargs,
         ).__aenter__()
 
         return Response[httpx.Response](
-            url=url,
+            url=request.url,
             status=response.status_code,
             headers=CIMultiDict(response.headers),
             backend_response=HttpxResponseWrap(response),

extapi/http/executors/metrics.py

@@ -0,0 +1,73 @@
+import warnings
+
+from extapi._meta import has_prometheus
+
+if not has_prometheus:
+    raise ImportError(  # pragma: no cover
+        "opentelemetry is not installed - run `pip install prometheus_client`"
+    )
+
+import time
+from typing import Generic, TypeVar
+
+from extapi.http.abc import AbstractExecutor
+from extapi.http.types import RequestData, Response
+
+from ..metrics.container import MetricsContainer
+from .wrapped import WrappedExecutor
+
+T = TypeVar("T", covariant=True)
+
+
+class PrometheusMetricsExecutor(WrappedExecutor[T], Generic[T]):
+    def __init__(
+        self, executor: AbstractExecutor[T], *, metrics_container: MetricsContainer
+    ):
+        super().__init__(executor)
+        self._metrics_container = metrics_container
+
+    async def execute(self, request: RequestData) -> Response[T]:
+        path_template = request.kwargs.pop("path_template", None)
+        if path_template is None:
+            warnings.warn(
+                "It is highly recommended to pass `path_template` "
+                "argument to the executor in order to to not blow the l"
+                "abel cardinality when path is customized",
+                UserWarning,
+                stacklevel=1,
+            )
+
+        path = path_template or request.url.path
+
+        method = request.method.upper()
+        started_at = time.monotonic()
+        try:
+            resp = await super().execute(request)
+        except Exception as e:
+            label_values = (
+                request.url.scheme,
+                request.url.host,
+                request.url.port,
+                method,
+                path,
+                e.__class__.__name__,
+            )
+            self._metrics_container.requests_error.labels(*label_values).inc()
+            self._metrics_container.requests_duration_error.labels(
+                *label_values
+            ).observe(time.monotonic() - started_at)
+            raise
+        else:
+            label_values = (
+                request.url.scheme,
+                request.url.host,
+                request.url.port,
+                method,
+                path,
+                str(resp.status),
+            )
+            self._metrics_container.requests.labels(*label_values).inc()
+            self._metrics_container.requests_duration.labels(*label_values).observe(
+                time.monotonic() - started_at
+            )
+            return resp

extapi/http/executors/trace.py

@@ -1,19 +1,16 @@
-from multidict import CIMultiDict
-from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
-
 from extapi._meta import has_open_telemetry
 
 if not has_open_telemetry:
     raise ImportError(  # pragma: no cover
         "opentelemetry is not installed - run `pip install opentelemetry-api opentelemetry-sdk`"
     )
 
-
 from typing import Generic, TypeVar
 
+from multidict import CIMultiDict
 from opentelemetry import trace
 from opentelemetry.semconv.trace import SpanAttributes
-from yarl import URL
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
 from extapi.http.abc import AbstractExecutor
 from extapi.http.types import RequestData, Response
@@ -60,9 +57,6 @@ async def execute(self, request: RequestData) -> Response[T]:
 
             span.set_attribute(SpanAttributes.HTTP_REQUEST_METHOD, request.method)
 
-            if not isinstance(request.url, URL):
-                request.url = URL(request.url)
-
             if request.url.host is not None:
                 span.set_attribute(SpanAttributes.SERVER_ADDRESS, request.url.host)
 

extapi/http/metrics/__init__.py

@@ -0,0 +1,60 @@
+from prometheus_client import REGISTRY, CollectorRegistry, Counter, Histogram
+
+from .helpers import DEFAULT_BUCKETS, with_prefix
+
+
+class MetricsContainer:
+    def __init__(
+        self,
+        *,
+        metrics_prefix: str,
+        metrics_registry: CollectorRegistry = REGISTRY,
+    ):
+        self.requests = Counter(
+            name=with_prefix("external_service_request", prefix=metrics_prefix),
+            documentation="Count of external requests",
+            labelnames=[
+                "scheme",
+                "domain",
+                "port",
+                "method",
+                "path",
+                "status",
+            ],
+            registry=metrics_registry,
+        )
+
+        self.requests_duration = Histogram(
+            name=with_prefix(
+                "external_service_request_duration_seconds", prefix=metrics_prefix
+            ),
+            documentation="External request duration in seconds",
+            labelnames=[
+                "scheme",
+                "domain",
+                "port",
+                "method",
+                "path",
+                "status",
+            ],
+            buckets=DEFAULT_BUCKETS,
+            registry=metrics_registry,
+        )
+
+        self.requests_error = Counter(
+            name=with_prefix("external_service_errored_request", prefix=metrics_prefix),
+            documentation="Count of errored external requests",
+            labelnames=["scheme", "domain", "port", "method", "path", "error_type"],
+            registry=metrics_registry,
+        )
+
+        self.requests_duration_error = Histogram(
+            name=with_prefix(
+                "external_service_errored_request_duration_seconds",
+                prefix=metrics_prefix,
+            ),
+            documentation="Errored external request duration in seconds",
+            labelnames=["scheme", "domain", "port", "method", "path", "error_type"],
+            buckets=DEFAULT_BUCKETS,
+            registry=metrics_registry,
+        )