POC: Retry with native grpc

dapr/clients/grpc/client.py

@@ -82,6 +82,41 @@
 )
 
 
+class RetryPolicy:
+    """RetryPolicy is a class that holds the retry policy configuration for a gRPC client.
+
+    Args:
+        max_attempts (int): The maximum number of retry attempts.
+        initial_backoff (str): The initial backoff duration.
+        max_backoff (str): The maximum backoff duration.
+        backoff_multiplier (float): The backoff multiplier.
+        retryable_status_codes (List[str]): The list of status codes that are retryable.
+    """
+
+    def __init__(
+        self,
+        max_attempts: Optional[int] = settings.DAPR_API_MAX_RETRIES,
+        initial_backoff: str = '1s',
+        max_backoff: str = '30s',
+        backoff_multiplier: float = 2,
+        retryable_status_codes: List[str] = ['UNAVAILABLE', 'DEADLINE_EXCEEDED'],
+    ):
+        self.max_attempts = max_attempts
+        self.initial_backoff = initial_backoff
+        self.max_backoff = max_backoff
+        self.backoff_multiplier = backoff_multiplier
+        self.retryable_status_codes = retryable_status_codes
+
+    def to_grpc_retry_policy(self):
+        return {
+            'maxAttempts': self.max_attempts,
+            'initialBackoff': self.initial_backoff,
+            'maxBackoff': self.max_backoff,
+            'backoffMultiplier': self.backoff_multiplier,
+            'retryableStatusCodes': self.retryable_status_codes,
+        }
+
+
 class DaprGrpcClient:
     """The convenient layer implementation of Dapr gRPC APIs.
 
@@ -116,6 +151,7 @@ def __init__(
             ]
         ] = None,
         max_grpc_message_length: Optional[int] = None,
+        retry_policy: Optional[RetryPolicy] = None,
     ):
         """Connects to Dapr Runtime and initialize gRPC client stub.
 
@@ -142,6 +178,20 @@ def __init__(
                 ('grpc.primary_user_agent', useragent),
             ]
 
+        retry_policy = retry_policy or RetryPolicy()
+        service_config = json.dumps(
+            {
+                'methodConfig': [
+                    {
+                        'name': [{'service': ''}],
+                        'retryPolicy': retry_policy.to_grpc_retry_policy(),
+                    }
+                ]
+            }
+        )
+        options.append(("grpc.enable_retries", 1))
+        options.append(('grpc.service_config', service_config))
+
         if not address:
             address = settings.DAPR_GRPC_ENDPOINT or (
                 f'{settings.DAPR_RUNTIME_HOST}:' f'{settings.DAPR_GRPC_PORT}'

dapr/conf/__init__.py

@@ -23,7 +23,10 @@ def __init__(self):
         for setting in dir(global_settings):
             default_value = getattr(global_settings, setting)
             env_variable = os.environ.get(setting)
-            setattr(self, setting, env_variable or default_value)
+            if env_variable:
+                setattr(self, setting, type(default_value)(env_variable))
+            else:
+                setattr(self, setting, default_value)
 
     def __getattr__(self, name):
         if name not in dir(global_settings):

dapr/conf/global_settings.py

@@ -27,6 +27,9 @@
 DAPR_API_VERSION = 'v1.0'
 DAPR_HEALTH_TIMEOUT = 60  # seconds
 
+DAPR_API_MAX_RETRIES = 0
+DAPR_API_TIMEOUT_SECONDS = 60
+
 DAPR_API_METHOD_INVOCATION_PROTOCOL = 'http'
 
 DAPR_HTTP_TIMEOUT_SECONDS = 60