add grpc helpers

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

dapr/conf/__init__.py

@@ -24,9 +24,7 @@ def __init__(self):
             default_value = getattr(global_settings, setting)
             env_variable = os.environ.get(setting)
             if env_variable:
-                val = (
-                    type(default_value)(env_variable) if default_value is not None else env_variable
-                )
+                val = self._coerce_env_value(default_value, env_variable)
                 setattr(self, setting, val)
             else:
                 setattr(self, setting, default_value)
@@ -36,5 +34,27 @@ def __getattr__(self, name):
             raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
         return getattr(self, name)
 
+    @staticmethod
+    def _coerce_env_value(default_value, env_variable: str):
+        if default_value is None:
+            return env_variable
+        # Handle booleans explicitly to avoid bool('false') == True
+        if isinstance(default_value, bool):
+            s = env_variable.strip().lower()
+            if s in ('1', 'true', 't', 'yes', 'y', 'on'):
+                return True
+            if s in ('0', 'false', 'f', 'no', 'n', 'off'):
+                return False
+            # Fallback: non-empty -> True for backward-compat
+            return bool(s)
+        # Integers
+        if isinstance(default_value, int) and not isinstance(default_value, bool):
+            return int(env_variable)
+        # Floats
+        if isinstance(default_value, float):
+            return float(env_variable)
+        # Other types: try to cast as before
+        return type(default_value)(env_variable)
+
 
 settings = Settings()

dapr/conf/global_settings.py

@@ -34,6 +34,23 @@
 
 DAPR_HTTP_TIMEOUT_SECONDS = 60
 
+# gRPC keepalive (disabled by default; enable via env to help with idle debugging sessions)
+DAPR_GRPC_KEEPALIVE_ENABLED: bool = False
+DAPR_GRPC_KEEPALIVE_TIME_MS: int = 120000  # send keepalive pings every 120s
+DAPR_GRPC_KEEPALIVE_TIMEOUT_MS: int = (
+    20000  # wait 20s for ack before considering the connection dead
+)
+DAPR_GRPC_KEEPALIVE_PERMIT_WITHOUT_CALLS: bool = False  # allow pings when there are no active calls
+
+# gRPC retries (disabled by default; enable via env to apply channel service config)
+DAPR_GRPC_RETRY_ENABLED: bool = False
+DAPR_GRPC_RETRY_MAX_ATTEMPTS: int = 4
+DAPR_GRPC_RETRY_INITIAL_BACKOFF_MS: int = 100
+DAPR_GRPC_RETRY_MAX_BACKOFF_MS: int = 1000
+DAPR_GRPC_RETRY_BACKOFF_MULTIPLIER: float = 2.0
+# Comma-separated list of status codes, e.g., 'UNAVAILABLE,DEADLINE_EXCEEDED'
+DAPR_GRPC_RETRY_CODES: str = 'UNAVAILABLE,DEADLINE_EXCEEDED'
+
 # ----- Conversation API settings ------
 
 # Configuration for handling large enums to avoid massive JSON schemas that can exceed LLM token limits

dapr/conf/helpers.py

@@ -1,6 +1,9 @@
+import json
 from urllib.parse import ParseResult, parse_qs, urlparse
 from warnings import warn
 
+from dapr.conf import settings
+
 
 class URIParseConfig:
     DEFAULT_SCHEME = 'dns'
@@ -126,9 +129,9 @@ def _preprocess_uri(self, url: str) -> str:
             # A URI like dns:mydomain:5000 or vsock:mycid:5000 was used
             url = url.replace(':', '://', 1)
         elif (
-            len(url_list) >= 2
-            and '://' not in url
-            and url_list[0] in URIParseConfig.ACCEPTED_SCHEMES
+                len(url_list) >= 2
+                and '://' not in url
+                and url_list[0] in URIParseConfig.ACCEPTED_SCHEMES
         ):
             # A URI like dns:mydomain or dns:[2001:db8:1f70::999:de8:7648:6e8]:mydomain was used
             # Possibly a URI like dns:[2001:db8:1f70::999:de8:7648:6e8]:mydomain was used
@@ -189,3 +192,68 @@ def _validate_path_and_query(self) -> None:
                     f'query parameters are not supported for gRPC endpoints:'
                     f" '{self._parsed_url.query}'"
                 )
+
+
+# ------------------------------
+# gRPC channel options helpers
+# ------------------------------
+
+
+def get_grpc_keepalive_options():
+    """Return a list of keepalive channel options if enabled, else empty list.
+
+    Options are tuples suitable for passing to grpc.{secure,insecure}_channel.
+    """
+    if not settings.DAPR_GRPC_KEEPALIVE_ENABLED:
+        return []
+    return [
+        ('grpc.keepalive_time_ms', int(settings.DAPR_GRPC_KEEPALIVE_TIME_MS)),
+        ('grpc.keepalive_timeout_ms', int(settings.DAPR_GRPC_KEEPALIVE_TIMEOUT_MS)),
+        (
+            'grpc.keepalive_permit_without_calls',
+            1 if settings.DAPR_GRPC_KEEPALIVE_PERMIT_WITHOUT_CALLS else 0,
+        ),
+    ]
+
+
+def get_grpc_retry_service_config_option():
+    """Return ('grpc.service_config', json) option if retry is enabled, else None.
+
+    Applies a universal retry policy via gRPC service config.
+    """
+    if not getattr(settings, 'DAPR_GRPC_RETRY_ENABLED', False):
+        return None
+    retry_policy = {
+        'maxAttempts': int(settings.DAPR_GRPC_RETRY_MAX_ATTEMPTS),
+        'initialBackoff': f'{int(settings.DAPR_GRPC_RETRY_INITIAL_BACKOFF_MS) / 1000.0}s',
+        'maxBackoff': f'{int(settings.DAPR_GRPC_RETRY_MAX_BACKOFF_MS) / 1000.0}s',
+        'backoffMultiplier': float(settings.DAPR_GRPC_RETRY_BACKOFF_MULTIPLIER),
+        'retryableStatusCodes': [
+            c.strip() for c in str(settings.DAPR_GRPC_RETRY_CODES).split(',') if c.strip()
+        ],
+    }
+    service_config = {
+        'methodConfig': [
+            {
+                'name': [{'service': ''}],  # apply to all services
+                'retryPolicy': retry_policy,
+            }
+        ]
+    }
+    return ('grpc.service_config', json.dumps(service_config))
+
+
+def build_grpc_channel_options(base_options=None):
+    """Combine base options with keepalive and retry policy options.
+
+    Args:
+        base_options: optional iterable of (key, value) tuples.
+    Returns:
+        list of (key, value) tuples.
+    """
+    options = list(base_options or [])
+    options.extend(get_grpc_keepalive_options())
+    retry_opt = get_grpc_retry_service_config_option()
+    if retry_opt is not None:
+        options.append(retry_opt)
+    return options