Reject deleted or scaled-to-zero workloads and unschedulable pods in mitigation eval

aiopslab/generators/fault/inject_otel.py

@@ -86,6 +86,14 @@ def recover_fault(self, feature_flag: str):
         )
         print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
+    def get_flag_variant(self, feature_flag: str) -> str:
+        """Return the default variant the feature flag is currently set to."""
+        configmap = self.kubectl.core_v1_api.read_namespaced_config_map(
+            self.configmap_name, self.namespace
+        )
+        flagd_data = json.loads(configmap.data["demo.flagd.json"])
+        return flagd_data["flags"][feature_flag]["defaultVariant"]
+
 
 # Example usage:
 # if __name__ == "__main__":

aiopslab/orchestrator/orchestrator.py

@@ -8,6 +8,7 @@
 from aiopslab.session import Session
 from aiopslab.orchestrator.problems.registry import ProblemRegistry
 from aiopslab.orchestrator.parser import ResponseParser
+from aiopslab.orchestrator.tasks.mitigation import MitigationTask
 from aiopslab.utils.status import *
 from aiopslab.utils.critical_section import CriticalSection
 from aiopslab.service.telemetry.prometheus import Prometheus
@@ -71,6 +72,11 @@ def init_problem(self, problem_id: str):
         prob.app.delete()
         prob.app.deploy()
 
+        # record the deployed workloads before the fault changes them, so that
+        # mitigation eval can tell a removed workload from a recovered one
+        if isinstance(prob, MitigationTask):
+            prob.record_deployed_workloads()
+
         # make sure is_fault_injected is correct to apply appropriate
         # function with atexit to recover fault
         with CriticalSection():

aiopslab/orchestrator/problems/kafka_queue_problems/kafka_queue_problems.py

@@ -107,6 +107,17 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
         print("== Evaluation ==")
         super().eval(soln, trace, duration)
 
+        # custom: the fault is a feature flag that leaves every pod healthy, so
+        # check that the flag is off again
+        try:
+            variant = self.injector.get_flag_variant("kafkaQueueProblems")
+        except Exception as e:
+            variant = f"unreadable ({e})"
+        if variant != "off":
+            print(f"Feature flag kafkaQueueProblems is {variant}, not off")
+            self.results["success"] = False
+            return self.results
+
         # Allow dependent workloads time to recover before judging instead of
         # taking a single snapshot at submit() time (see
         # MitigationTask.wait_until_pods_healthy).

aiopslab/orchestrator/problems/misconfig_app/misconfig_app_hotel_res.py

@@ -165,13 +165,8 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
         print("== Evaluation ==")
         super().eval(soln, trace, duration)
 
-        # Check if all services (not only faulty service) is back to normal (Running)
-        # Use wait_for_ready helper to poll for pod readiness
-        try:
-            self.kubectl.wait_for_ready(self.namespace, sleep=5, max_wait=60)
-            self.results["success"] = True
-        except Exception as e:
-            print(f"Pods are not all ready: {e}")
-            self.results["success"] = False
-
+        # Check if all services (not only faulty service) is back to normal (Running),
+        # allowing dependent workloads time to recover (see
+        # MitigationTask.wait_until_pods_healthy).
+        self.results["success"] = self.wait_until_pods_healthy(self.namespace)
         return self.results

aiopslab/orchestrator/problems/redeploy_without_pv/redeploy_without_pv.py

@@ -164,33 +164,8 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
         print("== Evaluation ==")
         super().eval(soln, trace, duration)
 
-        # Check if all services (not only faulty service) is back to normal (Running)
-        pod_list = self.kubectl.list_pods(self.namespace)
-        all_normal = True
-
-        for pod in pod_list.items:
-            # Check container statuses
-            for container_status in pod.status.container_statuses:
-                if (
-                    container_status.state.waiting
-                    and container_status.state.waiting.reason == "CrashLoopBackOff"
-                ):
-                    print(f"Container {container_status.name} is in CrashLoopBackOff")
-                    all_normal = False
-                elif (
-                    container_status.state.terminated
-                    and container_status.state.terminated.reason != "Completed"
-                ):
-                    print(
-                        f"Container {container_status.name} is terminated with reason: {container_status.state.terminated.reason}"
-                    )
-                    all_normal = False
-                elif not container_status.ready:
-                    print(f"Container {container_status.name} is not ready")
-                    all_normal = False
-
-            if not all_normal:
-                break
-
-        self.results["success"] = all_normal
+        # Check if all services (not only faulty service) is back to normal (Running),
+        # allowing dependent workloads time to recover (see
+        # MitigationTask.wait_until_pods_healthy).
+        self.results["success"] = self.wait_until_pods_healthy(self.namespace)
         return self.results

aiopslab/orchestrator/problems/scale_pod/scale_pod_social_net.py

@@ -177,7 +177,10 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
         print("== Evaluation ==")
         super().eval(soln, trace, duration)
 
-        all_normal = True
+        # Check if all services are running normally, allowing dependent
+        # workloads time to recover (see MitigationTask.wait_until_pods_healthy),
+        # then that the faulty service is back to one available replica.
+        all_normal = self.wait_until_pods_healthy(self.namespace)
 
         deployment = self.kubectl.get_deployment(self.faulty_service, self.namespace)
 
@@ -194,30 +197,5 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
                 )
                 all_normal = False
 
-        # Check if all services are running normally
-        pod_list = self.kubectl.list_pods(self.namespace)
-        for pod in pod_list.items:
-            for container_status in pod.status.container_statuses:
-                if (
-                    container_status.state.waiting
-                    and container_status.state.waiting.reason == "CrashLoopBackOff"
-                ):
-                    print(f"Container {container_status.name} is in CrashLoopBackOff")
-                    all_normal = False
-                elif (
-                    container_status.state.terminated
-                    and container_status.state.terminated.reason != "Completed"
-                ):
-                    print(
-                        f"Container {container_status.name} is terminated with reason: {container_status.state.terminated.reason}"
-                    )
-                    all_normal = False
-                elif not container_status.ready:
-                    print(f"Container {container_status.name} is not ready")
-                    all_normal = False
-
-            if not all_normal:
-                break
-
         self.results["success"] = all_normal
         return self.results

aiopslab/orchestrator/problems/wrong_bin_usage/wrong_bin_usage.py

@@ -163,29 +163,13 @@ def eval(self, soln: Any, trace: list[SessionItem], duration: float) -> dict:
         print("== Evaluation ==")
         super().eval(soln, trace, duration)
 
-        # Check if all services (not only faulty service) is back to normal (Running)
-        pod_list = self.kubectl.list_pods(self.namespace)
-        all_normal = True
-
-        for pod in pod_list.items:
-            # Check container statuses
-            for container_status in pod.status.container_statuses:
-                if container_status.state.waiting:
-                    reason = container_status.state.waiting.reason
-                    if reason in ["CrashLoopBackOff", "Error", "ImagePullBackOff", "ErrImagePull"]:
-                        print(f"Container {container_status.name} is in error state: {reason}")
-                        all_normal = False
-                elif container_status.state.terminated and container_status.state.terminated.reason != "Completed":
-                    print(f"Container {container_status.name} is terminated with reason: {container_status.state.terminated.reason}")
-                    all_normal = False
-                elif not container_status.ready:
-                    print(f"Container {container_status.name} is not ready")
-                    all_normal = False
-
-            if not all_normal:
-                print("Pods are not all in a good state.")
-                self.results["success"] = False
-                return self.results
+        # Check if all services (not only faulty service) is back to normal (Running),
+        # allowing dependent workloads time to recover (see
+        # MitigationTask.wait_until_pods_healthy).
+        if not self.wait_until_pods_healthy(self.namespace):
+            print("Pods are not all in a good state.")
+            self.results["success"] = False
+            return self.results
 
         # Check if the deployment was updated to use the right binary
         expected_command = "profile" # Command dictates which binary will be ran, we want to run /go/bin/profile and not /go/bin/geo

aiopslab/orchestrator/tasks/mitigation.py

@@ -93,12 +93,20 @@ def pods_unhealthy_reasons(self, namespace: str) -> list[str]:
         """Return human-readable reasons for any unhealthy container in `namespace`.
 
         An empty list means every container is healthy (no CrashLoopBackOff,
-        no abnormal termination, all ready).
+        no abnormal termination, all ready). A pod with no container statuses
+        has no running container (e.g. it is Pending because it cannot be
+        scheduled), so it is unhealthy unless it has already succeeded; this
+        matches ``KubeCtl._pod_is_ready_or_succeeded``.
         """
         pod_list = self.kubectl.list_pods(namespace)
         reasons: list[str] = []
         for pod in pod_list.items:
             if not pod.status.container_statuses:
+                if pod.status.phase != "Succeeded":
+                    reasons.append(
+                        f"Pod {pod.metadata.name} has no running containers "
+                        f"(phase: {pod.status.phase})"
+                    )
                 continue
             for container_status in pod.status.container_statuses:
                 waiting = container_status.state.waiting
@@ -116,6 +124,44 @@ def pods_unhealthy_reasons(self, namespace: str) -> list[str]:
                     reasons.append(f"Container {container_status.name} is not ready")
         return reasons
 
+    def _workload_replicas(self, namespace: str) -> dict[str, int]:
+        workloads = {}
+        for kind, list_workloads in (
+            ("Deployment", self.kubectl.list_deployments),
+            ("StatefulSet", self.kubectl.list_stateful_sets),
+        ):
+            for workload in list_workloads(namespace).items:
+                workloads[f"{kind} {workload.metadata.name}"] = workload.spec.replicas or 0
+        return workloads
+
+    def record_deployed_workloads(self):
+        """Record the Deployments and StatefulSets the application deployed.
+
+        The orchestrator calls this after deploying the application and before
+        injecting the fault. Deleting a workload or scaling it to zero removes
+        all of its pods, which the pod checks cannot see, so eval also checks
+        that every recorded workload still exists with at least one replica.
+        """
+        self.deployed_workloads = {
+            name: replicas
+            for name, replicas in self._workload_replicas(self.app.namespace).items()
+            if replicas > 0
+        }
+
+    def workloads_unhealthy_reasons(self, namespace: str) -> list[str]:
+        """Return reasons for any recorded workload that was deleted or scaled to zero."""
+        deployed = getattr(self, "deployed_workloads", None)
+        if not deployed:
+            return []
+        current = self._workload_replicas(namespace)
+        reasons: list[str] = []
+        for name, replicas in sorted(deployed.items()):
+            if name not in current:
+                reasons.append(f"{name} (deployed with {replicas} replicas) no longer exists")
+            elif current[name] == 0:
+                reasons.append(f"{name} (deployed with {replicas} replicas) is scaled to zero")
+        return reasons
+
     def wait_until_pods_healthy(
         self,
         namespace: str,
@@ -124,6 +170,8 @@ def wait_until_pods_healthy(
     ) -> bool:
         """Poll until all pods in `namespace` are healthy, or `timeout` elapses.
 
+        Healthy also requires every workload recorded by
+        ``record_deployed_workloads`` to still exist with at least one replica.
         Returns True as soon as the namespace is healthy. On timeout, prints the
         outstanding reasons (preserving the previous eval's diagnostic output)
         and returns False. The window is configurable via the
@@ -134,7 +182,9 @@ def wait_until_pods_healthy(
             timeout = float(os.getenv("AIOPSLAB_MITIGATION_SETTLE_SECONDS", "120"))
         deadline = time.monotonic() + max(timeout, 0.0)
         while True:
-            reasons = self.pods_unhealthy_reasons(namespace)
+            reasons = self.pods_unhealthy_reasons(namespace) + self.workloads_unhealthy_reasons(
+                namespace
+            )
             if not reasons:
                 return True
             if time.monotonic() >= deadline:

aiopslab/service/kubectl.py

@@ -49,6 +49,14 @@ def list_services(self, namespace):
         """Return a list of all services within a specified namespace."""
         return self.core_v1_api.list_namespaced_service(namespace)
 
+    def list_deployments(self, namespace):
+        """Return a list of all deployments within a specified namespace."""
+        return self.apps_v1_api.list_namespaced_deployment(namespace)
+
+    def list_stateful_sets(self, namespace):
+        """Return a list of all stateful sets within a specified namespace."""
+        return self.apps_v1_api.list_namespaced_stateful_set(namespace)
+
     def get_cluster_ip(self, service_name, namespace):
         """Retrieve the cluster IP address of a specified service within a namespace."""
         service_info = self.core_v1_api.read_namespaced_service(service_name, namespace)