todds changes

Signed-off-by: Lea Konvalinka <lea.konvalinka@dynatrace.com>

Author: toddbaert

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

@@ -68,6 +68,7 @@ def __init__(
         self.streamline_deadline_seconds = config.stream_deadline_ms * 0.001
         self.deadline = config.deadline_ms * 0.001
         self.connected = False
+        self._is_fatal = False
         self.channel = self._generate_channel(config)
         self.stub = evaluation_pb2_grpc.ServiceStub(self.channel)
 
@@ -153,9 +154,14 @@ def connect(self) -> None:
         ## block until ready or deadline reached
         timeout = self.deadline + time.monotonic()
         while not self.connected and time.monotonic() < timeout:
+            if self._is_fatal:
+                raise ProviderFatalError("fatal gRPC status code")
             time.sleep(0.05)
         logger.debug("Finished blocking gRPC state initialization")
 
+        if self._is_fatal:
+            raise ProviderFatalError("fatal gRPC status code")
+
         if not self.connected:
             raise ProviderNotReadyError(
                 "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
@@ -166,6 +172,8 @@ def monitor(self) -> None:
 
     def _state_change_callback(self, new_state: ChannelConnectivity) -> None:
         logger.debug(f"gRPC state change: {new_state}")
+        if self._is_fatal:
+            return
         if (
             new_state == grpc.ChannelConnectivity.READY
             or new_state == grpc.ChannelConnectivity.IDLE
@@ -197,6 +205,8 @@ def _state_change_callback(self, new_state: ChannelConnectivity) -> None:
             self.connected = False
 
     def emit_error(self) -> None:
+        if self._is_fatal:
+            return
         logger.debug("gRPC error emitted")
         if self.cache:
             self.cache.clear()
@@ -238,7 +248,15 @@ def listen(self) -> None:
                 # although it seems like this error log is not interesting, without it, the retry is not working as expected
                 logger.debug(f"SyncFlags stream error, {e.code()=} {e.details()=}")
                 if e.code().name in self.config.fatal_status_codes:
-                    raise ProviderFatalError("fatal error") from e
+                    self._is_fatal = True
+                    self.active = False
+                    self.emit_provider_error(
+                        ProviderEventDetails(
+                            message=f"Fatal gRPC status code: {e.code()}",
+                            error_code=ErrorCode.PROVIDER_FATAL,
+                        )
+                    )
+                    return
             except ParseError:
                 logger.exception(
                     f"Could not parse flag data using flagd syntax: {message=}"

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

@@ -55,6 +55,7 @@ def __init__(
         self.emit_provider_stale = emit_provider_stale
 
         self.connected = False
+        self._is_fatal = False
         self.thread: typing.Optional[threading.Thread] = None
         self.timer: typing.Optional[threading.Timer] = None
 
@@ -137,9 +138,14 @@ def connect(self) -> None:
         ## block until ready or deadline reached
         timeout = self.deadline + time.monotonic()
         while not self.connected and time.monotonic() < timeout:
+            if self._is_fatal:
+                raise ProviderFatalError("fatal gRPC status code")
             time.sleep(0.05)
         logger.debug("Finished blocking gRPC state initialization")
 
+        if self._is_fatal:
+            raise ProviderFatalError("fatal gRPC status code")
+
         if not self.connected:
             raise ProviderNotReadyError(
                 "Blocking init finished before data synced. Consider increasing startup deadline to avoid inconsistent evaluations."
@@ -150,6 +156,8 @@ def monitor(self) -> None:
 
     def _state_change_callback(self, new_state: grpc.ChannelConnectivity) -> None:
         logger.debug(f"gRPC state change: {new_state}")
+        if self._is_fatal:
+            return
         if (
             new_state == grpc.ChannelConnectivity.READY
             or new_state == grpc.ChannelConnectivity.IDLE
@@ -181,6 +189,8 @@ def _state_change_callback(self, new_state: grpc.ChannelConnectivity) -> None:
             self.connected = False
 
     def emit_error(self) -> None:
+        if self._is_fatal:
+            return
         logger.debug("gRPC error emitted")
         self.emit_provider_error(
             ProviderEventDetails(
@@ -274,7 +284,16 @@ def listen(self) -> None:
                         return
             except grpc.RpcError as e:  # noqa: PERF203
                 logger.warning(f"SyncFlags stream error, {e.code()=} {e.details()=}")
-                self._raise_on_fatal_status_code(e)
+                if e.code().name in self.config.fatal_status_codes:
+                    self._is_fatal = True
+                    self.active = False
+                    self.emit_provider_error(
+                        ProviderEventDetails(
+                            message=f"Fatal gRPC status code: {e.code()}",
+                            error_code=ErrorCode.PROVIDER_FATAL,
+                        )
+                    )
+                    return
             except json.JSONDecodeError:
                 logger.exception(
                     f"Could not parse JSON flag data from SyncFlags endpoint: {flag_str=}"
@@ -293,8 +312,3 @@ def generate_grpc_call_args(self) -> GrpcMultiCallableArgs:
         if metadata is not None:
             call_args["metadata"] = metadata
         return call_args
-
-    def _raise_on_fatal_status_code(self, e: grpc.RpcError) -> None:
-        if e.code().name in self.config.fatal_status_codes:
-            logger.error(f"Fatal gRPC status code received: {e.code()}")
-            raise ProviderFatalError(f"Fatal gRPC status code: {e.code()}") from e

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

@@ -1,70 +1,98 @@
-import os.path
+import logging
+import os
 import time
 import typing
 from pathlib import Path
 
 import grpc
 from grpc_health.v1 import health_pb2, health_pb2_grpc
-from testcontainers.core.container import DockerContainer
-from testcontainers.core.wait_strategies import LogMessageWaitStrategy
+from testcontainers.compose import DockerCompose
 
 from openfeature.contrib.provider.flagd.config import ResolverType
 
+logger = logging.getLogger(__name__)
+
 HEALTH_CHECK = 8014
 LAUNCHPAD = 8080
 FORBIDDEN = 9212
 
 
-class FlagdContainer(DockerContainer):
+class FlagdContainer:
+    """Manages the docker-compose environment for flagd e2e tests.
+
+    Uses docker-compose to start both flagd and envoy containers,
+    so the envoy forbidden endpoint (port 9212) returns a proper HTTP 403.
+    """
+
     def __init__(
         self,
         feature: typing.Optional[str] = None,
         **kwargs,
     ) -> None:
+        self._test_harness_dir = (
+            Path(__file__).parents[2] / "openfeature" / "test-harness"
+        )
+        self._version = (self._test_harness_dir / "version.txt").read_text().rstrip()
+
         image: str = "ghcr.io/open-feature/flagd-testbed"
         if feature is not None:
             image = f"{image}-{feature}"
-        path = Path(__file__).parents[2] / "openfeature/test-harness/version.txt"
-        data = path.read_text().rstrip()
-        super().__init__(f"{image}:v{data}", **kwargs)
-        self.rpc = 8013
-        self.ipr = 8015
+
         self.flagDir = Path("./flags")
         self.flagDir.mkdir(parents=True, exist_ok=True)
-        self.with_exposed_ports(self.rpc, self.ipr, HEALTH_CHECK, LAUNCHPAD, FORBIDDEN)
-        self.with_volume_mapping(os.path.abspath(self.flagDir.name), "/flags", "rw")
-        self.waiting_for(LogMessageWaitStrategy("listening").with_startup_timeout(5))
 
-    def get_port(self, resolver_type: ResolverType):
+        # Set environment variables for docker-compose substitution
+        os.environ["IMAGE"] = image
+        os.environ["VERSION"] = f"v{self._version}"
+        os.environ["FLAGS_DIR"] = str(self.flagDir.absolute())
+
+        self._compose = DockerCompose(
+            context=str(self._test_harness_dir),
+            compose_file_name="docker-compose.yaml",
+            wait=True,
+        )
+
+    def get_port(self, resolver_type: ResolverType) -> int:
         if resolver_type == ResolverType.RPC:
-            return self.get_exposed_port(self.rpc)
+            return self._compose.get_service_port("flagd", 8013)
         else:
-            return self.get_exposed_port(self.ipr)
+            return self._compose.get_service_port("flagd", 8015)
 
-    def get_launchpad_url(self):
-        return f"http://localhost:{self.get_exposed_port(LAUNCHPAD)}"
+    def get_exposed_port(self, port: int) -> int:
+        """Get mapped port. For FORBIDDEN (9212) returns envoy port, otherwise flagd port."""
+        if port == FORBIDDEN:
+            return self._compose.get_service_port("envoy", FORBIDDEN)
+        return self._compose.get_service_port("flagd", port)
+
+    def get_launchpad_url(self) -> str:
+        port = self._compose.get_service_port("flagd", LAUNCHPAD)
+        return f"http://localhost:{port}"
 
     def start(self) -> "FlagdContainer":
-        super().start()
-        self._checker(self.get_container_host_ip(), self.get_exposed_port(HEALTH_CHECK))
+        self._compose.start()
+        host = self._compose.get_service_host("flagd", HEALTH_CHECK) or "localhost"
+        port = self._compose.get_service_port("flagd", HEALTH_CHECK)
+        self._checker(host, port)
         return self
 
+    def stop(self) -> None:
+        self._compose.stop()
+
     def _checker(self, host: str, port: int) -> None:
         # Give an extra second before continuing
         time.sleep(1)
-        # Second we use the GRPC health check endpoint
-        with grpc.insecure_channel(host + ":" + str(port)) as channel:
+        # Use the GRPC health check endpoint
+        with grpc.insecure_channel(f"{host}:{port}") as channel:
             health_stub = health_pb2_grpc.HealthStub(channel)
 
             def health_check_call(stub: health_pb2_grpc.HealthStub):
                 request = health_pb2.HealthCheckRequest()
-                resp = stub.Check(request)
-                if resp.status == health_pb2.HealthCheckResponse.SERVING:
-                    return True
-                elif resp.status == health_pb2.HealthCheckResponse.NOT_SERVING:
+                try:
+                    resp = stub.Check(request)
+                    return resp.status == health_pb2.HealthCheckResponse.SERVING
+                except Exception:
                     return False
 
-            # Should succeed
             # Check health status every 1 second for 30 seconds
             ok = False
             for _ in range(30):
@@ -74,4 +102,4 @@ def health_check_call(stub: health_pb2_grpc.HealthStub):
                 time.sleep(1)
 
             if not ok:
-                raise ConnectionError("flagD not ready in time")
+                raise ConnectionError("flagd not ready in time")