Warn in jobby list if there are failed pods to a Kueue workload

backend/src/jobs_server/models.py

@@ -111,6 +111,7 @@ class WorkloadMetadata(BaseModel):
     termination_timestamp: datetime.datetime | None = None
     was_evicted: bool = False
     was_inadmissible: bool = False
+    has_failed_pods: bool = False
 
     @classmethod
     def from_kueue_workload(cls, workload: KueueWorkload) -> Self:
@@ -126,6 +127,7 @@ def from_kueue_workload(cls, workload: KueueWorkload) -> Self:
             termination_timestamp=workload.termination_timestamp,
             was_evicted=workload.was_evicted,
             was_inadmissible=workload.was_inadmissible,
+            has_failed_pods=workload.has_failed_pods,
         )
 
 

backend/src/jobs_server/services/k8s.py

@@ -122,6 +122,9 @@ def list_workloads(self, namespace: str | None = None) -> list[KueueWorkload]:
             namespace=namespace or self.namespace,
             plural="workloads",
         )
+        self._core_v1_api.list_namespaced_pod(
+            namespace=namespace or self.namespace,
+        )
         return [
             KueueWorkload.model_validate(workload)
             for workload in workloads.get("items", [])

backend/src/jobs_server/utils/kueue.py

@@ -247,6 +247,10 @@ def pods(self) -> list[client.V1Pod]:
         ).items
         return pods
 
+    @property
+    def has_failed_pods(self) -> bool:
+        return any(p.status.phase == "Failed" for p in self.pods)
+
     def stop(self, k8s: "KubernetesService") -> None:
         if not self.managed_resource:
             raise RuntimeError(

backend/tests/e2e/test_jobs.py

@@ -67,6 +67,9 @@ def test_job_lifecycle(
         assert str(status.managed_resource_id) == managed_resource_id.uid
         assert status.execution_status != JobStatus.FAILED
         assert status.kueue_status is not None and status.kueue_status.conditions != []
+        assert not status.was_evicted
+        assert not status.was_inadmissible
+        assert not status.has_failed_pods
 
         if status.execution_status != JobStatus.PENDING:
             break

backend/tests/integration/test_jobs.py

@@ -124,6 +124,9 @@ def test_success(
         mocker.patch.object(
             KueueWorkload, "for_managed_resource", return_value=workload
         )
+        # FIXME: This prevents the pod listing API call, which doesn't work in a integration
+        #  test scenario.
+        mocker.patch.object(KueueWorkload, "has_failed_pods", return_value=False)
 
         response = client.get(f"/jobs/{workload.metadata.uid}/status")
 
@@ -273,6 +276,10 @@ def test_list_jobs(
             return_value=[workload],
         )
 
+        # FIXME: This prevents the pod listing API call, which doesn't work in a integration
+        #  test scenario.
+        mocker.patch.object(KueueWorkload, "has_failed_pods", return_value=False)
+
         response = client.get("/jobs?include_metadata=true")
 
         assert response.is_success

client/src/cli/commands/list.py

@@ -30,6 +30,9 @@ def format_status(s: JobStatus) -> str:
     def status_flags(wl: openapi_client.WorkloadMetadata) -> str:
         if wl.was_evicted or wl.was_inadmissible:
             return "[bright_yellow] [!][/]"
+        # if the job is already failed, we don't really need to warn anymore.
+        elif wl.has_failed_pods and wl.execution_status != JobStatus.FAILED:
+            return "[bright_red] [!][/]"
         else:
             return ""
 

client/src/openapi_client/models/workload_metadata.py

@@ -39,6 +39,7 @@ class WorkloadMetadata(BaseModel):
     termination_timestamp: datetime | None = None
     was_evicted: StrictBool | None = False
     was_inadmissible: StrictBool | None = False
+    has_failed_pods: StrictBool | None = False
     __properties: ClassVar[list[str]] = [
         "managed_resource_id",
         "execution_status",
@@ -49,6 +50,7 @@ class WorkloadMetadata(BaseModel):
         "termination_timestamp",
         "was_evicted",
         "was_inadmissible",
+        "has_failed_pods",
     ]
 
     model_config = ConfigDict(
@@ -139,5 +141,8 @@ def from_dict(cls, obj: dict[str, Any] | None) -> Self | None:
             "was_inadmissible": obj.get("was_inadmissible")
             if obj.get("was_inadmissible") is not None
             else False,
+            "has_failed_pods": obj.get("has_failed_pods")
+            if obj.get("has_failed_pods") is not None
+            else False,
         })
         return _obj