open-feature · colebaileygit · Apr 27, 2024 · Apr 27, 2024 · May 1, 2024 · May 1, 2024
@@ -27,6 +27,28 @@ jobs:
           - "hooks/openfeature-hooks-opentelemetry"
           - "providers/openfeature-provider-flagd"
 
+    services:
+      # flagd-testbed for flagd RPC provider e2e tests
+      flagd:
+        image: ghcr.io/open-feature/flagd-testbed:v0.5.4
+        ports:
+          - 8013:8013
+      # flagd-testbed for flagd RPC provider reconnect e2e tests
+      flagd-unstable:
+        image: ghcr.io/open-feature/flagd-testbed-unstable:v0.5.4
+        ports:
+          - 8014:8013
+      # sync-testbed for flagd in-process provider e2e tests
+      sync:
+        image: ghcr.io/open-feature/sync-testbed:v0.5.4
+        ports:
+          - 9090:9090
+      # sync-testbed for flagd in-process provider reconnect e2e tests
+      sync-unstable:
+        image: ghcr.io/open-feature/sync-testbed-unstable:v0.5.4
+        ports:
+          - 9091:9090
+
     steps:
       - uses: actions/checkout@v4
         with:

@@ -26,7 +26,15 @@ We use `pytest` for our unit testing, making use of `parametrized` to inject cas
 
 ### Integration tests
 
-These are planned once the SDK has been stabilized and a Flagd provider implemented. At that point, we will utilize the [gherkin integration tests](https://github.com/open-feature/test-harness/blob/main/features/evaluation.feature) to validate against a live, seeded Flagd instance.
+The Flagd provider utilizes the [gherkin integration tests](https://github.com/open-feature/test-harness/blob/main/features/evaluation.feature) to validate against a live, seeded Flagd instance.
+
+To run the integration tests:
+
+```bash
+cd providers/openfeature-provider-flagd
+docker-compose up -d  # this runs the flagd sidecars
+hatch run test
+```
 
 ## Pull Request
 
@@ -62,7 +70,7 @@ To start working on a new feature or bugfix, create a new branch and start worki
 ```bash
 git checkout -b feat/NAME_OF_FEATURE
 # Make your changes
-git commit
+git commit -s -m "feat: my feature"
 git push fork feat/NAME_OF_FEATURE
 ```
 

@@ -19,6 +19,19 @@ from openfeature.contrib.provider.flagd import FlagdProvider
 api.set_provider(FlagdProvider())
 ```
 
+
+To use in-process evaluation with flagd gRPC sync service:
+
+```python
+from openfeature import api
+from openfeature.contrib.provider.flagd import FlagdProvider
+from openfeature.contrib.provider.flagd.config import ResolverType
+
+api.set_provider(FlagdProvider(
+    resolver_type=ResolverType.IN_PROCESS,
+))
+```
+
 To use in-process evaluation in offline mode with a file as source:
 
 ```python
@@ -36,12 +49,17 @@ api.set_provider(FlagdProvider(
 
 The default options can be defined in the FlagdProvider constructor.
 
-| Option name    | Type & Values | Default   |
-|----------------|---------------|-----------|
-| host           | str           | localhost |
-| port           | int           | 8013      |
-| schema         | str           | http      |
-| timeout        | int           | 2         |
+| Option name                   | Environment Variable Name           | Type & Values  | Default   |
+|-------------------------------|-------------------------------------|----------------|-----------|
+| resolver_type                 | FLAGD_RESOLVER_TYPE                 |  enum          | grpc      |
+| host                          | FLAGD_HOST                          |  str           | localhost |
+| port                          | FLAGD_PORT                          |  int           | 8013      |
+| tls                           | FLAGD_TLS                           |  bool          | false     |
+| timeout                       |                                     |  int           | 5         |
+| retry_backoff_seconds         | FLAGD_RETRY_BACKOFF_SECONDS         |  float         | 2.0       |
+| selector                      | FLAGD_SELECTOR                      |  str           | None      |
+| offline_flag_source_path      | FLAGD_OFFLINE_FLAG_SOURCE_PATH      |  str           | None      |
+| offline_poll_interval_seconds | FLAGD_OFFLINE_POLL_INTERVAL_SECONDS |  float         | 1.0       |
 
 ## License
 

@@ -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
@@ -58,6 +58,7 @@ cov = [
 exclude = [
   ".gitignore",
   "schemas",
+  "docker-compose.yaml",
 ]
 
 [tool.hatch.build.targets.wheel]

@@ -30,6 +30,8 @@ def __init__(  # noqa: PLR0913
         port: typing.Optional[int] = None,
         tls: typing.Optional[bool] = None,
         timeout: typing.Optional[int] = None,
+        retry_backoff_seconds: typing.Optional[float] = None,
+        selector: typing.Optional[str] = None,
         resolver_type: typing.Optional[ResolverType] = None,
         offline_flag_source_path: typing.Optional[str] = None,
         offline_poll_interval_seconds: typing.Optional[float] = None,
@@ -42,6 +44,14 @@ def __init__(  # noqa: PLR0913
             env_or_default("FLAGD_TLS", False, cast=str_to_bool) if tls is None else tls
         )
         self.timeout = 5 if timeout is None else timeout
+        self.retry_backoff_seconds: float = (
+            float(env_or_default("FLAGD_RETRY_BACKOFF_SECONDS", 2.0))
+            if retry_backoff_seconds is None
+            else retry_backoff_seconds
+        )
+        self.selector = (
+            env_or_default("FLAGD_SELECTOR", None) if selector is None else selector
+        )
         self.resolver_type = (
             ResolverType(env_or_default("FLAGD_RESOLVER_TYPE", "grpc"))
             if resolver_type is None

@@ -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):

@@ -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):

@@ -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):

@@ -43,6 +43,8 @@ def __init__(  # noqa: PLR0913
         port: typing.Optional[int] = None,
         tls: typing.Optional[bool] = None,
         timeout: typing.Optional[int] = None,
+        retry_backoff_seconds: typing.Optional[float] = None,
+        selector: typing.Optional[str] = None,
         resolver_type: typing.Optional[ResolverType] = None,
         offline_flag_source_path: typing.Optional[str] = None,
         offline_poll_interval_seconds: typing.Optional[float] = None,
@@ -60,6 +62,8 @@ def __init__(  # noqa: PLR0913
             port=port,
             tls=tls,
             timeout=timeout,
+            retry_backoff_seconds=retry_backoff_seconds,
+            selector=selector,
             resolver_type=resolver_type,
             offline_flag_source_path=offline_flag_source_path,
             offline_poll_interval_seconds=offline_poll_interval_seconds,
@@ -77,6 +81,10 @@ def setup_resolver(self) -> AbstractResolver:
                 f"`resolver_type` parameter invalid: {self.config.resolver_type}"
             )
 
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        if hasattr(self.resolver, "initialize"):
+            self.resolver.initialize(evaluation_context)
+
     def shutdown(self) -> None:
         if self.resolver:
             self.resolver.shutdown()

@@ -9,8 +9,11 @@
 from openfeature.provider.provider import AbstractProvider
 
 from ..config import Config
+from .process.connector import FlagStateConnector
+from .process.connector.file_watcher import FileWatcher
+from .process.connector.grpc_watcher import GrpcWatcher
 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,18 +30,23 @@
     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(provider)
+        self.connector: FlagStateConnector = (
+            FileWatcher(
+                self.config.offline_flag_source_path,
+                self.provider,
+                self.flag_store,
+                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 GrpcWatcher(self.config, self.provider, self.flag_store)
         )
 
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        self.connector.initialize(evaluation_context)
+
     def shutdown(self) -> None:
-        self.flag_store.shutdown()
+        self.connector.shutdown()
 
     def resolve_boolean_details(
         self,
@@ -62,15 +70,21 @@
         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,

@@ -0,0 +1,11 @@
+import typing
+
+from openfeature.evaluation_context import EvaluationContext
+
+
+class FlagStateConnector(typing.Protocol):
+    def initialize(
+        self, evaluation_context: EvaluationContext
+    ) -> None: ...  # pragma: no cover
+
+    def shutdown(self) -> None: ...  # pragma: no cover
@@ -0,0 +1,96 @@
+import json
+import logging
+import os
+import threading
+import time
+import typing
+
+import yaml
+
+from openfeature.evaluation_context import EvaluationContext
+from openfeature.event import ProviderEventDetails
+from openfeature.exception import ParseError, ProviderNotReadyError
+from openfeature.provider.provider import AbstractProvider
+
+from ..connector import FlagStateConnector
+from ..flags import FlagStore
+
+logger = logging.getLogger("openfeature.contrib")
+
+
+class FileWatcher(FlagStateConnector):
+    def __init__(
+        self,
+        file_path: str,
+        provider: AbstractProvider,
+        flag_store: FlagStore,
+        poll_interval_seconds: float = 1.0,
+    ):
+        self.file_path = file_path
+        self.provider = provider
+        self.poll_interval_seconds = poll_interval_seconds
+
+        self.last_modified = 0.0
+        self.flag_store = flag_store
+        self.should_emit_ready_on_success = False
+
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        self.active = True
+        self.thread = threading.Thread(target=self.refresh_file, daemon=True)
+        self.thread.start()
+
+        # Let this throw exceptions so that provider status is set correctly
+        try:
+            self._load_data()
+            self.should_emit_ready_on_success = True
+        except Exception as err:
+            raise ProviderNotReadyError from err
+
+    def shutdown(self) -> None:
+        self.active = False
+
+    def refresh_file(self) -> None:
+        while self.active:
+            time.sleep(self.poll_interval_seconds)
+            logger.debug("checking for new flag store contents from file")
+            self.safe_load_data()
+
+    def safe_load_data(self) -> None:
+        try:
+            last_modified = os.path.getmtime(self.file_path)
+            if last_modified > self.last_modified:
+                self._load_data(last_modified)
+        except FileNotFoundError:
+            self.handle_error("Provided file path not valid")
+        except json.JSONDecodeError:
+            self.handle_error("Could not parse JSON flag data from file")
+        except yaml.error.YAMLError:
+            self.handle_error("Could not parse YAML flag data from file")
+        except ParseError:
+            self.handle_error("Could not parse flag data using flagd syntax")
+        except Exception:
+            self.handle_error("Could not read flags from file")
+
+    def _load_data(self, modified_time: typing.Optional[float] = None) -> None:
+        with open(self.file_path) as file:
+            if self.file_path.endswith(".yaml"):
+                data = yaml.safe_load(file)
+            else:
+                data = json.load(file)
+
+            self.flag_store.update(data)
+
+            if self.should_emit_ready_on_success:
+                self.provider.emit_provider_ready(
+                    ProviderEventDetails(
+                        message="Reloading file contents recovered from error state"
+                    )
+                )
+                self.should_emit_ready_on_success = False
+
+        self.last_modified = modified_time or os.path.getmtime(self.file_path)
+
+    def handle_error(self, error_message: str) -> None:
+        logger.exception(error_message)
+        self.should_emit_ready_on_success = True
+        self.provider.emit_provider_error(ProviderEventDetails(message=error_message))