Merge branch 'feat/caching' into feat/grpc-sync-addition

Author: aepfli

providers/openfeature-provider-flagd/pyproject.toml

@@ -24,6 +24,7 @@ dependencies = [
   "panzi-json-logic>=1.0.1",
   "semver>=3,<4",
   "pyyaml>=6.0.1",
+  "cachebox"
 ]
 requires-python = ">=3.8"
 

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

@@ -2,6 +2,15 @@
 import typing
 from enum import Enum
 
+ENV_VAR_MAX_CACHE_SIZE = "FLAGD_MAX_CACHE_SIZE"
+ENV_VAR_CACHE_TYPE = "FLAGD_CACHE_TYPE"
+ENV_VAR_OFFLINE_POLL_INTERVAL_SECONDS = "FLAGD_OFFLINE_POLL_INTERVAL_SECONDS"
+ENV_VAR_OFFLINE_FLAG_SOURCE_PATH = "FLAGD_OFFLINE_FLAG_SOURCE_PATH"
+ENV_VAR_PORT = "FLAGD_PORT"
+ENV_VAR_RESOLVER_TYPE = "FLAGD_RESOLVER_TYPE"
+ENV_VAR_TLS = "FLAGD_TLS"
+ENV_VAR_HOST = "FLAGD_HOST"
+
 T = typing.TypeVar("T")
 
 
@@ -23,6 +32,11 @@ class ResolverType(Enum):
     IN_PROCESS = "in-process"
 
 
+class CacheType(Enum):
+    LRU = "lru"
+    DISABLED = "disabled"
+
+
 class Config:
     def __init__(  # noqa: PLR0913
         self,
@@ -35,13 +49,12 @@ def __init__(  # noqa: PLR0913
         resolver_type: typing.Optional[ResolverType] = None,
         offline_flag_source_path: typing.Optional[str] = None,
         offline_poll_interval_seconds: typing.Optional[float] = None,
+        cache_type: typing.Optional[CacheType] = None,
+        max_cache_size: typing.Optional[int] = None,
     ):
-        self.host = env_or_default("FLAGD_HOST", "localhost") if host is None else host
-        self.port = (
-            env_or_default("FLAGD_PORT", 8013, cast=int) if port is None else port
-        )
+        self.host = env_or_default(ENV_VAR_HOST, "localhost") if host is None else host
         self.tls = (
-            env_or_default("FLAGD_TLS", False, cast=str_to_bool) if tls is None else tls
+            env_or_default(ENV_VAR_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 = (
@@ -53,17 +66,36 @@ def __init__(  # noqa: PLR0913
             env_or_default("FLAGD_SELECTOR", None) if selector is None else selector
         )
         self.resolver_type = (
-            ResolverType(env_or_default("FLAGD_RESOLVER_TYPE", "grpc"))
+            ResolverType(env_or_default(ENV_VAR_RESOLVER_TYPE, "grpc"))
             if resolver_type is None
             else resolver_type
         )
+
+        default_port = 8013 if self.resolver_type is ResolverType.GRPC else 8015
+        self.port = (
+            env_or_default(ENV_VAR_PORT, default_port, cast=int)
+            if port is None
+            else port
+        )
         self.offline_flag_source_path = (
-            env_or_default("FLAGD_OFFLINE_FLAG_SOURCE_PATH", None)
+            env_or_default(ENV_VAR_OFFLINE_FLAG_SOURCE_PATH, None)
             if offline_flag_source_path is None
             else offline_flag_source_path
         )
         self.offline_poll_interval_seconds = (
-            float(env_or_default("FLAGD_OFFLINE_POLL_INTERVAL_SECONDS", 1.0))
+            float(env_or_default(ENV_VAR_OFFLINE_POLL_INTERVAL_SECONDS, 1.0))
             if offline_poll_interval_seconds is None
             else offline_poll_interval_seconds
         )
+
+        self.cache_type = (
+            CacheType(env_or_default(ENV_VAR_CACHE_TYPE, CacheType.DISABLED))
+            if cache_type is None
+            else cache_type
+        )
+
+        self.max_cache_size = (
+            env_or_default(ENV_VAR_MAX_CACHE_SIZE, 16, cast=int)
+            if max_cache_size is None
+            else max_cache_size
+        )

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

@@ -29,7 +29,7 @@
 from openfeature.provider.metadata import Metadata
 from openfeature.provider.provider import AbstractProvider
 
-from .config import Config, ResolverType
+from .config import CacheType, Config, ResolverType
 from .resolvers import AbstractResolver, GrpcResolver, InProcessResolver
 
 T = typing.TypeVar("T")
@@ -49,6 +49,8 @@ def __init__(  # noqa: PLR0913
         resolver_type: typing.Optional[ResolverType] = None,
         offline_flag_source_path: typing.Optional[str] = None,
         offline_poll_interval_seconds: typing.Optional[float] = None,
+        cache_type: typing.Optional[CacheType] = None,
+        max_cache_size: typing.Optional[int] = None,
     ):
         """
         Create an instance of the FlagdProvider
@@ -68,13 +70,20 @@ def __init__(  # noqa: PLR0913
             resolver_type=resolver_type,
             offline_flag_source_path=offline_flag_source_path,
             offline_poll_interval_seconds=offline_poll_interval_seconds,
+            cache_type=cache_type,
+            max_cache_size=max_cache_size,
         )
 
         self.resolver = self.setup_resolver()
 
     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,

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

@@ -1,6 +1,10 @@
+import logging
+import threading
+import time
 import typing
 
 import grpc
+from cachebox import LRUCache  # type:ignore[import-not-found]
 from google.protobuf.json_format import MessageToDict
 from google.protobuf.struct_pb2 import Struct
 from schemas.protobuf.flagd.evaluation.v1 import (  # type:ignore[import-not-found]
@@ -9,33 +13,123 @@
 )
 
 from openfeature.evaluation_context import EvaluationContext
+from openfeature.event import ProviderEventDetails
 from openfeature.exception import (
+    ErrorCode,
     FlagNotFoundError,
     GeneralError,
     InvalidContextError,
     ParseError,
     TypeMismatchError,
 )
-from openfeature.flag_evaluation import FlagResolutionDetails
+from openfeature.flag_evaluation import FlagResolutionDetails, Reason
 
-from ..config import Config
+from ..config import CacheType, Config
 from ..flag_type import FlagType
 from .protocol import AbstractResolver
 
 T = typing.TypeVar("T")
 
+logger = logging.getLogger("openfeature.contrib")
+
 
 class GrpcResolver(AbstractResolver):
-    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 = evaluation_pb2_grpc.ServiceStub(self.channel)
+        self.retry_backoff_seconds = 0.1
+        self.connected = False
+
+        self._cache = (
+            LRUCache(maxsize=self.config.max_cache_size)
+            if self.config.cache_type == CacheType.LRU
+            else None
+        )
+
+    def initialize(self, evaluation_context: EvaluationContext) -> None:
+        self.connect()
 
     def shutdown(self) -> None:
+        self.active = False
         self.channel.close()
+        if self._cache:
+            self._cache.clear()
+
+    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()
+            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
+
+                    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 * retry_delay, self.MAX_BACK_OFF)
+
+    def handle_changed_flags(self, data: typing.Any) -> None:
+        changed_flags = list(data["flags"].keys())
+
+        if self._cache:
+            for flag in changed_flags:
+                self._cache.pop(flag)
+
+        self.emit_provider_configuration_changed(ProviderEventDetails(changed_flags))
 
     def resolve_boolean_details(
         self,
@@ -77,13 +171,18 @@ def resolve_object_details(
     ) -> FlagResolutionDetails[typing.Union[dict, list]]:
         return self._resolve(key, FlagType.OBJECT, default_value, evaluation_context)
 
-    def _resolve(  # noqa: PLR0915
+    def _resolve(  # noqa: PLR0915 C901
         self,
         flag_key: str,
         flag_type: FlagType,
         default_value: T,
         evaluation_context: typing.Optional[EvaluationContext],
     ) -> FlagResolutionDetails[T]:
+        if self._cache is not None and flag_key in self._cache:
+            cached_flag: FlagResolutionDetails[T] = self._cache[flag_key]
+            cached_flag.reason = Reason.CACHED
+            return cached_flag
+
         context = self._convert_context(evaluation_context)
         call_args = {"timeout": self.config.timeout}
         try:
@@ -135,12 +234,17 @@ def _resolve(  # noqa: PLR0915
             raise GeneralError(message) from e
 
         # Got a valid flag and valid type. Return it.
-        return FlagResolutionDetails(
+        result = FlagResolutionDetails(
             value=value,
             reason=response.reason,
             variant=response.variant,
         )
 
+        if response.reason == Reason.STATIC and self._cache is not None:
+            self._cache.insert(flag_key, result)
+
+        return result
+
     def _convert_context(
         self, evaluation_context: typing.Optional[EvaluationContext]
     ) -> Struct:

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

@@ -9,8 +9,8 @@
 from .process.connector import FlagStateConnector
 from .process.connector.file_watcher import FileWatcher
 from .process.connector.grpc_watcher import GrpcWatcher
-from .process.targeting import targeting
 from .process.flags import FlagStore
+from .process.targeting import targeting
 
 T = typing.TypeVar("T")
 

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

@@ -130,8 +130,8 @@ def sem_ver(data: dict, *args: JsonLogicArg) -> typing.Optional[bool]:  # noqa:
     arg1, op, arg2 = args
 
     try:
-        v1 = semver.Version.parse(str(arg1))
-        v2 = semver.Version.parse(str(arg2))
+        v1 = parse_version(arg1)
+        v2 = parse_version(arg2)
     except ValueError as e:
         logger.exception(e)
         return None
@@ -155,3 +155,11 @@ def sem_ver(data: dict, *args: JsonLogicArg) -> typing.Optional[bool]:  # noqa:
     else:
         logger.error(f"Op not supported by sem_ver: {op}")
         return None
+
+
+def parse_version(arg: typing.Any) -> semver.Version:
+    version = str(arg)
+    if version.startswith(("v", "V")):
+        version = version[1:]
+
+    return semver.Version.parse(version)