feat: add Temporal metrics collection with Prometheus integration

- Add TemporalRuntimeManager singleton for Runtime lifecycle management
- Add prometheus_utils helper for shared Prometheus server management
- Add Temporal metrics configuration to BaseConfig and .env.test
- Integrate Runtime with metrics into TemporalAdapter and Worker
- Add BDD tests for Temporal metrics collection scenarios
- Fix Redis spop() type handling for async edge cases

Author: HosseinNejatiJavaremi

.env.test

@@ -24,6 +24,12 @@ TEMPORAL__HOST=localhost
 TEMPORAL__PORT=7233
 TEMPORAL__NAMESPACE=default
 TEMPORAL__TASK_QUEUE=test-task-queue
+TEMPORAL__ENABLE_METRICS=true
+TEMPORAL__METRICS_PORT=8201
+
+# Prometheus Configuration
+PROMETHEUS__IS_ENABLED=true
+PROMETHEUS__SERVER_PORT=8200
 
 # Redis Configuration
 REDIS__IMAGE=redis:8.4.0-alpine

archipy/adapters/redis/adapters.py

@@ -677,7 +677,10 @@ def spop(self, name: str, count: int | None = None) -> bytes | float | int | str
         Returns:
             bytes | float | int | str | list | None: Popped member(s) or None.
         """
-        return self.client.spop(name, count)
+        result = self.client.spop(name, count)
+        if isinstance(result, Awaitable):
+            raise TypeError("Unexpected awaitable from sync Redis client")
+        return result
 
     @override
     def srem(self, name: str, *values: bytes | str | float) -> RedisIntegerResponseType:

archipy/adapters/temporal/adapters.py

@@ -29,6 +29,7 @@
 from archipy.models.errors.base_error import BaseError
 
 from .ports import TemporalPort
+from .runtime import TemporalRuntimeManager
 
 T = TypeVar("T")
 
@@ -86,6 +87,16 @@ async def get_client(self) -> Client:
                     tls_config = self._build_tls_config()
                     connect_kwargs["tls"] = tls_config
 
+                # Configure Runtime with Prometheus telemetry if enabled
+                if self.config.ENABLE_METRICS:
+                    runtime_manager = TemporalRuntimeManager()
+                    runtime = runtime_manager.get_runtime(
+                        prometheus_enabled=True,
+                        prometheus_port=self.config.METRICS_PORT,
+                    )
+                    if runtime is not None:
+                        connect_kwargs["runtime"] = runtime
+
                 self._client = await Client.connect(
                     f"{self.config.HOST}:{self.config.PORT}",
                     **connect_kwargs,

archipy/adapters/temporal/runtime.py

@@ -0,0 +1,90 @@
+"""Temporal Runtime singleton for managing Runtime instances with telemetry.
+
+This module provides a singleton class for creating and managing Temporal Runtime
+instances with Prometheus metrics integration.
+"""
+
+import logging
+
+from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
+
+from archipy.helpers.metaclasses.singleton import Singleton
+
+logger = logging.getLogger(__name__)
+
+
+class TemporalRuntimeManager(metaclass=Singleton, thread_safe=True):
+    """Singleton manager for Temporal Runtime instances with telemetry configuration.
+
+    This class ensures only one Runtime instance is created and reused across all
+    Temporal clients and workers. Once created with metrics enabled, the Runtime
+    cannot be changed (Temporal SDK limitation).
+
+    Example:
+        ```python
+        from archipy.adapters.temporal.runtime import TemporalRuntimeManager
+
+        # Get the singleton manager
+        manager = TemporalRuntimeManager()
+
+        # Get Runtime with Prometheus enabled
+        runtime = manager.get_runtime(prometheus_enabled=True, prometheus_port=18201)
+        ```
+    """
+
+    def __init__(self) -> None:
+        """Initialize the TemporalRuntimeManager singleton."""
+        self._runtime: Runtime | None = None
+
+    def get_runtime(self, prometheus_enabled: bool = False, prometheus_port: int = 18201) -> Runtime | None:
+        """Get or create a Runtime with Prometheus telemetry.
+
+        Args:
+            prometheus_enabled (bool): Whether to enable Prometheus metrics collection.
+            prometheus_port (int): Port for the Prometheus metrics endpoint.
+
+        Returns:
+            Runtime | None: The configured Runtime instance if metrics are enabled,
+                None otherwise (uses default Runtime).
+
+        Note:
+            Once a Runtime is created with metrics enabled, it cannot be disabled
+            or recreated on a different port due to Temporal SDK limitations.
+            Subsequent calls will return the existing Runtime regardless of parameters.
+        """
+        if not prometheus_enabled:
+            logger.debug("Prometheus metrics disabled for Temporal, using default runtime")
+            return None
+
+        # If Runtime already created, return it (can't change once bound to port)
+        if self._runtime is not None:
+            logger.debug("Returning existing Temporal Runtime instance")
+            return self._runtime
+
+        logger.info("Creating Temporal Runtime with Prometheus metrics on port %d", prometheus_port)
+
+        try:
+            self._runtime = Runtime(
+                telemetry=TelemetryConfig(
+                    metrics=PrometheusConfig(bind_address=f"0.0.0.0:{prometheus_port}"),
+                ),
+            )
+            logger.info("Temporal Runtime created successfully with Prometheus telemetry")
+        except Exception:
+            logger.exception("Failed to create Temporal Runtime with Prometheus config")
+            # Return None so Temporal uses default Runtime
+            return None
+
+        return self._runtime
+
+    def reset_runtime(self) -> None:
+        """Reset the Runtime instance.
+
+        Warning:
+            This does NOT actually close the Runtime or release the port binding.
+            The Temporal SDK does not support Runtime cleanup. This method only
+            resets internal references for testing purposes. The port will remain
+            bound until the process exits.
+        """
+        logger.warning("Resetting Temporal Runtime reference (port remains bound until process exit)")
+        self._runtime = None

archipy/adapters/temporal/worker.py

@@ -297,6 +297,7 @@ async def start_worker(
 
         try:
             # Create the Temporal worker
+            # Note: Worker inherits Runtime (including Prometheus config) from the Client
             worker = Worker(
                 client,
                 task_queue=task_queue,

archipy/configs/config_template.py

@@ -31,30 +31,6 @@ class ElasticsearchConfig(BaseModel):
 
     Contains settings related to Elasticsearch server connectivity, authentication,
     TLS/SSL, request handling, node status management, and batch operation parameters.
-
-    Attributes:
-        HOSTS (list[str]): List of Elasticsearch server hosts (e.g., ['https://localhost:9200']).
-        HTTP_USER_NAME (str | None): Username for HTTP authentication.
-        HTTP_PASSWORD (SecretStr | None): Password for HTTP authentication.
-        CA_CERTS (str | None): Path to CA bundle for SSL verification.
-        SSL_ASSERT_FINGERPRINT (str | None): SSL certificate fingerprint for verification.
-        VERIFY_CERTS (bool): Whether to verify SSL certificates.
-        CLIENT_CERT (str | None): Path to client certificate for TLS authentication.
-        CLIENT_KEY (str | None): Path to client key for TLS authentication.
-        HTTP_COMPRESS (bool): Whether to enable HTTP compression (gzip).
-        REQUEST_TIMEOUT (float | None): Timeout for HTTP requests in seconds.
-        MAX_RETRIES (int): Maximum number of retries per request.
-        RETRY_ON_TIMEOUT (bool): Whether to retry on connection timeouts.
-        RETRY_ON_STATUS (tuple[int, ...]): HTTP status codes to retry on.
-        IGNORE_STATUS (tuple[int, ...]): HTTP status codes to ignore as errors.
-        SNIFF_ON_START (bool): Whether to sniff nodes on client instantiation.
-        SNIFF_BEFORE_REQUESTS (bool): Whether to sniff nodes before requests.
-        SNIFF_ON_NODE_FAILURE (bool): Whether to sniff nodes on node failure.
-        MIN_DELAY_BETWEEN_SNIFFING (float): Minimum delay between sniffing attempts in seconds.
-        NODE_SELECTOR_CLASS (str): Node selector strategy ('round_robin' or 'random').
-        CONNECTIONS_PER_NODE (int): Number of HTTP connections per node.
-        DEAD_NODE_BACKOFF_FACTOR (float): Factor for calculating node timeout duration after failures.
-        MAX_DEAD_NODE_BACKOFF (float): Maximum timeout duration for a dead node in seconds.
     """
 
     HOSTS: list[str] = Field(default=["https://localhost:9200"], description="List of Elasticsearch server hosts")
@@ -821,30 +797,25 @@ class TemporalConfig(BaseModel):
 
     Controls connection parameters, security settings, and timeout configurations
     for Temporal workflow orchestration services.
-
-    Attributes:
-        HOST (str): Temporal server host address.
-        PORT (int): Temporal server port number.
-        NAMESPACE (str): Temporal namespace for workflow isolation.
-        TASK_QUEUE (str): Default task queue for workflow and activity execution.
-        TLS_CA_CERT (str | None): Path to TLS CA certificate for secure connections.
-        TLS_CLIENT_CERT (str | None): Path to TLS client certificate for mutual authentication.
-        TLS_CLIENT_KEY (str | None): Path to TLS client private key.
-        WORKFLOW_EXECUTION_TIMEOUT (int): Maximum workflow execution time in seconds.
-        WORKFLOW_RUN_TIMEOUT (int): Maximum single workflow run time in seconds.
-        WORKFLOW_TASK_TIMEOUT (int): Maximum workflow task processing time in seconds.
-        ACTIVITY_START_TO_CLOSE_TIMEOUT (int): Maximum activity execution time in seconds.
-        ACTIVITY_HEARTBEAT_TIMEOUT (int): Activity heartbeat timeout in seconds.
-        RETRY_MAXIMUM_ATTEMPTS (int): Maximum retry attempts for failed activities.
-        RETRY_BACKOFF_COEFFICIENT (float): Backoff multiplier for retry delays.
-        RETRY_MAXIMUM_INTERVAL (int): Maximum retry interval in seconds.
     """
 
     HOST: str = Field(default="localhost", description="Temporal server host address")
     PORT: int = Field(default=7233, ge=1, le=65535, description="Temporal server port number")
     NAMESPACE: str = Field(default="default", description="Temporal namespace for workflow isolation")
     TASK_QUEUE: str = Field(default="task-queue", description="Default task queue name")
 
+    # Metrics Configuration
+    ENABLE_METRICS: bool = Field(
+        default=False,
+        description="Enable Prometheus metrics collection for Temporal workflows and activities",
+    )
+    METRICS_PORT: int = Field(
+        default=8201,
+        ge=1,
+        le=65535,
+        description="Port for Temporal Prometheus metrics endpoint (separate from main Prometheus port)",
+    )
+
     # TLS Configuration
     TLS_CA_CERT: str | None = Field(default=None, description="Path to TLS CA certificate")
     TLS_CLIENT_CERT: str | None = Field(default=None, description="Path to TLS client certificate")
@@ -925,29 +896,6 @@ class ScyllaDBConfig(BaseModel):
     Contains settings related to ScyllaDB cluster connectivity, authentication,
     compression, consistency levels, connection management, retry policies,
     prepared statement caching, and health checks.
-
-    Attributes:
-        CONTACT_POINTS (list[str]): List of ScyllaDB node addresses.
-        PORT (int): CQL native transport port number.
-        KEYSPACE (str | None): Default keyspace name.
-        USERNAME (str | None): Username for authentication.
-        PASSWORD (SecretStr | None): Password for authentication.
-        PROTOCOL_VERSION (int): Protocol version to use.
-        COMPRESSION (bool): Enable LZ4 compression.
-        CONNECT_TIMEOUT (int): Connection timeout in seconds.
-        REQUEST_TIMEOUT (int): Request timeout in seconds.
-        CONSISTENCY_LEVEL (Literal): Default consistency level.
-        DISABLE_SHARD_AWARENESS (bool): Disable shard awareness (default: False).
-        RETRY_POLICY (Literal): Retry policy type (default: "EXPONENTIAL_BACKOFF").
-            Options: "EXPONENTIAL_BACKOFF", "FALLTHROUGH", "DOWNGRADING_CONSISTENCY".
-        RETRY_MAX_NUM_RETRIES (float): Maximum number of retries for ExponentialBackoffRetryPolicy (default: 3.0).
-        RETRY_MIN_INTERVAL (float): Minimum interval in seconds between retries (default: 0.1).
-        RETRY_MAX_INTERVAL (float): Maximum interval in seconds between retries (default: 10.0).
-        ENABLE_PREPARED_STATEMENT_CACHE (bool): Enable prepared statement caching (default: True).
-        PREPARED_STATEMENT_CACHE_SIZE (int): Maximum cached prepared statements (default: 100).
-        PREPARED_STATEMENT_CACHE_TTL_SECONDS (int): TTL for cache in seconds (default: 3600).
-        HEALTH_CHECK_TIMEOUT (int): Timeout for health check queries in seconds (default: 5).
-        ENABLE_CONNECTION_POOL_MONITORING (bool): Enable pool monitoring (default: False).
     """
 
     CONTACT_POINTS: list[str] = Field(

archipy/helpers/utils/app_utils.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-import socket
 from collections.abc import Callable
 from concurrent import futures
 from contextlib import AbstractAsyncContextManager
@@ -12,6 +11,7 @@
 
 from archipy.configs.base_config import BaseConfig
 from archipy.helpers.utils.base_utils import BaseUtils
+from archipy.helpers.utils.prometheus_utils import is_prometheus_server_running
 from archipy.models.errors import (
     BaseError,
     InvalidArgumentError,
@@ -20,8 +20,8 @@
 )
 
 if TYPE_CHECKING:
-    from grpc.experimental import aio as grpc_aio
-    from grpc.experimental.aio import Server as GrpcAioServer
+    from grpc import aio as grpc_aio
+    from grpc.aio import Server as GrpcAioServer
 
     CreateGrpcServerType = Callable[..., GrpcAioServer]
 else:
@@ -32,7 +32,7 @@
 
 try:
     import grpc
-    from grpc.experimental import aio as grpc_aio
+    from grpc import aio as grpc_aio
 
     create_grpc_server: CreateGrpcServerType = grpc_aio.server
     GRPC_APP = True
@@ -53,19 +53,8 @@
     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
+# Backward compatibility alias
+_is_prometheus_server_running = is_prometheus_server_running
 
 
 class FastAPIExceptionHandler:
@@ -225,12 +214,10 @@ def setup_metric_interceptor(app: FastAPI, config: BaseConfig) -> None:
             return
 
         try:
-            from prometheus_client import start_http_server
-
             from archipy.helpers.interceptors.fastapi.metric.interceptor import FastAPIMetricInterceptor
+            from archipy.helpers.utils.prometheus_utils import start_prometheus_server_if_needed
 
-            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
-                start_http_server(config.PROMETHEUS.SERVER_PORT)
+            start_prometheus_server_if_needed(config.PROMETHEUS.SERVER_PORT)
 
             app.add_middleware(FastAPIMetricInterceptor)  # type: ignore[arg-type]
         except Exception:
@@ -317,12 +304,10 @@ def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
             return
 
         try:
-            from prometheus_client import start_http_server
-
             from archipy.helpers.interceptors.grpc.metric.server_interceptor import AsyncGrpcServerMetricInterceptor
+            from archipy.helpers.utils.prometheus_utils import start_prometheus_server_if_needed
 
-            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
-                start_http_server(config.PROMETHEUS.SERVER_PORT)
+            start_prometheus_server_if_needed(config.PROMETHEUS.SERVER_PORT)
 
             interceptors.append(AsyncGrpcServerMetricInterceptor())
 
@@ -363,12 +348,10 @@ def setup_metric_interceptor(config: BaseConfig, interceptors: list) -> None:
             return
 
         try:
-            from prometheus_client import start_http_server
-
             from archipy.helpers.interceptors.grpc.metric.server_interceptor import GrpcServerMetricInterceptor
+            from archipy.helpers.utils.prometheus_utils import start_prometheus_server_if_needed
 
-            if not _is_prometheus_server_running(config.PROMETHEUS.SERVER_PORT):
-                start_http_server(config.PROMETHEUS.SERVER_PORT)
+            start_prometheus_server_if_needed(config.PROMETHEUS.SERVER_PORT)
 
             interceptors.append(GrpcServerMetricInterceptor())
 
@@ -448,7 +431,7 @@ def create_async_grpc_app(
         AsyncGrpcAPIUtils.setup_metric_interceptor(config, async_interceptors)
 
         if create_grpc_server is None:
-            raise ImportError("grpc.experimental.aio is not available")
+            raise ImportError("grpc.aio is not available")
         app = create_grpc_server(
             futures.ThreadPoolExecutor(max_workers=config.GRPC.THREAD_WORKER_COUNT),
             interceptors=async_interceptors,