Prometheus Metrics

.pre-commit-config.yaml

@@ -163,6 +163,8 @@ repos:
     - cryptography==36.0.2; python_version <= '3.6'
     - types-setuptools == 57.4.2
     - pyyaml==5.3.1
+    # From requirements-metrics.txt
+    - prometheus_client==0.20.0
     args:
     # FIXME: get rid of missing imports ignore
     - --ignore-missing-imports

.readthedocs.yml

@@ -37,5 +37,6 @@ python:
     path: .
   - requirements: requirements-tunnel.txt
   - requirements: docs/requirements.txt
+  - requirements: requirements-metrics.txt
 
 ...

Makefile

@@ -104,7 +104,8 @@ lib-dep:
 	pip install \
 		-r requirements-testing.txt \
 		-r requirements-release.txt \
-		-r requirements-tunnel.txt && \
+		-r requirements-tunnel.txt \
+		-r requirements-metrics.txt && \
 	pip install "setuptools>=42"
 
 lib-pre-commit:

docs/requirements.txt

@@ -1,5 +1,5 @@
 #
-# This file is autogenerated by pip-compile with Python 3.10
+# This file is autogenerated by pip-compile with Python 3.11
 # by the following command:
 #
 #    pip-compile --allow-unsafe --generate-hashes --output-file=docs/requirements.txt --strip-extras docs/requirements.in

proxy/common/constants.py

@@ -137,6 +137,8 @@ def _env_threadless_compliant() -> bool:
 DEFAULT_OPEN_FILE_LIMIT = 1024
 DEFAULT_PAC_FILE = None
 DEFAULT_PAC_FILE_URL_PATH = b'/'
+DEFAULT_ENABLE_METRICS = False
+DEFAULT_METRICS_URL_PATH = b"/metrics"
 DEFAULT_PID_FILE = None
 DEFAULT_PORT_FILE = None
 DEFAULT_PLUGINS: List[Any] = []
@@ -172,6 +174,7 @@ def _env_threadless_compliant() -> bool:
 )
 DEFAULT_CACHE_REQUESTS = False
 DEFAULT_CACHE_BY_CONTENT_TYPE = False
+DEFAULT_METRICS_DIRECTORY_PATH = os.path.join(DEFAULT_DATA_DIRECTORY_PATH, "metrics")
 
 # Cor plugins enabled by default or via flags
 DEFAULT_ABC_PLUGINS = [
@@ -190,6 +193,7 @@ def _env_threadless_compliant() -> bool:
 PLUGIN_DEVTOOLS_PROTOCOL = 'proxy.http.inspector.devtools.DevtoolsProtocolPlugin'
 PLUGIN_INSPECT_TRAFFIC = 'proxy.http.inspector.inspect_traffic.InspectTrafficPlugin'
 PLUGIN_WEBSOCKET_TRANSPORT = 'proxy.http.websocket.transport.WebSocketTransport'
+PLUGIN_METRICS = "proxy.http.server.MetricsWebServerPlugin"
 
 PY2_DEPRECATION_MESSAGE = '''DEPRECATION: proxy.py no longer supports Python 2.7.  Kindly upgrade to Python 3+. '
                 'If for some reasons you cannot upgrade, use'

proxy/common/flag.py

@@ -24,12 +24,13 @@
 from .plugins import Plugins
 from .version import __version__
 from .constants import (
-    COMMA, PLUGIN_PAC_FILE, PLUGIN_DASHBOARD, PLUGIN_HTTP_PROXY,
-    PLUGIN_PROXY_AUTH, PLUGIN_WEB_SERVER, DEFAULT_NUM_WORKERS,
-    PLUGIN_REVERSE_PROXY, DEFAULT_NUM_ACCEPTORS, PLUGIN_INSPECT_TRAFFIC,
-    DEFAULT_DISABLE_HEADERS, PY2_DEPRECATION_MESSAGE, DEFAULT_DEVTOOLS_WS_PATH,
-    PLUGIN_DEVTOOLS_PROTOCOL, PLUGIN_WEBSOCKET_TRANSPORT,
-    DEFAULT_DATA_DIRECTORY_PATH, DEFAULT_MIN_COMPRESSION_LENGTH,
+    COMMA, PLUGIN_METRICS, PLUGIN_PAC_FILE, PLUGIN_DASHBOARD,
+    PLUGIN_HTTP_PROXY, PLUGIN_PROXY_AUTH, PLUGIN_WEB_SERVER,
+    DEFAULT_NUM_WORKERS, PLUGIN_REVERSE_PROXY, DEFAULT_NUM_ACCEPTORS,
+    PLUGIN_INSPECT_TRAFFIC, DEFAULT_DISABLE_HEADERS, PY2_DEPRECATION_MESSAGE,
+    DEFAULT_DEVTOOLS_WS_PATH, PLUGIN_DEVTOOLS_PROTOCOL,
+    PLUGIN_WEBSOCKET_TRANSPORT, DEFAULT_DATA_DIRECTORY_PATH,
+    DEFAULT_MIN_COMPRESSION_LENGTH,
 )
 
 
@@ -182,6 +183,13 @@
                 args.enable_events,
             ),
         )
+        args.enable_metrics = cast(
+            bool,
+            opts.get(
+                'enable_metrics',
+                args.enable_metrics,
+            ),
+        )
 
         # Load default plugins along with user provided --plugins
         default_plugins = [
@@ -195,6 +203,9 @@
             default_plugins + auth_plugins + requested_plugins,
         )
 
+        if bytes_(PLUGIN_METRICS) in default_plugins:
+            args.metrics_lock = multiprocessing.Lock()
+
         # https://github.com/python/mypy/issues/5865
         #
         # def option(t: object, key: str, default: Any) -> Any:
@@ -422,6 +433,10 @@
             default_plugins.append(PLUGIN_INSPECT_TRAFFIC)
             args.enable_events = True
             args.enable_devtools = True
+        if hasattr(args, 'enable_metrics') and args.enable_metrics:
+            default_plugins.append(PLUGIN_WEB_SERVER)
+            default_plugins.append(PLUGIN_METRICS)
+            args.enable_events = True
         if hasattr(args, 'enable_devtools') and args.enable_devtools:
             default_plugins.append(PLUGIN_DEVTOOLS_PROTOCOL)
             default_plugins.append(PLUGIN_WEB_SERVER)

proxy/core/event/metrics.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import os
+import glob
+from typing import Any, Dict
+from pathlib import Path
+from multiprocessing.synchronize import Lock
+
+from ...core.event import EventQueue, EventSubscriber, eventNames
+from ...common.constants import DEFAULT_METRICS_DIRECTORY_PATH
+
+
+class MetricsStorage:
+
+    def __init__(self, lock: Lock) -> None:
+        self._lock = lock
+
+    def get_counter(self, name: str) -> float:
+        with self._lock:
+            return self._get_counter(name)
+
+    def _get_counter(self, name: str) -> float:
+        path = os.path.join(DEFAULT_METRICS_DIRECTORY_PATH, f'{name}.counter')
+        if not os.path.exists(path):
+            return 0
+        return float(Path(path).read_text(encoding='utf-8').strip())
+
+    def incr_counter(self, name: str, by: float = 1.0) -> None:
+        with self._lock:
+            self._incr_counter(name, by)
+
+    def _incr_counter(self, name: str, by: float = 1.0) -> None:
+        current = self._get_counter(name)
+        path = os.path.join(DEFAULT_METRICS_DIRECTORY_PATH, f'{name}.counter')
+        Path(path).write_text(str(current + by), encoding='utf-8')
+
+    def get_gauge(self, name: str) -> float:
+        with self._lock:
+            return self._get_gauge(name)
+
+    def _get_gauge(self, name: str) -> float:
+        path = os.path.join(DEFAULT_METRICS_DIRECTORY_PATH, f'{name}.gauge')
+        if not os.path.exists(path):
+            return 0
+        return float(Path(path).read_text(encoding='utf-8').strip())
+
+    def set_gauge(self, name: str, value: float) -> None:
+        """Stores a single values."""
+        with self._lock:
+            self._set_gauge(name, value)
+
+    def _set_gauge(self, name: str, value: float) -> None:
+        path = os.path.join(DEFAULT_METRICS_DIRECTORY_PATH, f'{name}.gauge')
+        with open(path, 'w', encoding='utf-8') as g:
+            g.write(str(value))
+
+
+class MetricsEventSubscriber:
+
+    def __init__(self, event_queue: EventQueue, metrics_lock: Lock) -> None:
+        """Aggregates metric events pushed by proxy.py core and plugins.
+
+        1) Metrics are stored and managed by multiprocessing safe MetricsStorage
+        2) Collection must be done via MetricsWebServerPlugin endpoint
+        """
+        self.storage = MetricsStorage(metrics_lock)
+        self.subscriber = EventSubscriber(
+            event_queue,
+            callback=lambda event: MetricsEventSubscriber.callback(self.storage, event),
+        )
+
+    def setup(self) -> None:
+        self._setup_metrics_directory()
+        self.subscriber.setup()
+
+    def shutdown(self) -> None:
+        self.subscriber.shutdown()
+
+    def __enter__(self) -> 'MetricsEventSubscriber':
+        self.setup()
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.shutdown()
+
+    @staticmethod
+    def callback(storage: MetricsStorage, event: Dict[str, Any]) -> None:
+        if event['event_name'] == eventNames.WORK_STARTED:
+            storage.incr_counter('work_started')
+        elif event['event_name'] == eventNames.REQUEST_COMPLETE:
+            storage.incr_counter('request_complete')
+        elif event['event_name'] == eventNames.WORK_FINISHED:
+            storage.incr_counter('work_finished')
+        else:
+            print('Unhandled', event)
+
+    def _setup_metrics_directory(self) -> None:
+        os.makedirs(DEFAULT_METRICS_DIRECTORY_PATH, exist_ok=True)
+        patterns = ['*.counter', '*.gauge']
+        for pattern in patterns:
+            files = glob.glob(os.path.join(DEFAULT_METRICS_DIRECTORY_PATH, pattern))
+            for file_path in files:
+                try:
+                    os.remove(file_path)
+                except OSError as e:
+                    print(f'Error deleting file {file_path}: {e}')

proxy/http/server/__init__.py

@@ -10,6 +10,7 @@
 """
 from .web import HttpWebServerPlugin
 from .plugin import ReverseProxyBasePlugin, HttpWebServerBasePlugin
+from .metrics import MetricsWebServerPlugin
 from .protocols import httpProtocolTypes
 from .pac_plugin import HttpWebServerPacFilePlugin
 
@@ -20,4 +21,5 @@
     'HttpWebServerBasePlugin',
     'httpProtocolTypes',
     'ReverseProxyBasePlugin',
+    'MetricsWebServerPlugin',
 ]