feat: add grpc sync flag store

providers/openfeature-provider-flagd/docker-compose.yaml

@@ -0,0 +1,25 @@
+services:
+  flagd:
+    build:
+      context: test-harness
+      dockerfile: flagd/Dockerfile
+    ports:
+      - 8013:8013
+  flagd-unstable:
+    build:
+      context: test-harness
+      dockerfile: flagd/Dockerfile.unstable
+    ports:
+      - 8014:8013
+  flagd-sync:
+    build:
+      context: test-harness
+      dockerfile: sync/Dockerfile
+    ports:
+      - 9090:9090
+  flagd-sync-unstable:
+    build:
+      context: test-harness
+      dockerfile: sync/Dockerfile.unstable
+    ports:
+      - 9091:9090

providers/openfeature-provider-flagd/pyproject.toml

@@ -69,3 +69,8 @@ omit = [
   "src/openfeature/contrib/provider/flagd/proto/*",
   "tests/**",
 ]
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib",
+]

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/proto/flagd/sync/v1/sync_pb2_grpc.py

@@ -2,7 +2,7 @@
 """Client and server classes corresponding to protobuf-defined services."""
 import grpc
 
-from flagd.sync.v1 import sync_pb2 as flagd_dot_sync_dot_v1_dot_sync__pb2
+from . import sync_pb2 as flagd_dot_sync_dot_v1_dot_sync__pb2
 
 
 class FlagSyncServiceStub(object):

providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/proto/sync/v1/sync_service_pb2_grpc.py

@@ -2,7 +2,7 @@
 """Client and server classes corresponding to protobuf-defined services."""
 import grpc
 
-from sync.v1 import sync_service_pb2 as sync_dot_v1_dot_sync__service__pb2
+from . import sync_service_pb2 as sync_dot_v1_dot_sync__service__pb2
 
 
 class FlagSyncServiceStub(object):

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

@@ -9,8 +9,10 @@
 from openfeature.provider.provider import AbstractProvider
 
 from ..config import Config
+from .process.connector.file_watcher import FileWatcherFlagStore
+from .process.connector.grpc_watcher import GrpcWatcherFlagStore
 from .process.custom_ops import ends_with, fractional, sem_ver, starts_with
-from .process.file_watcher import FileWatcherFlagStore
+from .process.flags import FlagStore
 
 T = typing.TypeVar("T")
 
@@ -27,14 +29,14 @@ class InProcessResolver:
     def __init__(self, config: Config, provider: AbstractProvider):
         self.config = config
         self.provider = provider
-        if not self.config.offline_flag_source_path:
-            raise ValueError(
-                "offline_flag_source_path must be provided when using in-process resolver"
+        self.flag_store: FlagStore = (
+            FileWatcherFlagStore(
+                self.config.offline_flag_source_path,
+                self.provider,
+                self.config.offline_poll_interval_seconds,
             )
-        self.flag_store = FileWatcherFlagStore(
-            self.config.offline_flag_source_path,
-            self.provider,
-            self.config.offline_poll_interval_seconds,
+            if self.config.offline_flag_source_path
+            else GrpcWatcherFlagStore(self.config, self.provider)
         )
 
     def shutdown(self) -> None:
@@ -62,15 +64,21 @@ def resolve_float_details(
         default_value: float,
         evaluation_context: typing.Optional[EvaluationContext] = None,
     ) -> FlagResolutionDetails[float]:
-        return self._resolve(key, default_value, evaluation_context)
+        result = self._resolve(key, default_value, evaluation_context)
+        if not isinstance(result.value, float):
+            result.value = float(result.value)
+        return result
 
     def resolve_integer_details(
         self,
         key: str,
         default_value: int,
         evaluation_context: typing.Optional[EvaluationContext] = None,
     ) -> FlagResolutionDetails[int]:
-        return self._resolve(key, default_value, evaluation_context)
+        result = self._resolve(key, default_value, evaluation_context)
+        if not isinstance(result.value, int):
+            result.value = int(result.value)
+        return result
 
     def resolve_object_details(
         self,

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

@@ -1,7 +1,6 @@
 import json
 import logging
 import os
-import re
 import threading
 import time
 import typing
@@ -12,12 +11,12 @@
 from openfeature.exception import ParseError
 from openfeature.provider.provider import AbstractProvider
 
-from .flags import Flag
+from ..flags import Flag, FlagStore
 
 logger = logging.getLogger("openfeature.contrib")
 
 
-class FileWatcherFlagStore:
+class FileWatcherFlagStore(FlagStore):
     def __init__(
         self,
         file_path: str,
@@ -56,7 +55,7 @@ def load_data(self, modified_time: typing.Optional[float] = None) -> None:
                 else:
                     data = json.load(file)
 
-                self.flag_data = self.parse_flags(data)
+                self.flag_data = Flag.parse_flags(data)
                 logger.debug(f"{self.flag_data=}")
                 self.provider.emit_provider_configuration_changed(
                     ProviderEventDetails(flags_changed=list(self.flag_data.keys()))
@@ -72,18 +71,3 @@ def load_data(self, modified_time: typing.Optional[float] = None) -> None:
             logger.exception("Could not parse flag data using flagd syntax")
         except Exception:
             logger.exception("Could not read flags from file")
-
-    def parse_flags(self, flags_data: dict) -> dict:
-        flags = flags_data.get("flags", {})
-        evaluators: typing.Optional[dict] = flags_data.get("$evaluators")
-        if evaluators:
-            transposed = json.dumps(flags)
-            for name, rule in evaluators.items():
-                transposed = re.sub(
-                    rf"{{\s*\"\$ref\":\s*\"{name}\"\s*}}", json.dumps(rule), transposed
-                )
-            flags = json.loads(transposed)
-
-        if not isinstance(flags, dict):
-            raise ParseError("`flags` key of configuration must be a dictionary")
-        return {key: Flag.from_dict(key, data) for key, data in flags.items()}

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

@@ -0,0 +1,92 @@
+import json
+import logging
+import threading
+import time
+import typing
+
+import grpc
+
+from openfeature.event import ProviderEventDetails
+from openfeature.exception import ParseError
+from openfeature.provider.provider import AbstractProvider
+
+from ....config import Config
+from ....proto.flagd.sync.v1 import sync_pb2, sync_pb2_grpc
+from ..flags import Flag, FlagStore
+
+logger = logging.getLogger("openfeature.contrib")
+
+
+class GrpcWatcherFlagStore(FlagStore):
+    INIT_BACK_OFF = 2
+    MAX_BACK_OFF = 120
+
+    def __init__(self, config: Config, provider: AbstractProvider):
+        self.provider = provider
+        self.flag_data: typing.Mapping[str, Flag] = {}
+        channel_factory = grpc.secure_channel if config.tls else grpc.insecure_channel
+        self.channel = channel_factory(f"{config.host}:{config.port}")
+        self.stub = sync_pb2_grpc.FlagSyncServiceStub(self.channel)
+
+        self.connected = False
+
+        # TODO: Add selector
+
+        self.thread = threading.Thread(target=self.sync_flags, daemon=True)
+        self.thread.start()
+
+        ## block until ready or deadline reached
+
+        # TODO: get deadline from user
+        deadline = 2 + time.time()
+        while not self.connected and time.time() < deadline:
+            logger.debug("blocking on init")
+            time.sleep(0.05)
+
+        if not self.connected:
+            logger.warning(
+                "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
+            )
+
+    def shutdown(self) -> None:
+        pass
+
+    def get_flag(self, key: str) -> typing.Optional[Flag]:
+        return self.flag_data.get(key)
+
+    def sync_flags(self) -> None:
+        request = sync_pb2.SyncFlagsRequest()  # type:ignore[attr-defined]
+
+        retry_delay = self.INIT_BACK_OFF
+        while True:
+            try:
+                logger.debug("Setting up gRPC sync flags connection")
+                for flag_rsp in self.stub.SyncFlags(request):
+                    flag_str = flag_rsp.flag_configuration
+                    logger.debug(
+                        f"Received flag configuration - {abs(hash(flag_str)) % (10 ** 8)}"
+                    )
+                    self.flag_data = Flag.parse_flags(json.loads(flag_str))
+
+                    self.connected = True
+                    self.provider.emit_provider_configuration_changed(
+                        ProviderEventDetails(flags_changed=list(self.flag_data.keys()))
+                    )
+
+                    # reset retry delay after successsful read
+                    retry_delay = self.INIT_BACK_OFF
+            except grpc.RpcError as e:  # noqa: PERF203
+                logger.error(f"SyncFlags stream error, {e.code()=} {e.details()=}")
+            except json.JSONDecodeError:
+                logger.exception(
+                    f"Could not parse JSON flag data from SyncFlags endpoint: {flag_str=}"
+                )
+            except ParseError:
+                logger.exception(
+                    f"Could not parse flag data using flagd syntax: {flag_str=}"
+                )
+            finally:
+                # self.connected = False
+                logger.info(f"Reconnecting in {retry_delay}s")
+                time.sleep(retry_delay)
+                retry_delay = min(2 * retry_delay, self.MAX_BACK_OFF)

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

@@ -1,9 +1,19 @@
+import json
+import re
 import typing
 from dataclasses import dataclass
 
 from openfeature.exception import ParseError
 
 
+class FlagStore(typing.Protocol):
+    def get_flag(self, key: str) -> typing.Optional["Flag"]:
+        pass
+
+    def shutdown(self) -> None:
+        pass
+
+
 @dataclass
 class Flag:
     key: str
@@ -49,3 +59,19 @@ def get_variant(
             variant_key = str(variant_key).lower()
 
         return variant_key, self.variants.get(variant_key)
+
+    @classmethod
+    def parse_flags(cls, flags_data: dict) -> typing.Dict[str, "Flag"]:
+        flags = flags_data.get("flags", {})
+        evaluators: typing.Optional[dict] = flags_data.get("$evaluators")
+        if evaluators:
+            transposed = json.dumps(flags)
+            for name, rule in evaluators.items():
+                transposed = re.sub(
+                    rf"{{\s*\"\$ref\":\s*\"{name}\"\s*}}", json.dumps(rule), transposed
+                )
+            flags = json.loads(transposed)
+
+        if not isinstance(flags, dict):
+            raise ParseError("`flags` key of configuration must be a dictionary")
+        return {key: Flag.from_dict(key, data) for key, data in flags.items()}

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

@@ -1,35 +1,23 @@
 import typing
 
 import pytest
-from pytest_bdd import given, parsers, then, when
-from tests.e2e.parsers import to_bool
+from pytest_bdd import parsers, then, when
 
-from openfeature import api
 from openfeature.client import OpenFeatureClient
-from openfeature.contrib.provider.flagd import FlagdProvider
-from openfeature.contrib.provider.flagd.config import ResolverType
 from openfeature.evaluation_context import EvaluationContext
 
 JsonPrimitive = typing.Union[str, bool, float, int]
 
 
+def to_bool(s: str) -> bool:
+    return s.lower() == "true"
+
+
 @pytest.fixture
 def evaluation_context() -> EvaluationContext:
     return EvaluationContext()
 
 
-@given("a flagd provider is set", target_fixture="client")
-def setup_provider(flag_file) -> OpenFeatureClient:
-    api.set_provider(
-        FlagdProvider(
-            resolver_type=ResolverType.IN_PROCESS,
-            offline_flag_source_path=flag_file,
-            offline_poll_interval_seconds=0.1,
-        )
-    )
-    return api.get_client()
-
-
 @when(
     parsers.cfparse(
         'a zero-value boolean flag with key "{key}" is evaluated with default value "{default:bool}"',

providers/openfeature-provider-flagd/tests/e2e/inprocess/file/conftest.py

@@ -0,0 +1,18 @@
+from pytest_bdd import given
+
+from openfeature import api
+from openfeature.client import OpenFeatureClient
+from openfeature.contrib.provider.flagd import FlagdProvider
+from openfeature.contrib.provider.flagd.config import ResolverType
+
+
+@given("a flagd provider is set", target_fixture="client")
+def setup_provider(flag_file) -> OpenFeatureClient:
+    api.set_provider(
+        FlagdProvider(
+            resolver_type=ResolverType.IN_PROCESS,
+            offline_flag_source_path=flag_file,
+            offline_poll_interval_seconds=0.1,
+        )
+    )
+    return api.get_client()