feat: add ACTIVE_REQUESTS metrics and Prometheus server deduplication

- Add ACTIVE_REQUESTS gauge to gRPC sync/async interceptors
- Create _is_prometheus_server_running() helper in app_utils
- Prevent duplicate Prometheus server starts across FastAPI/gRPC
- Unify metric interceptor tests with scenario outlines
- Support FastAPI, gRPC, and AsyncgRPC in parameterized tests
- Remove duplicate test scenarios

Author: HosseinNejatiJavaremi

archipy/helpers/interceptors/grpc/metric/server_interceptor.py

@@ -18,10 +18,11 @@ class GrpcServerMetricInterceptor(BaseGrpcServerInterceptor):
     """A gRPC server interceptor for collecting and reporting metrics using Prometheus.
 
     This interceptor measures the response time of gRPC methods and records it in a Prometheus histogram.
+    It also tracks the number of active requests using a Prometheus gauge.
     It also captures errors and logs them for monitoring purposes.
     """
 
-    from prometheus_client import Histogram
+    from prometheus_client import Gauge, Histogram
 
     "Buckets for measuring response times between 0 and 1 second."
     ZERO_TO_ONE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 1000 for i in range(0, 1000, 5)]
@@ -39,20 +40,27 @@ class GrpcServerMetricInterceptor(BaseGrpcServerInterceptor):
 
     "Prometheus histogram for tracking response times of gRPC methods."
     RESPONSE_TIME_SECONDS = Histogram(
-        "response_time_seconds",
-        "Time spent processing request",
+        "grpc_response_time_seconds",
+        "Time spent processing gRPC request",
         labelnames=("package", "service", "method", "status_code"),
         buckets=TOTAL_BUCKETS,
     )
 
+    "Prometheus gauge for tracking active gRPC requests."
+    ACTIVE_REQUESTS = Gauge(
+        "grpc_active_requests",
+        "Number of active gRPC requests",
+        labelnames=("package", "service", "method"),
+    )
+
     def intercept(
         self,
         method: Callable,
         request: object,
         context: grpc.ServicerContext,
         method_name_model: MethodName,
     ) -> object:
-        """Intercepts a gRPC server call to measure response time and capture errors.
+        """Intercepts a gRPC server call to measure response time and track active requests.
 
         Args:
             method (Callable): The gRPC method being intercepted.
@@ -66,47 +74,53 @@ def intercept(
         Raises:
             Exception: If an exception occurs during the method execution, it is captured and logged.
         """
-        try:
-            # Skip metric collection if Prometheus is disabled
-            if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
-                return method(request, context)
+        if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
+            return method(request, context)
+
+        package = method_name_model.package
+        service = method_name_model.service
+        method_name = method_name_model.method
 
-            # Measure the start time
-            start_time = time.time()
+        self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()
 
-            # Execute the gRPC method
+        start_time = time.time()
+        status_code = "OK"
+
+        try:
             result = method(request, context)
 
-            # Record the response time in the Prometheus histogram
-            status_code = "OK"
             if hasattr(context, "code") and callable(context.code):
                 code_method = cast("Callable[[], Any]", context.code)
                 code_obj = code_method()
                 if code_obj is not None:
                     code_name = getattr(code_obj, "name", None)
                     if code_name is not None:
                         status_code = code_name
-            self.RESPONSE_TIME_SECONDS.labels(
-                package=method_name_model.package,
-                service=method_name_model.service,
-                method=method_name_model.method,
-                status_code=status_code,
-            ).observe(time.time() - start_time)
         except Exception as exception:
             BaseUtils.capture_exception(exception)
             raise
         else:
             return result
+        finally:
+            duration = time.time() - start_time
+            self.RESPONSE_TIME_SECONDS.labels(
+                package=package,
+                service=service,
+                method=method_name,
+                status_code=status_code,
+            ).observe(duration)
+            self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()
 
 
 class AsyncGrpcServerMetricInterceptor(BaseAsyncGrpcServerInterceptor):
     """An async gRPC server interceptor for collecting and reporting metrics using Prometheus.
 
     This interceptor measures the response time of async gRPC methods and records it in a Prometheus histogram.
+    It also tracks the number of active requests using a Prometheus gauge.
     It also captures errors and logs them for monitoring purposes.
     """
 
-    from prometheus_client import Histogram
+    from prometheus_client import Gauge, Histogram
 
     "Buckets for measuring response times between 0 and 1 second."
     ZERO_TO_ONE_SECONDS_BUCKETS: ClassVar[list[float]] = [i / 1000 for i in range(0, 1000, 5)]
@@ -124,20 +138,27 @@ class AsyncGrpcServerMetricInterceptor(BaseAsyncGrpcServerInterceptor):
 
     "Prometheus histogram for tracking response times of async gRPC methods."
     RESPONSE_TIME_SECONDS = Histogram(
-        "grpc_async_server_response_time_seconds",
+        "grpc_async_response_time_seconds",
         "Time spent processing async gRPC request",
         labelnames=("package", "service", "method", "status_code"),
         buckets=TOTAL_BUCKETS,
     )
 
+    "Prometheus gauge for tracking active async gRPC requests."
+    ACTIVE_REQUESTS = Gauge(
+        "grpc_async_active_requests",
+        "Number of active async gRPC requests",
+        labelnames=("package", "service", "method"),
+    )
+
     async def intercept(
         self,
         method: Callable,
         request: object,
         context: grpc.aio.ServicerContext,
         method_name_model: MethodName,
     ) -> object:
-        """Intercepts an async gRPC server call to measure response time and capture errors.
+        """Intercepts an async gRPC server call to measure response time and track active requests.
 
         Args:
             method (Callable): The async gRPC method being intercepted.
@@ -151,24 +172,25 @@ async def intercept(
         Raises:
             Exception: If an exception occurs during the method execution, it is captured and logged.
         """
-        try:
-            # Skip metric collection if Prometheus is disabled
-            if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
-                return await method(request, context)
+        if not BaseConfig.global_config().PROMETHEUS.IS_ENABLED:
+            return await method(request, context)
+
+        package = method_name_model.package
+        service = method_name_model.service
+        method_name = method_name_model.method
 
-            # Measure the start time using asyncio event loop time for better precision
-            start_time = asyncio.get_event_loop().time()
-            status_code = "OK"
+        self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).inc()
 
+        start_time = asyncio.get_event_loop().time()
+        status_code = "OK"
+
+        try:
             try:
-                # Execute the async gRPC method
                 result = await method(request, context)
 
-                # Get the actual status code from context
                 if hasattr(context, "code") and context.code():
                     status_code = context.code().name
             except Exception as e:
-                # Determine error status code
                 if isinstance(e, grpc.aio.AioRpcError):
                     code_obj = e.code()
                     if code_obj is not None:
@@ -188,14 +210,14 @@ async def intercept(
             else:
                 return result
             finally:
-                # Record the response time in the Prometheus histogram
                 duration = asyncio.get_event_loop().time() - start_time
                 self.RESPONSE_TIME_SECONDS.labels(
-                    package=method_name_model.package,
-                    service=method_name_model.service,
-                    method=method_name_model.method,
+                    package=package,
+                    service=service,
+                    method=method_name,
                     status_code=status_code,
                 ).observe(duration)
+                self.ACTIVE_REQUESTS.labels(package=package, service=service, method=method_name).dec()
 
         except Exception as exception:
             BaseUtils.capture_exception(exception)

archipy/helpers/utils/app_utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import socket
 from collections.abc import Callable
 from concurrent import futures
 from contextlib import AbstractAsyncContextManager
@@ -52,6 +53,21 @@
     FASTAPI_APP = False
 
 
+def _is_prometheus_server_running(port: int) -> bool:
+    """Check if Prometheus server is already running on the specified port.
+
+    Args:
+        port (int): The port number to check.
+
+    Returns:
+        bool: True if server is running, False otherwise.
+    """
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    result = sock.connect_ex(("localhost", port))
+    sock.close()
+    return result == 0
+
+
 class FastAPIExceptionHandler:
     """Handles various types of errors and converts them to appropriate JSON responses."""
 
@@ -209,18 +225,11 @@ def setup_metric_interceptor(app: FastAPI, config: BaseConfig) -> None:
             return
 
         try:
-            import socket
-
             from prometheus_client import start_http_server
 
             from archipy.helpers.interceptors.fastapi.metric.interceptor import FastAPIMetricInterceptor
 
-            # Conditionally start Prometheus server (check if already running)
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            result = sock.connect_ex(("localhost", config.PROMETHEUS.SERVER_PORT))
-            sock.close()
-
-            if result != 0:
+            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
                 start_http_server(config.PROMETHEUS.SERVER_PORT)
 
             app.add_middleware(FastAPIMetricInterceptor)  # type: ignore[arg-type]
@@ -312,7 +321,9 @@ def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
 
             from archipy.helpers.interceptors.grpc.metric.server_interceptor import AsyncGrpcServerMetricInterceptor
 
-            start_http_server(config.PROMETHEUS.SERVER_PORT)
+            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
+                start_http_server(config.PROMETHEUS.SERVER_PORT)
+
             interceptors.append(AsyncGrpcServerMetricInterceptor())
 
         except Exception:
@@ -356,7 +367,9 @@ def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
 
             from archipy.helpers.interceptors.grpc.metric.server_interceptor import GrpcServerMetricInterceptor
 
-            start_http_server(config.PROMETHEUS.SERVER_PORT)
+            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
+                start_http_server(config.PROMETHEUS.SERVER_PORT)
+
             interceptors.append(GrpcServerMetricInterceptor())
 
         except Exception:

…tures/fastapi_metric_interceptor.feature features/metric_interceptor.featurefeatures/fastapi_metric_interceptor.feature renamed to features/metric_interceptor.feature features/fastapi_metric_interceptor.feature renamed to features/metric_interceptor.feature

@@ -1,39 +1,67 @@
-Feature: FastAPI Metric Interceptor
+Feature: Metric Interceptor
 
-  Scenario: Interceptor is added when Prometheus is enabled
-    Given a FastAPI app with Prometheus enabled
-    When the metric interceptor is setup
-    Then the FastAPI app should have the metric interceptor
+  Scenario Outline: Interceptor is added when Prometheus is enabled
+    Given a <framework> app with Prometheus enabled
+    When the <framework> metric interceptor is setup
+    Then the <framework> app should have the metric interceptor
 
-  Scenario: Interceptor is skipped when Prometheus is disabled
-    Given a FastAPI app with Prometheus disabled
-    When the metric interceptor is setup
-    Then the FastAPI app should not have the metric interceptor
+    Examples:
+      | framework |
+      | FastAPI   |
+      | gRPC      |
+      | AsyncgRPC |
 
-  Scenario: Response time is recorded for successful requests
-    Given a FastAPI app with Prometheus enabled and metric interceptor
-    When a GET request is made to "/test" endpoint
-    Then the response time metric should be recorded
-    And the metric should have method label "GET"
-    And the metric should have status_code label "200"
-    And the metric should have path_template label "/test"
+  Scenario Outline: Interceptor is skipped when Prometheus is disabled
+    Given a <framework> app with Prometheus disabled
+    When the <framework> metric interceptor is setup
+    Then the <framework> app should not have the metric interceptor
+
+    Examples:
+      | framework |
+      | FastAPI   |
+      | gRPC      |
+      | AsyncgRPC |
+
+  Scenario Outline: Response time is recorded for successful requests
+    Given a <framework> app with Prometheus enabled and metric interceptor
+    When a <framework> request is made
+    Then the <framework> response time metric should be recorded
+    And the <framework> metric should have correct labels
+
+    Examples:
+      | framework |
+      | FastAPI   |
+      | gRPC      |
+      | AsyncgRPC |
+
+  Scenario Outline: Active requests gauge increments and decrements correctly
+    Given a <framework> app with Prometheus enabled and metric interceptor
+    When a <framework> request is made
+    Then the <framework> active requests gauge should increment before processing
+    And the <framework> active requests gauge should decrement after processing
+
+    Examples:
+      | framework |
+      | FastAPI   |
+      | gRPC      |
+      | AsyncgRPC |
+
+  Scenario Outline: Prometheus server starts only once
+    Given Prometheus is enabled for <framework>
+    When multiple <framework> apps are created
+    Then the Prometheus server should only start once
+
+    Examples:
+      | framework |
+      | FastAPI   |
+      | gRPC      |
+      | AsyncgRPC |
 
   Scenario: Response time is recorded for failed requests
     Given a FastAPI app with Prometheus enabled and metric interceptor
     When a GET request is made to an endpoint that raises an error
     Then the response time metric should be recorded with status code 500
 
-  Scenario: Active requests gauge increments and decrements correctly
-    Given a FastAPI app with Prometheus enabled and metric interceptor
-    When a GET request is made to "/test" endpoint
-    Then the active requests gauge should increment before processing
-    And the active requests gauge should decrement after processing
-
-  Scenario: Prometheus server starts only once
-    Given Prometheus is enabled
-    When multiple FastAPI apps are created
-    Then the Prometheus server should only start once
-
   Scenario Outline: Metrics include correct labels for parameterized routes
     Given a FastAPI app with Prometheus enabled and metric interceptor with routes
     When a <method> request is made to "<actual_path>" with route pattern "<route_pattern>"