fix(flagd): adding events for rpc mode

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

Author: aepfli

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/proto/flagd/evaluation/v1/evaluation_pb2_grpc.py

@@ -2,7 +2,7 @@
 """Client and server classes corresponding to protobuf-defined services."""
 import grpc
 
-from flagd.evaluation.v1 import evaluation_pb2 as flagd_dot_evaluation_dot_v1_dot_evaluation__pb2
+from . import evaluation_pb2 as flagd_dot_evaluation_dot_v1_dot_evaluation__pb2
 
 
 class ServiceStub(object):

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

@@ -69,14 +69,22 @@ def __init__(  # noqa: PLR0913
 
     def setup_resolver(self) -> AbstractResolver:
         if self.config.resolver_type == ResolverType.GRPC:
-            return GrpcResolver(self.config)
+            return GrpcResolver(
+                self.config,
+                self.emit_provider_ready,
+                self.emit_provider_error,
+                self.emit_provider_configuration_changed,
+            )
         elif self.config.resolver_type == ResolverType.IN_PROCESS:
             return InProcessResolver(self.config, self)
         else:
             raise ValueError(
                 f"`resolver_type` parameter invalid: {self.config.resolver_type}"
             )
 
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        self.resolver.initialize(evaluation_context)
+
     def shutdown(self) -> None:
         if self.resolver:
             self.resolver.shutdown()

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

@@ -8,6 +8,9 @@
 
 
 class AbstractResolver(typing.Protocol):
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        return
+
     def shutdown(self) -> None: ...
 
     def resolve_boolean_details(

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

@@ -1,11 +1,16 @@
+import logging
+import threading
+import time
 import typing
 
 import grpc
 from google.protobuf.json_format import MessageToDict
 from google.protobuf.struct_pb2 import Struct
 
 from openfeature.evaluation_context import EvaluationContext
+from openfeature.event import ProviderEventDetails
 from openfeature.exception import (
+    ErrorCode,
     FlagNotFoundError,
     GeneralError,
     InvalidContextError,
@@ -16,23 +21,101 @@
 
 from ..config import Config
 from ..flag_type import FlagType
-from ..proto.schema.v1 import schema_pb2, schema_pb2_grpc
+from ..proto.flagd.evaluation.v1 import evaluation_pb2, evaluation_pb2_grpc
 
 T = typing.TypeVar("T")
 
+logger = logging.getLogger("openfeature.contrib")
+
 
 class GrpcResolver:
-    def __init__(self, config: Config):
+    MAX_BACK_OFF = 120
+
+    def __init__(
+        self,
+        config: Config,
+        emit_provider_ready: typing.Callable[[ProviderEventDetails], None],
+        emit_provider_error: typing.Callable[[ProviderEventDetails], None],
+        emit_provider_configuration_changed: typing.Callable[
+            [ProviderEventDetails], None
+        ],
+    ):
         self.config = config
+        self.emit_provider_ready = emit_provider_ready
+        self.emit_provider_error = emit_provider_error
+        self.emit_provider_configuration_changed = emit_provider_configuration_changed
         channel_factory = (
             grpc.secure_channel if self.config.tls else grpc.insecure_channel
         )
         self.channel = channel_factory(f"{self.config.host}:{self.config.port}")
-        self.stub = schema_pb2_grpc.ServiceStub(self.channel)
+        self.stub = evaluation_pb2_grpc.ServiceStub(self.channel)
+        self.retry_backoff_seconds = 0.1
+        self.connected = False
+
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        self.connect()
 
     def shutdown(self) -> None:
         self.channel.close()
 
+    def connect(self) -> None:
+        self.active = True
+        self.thread = threading.Thread(
+            target=self.listen, daemon=True, name="FlagdGrpcServiceWorkerThread"
+        )
+        self.thread.start()
+
+    def listen(self) -> None:
+        retry_delay = self.retry_backoff_seconds
+        while self.active:
+            request = evaluation_pb2.EventStreamRequest()  # type:ignore[attr-defined]
+            try:
+                logger.debug("Setting up gRPC sync flags connection")
+                for message in self.stub.EventStream(request):
+                    if message.type == "provider_ready":
+                        if not self.connected:
+                            self.emit_provider_ready(
+                                ProviderEventDetails(
+                                    message="gRPC sync connection established"
+                                )
+                            )
+                            self.connected = True
+                            # reset retry delay after successsful read
+                            retry_delay = self.retry_backoff_seconds
+                        else:
+                            self.emit_provider_error(ProviderEventDetails())
+                            self.connected = False
+
+                    elif message.type == "configuration_change":
+                        data = MessageToDict(message)["data"]
+                        self.handle_changed_flags(data)
+
+                    if not self.active:
+                        logger.info("Terminating gRPC sync thread")
+                        return
+            except grpc.RpcError as e:
+                logger.error(f"SyncFlags stream error, {e.code()=} {e.details()=}")
+            except ParseError:
+                logger.exception(
+                    f"Could not parse flag data using flagd syntax: {message=}"
+                )
+
+            self.connected = False
+            self.emit_provider_error(
+                ProviderEventDetails(
+                    message=f"gRPC sync disconnected, reconnecting in {retry_delay}s",
+                    error_code=ErrorCode.GENERAL,
+                )
+            )
+            logger.info(f"gRPC sync disconnected, reconnecting in {retry_delay}s")
+            time.sleep(retry_delay)
+            retry_delay = min(2, self.MAX_BACK_OFF)
+
+    def handle_changed_flags(self, data: typing.Any) -> None:
+        changed_flags = list(data["flags"].keys())
+
+        self.emit_provider_configuration_changed(ProviderEventDetails(changed_flags))
+
     def resolve_boolean_details(
         self,
         key: str,
@@ -84,33 +167,33 @@ def _resolve(  # noqa: PLR0915
         call_args = {"timeout": self.config.timeout}
         try:
             if flag_type == FlagType.BOOLEAN:
-                request = schema_pb2.ResolveBooleanRequest(  # type:ignore[attr-defined]
+                request = evaluation_pb2.ResolveBooleanRequest(  # type:ignore[attr-defined]
                     flag_key=flag_key, context=context
                 )
                 response = self.stub.ResolveBoolean(request, **call_args)
                 value = response.value
             elif flag_type == FlagType.STRING:
-                request = schema_pb2.ResolveStringRequest(  # type:ignore[attr-defined]
+                request = evaluation_pb2.ResolveStringRequest(  # type:ignore[attr-defined]
                     flag_key=flag_key, context=context
                 )
                 response = self.stub.ResolveString(request, **call_args)
                 value = response.value
             elif flag_type == FlagType.OBJECT:
-                request = schema_pb2.ResolveObjectRequest(  # type:ignore[attr-defined]
+                request = evaluation_pb2.ResolveObjectRequest(  # type:ignore[attr-defined]
                     flag_key=flag_key, context=context
                 )
                 response = self.stub.ResolveObject(request, **call_args)
                 value = MessageToDict(response, preserving_proto_field_name=True)[
                     "value"
                 ]
             elif flag_type == FlagType.FLOAT:
-                request = schema_pb2.ResolveFloatRequest(  # type:ignore[attr-defined]
+                request = evaluation_pb2.ResolveFloatRequest(  # type:ignore[attr-defined]
                     flag_key=flag_key, context=context
                 )
                 response = self.stub.ResolveFloat(request, **call_args)
                 value = response.value
             elif flag_type == FlagType.INTEGER:
-                request = schema_pb2.ResolveIntRequest(  # type:ignore[attr-defined]
+                request = evaluation_pb2.ResolveIntRequest(  # type:ignore[attr-defined]
                     flag_key=flag_key, context=context
                 )
                 response = self.stub.ResolveInt(request, **call_args)

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

@@ -3,7 +3,7 @@
 from openfeature.evaluation_context import EvaluationContext
 from openfeature.exception import FlagNotFoundError, ParseError
 from openfeature.flag_evaluation import FlagResolutionDetails, Reason
-from openfeature.provider.provider import AbstractProvider
+from openfeature.provider import AbstractProvider
 
 from ..config import Config
 from .process.file_watcher import FileWatcherFlagStore
@@ -26,6 +26,9 @@ def __init__(self, config: Config, provider: AbstractProvider):
             self.config.offline_poll_interval_seconds,
         )
 
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        pass
+
     def shutdown(self) -> None:
         self.flag_store.shutdown()
 

providers/openfeature-provider-flagd/tests/e2e/steps.py

@@ -508,6 +508,7 @@ def provider_ready_was_executed(client: OpenFeatureClient, context):
 def provider_changed_add(client: OpenFeatureClient, context):
     def provider_changed_handler(event_details: EventDetails):
         context["provider_changed_ran"] = True
+        context["changed_flags"] = event_details.flags_changed
 
     client.add_handler(
         ProviderEvent.PROVIDER_CONFIGURATION_CHANGED, provider_changed_handler

providers/openfeature-provider-flagd/tests/e2e/test_rpc.py

@@ -1,5 +1,5 @@
 import pytest
-from pytest_bdd import scenario, scenarios
+from pytest_bdd import scenarios
 
 from openfeature.contrib.provider.flagd.config import ResolverType
 
@@ -19,12 +19,6 @@ def image():
     return "ghcr.io/open-feature/flagd-testbed:v0.5.13"
 
 
-@pytest.mark.skip(reason="Eventing not implemented")
-@scenario("../../test-harness/gherkin/flagd.feature", "Flag change event")
-def test_flag_change_event():
-    """not implemented"""
-
-
 scenarios(
     "../../test-harness/gherkin/flagd.feature",
     "../../test-harness/gherkin/flagd-json-evaluator.feature",