Merge pull request #90 from mahlaghakafili/expose-prometheous-logs

Expose prometheous logs

Author: HosseinNejatiJavaremi

archipy/configs/config_template.py

@@ -193,6 +193,7 @@ class GrpcConfig(BaseModel):
     SERVE_PORT: int = Field(default=8100, description="Port to serve gRPC on")
     SERVE_HOST: str = Field(default="[::]", description="Host to serve gRPC on")  # IPv6 equivalent of 0.0.0.0
     THREAD_WORKER_COUNT: int | None = Field(default=None, description="Number of worker threads")
+    MAX_CONCURRENT_RPCS: int | None = Field(default=None, description="Maximum number of concurrent requests")
     THREAD_PER_CPU_CORE: int = Field(
         default=40,
         description="Threads per CPU core",

archipy/helpers/decorators/__init__.py

@@ -1,4 +1,3 @@
-from .cache import ttl_cache_decorator
 from .deprecation_exception import class_deprecation_error, method_deprecation_error
 from .deprecation_warnings import class_deprecation_warning, method_deprecation_warning
 from .retry import retry_decorator

archipy/helpers/utils/app_utils.py

@@ -2,16 +2,12 @@
 
 import logging
 from collections.abc import Awaitable, Callable
+from concurrent import futures
 from contextlib import AbstractAsyncContextManager
 from http import HTTPStatus
 from typing import Any, cast
 
-from fastapi import FastAPI, Request, Response
-from fastapi.exceptions import RequestValidationError
-from fastapi.responses import JSONResponse
-from fastapi.routing import APIRoute
 from pydantic import ValidationError
-from starlette.middleware.cors import CORSMiddleware
 
 from archipy.configs.base_config import BaseConfig
 from archipy.helpers.utils.base_utils import BaseUtils
@@ -22,6 +18,25 @@
     UnknownError,
 )
 
+try:
+    import grpc
+    from grpc.experimental.aio import server  # type: ignore[import-not-found]
+
+    GRPC_APP = True
+except ImportError:
+    GRPC_APP = False
+
+try:
+    from fastapi import FastAPI, Request, Response
+    from fastapi.exceptions import RequestValidationError
+    from fastapi.responses import JSONResponse
+    from fastapi.routing import APIRoute
+    from starlette.middleware.cors import CORSMiddleware
+
+    FASTAPI_APP = True
+except ImportError:
+    FASTAPI_APP = False
+
 
 class FastAPIExceptionHandler:
     """Handles various types of errors and converts them to appropriate JSON responses."""
@@ -199,6 +214,94 @@ def setup_exception_handlers(app: FastAPI) -> None:
         app.add_exception_handler(Exception, generic_handler)
 
 
+class AsyncGrpcAPIUtils:
+    """async grpc api utilities."""
+
+    @staticmethod
+    def setup_trace_interceptor(config: BaseConfig, interceptors: list) -> None:
+        """Configures trace interceptor for gRPC server if tracing is enabled.
+
+        Args:
+            config (BaseConfig): The configuration object containing tracing settings.
+            interceptors (List): List of gRPC interceptors to add the trace interceptor to.
+        """
+        if not config.ELASTIC_APM.IS_ENABLED and not config.SENTRY.IS_ENABLED:
+            return
+
+        try:
+            from archipy.helpers.interceptors.grpc.trace.server_interceptor import AsyncGrpcServerTraceInterceptor
+
+            interceptors.append(AsyncGrpcServerTraceInterceptor())
+        except Exception:
+            logging.exception("Failed to initialize Trace Interceptor")
+
+    @staticmethod
+    def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
+        """Configures metric interceptor for gRPC server if Prometheus is enabled.
+
+        Args:
+            config (BaseConfig): The configuration object containing Prometheus settings.
+            interceptors (List): List of gRPC interceptors to add the metric interceptor to.
+        """
+        if not config.PROMETHEUS.IS_ENABLED:
+            return
+
+        try:
+            from prometheus_client import start_http_server
+
+            from archipy.helpers.interceptors.grpc.metric.server_interceptor import AsyncGrpcServerMetricInterceptor
+
+            start_http_server(config.PROMETHEUS.SERVER_PORT)
+            interceptors.append(AsyncGrpcServerMetricInterceptor())
+
+        except Exception:
+            logging.exception("Failed to initialize Metric Interceptor")
+
+
+class GrpcAPIUtils:
+    """grpc api utilities."""
+
+    @staticmethod
+    def setup_trace_interceptor(config: BaseConfig, interceptors: list) -> None:
+        """Configures trace interceptor for gRPC server if tracing is enabled.
+
+        Args:
+            config (BaseConfig): The configuration object containing tracing settings.
+            interceptors (List): List of gRPC interceptors to add the trace interceptor to.
+        """
+        if not config.ELASTIC_APM.IS_ENABLED and not config.SENTRY.IS_ENABLED:
+            return
+
+        try:
+            from archipy.helpers.interceptors.grpc.trace.server_interceptor import GrpcServerTraceInterceptor
+
+            interceptors.append(GrpcServerTraceInterceptor())
+        except Exception:
+            logging.exception("Failed to initialize Trace Interceptor")
+
+    @staticmethod
+    def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
+        """Configures metric interceptor for gRPC server if Prometheus is enabled.
+
+        Args:
+            config (BaseConfig): The configuration object containing Prometheus settings.
+            interceptors (List): List of gRPC interceptors to add the metric interceptor to.
+        """
+        if not config.PROMETHEUS.IS_ENABLED:
+            return
+
+        try:
+            from prometheus_client import start_http_server
+
+            from archipy.helpers.interceptors.grpc.metric.server_interceptor import GrpcServerMetricInterceptor
+
+            start_http_server(config.PROMETHEUS.SERVER_PORT)
+            interceptors.append(GrpcServerMetricInterceptor())
+
+        except Exception:
+            logging.exception("Failed to initialize Metric Interceptor")
+
+
 class AppUtils:
     """Utility class for creating and configuring FastAPI applications."""
 
@@ -249,3 +352,56 @@ def create_fastapi_app(
             FastAPIUtils.setup_exception_handlers(app)
 
         return app
+
+    @classmethod
+    def create_async_grpc_app(
+        cls,
+        config: BaseConfig,
+        customized_interceptors: set[Any] | None = None,
+        compression: grpc.Compression | None = None,
+    ) -> server:
+        """Create and configure an async gRPC application."""
+        from archipy.helpers.interceptors.grpc.exception import AsyncGrpcServerExceptionInterceptor
+
+        async_interceptors = [AsyncGrpcServerExceptionInterceptor()]
+
+        AsyncGrpcAPIUtils.setup_trace_interceptor(config, async_interceptors)
+        AsyncGrpcAPIUtils.setup_metric_interceptor(config, async_interceptors)
+
+        app = server(
+            futures.ThreadPoolExecutor(max_workers=config.GRPC.THREAD_WORKER_COUNT),
+            interceptors=async_interceptors,
+            compression=compression,
+            options=config.GRPC.SERVER_OPTIONS_CONFIG_LIST,
+            maximum_concurrent_rpcs=config.GRPC.MAX_CONCURRENT_RPCS,
+        )
+        if customized_interceptors:
+            async_interceptors.extend(customized_interceptors)
+        return app
+
+    @classmethod
+    def create_grpc_app(
+        cls,
+        config: BaseConfig,
+        customized_interceptors: set[Any] | None = None,
+        compression: grpc.Compression | None = None,
+    ) -> grpc.Server:
+        """Create and configure an async gRPC application."""
+        from archipy.helpers.interceptors.grpc.exception import GrpcServerExceptionInterceptor
+
+        interceptors = [GrpcServerExceptionInterceptor()]
+
+        GrpcAPIUtils.setup_trace_interceptor(config, interceptors)
+        GrpcAPIUtils.setup_metric_interceptor(config, interceptors)
+        if customized_interceptors:
+            interceptors.extend(customized_interceptors)
+
+        app = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=config.GRPC.THREAD_WORKER_COUNT),
+            interceptors=interceptors,  # type: ignore
+            compression=compression,
+            options=config.GRPC.SERVER_OPTIONS_CONFIG_LIST,
+            maximum_concurrent_rpcs=config.GRPC.MAX_CONCURRENT_RPCS,
+        )
+
+        return app