From 87cc786f46510ec12aa814ffa15179c7ec6a5593 Mon Sep 17 00:00:00 2001 From: Vanshika Date: Wed, 5 Aug 2026 15:48:43 +0530 Subject: [PATCH] feat: surface generated workload failures in ModelBooster/ModelServing status ModelServing's Progressing/UpdateInProgress conditions only ever carried a generic "some groups are progressing" message, whether a ServingGroup was still starting up normally or permanently stuck (bad image, crash-looping container, unschedulable Pod). Pod-level failure signals were read only coarsely (Phase==Failed or restart count>0) to drive the existing delete-and-recreate recovery loop, and the diagnostic detail was discarded rather than surfaced anywhere in status. Extract actionable failure detail from live Pod status (scheduler PodScheduled condition, container/init-container Waiting/Terminated reasons) using only structured, stable fields kubelet/the scheduler themselves populate, never Kubernetes Event text. Record it per-Role in the model-serving-controller's in-memory store, clearing it once the Role's Pod is ready again so recovered failures don't linger. When a ServingGroup is progressing because of a recorded failure, replace the condition's generic Reason with the specific one and append the detail to Message; reuse the existing Progressing/UpdateInProgress condition types rather than adding a new one. ModelBooster's Active condition already toggles False while its generated ModelServing isn't Available, but the message was always the static "ModelBooster not ready yet". When the blocking ModelServing condition's Reason is one of these new actionable ones (not the generic startup/rollout reasons), propagate it onto ModelBooster's Active condition too, so the failure is visible without inspecting the ModelServing separately. This also fixes SetCondition silently skipping Reason/Message-only updates when a condition's Status doesn't change, which would otherwise have kept the enriched message from ever reaching the API for an already-True Progressing condition. Signed-off-by: Vanshika --- .../controller/condition.go | 29 +++ .../controller/model_booster_controller.go | 39 +++- .../model_booster_controller_test.go | 157 ++++++++++++++ .../controller/model_serving_controller.go | 70 +++++- .../model_serving_controller_test.go | 196 +++++++++++++++++ .../datastore/store.go | 44 ++++ .../datastore/store_test.go | 48 +++++ pkg/model-serving-controller/utils/utils.go | 149 ++++++++++++- .../utils/utils_test.go | 203 +++++++++++++++++- 9 files changed, 921 insertions(+), 14 deletions(-) diff --git a/pkg/model-booster-controller/controller/condition.go b/pkg/model-booster-controller/controller/condition.go index ee5dc3679..6b8de9d82 100644 --- a/pkg/model-booster-controller/controller/condition.go +++ b/pkg/model-booster-controller/controller/condition.go @@ -32,6 +32,17 @@ const ( ModelFailedReason = "ModelAbnormal" ) +// genericModelServingProgressReasons are the Reason values the model-serving-controller uses +// on ModelServing's Progressing/UpdateInProgress conditions for ordinary startup/rollout +// progress (see pkg/model-serving-controller/utils/utils.go's newCondition). Any other Reason +// on those condition types means model-serving-controller identified a specific, actionable +// Pod-level failure (e.g. scheduling, image-pull, or crash-loop) that's worth surfacing here +// instead of the generic "ModelBooster not ready yet" message. +var genericModelServingProgressReasons = map[string]bool{ + "GroupProgressing": true, + "GroupsUpdating": true, +} + // setModelInitCondition sets model condition to initialized func (mc *ModelBoosterController) setModelInitCondition(ctx context.Context, model *workloadv1alpha1.ModelBooster) error { meta.SetStatusCondition(&model.Status.Conditions, newCondition(string(workloadv1alpha1.ModelStatusConditionTypeInitialized), @@ -63,6 +74,24 @@ func (mc *ModelBoosterController) setModelFailedCondition(ctx context.Context, m } } +// setModelWorkloadDegradedCondition sets ModelBooster's Active condition to False using the +// Reason/Message propagated from the generated ModelServing's own condition, so a Pod-level +// failure (scheduling, image pull, crash loop, etc.) is visible on ModelBooster without +// inspecting the ModelServing separately. It reuses the existing Active condition rather than +// introducing a new one: the next reconcile naturally supersedes it, either with +// setModelActiveCondition (True, once the ModelServing recovers) or with a fresh call to this +// function (updated detail) or setModelProcessingCondition's generic message (once the +// ModelServing's condition Reason is no longer an actionable failure) — so it never goes stale. +func (mc *ModelBoosterController) setModelWorkloadDegradedCondition(ctx context.Context, model *workloadv1alpha1.ModelBooster, reason, message string) error { + meta.SetStatusCondition(&model.Status.Conditions, newCondition(string(workloadv1alpha1.ModelStatusConditionTypeActive), + metav1.ConditionFalse, reason, message)) + if err := mc.updateModelBoosterStatus(ctx, model); err != nil { + klog.Errorf("update ModelBooster status failed: %v", err) + return err + } + return nil +} + // setModelActiveCondition sets ModelBooster conditions to active func (mc *ModelBoosterController) setModelActiveCondition(ctx context.Context, model *workloadv1alpha1.ModelBooster) error { meta.SetStatusCondition(&model.Status.Conditions, newCondition(string(workloadv1alpha1.ModelStatusConditionTypeActive), diff --git a/pkg/model-booster-controller/controller/model_booster_controller.go b/pkg/model-booster-controller/controller/model_booster_controller.go index 9e0d9d956..495461ead 100644 --- a/pkg/model-booster-controller/controller/model_booster_controller.go +++ b/pkg/model-booster-controller/controller/model_booster_controller.go @@ -219,10 +219,23 @@ func (mc *ModelBoosterController) reconcile(ctx context.Context, namespaceAndNam mc.setModelFailedCondition(ctx, model, err) return err } - modelServingActive, err := mc.isModelServingActive(model) - if err != nil || !modelServingActive { + modelServingActive, blockingReason, blockingMessage, err := mc.isModelServingActive(model) + if err != nil { return err } + if !modelServingActive { + // If the generated ModelServing's own condition already identifies an actionable, + // non-generic cause (e.g. a Pod scheduling/image-pull/crash failure surfaced by + // model-serving-controller), propagate it here so users don't have to inspect the + // ModelServing separately. Otherwise leave the generic "still initializing" Active + // condition set by setModelProcessingCondition above untouched. + if blockingReason != "" && !genericModelServingProgressReasons[blockingReason] { + if err := mc.setModelWorkloadDegradedCondition(ctx, model, blockingReason, blockingMessage); err != nil { + return err + } + } + return nil + } if err := mc.setModelActiveCondition(ctx, model); err != nil { return err } @@ -230,26 +243,36 @@ func (mc *ModelBoosterController) reconcile(ctx context.Context, namespaceAndNam return nil } -// isModelServingActive returns true if all ModelServings are available. -func (mc *ModelBoosterController) isModelServingActive(model *workload.ModelBooster) (bool, error) { +// isModelServingActive returns true if all ModelServings are available. When a ModelServing +// isn't available, it also returns the Reason/Message of that ModelServing's blocking +// Progressing or UpdateInProgress condition (if present), so the caller can decide whether it +// describes normal startup progress or an actionable failure worth surfacing on ModelBooster. +func (mc *ModelBoosterController) isModelServingActive(model *workload.ModelBooster) (active bool, blockingReason string, blockingMessage string, err error) { modelServings, err := mc.listModelServingsByLabel(model) if err != nil { - return false, err + return false, "", "", err } // Ensure exactly one ModelServing exists for the single backend if len(modelServings) != 1 { klog.Infof("Number of ModelServings: %d, expected: %d", len(modelServings), 1) - return false, fmt.Errorf("ModelServing number not equal to backend number") + return false, "", "", fmt.Errorf("ModelServing number not equal to backend number") } // Check if all ModelServings are available for _, modelServing := range modelServings { if !meta.IsStatusConditionPresentAndEqual(modelServing.Status.Conditions, string(workload.ModelServingAvailable), metav1.ConditionTrue) { // requeue until all ModelServings are active klog.InfoS("ModelServing is not available", "ModelServing", klog.KObj(modelServing)) - return false, nil + cond := meta.FindStatusCondition(modelServing.Status.Conditions, string(workload.ModelServingProgressing)) + if cond == nil { + cond = meta.FindStatusCondition(modelServing.Status.Conditions, string(workload.ModelServingUpdateInProgress)) + } + if cond != nil { + return false, cond.Reason, cond.Message, nil + } + return false, "", "", nil } } - return true, nil + return true, "", "", nil } // updateModelBoosterStatus updates model status. diff --git a/pkg/model-booster-controller/controller/model_booster_controller_test.go b/pkg/model-booster-controller/controller/model_booster_controller_test.go index f58f8ee84..b9c2f58a1 100644 --- a/pkg/model-booster-controller/controller/model_booster_controller_test.go +++ b/pkg/model-booster-controller/controller/model_booster_controller_test.go @@ -24,6 +24,7 @@ import ( "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" kthenafake "github.com/volcano-sh/kthena/client-go/clientset/versioned/fake" workload "github.com/volcano-sh/kthena/pkg/apis/workload/v1alpha1" "github.com/volcano-sh/kthena/pkg/model-booster-controller/convert" @@ -259,6 +260,162 @@ func TestReconcile_ReturnsError(t *testing.T) { }) } +// TestIsModelServingActivePropagatesBlockingCondition verifies that when the generated +// ModelServing isn't Available, isModelServingActive surfaces its blocking condition's +// Reason/Message only when that Reason is a specific, actionable one (as opposed to the +// generic "still starting up" reasons model-serving-controller uses for ordinary progress), +// so callers can distinguish "still starting" from "stuck due to a Pod-level failure". +func TestIsModelServingActivePropagatesBlockingCondition(t *testing.T) { + tests := []struct { + name string + conditions []metav1.Condition + expectActive bool + expectReason string + expectMessage string + }{ + { + name: "available", + conditions: []metav1.Condition{{Type: string(workload.ModelServingAvailable), Status: metav1.ConditionTrue, Reason: "AllGroupsReady"}}, + expectActive: true, + }, + { + name: "generic progressing reason is not treated as a failure", + conditions: []metav1.Condition{{ + Type: string(workload.ModelServingProgressing), Status: metav1.ConditionTrue, + Reason: "GroupProgressing", Message: "Some groups is progressing: [0]", + }}, + expectActive: false, + expectReason: "GroupProgressing", + }, + { + name: "pod-level failure reason is propagated", + conditions: []metav1.Condition{{ + Type: string(workload.ModelServingProgressing), Status: metav1.ConditionTrue, + Reason: "ImagePullBackOff", + Message: "Some groups is progressing: [0]; pod p-0 init container downloader: back-off pulling image", + }}, + expectActive: false, + expectReason: "ImagePullBackOff", + expectMessage: "downloader", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + kubeClient := fake.NewClientset() + kthenaClient := kthenafake.NewSimpleClientset() + controller := NewModelBoosterController(kubeClient, kthenaClient) + + model := &workload.ModelBooster{ + ObjectMeta: metav1.ObjectMeta{Name: "m1", Namespace: "default", UID: "model-uid-1"}, + } + modelServing := &workload.ModelServing{ + ObjectMeta: metav1.ObjectMeta{ + Name: "ms1", + Namespace: "default", + Labels: map[string]string{utils.OwnerUIDKey: string(model.UID)}, + }, + Status: workload.ModelServingStatus{Conditions: tt.conditions}, + } + assert.NoError(t, controller.modelServingInformer.GetIndexer().Add(modelServing)) + + active, reason, message, err := controller.isModelServingActive(model) + assert.NoError(t, err) + assert.Equal(t, tt.expectActive, active) + assert.Equal(t, tt.expectReason, reason) + if tt.expectMessage != "" { + assert.Contains(t, message, tt.expectMessage) + } + }) + } +} + +// TestReconcileSurfacesModelServingPodFailure verifies the end-to-end propagation: when +// reconcile finds the ModelServing blocked on an actionable (non-generic) Progressing reason, +// ModelBooster's Active condition is set to False with that same Reason/Message instead of the +// generic "ModelBooster not ready yet" message -- and that once the ModelServing recovers to +// Available, the next reconcile clears it back to Active=True (the condition never sticks). +func TestReconcileSurfacesModelServingPodFailure(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + kubeClient := fake.NewClientset() + kthenaClient := kthenafake.NewSimpleClientset() + controller := NewModelBoosterController(kubeClient, kthenaClient) + go controller.Run(ctx, 1) + assert.True(t, waitForControllerCacheSync(controller)) + + model := loadYaml[workload.ModelBooster](t, "../convert/testdata/input/model.yaml") + _, err := kthenaClient.WorkloadV1alpha1().ModelBoosters(model.Namespace).Create(ctx, model, metav1.CreateOptions{}) + assert.NoError(t, err) + + // Let the controller create the real ModelServing itself (correct name/labels/revision), + // matching how it would exist in a real cluster. + var modelServingName string + assert.True(t, waitForCondition(func() bool { + list, err := kthenaClient.WorkloadV1alpha1().ModelServings(model.Namespace).List(ctx, metav1.ListOptions{}) + if err != nil || len(list.Items) != 1 { + return false + } + modelServingName = list.Items[0].Name + return true + })) + + // Simulate model-serving-controller surfacing a Pod-level failure: patch only the Status + // subresource so the Spec/labels the running controller compares against on future passes + // are untouched (i.e. it will see nothing to update and leave this Status alone). + msToUpdate, err := kthenaClient.WorkloadV1alpha1().ModelServings(model.Namespace).Get(ctx, modelServingName, metav1.GetOptions{}) + assert.NoError(t, err) + msToUpdate.Status.Conditions = []metav1.Condition{{ + Type: string(workload.ModelServingProgressing), Status: metav1.ConditionTrue, + Reason: "CrashLoopBackOff", + Message: "Some groups is progressing: [0]; pod p-0 container engine: back-off restarting failed container", + }} + _, err = kthenaClient.WorkloadV1alpha1().ModelServings(model.Namespace).UpdateStatus(ctx, msToUpdate, metav1.UpdateOptions{}) + assert.NoError(t, err) + + // The ModelServing status update triggers the controller (via triggerModel) to reconcile the + // ModelBooster; wait for the propagated Reason/Message to land on its Active condition. + assert.True(t, waitForCondition(func() bool { + get, err := kthenaClient.WorkloadV1alpha1().ModelBoosters(model.Namespace).Get(ctx, model.Name, metav1.GetOptions{}) + if err != nil { + return false + } + cond := meta.FindStatusCondition(get.Status.Conditions, string(workload.ModelStatusConditionTypeActive)) + return cond != nil && cond.Reason == "CrashLoopBackOff" + }), "ModelBooster's Active condition should surface the ModelServing's Pod failure reason") + + get, err := kthenaClient.WorkloadV1alpha1().ModelBoosters(model.Namespace).Get(ctx, model.Name, metav1.GetOptions{}) + assert.NoError(t, err) + activeCond := meta.FindStatusCondition(get.Status.Conditions, string(workload.ModelStatusConditionTypeActive)) + require.NotNil(t, activeCond) + assert.Equal(t, metav1.ConditionFalse, activeCond.Status) + assert.Contains(t, activeCond.Message, "back-off restarting") + + // The ModelServing recovers: the controller must clear the propagated failure, not leave it + // stale on ModelBooster. + msToUpdate, err = kthenaClient.WorkloadV1alpha1().ModelServings(model.Namespace).Get(ctx, modelServingName, metav1.GetOptions{}) + assert.NoError(t, err) + msToUpdate.Status.Conditions = []metav1.Condition{{Type: string(workload.ModelServingAvailable), Status: metav1.ConditionTrue, Reason: "AllGroupsReady"}} + _, err = kthenaClient.WorkloadV1alpha1().ModelServings(model.Namespace).UpdateStatus(ctx, msToUpdate, metav1.UpdateOptions{}) + assert.NoError(t, err) + + assert.True(t, waitForCondition(func() bool { + get, err := kthenaClient.WorkloadV1alpha1().ModelBoosters(model.Namespace).Get(ctx, model.Name, metav1.GetOptions{}) + if err != nil { + return false + } + cond := meta.FindStatusCondition(get.Status.Conditions, string(workload.ModelStatusConditionTypeActive)) + return cond != nil && cond.Status == metav1.ConditionTrue + }), "the stale failure reason must not survive recovery") + + get, err = kthenaClient.WorkloadV1alpha1().ModelBoosters(model.Namespace).Get(ctx, model.Name, metav1.GetOptions{}) + assert.NoError(t, err) + activeCond = meta.FindStatusCondition(get.Status.Conditions, string(workload.ModelStatusConditionTypeActive)) + require.NotNil(t, activeCond) + assert.Equal(t, ModelActiveReason, activeCond.Reason) +} + func TestCreateModel(t *testing.T) { kubeClient := fake.NewClientset() kthenaClient := kthenafake.NewClientset() diff --git a/pkg/model-serving-controller/controller/model_serving_controller.go b/pkg/model-serving-controller/controller/model_serving_controller.go index aebfac3c2..ea3db73b8 100644 --- a/pkg/model-serving-controller/controller/model_serving_controller.go +++ b/pkg/model-serving-controller/controller/model_serving_controller.go @@ -367,6 +367,21 @@ func (c *ModelServingController) updatePod(_, newObj interface{}) { Name: ms.Name, }, servingGroupName, utils.ObjectRevision(newPod), roleTemplateHash, roleName, utils.GetRoleID(newPod)) } + + // Some actionable failures (e.g. a Pod that can't be scheduled at all) neither make the + // Pod Ready nor Failed nor restart a container, so they never reach the two cases above. + // They also should NOT trigger handleErrorPod's delete-and-recreate behavior: recreating + // a Pod that can't be scheduled would not help and would just add churn. Instead, just + // record/clear the failure detail for status-surfacing purposes. + reason, message := "", "" + if detail, ok := utils.ExtractPodFailureDetail(newPod); ok { + reason, message = detail.Reason, detail.Message + } + roleName := utils.GetRoleName(newPod) + roleID := utils.GetRoleID(newPod) + if c.store.SetRoleFailure(utils.GetNamespaceName(ms), servingGroupName, roleName, roleID, reason, message) { + c.enqueueModelServing(ms) + } } } @@ -1645,6 +1660,10 @@ func (c *ModelServingController) handleReadyPod(ms *workloadv1alpha1.ModelServin Name: ms.Name, }, servingGroupName, newPod.Name, utils.ObjectRevision(newPod), roleTemplateHash, roleName, roleID) + // The pod backing this role is healthy again: clear any previously recorded failure so a + // resolved problem (e.g. old failed Pod replaced by a healthy one) doesn't linger in status. + c.store.SetRoleFailure(utils.GetNamespaceName(ms), servingGroupName, roleName, roleID, "", "") + // Check and update role status to Running when all pods in the role are ready roleReady, err := c.checkRoleReady(ms, servingGroupName, roleName, roleID) if err != nil { @@ -1699,6 +1718,14 @@ func (c *ModelServingController) handleErrorPod(ms *workloadv1alpha1.ModelServin // Update role status back to Creating when pod fails roleName := utils.GetRoleName(errPod) roleID := utils.GetRoleID(errPod) + + // Record the actionable failure detail (if any) while the Pod object is still live, so it + // can be surfaced on the ModelServing's Progressing/UpdateInProgress condition. This is the + // only point where the raw Pod status is available before handlePodAfterGraceTime deletes it. + if detail, ok := utils.ExtractPodFailureDetail(errPod); ok { + c.store.SetRoleFailure(utils.GetNamespaceName(ms), servingGroupName, roleName, roleID, detail.Reason, detail.Message) + } + if roleStatus := c.store.GetRoleStatus(utils.GetNamespaceName(ms), servingGroupName, roleName, roleID); roleStatus == datastore.RoleRunning { err := c.store.UpdateRoleStatus(utils.GetNamespaceName(ms), servingGroupName, roleName, roleID, datastore.RoleCreating) klog.V(4).Infof("Setting role %s/%s status to Creating when pod fails", ms.GetName(), roleID) @@ -1796,6 +1823,39 @@ func (c *ModelServingController) handleDeletedPod(ms *workloadv1alpha1.ModelServ return nil } +// firstRoleFailure returns the most actionable Pod-level failure recorded for any Role in the +// given ServingGroup, if any. Roles are scanned in a deterministic (sorted) order so that, when +// multiple roles/pods are failing simultaneously, the choice of which one to surface is stable +// across calls instead of depending on Go's randomized map iteration order. +func (c *ModelServingController) firstRoleFailure(ms *workloadv1alpha1.ModelServing, groupName string) *utils.PodFailureDetail { + rolesByName, err := c.store.GetRolesByGroup(utils.GetNamespaceName(ms), groupName) + if err != nil { + return nil + } + + roleNames := make([]string, 0, len(rolesByName)) + for name := range rolesByName { + roleNames = append(roleNames, name) + } + slices.Sort(roleNames) + + for _, roleName := range roleNames { + roleIDs := make([]string, 0, len(rolesByName[roleName])) + for id := range rolesByName[roleName] { + roleIDs = append(roleIDs, id) + } + slices.Sort(roleIDs) + + for _, roleID := range roleIDs { + role := rolesByName[roleName][roleID] + if role != nil && role.FailureReason != "" { + return &utils.PodFailureDetail{Reason: role.FailureReason, Message: role.FailureMessage} + } + } + } + return nil +} + func (c *ModelServingController) checkServingGroupReady(ms *workloadv1alpha1.ModelServing, servingGroupName string) (bool, error) { // TODO: modify ServingGroupReady logic after rolling update functionality is implemented klog.V(4).Infof("checkServingGroupReady: modelServing=%s/%s, servingGroup=%s", ms.Namespace, ms.Name, servingGroupName) @@ -2130,6 +2190,11 @@ func (c *ModelServingController) UpdateModelServingStatus(ms *workloadv1alpha1.M available, updated, current := 0, 0, 0 progressingGroups, updatedGroups, currentGroups := []int{}, []int{}, []int{} + // The most actionable (lowest-ordinal) Pod-level failure among the progressing groups, if + // any. Groups are already iterated in ordinal order (GetServingGroupByModelServing sorts + // them), so keeping only the first one found gives a deterministic, low-noise result + // instead of arbitrarily picking among several simultaneously-failing groups. + var progressingFailure *utils.PodFailureDetail // Track revision counts to determine the most common non-updated revision (CurrentRevision) revisionCount := make(map[string]int) for index := range groups { @@ -2154,6 +2219,9 @@ func (c *ModelServingController) UpdateModelServingStatus(ms *workloadv1alpha1.M klog.V(2).Infof("Update servingGroup %s status to Running", groups[index].Name) } else { progressingGroups = append(progressingGroups, index) + if progressingFailure == nil { + progressingFailure = c.firstRoleFailure(latestMS, groups[index].Name) + } } if groups[index].Revision == revision { @@ -2168,7 +2236,7 @@ func (c *ModelServingController) UpdateModelServingStatus(ms *workloadv1alpha1.M } copy := latestMS.DeepCopy() - shouldUpdate := utils.SetCondition(copy, progressingGroups, updatedGroups, currentGroups) + shouldUpdate := utils.SetCondition(copy, progressingGroups, updatedGroups, currentGroups, progressingFailure) if copy.Status.Replicas != int32(len(groups)) || copy.Status.AvailableReplicas != int32(available) || copy.Status.UpdatedReplicas != int32(updated) || copy.Status.CurrentReplicas != int32(current) { shouldUpdate = true copy.Status.Replicas = int32(len(groups)) diff --git a/pkg/model-serving-controller/controller/model_serving_controller_test.go b/pkg/model-serving-controller/controller/model_serving_controller_test.go index f2b7bf4be..221dae036 100644 --- a/pkg/model-serving-controller/controller/model_serving_controller_test.go +++ b/pkg/model-serving-controller/controller/model_serving_controller_test.go @@ -5836,6 +5836,202 @@ func TestSyncAllWithMixedPods(t *testing.T) { assert.NotEmpty(t, servingGroups, "ServingGroups should exist in store") } +// newTestModelServingWithSinglePod builds a minimal single-role, single-replica ModelServing +// and a matching Pod (owned by it, with the labels the controller uses to map Pods back to +// their ServingGroup/Role) for use by the Pod-failure-surfacing tests below. +func newTestModelServingWithSinglePod(ns, msName, groupName, roleName, roleID, revision, podName string) (*workloadv1alpha1.ModelServing, *corev1.Pod) { + ms := &workloadv1alpha1.ModelServing{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: msName, + UID: "test-ms-uid-123", + }, + Spec: workloadv1alpha1.ModelServingSpec{ + Replicas: ptr.To[int32](1), + Template: workloadv1alpha1.ServingGroup{ + Roles: []workloadv1alpha1.Role{ + { + Name: roleName, + Replicas: ptr.To[int32](1), + EntryTemplate: workloadv1alpha1.PodTemplateSpec{ + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "main", Image: "nginx"}}}, + }, + }, + }, + }, + }, + } + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: podName, + Labels: map[string]string{ + workloadv1alpha1.ModelServingNameLabelKey: msName, + workloadv1alpha1.GroupNameLabelKey: groupName, + workloadv1alpha1.RoleLabelKey: roleName, + workloadv1alpha1.RoleIDKey: roleID, + workloadv1alpha1.RevisionLabelKey: revision, + }, + OwnerReferences: []metav1.OwnerReference{ + {APIVersion: workloadv1alpha1.GroupVersion.String(), Kind: "ModelServing", Name: msName, UID: ms.UID}, + }, + }, + } + return ms, pod +} + +// TestUpdatePodSchedulingFailureRecordedWithoutDeletion verifies that a Pod which can't be +// scheduled at all (PodScheduled condition False) has its failure surfaced into the datastore +// (for later propagation onto ModelServing's status) without going through the aggressive +// delete-and-recreate path used for Failed/crash-looping pods: recreating an unschedulable Pod +// would not fix the underlying scheduling problem and would just add churn. +func TestUpdatePodSchedulingFailureRecordedWithoutDeletion(t *testing.T) { + ns, msName, groupName, roleName, roleID, revision := "default", "test-ms", "test-ms-0", "prefill", "prefill-0", "hash123" + + kubeClient := kubefake.NewSimpleClientset() + kthenaClient := kthenafake.NewSimpleClientset() + volcanoClient := volcanofake.NewSimpleClientset() + apiextClient := apiextfake.NewSimpleClientset() + + controller, err := NewModelServingController(kubeClient, kthenaClient, volcanoClient, apiextClient) + assert.NoError(t, err) + + ms, pod := newTestModelServingWithSinglePod(ns, msName, groupName, roleName, roleID, revision, "test-pod-pending") + pod.Status = corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{ + {Type: corev1.PodScheduled, Status: corev1.ConditionFalse, Reason: "Unschedulable", Message: "0/3 nodes are available: insufficient cpu"}, + }, + } + + require.NoError(t, controller.modelServingsInformer.GetIndexer().Add(ms)) + + startActions := len(kubeClient.Actions()) + controller.addPod(pod) + + for _, action := range kubeClient.Actions()[startActions:] { + assert.False(t, action.Matches("delete", "pods"), "an unschedulable pod must not be deleted/recreated") + } + + roles, err := controller.store.GetRoleList(types.NamespacedName{Namespace: ns, Name: msName}, groupName, roleName) + assert.NoError(t, err) + require.Len(t, roles, 1) + assert.Equal(t, "Unschedulable", roles[0].FailureReason) + assert.Contains(t, roles[0].FailureMessage, "insufficient cpu") +} + +// TestUpdatePodCrashLoopFailureRecordedAndClearedOnRecovery verifies that a crash-looping Pod's +// failure is recorded in the datastore, and that once the (replacement) Pod for the same Role +// becomes ready, the previously recorded failure is cleared rather than lingering forever. +func TestUpdatePodCrashLoopFailureRecordedAndClearedOnRecovery(t *testing.T) { + ns, msName, groupName, roleName, roleID, revision := "default", "test-ms", "test-ms-0", "prefill", "prefill-0", "hash123" + + kubeClient := kubefake.NewSimpleClientset() + kthenaClient := kthenafake.NewSimpleClientset() + volcanoClient := volcanofake.NewSimpleClientset() + apiextClient := apiextfake.NewSimpleClientset() + + controller, err := NewModelServingController(kubeClient, kthenaClient, volcanoClient, apiextClient) + assert.NoError(t, err) + + ms, crashingPod := newTestModelServingWithSinglePod(ns, msName, groupName, roleName, roleID, revision, "test-pod-crash") + crashingPod.Status = corev1.PodStatus{ + Phase: corev1.PodRunning, + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "main", + RestartCount: 3, + State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{ + Reason: "CrashLoopBackOff", Message: "back-off restarting failed container", + }}, + }, + }, + } + + require.NoError(t, controller.modelServingsInformer.GetIndexer().Add(ms)) + _, err = kubeClient.CoreV1().Pods(ns).Create(context.Background(), crashingPod.DeepCopy(), metav1.CreateOptions{}) + require.NoError(t, err) + + // Prime the store as if the Pod had already gone through its normal Pending/Creating + // startup (as real Pod events would have done before it started crashing). + nsName := types.NamespacedName{Namespace: ns, Name: msName} + controller.store.AddServingGroupAndRole(nsName, groupName, revision, "role-hash", roleName, roleID) + + // The crashing pod is recorded as an error pod: this triggers the existing delete-and-recreate + // recovery flow (grace period, then deletion) in addition to recording the failure detail. + controller.addPod(crashingPod) + + roles, err := controller.store.GetRoleList(nsName, groupName, roleName) + assert.NoError(t, err) + require.Len(t, roles, 1) + assert.Equal(t, "CrashLoopBackOff", roles[0].FailureReason) + assert.Contains(t, roles[0].FailureMessage, "back-off restarting") + + // The replacement pod (same deterministic name/labels) comes up healthy. + _, healthyPod := newTestModelServingWithSinglePod(ns, msName, groupName, roleName, roleID, revision, "test-pod-crash") + healthyPod.Status = corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{ + {Type: corev1.PodReady, Status: corev1.ConditionTrue}, + }, + ContainerStatuses: []corev1.ContainerStatus{ + {Name: "main", Ready: true, State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}, + }, + } + controller.addPod(healthyPod) + + roles, err = controller.store.GetRoleList(nsName, groupName, roleName) + assert.NoError(t, err) + require.Len(t, roles, 1) + assert.Empty(t, roles[0].FailureReason, "recorded failure must be cleared once the role's pod is ready again") + assert.Empty(t, roles[0].FailureMessage) +} + +// TestUpdateModelServingStatusSurfacesPodFailureReason verifies that once a Pod failure has been +// recorded for a progressing ServingGroup, UpdateModelServingStatus propagates it onto the +// Progressing condition's Reason/Message instead of the generic "GroupProgressing" reason. +func TestUpdateModelServingStatusSurfacesPodFailureReason(t *testing.T) { + ns, msName, groupName, roleName, roleID, revision := "default", "test-ms", "test-ms-0", "prefill", "prefill-0", "hash123" + + kubeClient := kubefake.NewSimpleClientset() + kthenaClient := kthenafake.NewSimpleClientset() + volcanoClient := volcanofake.NewSimpleClientset() + apiextClient := apiextfake.NewSimpleClientset() + + controller, err := NewModelServingController(kubeClient, kthenaClient, volcanoClient, apiextClient) + assert.NoError(t, err) + + ms, _ := newTestModelServingWithSinglePod(ns, msName, groupName, roleName, roleID, revision, "test-pod") + created, err := kthenaClient.WorkloadV1alpha1().ModelServings(ns).Create(context.Background(), ms, metav1.CreateOptions{}) + require.NoError(t, err) + require.NoError(t, controller.modelServingsInformer.GetIndexer().Add(created)) + + nsName := types.NamespacedName{Namespace: ns, Name: msName} + controller.store.AddServingGroupAndRole(nsName, groupName, revision, "role-hash", roleName, roleID) + controller.store.SetRoleFailure(nsName, groupName, roleName, roleID, "ImagePullBackOff", + "pod test-pod init container downloader: back-off pulling image") + + err = controller.UpdateModelServingStatus(created, revision) + assert.NoError(t, err) + + updated, err := kthenaClient.WorkloadV1alpha1().ModelServings(ns).Get(context.Background(), msName, metav1.GetOptions{}) + require.NoError(t, err) + cond := findCondition(updated.Status.Conditions, string(workloadv1alpha1.ModelServingProgressing)) + require.NotNil(t, cond, "expected a Progressing condition") + assert.Equal(t, metav1.ConditionTrue, cond.Status) + assert.Equal(t, "ImagePullBackOff", cond.Reason) + assert.Contains(t, cond.Message, "downloader") +} + +func findCondition(conditions []metav1.Condition, condType string) *metav1.Condition { + for i := range conditions { + if conditions[i].Type == condType { + return &conditions[i] + } + } + return nil +} + // TestSyncAllBeforeFixBehavior documents the previous buggy behavior where // failed pods at startup were silently ignored when initialSync was false. // This test verifies that the fix properly addresses this issue. diff --git a/pkg/model-serving-controller/datastore/store.go b/pkg/model-serving-controller/datastore/store.go index 448fbc332..307cc5fe1 100644 --- a/pkg/model-serving-controller/datastore/store.go +++ b/pkg/model-serving-controller/datastore/store.go @@ -39,6 +39,11 @@ type Store interface { GetRolesByGroup(modelServingName types.NamespacedName, groupName string) (map[string]map[string]*Role, error) GetRoleStatus(modelServingName types.NamespacedName, groupName, roleName, roleID string) RoleStatus UpdateRoleStatus(modelServingName types.NamespacedName, groupName, roleName, roleID string, status RoleStatus) error + // SetRoleFailure records (or, when reason is empty, clears) the last observed Pod-level failure + // for a Role, and reports whether that changed the previously recorded value. It is best-effort: + // if the Role is not found (e.g. it was already deleted/recreated) the call is a no-op, since the + // failure no longer applies to any live Role. + SetRoleFailure(modelServingName types.NamespacedName, groupName, roleName, roleID, reason, message string) bool DeleteRole(modelServingName types.NamespacedName, groupName, roleName, roleID string) DeleteModelServing(modelServingName types.NamespacedName) DeleteServingGroup(modelServingName types.NamespacedName, groupName string) @@ -75,6 +80,13 @@ type Role struct { Revision string // Revision of the ServingGroup RoleTemplateHash string // Revision of the Role, used for RoleRollingUpdate strategy Status RoleStatus + // FailureReason is a stable, programmatic identifier for the last observed Pod-level + // failure of this Role (e.g. a container waiting/terminated reason reported by kubelet, + // or "Unschedulable"). Empty when no failure is currently observed for the Role. + FailureReason string + // FailureMessage is a short, human-readable detail for FailureReason (Pod/container name, + // exit code, etc). Empty when FailureReason is empty. + FailureMessage string } type ServingGroupStatus string @@ -219,6 +231,38 @@ func (s *store) UpdateRoleStatus(modelServingName types.NamespacedName, groupNam return nil } +// SetRoleFailure records or clears the last observed Pod-level failure for a Role. +// Passing an empty reason clears any previously recorded failure (e.g. once the Role's +// Pod becomes ready again). If the Role can't be found, the call is silently ignored: +// a Role that no longer exists in the store has no failure to surface either. +func (s *store) SetRoleFailure(modelServingName types.NamespacedName, groupName, roleName, roleID, reason, message string) bool { + s.mutex.Lock() + defer s.mutex.Unlock() + + servingGroups, ok := s.servingGroup[modelServingName] + if !ok { + return false + } + servingGroup, ok := servingGroups[groupName] + if !ok { + return false + } + roleMap, ok := servingGroup.roles[roleName] + if !ok { + return false + } + role, ok := roleMap[roleID] + if !ok { + return false + } + if role.FailureReason == reason && role.FailureMessage == message { + return false + } + role.FailureReason = reason + role.FailureMessage = message + return true +} + // GetRoleStatus returns the status of a specific role func (s *store) GetRoleStatus(modelServingName types.NamespacedName, groupName, roleName, roleID string) RoleStatus { s.mutex.RLock() diff --git a/pkg/model-serving-controller/datastore/store_test.go b/pkg/model-serving-controller/datastore/store_test.go index bd8502024..013f707ea 100644 --- a/pkg/model-serving-controller/datastore/store_test.go +++ b/pkg/model-serving-controller/datastore/store_test.go @@ -241,6 +241,54 @@ func TestUpdateRoleStatus(t *testing.T) { assert.Error(t, err) } +func TestSetRoleFailure(t *testing.T) { + key := types.NamespacedName{Namespace: "ns1", Name: "model1"} + + s := &store{ + mutex: sync.RWMutex{}, + servingGroup: map[types.NamespacedName]map[string]*ServingGroup{ + key: { + "group0": &ServingGroup{ + Name: "group0", + roles: map[string]map[string]*Role{ + "prefill": { + "prefill-0": &Role{Name: "prefill-0", Status: RoleCreating}, + }, + }, + }, + }, + }, + } + + // 1. Recording a new failure updates the Role and reports a change. + changed := s.SetRoleFailure(key, "group0", "prefill", "prefill-0", "ImagePullBackOff", "back-off pulling image") + assert.True(t, changed) + role := s.servingGroup[key]["group0"].roles["prefill"]["prefill-0"] + assert.Equal(t, "ImagePullBackOff", role.FailureReason) + assert.Equal(t, "back-off pulling image", role.FailureMessage) + + // 2. Recording the exact same failure again is a no-op (no change reported). + changed = s.SetRoleFailure(key, "group0", "prefill", "prefill-0", "ImagePullBackOff", "back-off pulling image") + assert.False(t, changed) + + // 3. Clearing the failure (empty reason/message) once the Pod recovers updates the Role. + changed = s.SetRoleFailure(key, "group0", "prefill", "prefill-0", "", "") + assert.True(t, changed) + assert.Empty(t, role.FailureReason) + assert.Empty(t, role.FailureMessage) + + // 4. Clearing an already-clear failure is a no-op. + changed = s.SetRoleFailure(key, "group0", "prefill", "prefill-0", "", "") + assert.False(t, changed) + + // 5. Setting a failure for a Role/group/modelServing that no longer exists is a silent no-op: + // there's no live Role to attach the failure to (it may have been recreated already). + assert.False(t, s.SetRoleFailure(key, "group0", "prefill", "nonexistent", "X", "Y")) + assert.False(t, s.SetRoleFailure(key, "nonexistgroup", "prefill", "prefill-0", "X", "Y")) + nonExistKey := types.NamespacedName{Namespace: "ns2", Name: "model2"} + assert.False(t, s.SetRoleFailure(nonExistKey, "group0", "prefill", "prefill-0", "X", "Y")) +} + func TestDeleteRole(t *testing.T) { key := types.NamespacedName{Namespace: "ns1", Name: "model1"} diff --git a/pkg/model-serving-controller/utils/utils.go b/pkg/model-serving-controller/utils/utils.go index 492afe628..6cd29d8d8 100644 --- a/pkg/model-serving-controller/utils/utils.go +++ b/pkg/model-serving-controller/utils/utils.go @@ -24,6 +24,7 @@ import ( "net/http" "regexp" "strconv" + "strings" admissionv1 "k8s.io/api/admission/v1" corev1 "k8s.io/api/core/v1" @@ -391,6 +392,132 @@ func ContainerRestarted(pod *corev1.Pod) bool { return false } +// problematicWaitingReasons are container Waiting reasons reported by kubelet that indicate +// an actionable failure rather than a normal startup step. Reasons such as "ContainerCreating" +// or "PodInitializing" are deliberately excluded so ordinary startup is never reported as failed. +var problematicWaitingReasons = map[string]bool{ + "ImagePullBackOff": true, + "ErrImagePull": true, + "InvalidImageName": true, + "CreateContainerConfigError": true, + "CreateContainerError": true, + "CrashLoopBackOff": true, +} + +const maxPodFailureMessageLen = 300 + +// PodFailureDetail describes an actionable Pod-level failure extracted from live Pod status. +type PodFailureDetail struct { + // Reason is a stable, programmatic identifier for the failure: either a well-known + // reason reported directly by the scheduler/kubelet (e.g. "Unschedulable", + // "ImagePullBackOff", "CrashLoopBackOff", "OOMKilled"), or one of our own stable + // fallback identifiers ("PodFailed", "ContainerRestarted") when kubelet hasn't + // reported a more specific one. + Reason string + // Message is a short, human-readable detail: Pod name, container/init-container name, + // exit code, and a truncated excerpt of any message kubelet/the scheduler attached. + Message string +} + +// ExtractPodFailureDetail inspects a Pod's live status and returns the most actionable +// failure it can find, if any. It relies only on structured fields that kubelet/the +// scheduler themselves populate (Pod/container conditions, waiting/terminated reasons) — +// never on Kubernetes Event text, which is free-form and not a stable API contract. +// +// Normal startup states (e.g. Pending while an image is still being pulled for the first +// time, "ContainerCreating", "PodInitializing") are intentionally not treated as failures. +func ExtractPodFailureDetail(pod *corev1.Pod) (PodFailureDetail, bool) { + // Scheduling failure: the scheduler could not place the Pod at all. This is reported + // directly on the Pod via the PodScheduled condition and requires no Events. + if pod.Status.Phase == corev1.PodPending { + for _, cond := range pod.Status.Conditions { + if cond.Type == corev1.PodScheduled && cond.Status == corev1.ConditionFalse { + reason := cond.Reason + if reason == "" { + reason = string(corev1.PodReasonUnschedulable) + } + return PodFailureDetail{ + Reason: reason, + Message: truncatePodFailureMessage(fmt.Sprintf("pod %s: %s", pod.Name, cond.Message)), + }, true + } + } + } + + // Init container problems (e.g. downloader/model-path/image-pull failures) take + // priority over main container problems, since init containers run first and block + // the rest of the Pod from starting. + if detail, ok := extractContainerFailure(pod.Name, pod.Status.InitContainerStatuses, "init container"); ok { + return detail, true + } + + // Main container problems. + if detail, ok := extractContainerFailure(pod.Name, pod.Status.ContainerStatuses, "container"); ok { + return detail, true + } + + // Pod already terminated in failure without a more specific per-container signal. + if pod.Status.Phase == corev1.PodFailed { + reason := pod.Status.Reason + if reason == "" { + reason = "PodFailed" + } + return PodFailureDetail{ + Reason: reason, + Message: truncatePodFailureMessage(fmt.Sprintf("pod %s failed: %s", pod.Name, pod.Status.Message)), + }, true + } + + return PodFailureDetail{}, false +} + +// extractContainerFailure scans a set of container statuses for a currently-actionable +// problem: a "problem" Waiting reason, a non-zero-exit Terminated state, or (when the +// container is currently up again) a Terminated LastTerminationState left by a previous +// crash. It ignores benign/expected Waiting reasons such as "ContainerCreating". +func extractContainerFailure(podName string, statuses []corev1.ContainerStatus, kind string) (PodFailureDetail, bool) { + for _, status := range statuses { + switch { + case status.State.Waiting != nil && problematicWaitingReasons[status.State.Waiting.Reason]: + return PodFailureDetail{ + Reason: status.State.Waiting.Reason, + Message: truncatePodFailureMessage(fmt.Sprintf("pod %s %s %s: %s", podName, kind, status.Name, + status.State.Waiting.Message)), + }, true + case status.State.Terminated != nil && status.State.Terminated.ExitCode != 0: + reason := status.State.Terminated.Reason + if reason == "" { + reason = "Error" + } + return PodFailureDetail{ + Reason: reason, + Message: truncatePodFailureMessage(fmt.Sprintf("pod %s %s %s exited with code %d: %s", podName, kind, + status.Name, status.State.Terminated.ExitCode, status.State.Terminated.Message)), + }, true + case status.RestartCount > 0 && status.LastTerminationState.Terminated != nil: + last := status.LastTerminationState.Terminated + reason := last.Reason + if reason == "" { + reason = "ContainerRestarted" + } + return PodFailureDetail{ + Reason: reason, + Message: truncatePodFailureMessage(fmt.Sprintf("pod %s %s %s restarted %d time(s), last exit code %d: %s", + podName, kind, status.Name, status.RestartCount, last.ExitCode, last.Message)), + }, true + } + } + return PodFailureDetail{}, false +} + +func truncatePodFailureMessage(s string) string { + s = strings.TrimSpace(s) + if len(s) <= maxPodFailureMessageLen { + return s + } + return s[:maxPodFailureMessageLen] + "..." +} + func newCondition(condType workloadv1alpha1.ModelServingConditionType, message string) metav1.Condition { var conditionType, reason string switch condType { @@ -414,7 +541,14 @@ func newCondition(condType workloadv1alpha1.ModelServingConditionType, message s } } -func SetCondition(ms *workloadv1alpha1.ModelServing, progressingGroups, updatedGroups, currentGroups []int) bool { +// SetCondition computes and applies the ModelServing's Available/Progressing/UpdateInProgress +// condition from the given ServingGroup index buckets. When failure is non-nil, it describes +// the most actionable Pod-level failure found among the progressing groups (see +// ExtractPodFailureDetail): its Reason replaces the generic "GroupProgressing"/"GroupsUpdating" +// reason, and its Message is appended, so the condition explains *why* the group isn't +// progressing rather than only *that* it isn't. Pass nil when no such failure is observed +// (e.g. groups are progressing through a normal startup) to keep the existing generic reason. +func SetCondition(ms *workloadv1alpha1.ModelServing, progressingGroups, updatedGroups, currentGroups []int, failure *PodFailureDetail) bool { var newCond metav1.Condition found := false shouldUpdate := false @@ -445,6 +579,9 @@ func SetCondition(ms *workloadv1alpha1.ModelServing, progressingGroups, updatedG newCond = newCondition(workloadv1alpha1.ModelServingAvailable, AllGroupsIsReady) } else { message := SomeGroupsAreProgressing + ": " + fmt.Sprintf("%v", progressingGroups) + if failure != nil && failure.Reason != "" { + message = message + "; " + failure.Message + } // If the number of current groups is greater than the Partition, modelServing is still updating. if len(currentGroups) > partition { message = message + ", " + SomeGroupsAreUpdated + ": " + fmt.Sprintf("%v", updatedGroups) @@ -452,12 +589,20 @@ func SetCondition(ms *workloadv1alpha1.ModelServing, progressingGroups, updatedG } else { newCond = newCondition(workloadv1alpha1.ModelServingProgressing, message) } + if failure != nil && failure.Reason != "" { + newCond.Reason = failure.Reason + } } newCond.LastTransitionTime = metav1.Now() for i, curCondition := range ms.Status.Conditions { if newCond.Type == curCondition.Type { - if newCond.Status != curCondition.Status { + if newCond.Status == curCondition.Status { + // Status unchanged: keep the original transition time, but still pick up + // Reason/Message changes (e.g. newly observed or cleared Pod failure detail). + newCond.LastTransitionTime = curCondition.LastTransitionTime + } + if newCond.Status != curCondition.Status || newCond.Reason != curCondition.Reason || newCond.Message != curCondition.Message { ms.Status.Conditions[i] = newCond shouldUpdate = true } diff --git a/pkg/model-serving-controller/utils/utils_test.go b/pkg/model-serving-controller/utils/utils_test.go index 4ac12002e..271b61d33 100644 --- a/pkg/model-serving-controller/utils/utils_test.go +++ b/pkg/model-serving-controller/utils/utils_test.go @@ -18,6 +18,7 @@ package utils import ( "testing" + "time" "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" @@ -107,7 +108,7 @@ func TestSetCondition(t *testing.T) { updatedGroups := []int{2, 3} currentGroups := []int{0, 1} - shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups) + shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups, nil) assert.True(t, shouldUpdate) assert.Len(t, ms.Status.Conditions, 1) cond := ms.Status.Conditions[0] @@ -128,7 +129,7 @@ func TestSetCondition(t *testing.T) { updatedGroups := []int{2, 3} currentGroups := []int{0, 1} - shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups) + shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups, nil) assert.True(t, shouldUpdate) assert.Len(t, ms.Status.Conditions, 1) cond := ms.Status.Conditions[0] @@ -158,7 +159,7 @@ func TestSetCondition(t *testing.T) { updatedGroups := []int{2} currentGroups := []int{0, 1} - shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups) + shouldUpdate := SetCondition(ms, progressingGroups, updatedGroups, currentGroups, nil) assert.True(t, shouldUpdate) assert.Len(t, ms.Status.Conditions, 1) cond := ms.Status.Conditions[0] @@ -166,6 +167,202 @@ func TestSetCondition(t *testing.T) { assert.Equal(t, metav1.ConditionTrue, cond.Status) assert.Contains(t, cond.Message, SomeGroupsAreProgressing) }) + + t.Run("progressing with pod failure detail uses the specific reason and message", func(t *testing.T) { + ms := &workloadv1alpha1.ModelServing{ + Spec: workloadv1alpha1.ModelServingSpec{}, + Status: workloadv1alpha1.ModelServingStatus{ + Conditions: []metav1.Condition{}, + }, + } + + progressingGroups := []int{0} + failure := &PodFailureDetail{ + Reason: "ImagePullBackOff", + Message: "pod test-ms-0-prefill-0-0 init container downloader: back-off pulling image", + } + + shouldUpdate := SetCondition(ms, progressingGroups, nil, nil, failure) + assert.True(t, shouldUpdate) + assert.Len(t, ms.Status.Conditions, 1) + cond := ms.Status.Conditions[0] + assert.Equal(t, string(workloadv1alpha1.ModelServingProgressing), cond.Type) + assert.Equal(t, metav1.ConditionTrue, cond.Status) + // The specific, stable failure reason replaces the generic "GroupProgressing" reason. + assert.Equal(t, "ImagePullBackOff", cond.Reason) + assert.Contains(t, cond.Message, SomeGroupsAreProgressing) + assert.Contains(t, cond.Message, failure.Message) + }) + + t.Run("re-evaluating with an unchanged status still refreshes reason/message", func(t *testing.T) { + ms := &workloadv1alpha1.ModelServing{ + Status: workloadv1alpha1.ModelServingStatus{ + Conditions: []metav1.Condition{ + { + Type: string(workloadv1alpha1.ModelServingProgressing), + Status: metav1.ConditionTrue, + Reason: "GroupProgressing", + Message: "stale message", + LastTransitionTime: metav1.NewTime(metav1.Now().Add(-time.Hour)), + }, + }, + }, + } + originalTransitionTime := ms.Status.Conditions[0].LastTransitionTime + + failure := &PodFailureDetail{Reason: "CrashLoopBackOff", Message: "pod p container c: crash looping"} + shouldUpdate := SetCondition(ms, []int{0}, nil, nil, failure) + assert.True(t, shouldUpdate, "message/reason changed even though Status stayed True, so an update is still required") + + cond := ms.Status.Conditions[0] + assert.Equal(t, "CrashLoopBackOff", cond.Reason) + assert.Contains(t, cond.Message, failure.Message) + // Status didn't actually transition (True -> True), so the transition time must be preserved. + assert.Equal(t, originalTransitionTime, cond.LastTransitionTime) + }) +} + +func TestExtractPodFailureDetail(t *testing.T) { + tests := []struct { + name string + pod *corev1.Pod + expectFailure bool + expectedReason string + }{ + { + name: "unschedulable pod is reported as a scheduling failure", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{ + {Type: corev1.PodScheduled, Status: corev1.ConditionFalse, Reason: "Unschedulable", Message: "0/3 nodes are available: insufficient cpu"}, + }, + }, + }, + expectFailure: true, + expectedReason: "Unschedulable", + }, + { + name: "pod still pending on normal container creation is not a failure", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{ + {Type: corev1.PodScheduled, Status: corev1.ConditionTrue}, + }, + ContainerStatuses: []corev1.ContainerStatus{ + {Name: "main", State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{Reason: "ContainerCreating"}}}, + }, + }, + }, + expectFailure: false, + }, + { + name: "init container image pull failure is reported", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + InitContainerStatuses: []corev1.ContainerStatus{ + {Name: "downloader", State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{ + Reason: "ImagePullBackOff", Message: "back-off pulling image \"bad-registry/model:latest\"", + }}}, + }, + }, + }, + expectFailure: true, + expectedReason: "ImagePullBackOff", + }, + { + name: "init container non-zero exit (e.g. downloader/model-path failure) is reported", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + InitContainerStatuses: []corev1.ContainerStatus{ + {Name: "downloader", State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + ExitCode: 1, Reason: "Error", Message: "model path not found", + }}}, + }, + }, + }, + expectFailure: true, + expectedReason: "Error", + }, + { + name: "main container crash loop is reported", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + ContainerStatuses: []corev1.ContainerStatus{ + {Name: "engine", RestartCount: 3, State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{ + Reason: "CrashLoopBackOff", Message: "back-off restarting failed container", + }}}, + }, + }, + }, + expectFailure: true, + expectedReason: "CrashLoopBackOff", + }, + { + name: "main container OOMKilled after restart is reported from LastTerminationState", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: "engine", + RestartCount: 1, + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + LastTerminationState: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "OOMKilled", ExitCode: 137, + }}, + }, + }, + }, + }, + expectFailure: true, + expectedReason: "OOMKilled", + }, + { + name: "pod failed phase without container detail falls back to PodFailed", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{Phase: corev1.PodFailed}, + }, + expectFailure: true, + expectedReason: "PodFailed", + }, + { + name: "ready running pod has no failure", + pod: &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + ContainerStatuses: []corev1.ContainerStatus{ + {Name: "engine", State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}, + }, + }, + }, + expectFailure: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + detail, ok := ExtractPodFailureDetail(tt.pod) + assert.Equal(t, tt.expectFailure, ok) + if tt.expectFailure { + assert.Equal(t, tt.expectedReason, detail.Reason) + assert.NotEmpty(t, detail.Message) + assert.Contains(t, detail.Message, "p") + } + }) + } } func TestGetMaxUnavailable(t *testing.T) {