fix: add retry policy to grpc calls.

Some error codes are not on the stream, eg. istio is sending some special ones, and we already used to implement the same logic in java to bypass this issue.

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/grpc.py

@@ -80,6 +80,43 @@ def _generate_channel(self, config: Config) -> grpc.Channel:
             ("grpc.initial_reconnect_backoff_ms", config.retry_backoff_ms),
             ("grpc.max_reconnect_backoff_ms", config.retry_backoff_max_ms),
             ("grpc.min_reconnect_backoff_ms", config.deadline_ms),
+            ("grpc.server_config", {
+                "methodConfig": [
+                    {
+                        "name": [
+                            {
+                                "service": "flagd.sync.v1.FlagSyncService"
+                            },
+                            {
+                                "service": "flagd.evaluation.v1.Service"
+                            }
+                        ],
+                        "retryPolicy": {
+                            "maxAttempts": 3.0,
+                            "initialBackoff": "1s",
+                            "maxBackoff": "5s",
+                            "backoffMultiplier": 2.0,
+                            "retryableStatusCodes": [
+                                "CANCELLED",
+                                "UNKNOWN",
+                                "INVALID_ARGUMENT",
+                                "NOT_FOUND",
+                                "ALREADY_EXISTS",
+                                "PERMISSION_DENIED",
+                                "RESOURCE_EXHAUSTED",
+                                "FAILED_PRECONDITION",
+                                "ABORTED",
+                                "OUT_OF_RANGE",
+                                "UNIMPLEMENTED",
+                                "INTERNAL",
+                                "UNAVAILABLE",
+                                "DATA_LOSS",
+                                "UNAUTHENTICATED"
+                            ]
+                        }
+                    }
+                ]
+            })
         ]
         if config.tls:
             channel_args = {

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py

@@ -25,12 +25,12 @@
 
 class GrpcWatcher(FlagStateConnector):
     def __init__(
-        self,
-        config: Config,
-        flag_store: FlagStore,
-        emit_provider_ready: typing.Callable[[ProviderEventDetails, dict], None],
-        emit_provider_error: typing.Callable[[ProviderEventDetails], None],
-        emit_provider_stale: typing.Callable[[ProviderEventDetails], None],
+            self,
+            config: Config,
+            flag_store: FlagStore,
+            emit_provider_ready: typing.Callable[[ProviderEventDetails, dict], None],
+            emit_provider_error: typing.Callable[[ProviderEventDetails], None],
+            emit_provider_stale: typing.Callable[[ProviderEventDetails], None],
     ):
         self.flag_store = flag_store
         self.config = config
@@ -62,6 +62,43 @@ def _generate_channel(self, config: Config) -> grpc.Channel:
             ("grpc.initial_reconnect_backoff_ms", config.retry_backoff_ms),
             ("grpc.max_reconnect_backoff_ms", config.retry_backoff_max_ms),
             ("grpc.min_reconnect_backoff_ms", config.stream_deadline_ms),
+            ("grpc.server_config", {
+                "methodConfig": [
+                    {
+                        "name": [
+                            {
+                                "service": "flagd.sync.v1.FlagSyncService"
+                            },
+                            {
+                                "service": "flagd.evaluation.v1.Service"
+                            }
+                        ],
+                        "retryPolicy": {
+                            "maxAttempts": 3.0,
+                            "initialBackoff": "1s",
+                            "maxBackoff": "5s",
+                            "backoffMultiplier": 2.0,
+                            "retryableStatusCodes": [
+                                "CANCELLED",
+                                "UNKNOWN",
+                                "INVALID_ARGUMENT",
+                                "NOT_FOUND",
+                                "ALREADY_EXISTS",
+                                "PERMISSION_DENIED",
+                                "RESOURCE_EXHAUSTED",
+                                "FAILED_PRECONDITION",
+                                "ABORTED",
+                                "OUT_OF_RANGE",
+                                "UNIMPLEMENTED",
+                                "INTERNAL",
+                                "UNAVAILABLE",
+                                "DATA_LOSS",
+                                "UNAUTHENTICATED"
+                            ]
+                        }
+                    }
+                ]
+            })
         ]
         if config.default_authority is not None:
             options.append(("grpc.default_authority", config.default_authority))
@@ -120,8 +157,8 @@ def monitor(self) -> None:
     def _state_change_callback(self, new_state: grpc.ChannelConnectivity) -> None:
         logger.debug(f"gRPC state change: {new_state}")
         if (
-            new_state == grpc.ChannelConnectivity.READY
-            or new_state == grpc.ChannelConnectivity.IDLE
+                new_state == grpc.ChannelConnectivity.READY
+                or new_state == grpc.ChannelConnectivity.IDLE
         ):
             if not self.thread or not self.thread.is_alive():
                 self.thread = threading.Thread(
@@ -199,11 +236,11 @@ def listen(self) -> None:
 
                 logger.debug("Setting up gRPC sync flags connection")
                 for flag_rsp in self.stub.SyncFlags(
-                    request, wait_for_ready=True, **call_args
+                        request, wait_for_ready=True, **call_args
                 ):
                     flag_str = flag_rsp.flag_configuration
                     logger.debug(
-                        f"Received flag configuration - {abs(hash(flag_str)) % (10**8)}"
+                        f"Received flag configuration - {abs(hash(flag_str)) % (10 ** 8)}"
                     )
                     self.flag_store.update(json.loads(flag_str))