feat(#739): add probe check in client and operator

  - add e2e unit tests for operator

Author: liquidiert

client/gefyra/api/bridge.py

@@ -3,6 +3,7 @@
 from typing import List, Dict, TYPE_CHECKING
 
 from gefyra.exceptions import CommandTimeoutError, GefyraBridgeError
+from kubernetes.client.exceptions import ApiException
 
 if TYPE_CHECKING:
     from gefyra.configuration import ClientConfiguration
@@ -37,12 +38,12 @@ def get_pods_to_intercept(
 
 
 def check_workloads(
-    pods_to_intercept,
+    pods_to_intercept: dict,
     workload_type: str,
     workload_name: str,
     container_name: str,
     namespace: str,
-    config,
+    config: "ClientConfiguration",
 ):
     from gefyra.cluster.resources import check_pod_valid_for_bridge
 
@@ -57,11 +58,45 @@ def check_workloads(
             f"Could not find {workload_type}/{workload_name} to bridge. Available"
             f" {workload_type}: {', '.join(cleaned_names)}"
         )
+    
     if container_name not in [
         container for c_list in pods_to_intercept.values() for container in c_list
     ]:
         raise RuntimeError(f"Could not find container {container_name} to bridge.")
 
+    # Validate workload and probes
+    api = config.K8S_APP_API
+    try:
+        if workload_type == "deployment":
+            workload = api.read_namespaced_deployment(workload_name, namespace)
+        elif workload_type == "statefulset":
+            workload = api.read_namespaced_stateful_set(workload_name, namespace)
+        elif workload_type == "daemonset":
+            workload = api.read_namespaced_daemon_set(workload_name, namespace)
+        else:
+            raise RuntimeError(f"Unsupported workload type: {workload_type}")
+    except ApiException as e:
+        raise RuntimeError(f"Error fetching workload {workload_type}/{workload_name}: {e}")
+
+    containers = workload.spec.template.spec.containers
+    target_container = next(
+        (c for c in containers if c.name == container_name), None
+    )
+    if not target_container:
+        raise RuntimeError(f"Container {container_name} not found in workload {workload_type}/{workload_name}.")
+
+    def validate_http_probe(probe, probe_type):
+        if probe and probe.http_get is None:
+            raise RuntimeError(
+                f"{probe_type} in container {container_name} does not use httpGet. "
+                f"Only HTTP-based probes are supported."
+            )
+
+    # Check for HTTP probes only
+    validate_http_probe(target_container.liveness_probe, "LivenessProbe")
+    validate_http_probe(target_container.readiness_probe, "ReadinessProbe")
+    validate_http_probe(target_container.startup_probe, "StartupProbe")
+
     for name in pod_names:
         check_pod_valid_for_bridge(config, name, namespace, container_name)
 

operator/gefyra/bridge/carrier/__init__.py

@@ -1,6 +1,6 @@
 import json
 from typing import Any, Dict, List, Optional
-from gefyra.utils import exec_command_pod
+from gefyra.utils import BridgeException, exec_command_pod
 import kubernetes as k8s
 
 from gefyra.bridge.abstract import AbstractGefyraBridgeProvider
@@ -33,7 +33,10 @@ def __init__(
 
     def install(self, parameters: Optional[Dict[Any, Any]] = None):
         parameters = parameters or {}
-        self._patch_pod_with_carrier(handle_probes=parameters.get("handleProbes", True))
+        try:
+            self._patch_pod_with_carrier(handle_probes=parameters.get("handleProbes", True))
+        except BridgeException as be:
+            raise BridgeException from be
 
     def _ensure_probes(self, container: k8s.client.V1Container) -> bool:
         probes = self._get_all_probes(container)
@@ -143,7 +146,7 @@ def _patch_pod_with_carrier(
                             "Not all of the probes to be handled are currently"
                             " supported by Gefyra"
                         )
-                        return False, pod
+                        raise BridgeException()
                 if (
                     container.image
                     == f"{self.configuration.CARRIER_IMAGE}:{self.configuration.CARRIER_IMAGE_TAG}"

operator/gefyra/bridgestate.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, UTC
 from typing import Any, Optional
 from gefyra.bridge.abstract import AbstractGefyraBridgeProvider
 from gefyra.bridge.factory import BridgeProviderType, bridge_provider_factory
@@ -7,10 +7,11 @@
 import kubernetes as k8s
 from statemachine import State, StateMachine
 
-
 from gefyra.base import GefyraStateObject, StateControllerMixin
 from gefyra.configuration import OperatorConfiguration
 
+from gefyra.utils import BridgeException
+
 
 class GefyraBridgeObject(GefyraStateObject):
     plural = "gefyrabridges"
@@ -37,7 +38,7 @@ class GefyraBridge(StateMachine, StateControllerMixin):
 
     install = (
         requested.to(installing, on="_install_provider")
-        | error.to(installing)
+        | installing.to(error)
         | installing.to.itself(on="_wait_for_provider")
     )
     set_installed = (
@@ -106,7 +107,7 @@ def sunset(self) -> Optional[datetime]:
 
     @property
     def should_terminate(self) -> bool:
-        if self.sunset and self.sunset <= datetime.utcnow():
+        if self.sunset and self.sunset <= datetime.now(UTC):
             # remove this bridge because the sunset time is in the past
             self.logger.warning(
                 f"Bridge '{self.object_name}' should be terminated "
@@ -121,7 +122,10 @@ def _install_provider(self):
         It installs the bridge provider
         :return: Nothing
         """
-        self.bridge_provider.install()
+        try:
+            self.bridge_provider.install()
+        except BridgeException:
+            self.send("impair")
 
     def _wait_for_provider(self):
         if not self.bridge_provider.ready():

operator/gefyra/utils.py

@@ -13,6 +13,10 @@
 logger = logging.getLogger("gefyra.utils")
 
 
+class BridgeException(Exception):
+    pass
+
+
 def get_label_selector(labels: dict[str, str]) -> str:
     return ",".join(["{0}={1}".format(*label) for label in list(labels.items())])
 

operator/poetry.lock

@@ -10,6 +10,7 @@ kopf = "^1.37.3"
 kubernetes = "^31.0.0"
 python-decouple = "^3.8"
 python-statemachine = "^2.4.0"
+pydot = "^3.0.3"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4"

operator/pyproject.toml

@@ -1,5 +1,7 @@
 import json
 import logging
+import subprocess
+import pytest
 from pytest_kubernetes.providers import AClusterManager
 from .utils import GefyraDockerClient
 
@@ -101,7 +103,6 @@ def test_a_bridge(
 
     gclient_a.delete()
 
-
 def test_b_cleanup_bridges_routes(
     carrier_image,
     operator: AClusterManager,
@@ -132,3 +133,44 @@ def test_b_cleanup_bridges_routes(
         namespace="gefyra",
         timeout=60,
     )
+
+def test_c_fail_create_not_supported_bridges(
+    demo_backend_image,
+    demo_frontend_image,
+    carrier_image,
+    operator: AClusterManager
+):
+    k3d = operator
+    k3d.load_image(demo_backend_image)
+    k3d.load_image(demo_frontend_image)
+    k3d.load_image(carrier_image)
+
+    k3d.kubectl(["create", "namespace", "demo-failing"])
+    k3d.wait("ns/demo-failing", "jsonpath='{.status.phase}'=Active")
+    k3d.apply("tests/fixtures/demo_pods_not_supported.yaml")
+    k3d.wait(
+        "pod/backend",
+        "condition=ready",
+        namespace="demo-failing",
+        timeout=60,
+    )
+
+    k3d.apply("tests/fixtures/a_gefyra_bridge_failing.yaml")
+    # bridge should be in error state
+    k3d.wait(
+        "gefyrabridges.gefyra.dev/bridge-a",
+        "jsonpath=.state=ERROR",
+        namespace="gefyra",
+        timeout=20,
+    )
+
+    # applying the bridge shouldn't have worked
+    with pytest.raises(subprocess.TimeoutExpired):
+        k3d.wait(
+            "pod/frontend",
+            "jsonpath=.status.containerStatuses[0].image=docker.io/library/"
+            + carrier_image,
+            namespace="demo-failing",
+            timeout=60,
+        )
+    

operator/tests/e2e/test_create_bridge.py

@@ -0,0 +1,14 @@
+apiVersion: gefyra.dev/v1
+kind: gefyrabridge
+metadata:
+  name: bridge-a
+  namespace: gefyra
+provider: carrier
+connectionProvider: stowaway
+client: client-a
+targetNamespace: demo-failing
+targetPod: frontend
+targetContainer: frontend
+portMappings:
+  - "8080:80"
+destinationIP: "192.168.101.1"

operator/tests/fixtures/a_gefyra_bridge_failing.yaml

@@ -0,0 +1,81 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: backend
+  namespace: demo-failing
+  labels:
+    app: backend
+spec:
+  securityContext:
+    runAsUser: 1000
+    runAsGroup: 1000
+    fsGroup: 1000
+  containers:
+    - name: backend
+      image: quay.io/gefyra/gefyra-demo-backend
+      imagePullPolicy: IfNotPresent
+      ports:
+        - name: web
+          containerPort: 5002
+          protocol: TCP
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: frontend
+  namespace: demo-failing
+  labels:
+    app: frontend
+spec:
+  containers:
+    - name: frontend
+      image: quay.io/gefyra/gefyra-demo-frontend
+      imagePullPolicy: IfNotPresent
+      ports:
+        - name: web
+          containerPort: 5003
+          protocol: TCP
+      env:
+        - name: SVC_URL
+          value: "backend.demo.svc.cluster.local:5002"
+      livenessProbe:
+        exec:
+          command:
+            - cat
+            - /tmp/healthy
+        initialDelaySeconds: 30
+        periodSeconds: 10
+      readinessProbe:
+        exec:
+          command:
+            - cat
+            - /tmp/ready
+        initialDelaySeconds: 5
+        periodSeconds: 5
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: backend
+  namespace: demo-failing
+spec:
+  selector:
+    app: backend
+  ports:
+  - protocol: TCP
+    port: 5002
+    targetPort: 5002
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: frontend
+  namespace: demo-failing
+spec:
+  selector:
+    app: frontend
+  ports:
+  - protocol: "TCP"
+    port: 80
+    targetPort: 5003
+  type: LoadBalancer