diff --git a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel b/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel index 01be42ef9..67390b64d 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel +++ b/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel @@ -60,6 +60,7 @@ go_library( "//vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/types/nvca/config", "//vendor/github.com/go-logr/logr", "//vendor/github.com/google/go-cmp/cmp", + "//vendor/github.com/google/go-cmp/cmp/cmpopts", "//vendor/github.com/google/go-containerregistry/pkg/name", "//vendor/github.com/hashicorp/go-retryablehttp", "//vendor/github.com/imdario/mergo", diff --git a/src/compute-plane-services/nvca/internal/miniservice/controller.go b/src/compute-plane-services/nvca/internal/miniservice/controller.go index 9033fca30..1e744691b 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/controller.go +++ b/src/compute-plane-services/nvca/internal/miniservice/controller.go @@ -136,6 +136,8 @@ type registrationInstanceTypeCache interface { } const ( + controllerName = "miniservice-controller" + defaultCacheDir = "/var/run/nvca/reval-rendered-helmcharts" ) @@ -185,7 +187,7 @@ func BuildController(ctx context.Context, tracer: otel.NewTracer(), Decoder: newFlexibleDecoder(mgr.GetScheme(), extraGVKs...), NFClient: nflient, - eventRecorder: mgr.GetEventRecorderFor("miniservice-controller"), + eventRecorder: mgr.GetEventRecorderFor(controllerName), chartCache: chartcache.New(opts.cacheDir), regITCache: regITCache, enabledAttrs: enabledAttrs, diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile.go index 117c945bd..ace79ec11 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go +++ b/src/compute-plane-services/nvca/internal/miniservice/reconcile.go @@ -39,6 +39,8 @@ import ( translateutil "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/util" cmnotel "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/otel" nvcaconfig "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/types/nvca/config" + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" "github.com/google/go-containerregistry/pkg/name" otelattr "go.opentelemetry.io/otel/attribute" otelcodes "go.opentelemetry.io/otel/codes" @@ -443,6 +445,41 @@ func (r *Reconciler) patchMiniService(ctx context.Context, oldObj, newObj *v1alp return nil } +// saveWorkloadConfig persists the workload config decoded from the rendered workload config +// ConfigMap onto the MiniService spec, patching only that field inline. It is a no-op when the +// config is unchanged. On success ms is updated in place so callers see the persisted value +// and current ResourceVersion. +func (r *Reconciler) saveWorkloadConfig( + ctx context.Context, + ms *v1alpha1.MiniService, + desired *v1alpha1.WorkloadConfig, +) error { + if cmp.Equal(ms.Spec.WorkloadConfig, desired, cmpopts.EquateEmpty()) { + return nil + } + + base := &v1alpha1.MiniService{ + TypeMeta: metav1.TypeMeta{ + APIVersion: v1alpha1.SchemeGroupVersion.String(), + Kind: "MiniService", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: ms.Name, + ResourceVersion: ms.ResourceVersion, + }, + Spec: v1alpha1.MiniServiceSpec{ + WorkloadConfig: desired.DeepCopy(), + }, + } + if err := r.Client.Patch(ctx, base, client.Apply, client.ForceOwnership, client.FieldOwner(managedByValue)); err != nil { + return fmt.Errorf("patch miniservice %s workload config: %w", ms.Name, err) + } + + ms.Spec.WorkloadConfig = desired + ms.ResourceVersion = base.ResourceVersion + return nil +} + // prepareUpdateIfNeeded detects spec changes by comparing metadata.generation // to status.observedGeneration. Because generation increments on any spec field change, // this method additionally compares helm values against the latest revision ConfigMap. @@ -560,10 +597,15 @@ func (r *Reconciler) doInstall(ctx context.Context, } } - workloadObjs, resources, err := decodeObjects(ctx, r.Decoder, objsData) + workloadObjs, resources, workloadConfig, err := decodeObjects(ctx, r.Decoder, objsData) if err != nil { return reconcile.Result{}, err } + // Persist workload config decoded from the rendered workload config ConfigMap so later + // reconciles (e.g. status) can source it from the spec. + if err := r.saveWorkloadConfig(ctx, ms, workloadConfig); err != nil { + return reconcile.Result{}, err + } // Update the resources status in the MiniService status. updateResourcesStatus(ms, resources) // ReVal may render filtered workload pull secrets, which should be used if possible. @@ -911,7 +953,7 @@ func (r *Reconciler) prepareUpdateWorkload(ctx context.Context, } } - workloadObjs, resources, err := decodeObjects(ctx, r.Decoder, objsData) + workloadObjs, resources, _, err := decodeObjects(ctx, r.Decoder, objsData) if err != nil { return nil, nil, "", "", err } @@ -1274,7 +1316,7 @@ func (r *Reconciler) doCleanup(ctx context.Context, //nolint:gocyclo } else { log.V(1).Info("Using cached rendered object data in cleanup") - if objs, _, err = decodeObjects(ctx, r.Decoder, objsData); err != nil { + if objs, _, _, err = decodeObjects(ctx, r.Decoder, objsData); err != nil { return reconcile.Result{}, err } for i := 0; i < len(objs); i++ { @@ -1511,8 +1553,6 @@ func (r *Reconciler) applyWorkload(ctx context.Context, return r.create(ctx, ms, objectMutatorSet{}, genericMutator, c, objs...) } -const fieldManagerName = "miniservice-controller" - // applySSAWorkload applies workload objects using server-side apply for updates. // Like applyWorkload, it may use an impersonating client for RBAC enforcement. func (r *Reconciler) applySSAWorkload(ctx context.Context, @@ -1587,7 +1627,7 @@ func (r *Reconciler) applySSA(ctx context.Context, obj.SetResourceVersion("") obj.GetObjectKind().SetGroupVersionKind(gvk) - if err := c.Patch(ctx, obj, client.Apply, client.FieldOwner(fieldManagerName), client.ForceOwnership); err != nil { + if err := c.Patch(ctx, obj, client.Apply, client.FieldOwner(managedByValue), client.ForceOwnership); err != nil { if apierrors.IsInvalid(err) || apierrors.IsForbidden(err) { err = reconcile.TerminalError(err) } @@ -1802,29 +1842,46 @@ func weighObject(obj client.Object) int8 { } } -func decodeObjects(ctx context.Context, decoder runtime.Decoder, objsData []byte) ([]client.Object, []v1alpha1.ResourceStatus, error) { +// decodeObjects decodes the ReVal-rendered object data into typed client objects and a +// per-GVK resource summary. The workload config ConfigMap +// (featureflag.WorkloadConfigConfigMapName) is a control signal from the chart and is never +// created on-cluster: it is extracted and returned as a decoded WorkloadConfig, and excluded +// from the returned objects and resource summary so it is never applied or status-checked. +func decodeObjects(ctx context.Context, decoder runtime.Decoder, objsData []byte) ( + []client.Object, []v1alpha1.ResourceStatus, *v1alpha1.WorkloadConfig, error, +) { log := logf.FromContext(ctx) var rawObjs []json.RawMessage if err := json.Unmarshal(objsData, &rawObjs); err != nil { - return nil, nil, err + return nil, nil, nil, err } - objs := make([]client.Object, len(rawObjs)) + objs := make([]client.Object, 0, len(rawObjs)) resourcesByGVK := map[string]v1alpha1.ResourceStatus{} + var workloadConfig *v1alpha1.WorkloadConfig for i, rawObj := range rawObjs { robj, _, err := decoder.Decode(rawObj, nil, nil) if err != nil { log.Error(err, "Error decoding object from ReVal", "index", i) - return nil, nil, reconcile.TerminalError(err) + return nil, nil, nil, reconcile.TerminalError(err) } cobj, ok := robj.(client.Object) if !ok { log.Error(nil, "Object does not implement client.Object", "index", i) - return nil, nil, reconcile.TerminalError(fmt.Errorf("bad object type")) + return nil, nil, nil, reconcile.TerminalError(fmt.Errorf("bad object type")) + } + + // Extract the workload config ConfigMap. It must never be applied on-cluster, + // so drop it from the objects to create and status-check. + if cm, ok := cobj.(*corev1.ConfigMap); ok && cm.Name == featureflag.WorkloadConfigConfigMapName { + if workloadConfig, err = featureflag.DecodeWorkloadConfig(ctx, log, cm); err != nil { + return nil, nil, nil, reconcile.TerminalError(fmt.Errorf("decode workload config: %w", err)) + } + continue } - objs[i] = cobj + objs = append(objs, cobj) gvk := cobj.GetObjectKind().GroupVersionKind().String() if _, ok := resourcesByGVK[gvk]; ok { resource := resourcesByGVK[gvk] @@ -1846,7 +1903,7 @@ func decodeObjects(ctx context.Context, decoder runtime.Decoder, objsData []byte log.V(1).Info("Decoded object", "gvk", resource.GVK, "names", resource.Names, "count", resource.Count) } - return objs, resources, nil + return objs, resources, workloadConfig, nil } // getObjectGVK returns the GVK for an object using the gvkCache from context for better performance. diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go index ae4341afd..b5e24abc7 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go +++ b/src/compute-plane-services/nvca/internal/miniservice/reconcile_test.go @@ -4340,7 +4340,7 @@ func TestDecodeObjects_Success(t *testing.T) { data, err := json.Marshal(helmObjs) require.NoError(t, err) - objs, resources, derr := decodeObjects(ctx, decoder, data) + objs, resources, _, derr := decodeObjects(ctx, decoder, data) require.NoError(t, derr) require.Len(t, objs, 2) require.Len(t, resources, 2) @@ -4366,6 +4366,57 @@ func TestDecodeObjects_Success(t *testing.T) { }) } +func TestDecodeObjects_ExtractsWorkloadConfig(t *testing.T) { + ctx := newTestContext() + s := mgrScheme + decoder := serializer.NewCodecFactory(s).UniversalDeserializer() + + helmObjs := []client.Object{ + &corev1.ConfigMap{ + TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"}, + ObjectMeta: metav1.ObjectMeta{ + Name: featureflag.WorkloadConfigConfigMapName, + }, + Data: map[string]string{ + featureflag.WorkloadConfigDataKey: "featureFlags:\n" + + " " + featureflag.StatusByWorkerReadiness + ": true\n" + + " SOME_UNKNOWN_FLAG: true\n", + }, + }, + &appsv1.Deployment{ + TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"}, + ObjectMeta: metav1.ObjectMeta{ + Name: "dep-1", + }, + }, + } + + data, err := json.Marshal(helmObjs) + require.NoError(t, err) + + objs, resources, workloadConfig, derr := decodeObjects(ctx, decoder, data) + require.NoError(t, derr) + + // The workload config ConfigMap must be stripped from the applied/status-checked objects + // and from the resource summary. + require.Len(t, objs, 1) + if dep, ok := objs[0].(*appsv1.Deployment); assert.True(t, ok) { + assert.Equal(t, "dep-1", dep.Name) + } + require.Len(t, resources, 1) + assert.Contains(t, resources, v1alpha1.ResourceStatus{ + GVK: "apps/v1, Kind=Deployment", + Names: []string{"dep-1"}, + Count: 1, + }) + + // Recognized flags are decoded; unknown keys are dropped. + assert.Equal(t, &v1alpha1.WorkloadConfig{ + FeatureFlags: map[string]bool{featureflag.StatusByWorkerReadiness: true}, + }, workloadConfig) + assert.True(t, workloadConfig.IsFeatureFlagEnabled(featureflag.StatusByWorkerReadiness)) +} + func TestDecodeObjects_JSONUnmarshalError(t *testing.T) { ctx := newTestContext() s := mgrScheme @@ -4374,7 +4425,7 @@ func TestDecodeObjects_JSONUnmarshalError(t *testing.T) { // Invalid JSON data := []byte("not-json") - objs, resources, err := decodeObjects(ctx, decoder, data) + objs, resources, _, err := decodeObjects(ctx, decoder, data) assert.Error(t, err) assert.Nil(t, objs) assert.Nil(t, resources) @@ -4394,7 +4445,7 @@ func TestDecodeObjects_NonClientObject(t *testing.T) { data, err := json.Marshal(objs) require.NoError(t, err) - gotObjs, gotResources, derr := decodeObjects(ctx, decoder, data) + gotObjs, gotResources, _, derr := decodeObjects(ctx, decoder, data) assert.EqualError(t, derr, "terminal error: bad object type") assert.Nil(t, gotObjs) assert.Nil(t, gotResources) diff --git a/src/compute-plane-services/nvca/internal/miniservice/revision.go b/src/compute-plane-services/nvca/internal/miniservice/revision.go index c7179830a..684919d00 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/revision.go +++ b/src/compute-plane-services/nvca/internal/miniservice/revision.go @@ -40,7 +40,7 @@ const ( revisionConfigMapPrefix = "miniservice-revision-v" revisionLabel = "nvca.nvcf.nvidia.io/revision" managedByLabel = "app.kubernetes.io/managed-by" - managedByValue = "miniservice-controller" + managedByValue = controllerName revisionDataKeyValues = "values" revisionDataKeyRenderHash = "renderHash" diff --git a/src/compute-plane-services/nvca/internal/miniservice/status.go b/src/compute-plane-services/nvca/internal/miniservice/status.go index 7db674010..2f2cb1020 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/status.go +++ b/src/compute-plane-services/nvca/internal/miniservice/status.go @@ -27,6 +27,7 @@ import ( "time" "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common" + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/function" translateutil "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/util" "github.com/google/go-cmp/cmp" appsv1 "k8s.io/api/apps/v1" @@ -48,6 +49,7 @@ import ( nvcainternaltranslate "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/internal/util/translate" "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1" nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/featureflag" nvcastorage "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/storage" nvcatypes "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" ) @@ -121,39 +123,34 @@ func filterEventsForObject(ctx context.Context, sch *runtime.Scheme, events []co type checkStatusFunc func(context.Context, client.Object, *nvcav2beta1.ICMSRequest) (ObjectStatus, error) //nolint:gocyclo -func (r *Reconciler) doStatus( +func (r *Reconciler) collectObjectStatuses( ctx context.Context, ms *v1alpha1.MiniService, icmsReq *nvcav2beta1.ICMSRequest, -) (reconcile.Result, error) { +) (ObjectStatuses, []error, error) { log := logf.FromContext(ctx) objsData, isRendered, err := r.getRenderedData(ctx, ms) if err != nil { - return reconcile.Result{}, err + return nil, nil, err } if !isRendered { if objsData, err = r.render(ctx, ms, icmsReq); err != nil { - return reconcile.Result{}, err + return nil, nil, err } if err := r.saveRenderedData(ctx, ms, objsData); err != nil { - return reconcile.Result{}, err + return nil, nil, err } } - objs, resources, err := decodeObjects(ctx, r.Decoder, objsData) + objs, resources, _, err := decodeObjects(ctx, r.Decoder, objsData) if err != nil { - return reconcile.Result{}, err + return nil, nil, err } updateResourcesStatus(ms, resources) - // Check utils pod too. - utilsPod := &corev1.Pod{} - utilsPod.Name = common.UtilsPodName - objs = append(objs, utilsPod) - var ( - errs []error + serrs []error statuses ObjectStatuses ) @@ -165,7 +162,7 @@ func (r *Reconciler) doStatus( // The list functions will transparently return cached data when statusContext is in context. ctx, err = r.buildStatusContext(ctx, ms.Spec.Namespace) if err != nil { - return reconcile.Result{}, err + return nil, nil, err } // Use the main client for type info since impersonated clients will not have access @@ -174,25 +171,24 @@ func (r *Reconciler) doStatus( // Observe the status of only rendered objects. Owned objects created by controllers // will be retrieved by owner reference. - utilsPodFound := false // Track pods that have explicit status checks so they are skipped in the general pod check below. - // Let StorageRequest controller handle the SMB pod. - seenPods := sets.New(nvcastorage.SMBServerPodName) + // Let StorageRequest controller handle the SMB pod, and caller handle the utils pod. + seenPods := sets.New(common.UtilsPodName, nvcastorage.SMBServerPodName) for _, o := range objs { gvk, err := r.getObjectGVK(ctx, o) if err != nil { - return reconcile.Result{}, reconcile.TerminalError(err) + return nil, serrs, reconcile.TerminalError(err) } if isNamespaced, err := apiutil.IsGVKNamespaced(gvk, rm); isNamespaced { o.SetNamespace(ms.Spec.Namespace) } else if err != nil { log.Error(err, "Failed to check if object is namespaced") - return reconcile.Result{}, reconcile.TerminalError(err) + return nil, serrs, reconcile.TerminalError(err) } if err := checkPermissions(ctx, r.Client, gvk, o.GetNamespace()); err != nil { - return reconcile.Result{}, err + return nil, serrs, err } if err := r.Client.Get(ctx, client.ObjectKeyFromObject(o), o); err != nil { @@ -203,7 +199,7 @@ func (r *Reconciler) doStatus( TerminalBad: true, }) } else { - errs = append(errs, err) + serrs = append(serrs, err) log.Error(err, "Failed to get live object from rendered") } continue @@ -223,18 +219,13 @@ func (r *Reconciler) doStatus( } objStatus, err := check(ctx, o, icmsReq) if err != nil { - errs = append(errs, err) + serrs = append(serrs, err) continue } statuses = append(statuses, objStatus) if pod, ok := o.(*corev1.Pod); ok { - // The utils pod status is added above. seenPods.Insert(pod.Name) - if nvcak8sutil.IsUtilsPod(pod) { - utilsPodFound = true - utilsPod = pod - } } } @@ -242,7 +233,7 @@ func (r *Reconciler) doStatus( // for awhile by their owning object. allPods, err := r.listPods(ctx, ms.Spec.Namespace) if err != nil { - return reconcile.Result{}, err + return nil, serrs, err } for _, pod := range allPods { if seenPods.Has(pod.Name) || nvcatypes.IsInfraOwnedObject(pod) { @@ -250,7 +241,7 @@ func (r *Reconciler) doStatus( } podStatus, err := r.checkPodStatus(ctx, pod, icmsReq) if err != nil { - errs = append(errs, err) + serrs = append(serrs, err) continue } switch podStatus.Status { @@ -259,6 +250,55 @@ func (r *Reconciler) doStatus( } } + return statuses, serrs, nil +} + +func (r *Reconciler) doStatus( + ctx context.Context, + ms *v1alpha1.MiniService, + icmsReq *nvcav2beta1.ICMSRequest, +) (reconcile.Result, error) { + // The workload feature-flags ConfigMap may opt a workload into worker-readiness-based + // status, which only considers worker container readiness rather than aggressively + // accounting for the health of all workload objects. + if ms.Spec.WorkloadConfig.IsFeatureFlagEnabled(featureflag.StatusByWorkerReadiness) { + return r.doStatusByWorkerReadiness(ctx, ms, icmsReq) + } + return r.doStatusAggressive(ctx, ms, icmsReq) +} + +// doStatusAggressive checks the status of all workload objects and the utils pod. +// If any object has a terminal bad status, the function will return a terminal error. +func (r *Reconciler) doStatusAggressive( + ctx context.Context, + ms *v1alpha1.MiniService, + icmsReq *nvcav2beta1.ICMSRequest, +) (reconcile.Result, error) { + log := logf.FromContext(ctx) + + statuses, serrs, err := r.collectObjectStatuses(ctx, ms, icmsReq) + if err != nil { + return reconcile.Result{}, err + } + + // Check utils pod too. + utilsPodFound := true + utilsPod := &corev1.Pod{} + if err := r.Client.Get(ctx, client.ObjectKey{Namespace: ms.Spec.Namespace, Name: common.UtilsPodName}, utilsPod); err != nil { + if !apierrors.IsNotFound(err) { + return reconcile.Result{}, err + } + utilsPodFound = false + } + if utilsPodFound { + utilsStatus, err := r.checkPodStatus(ctx, utilsPod, icmsReq) + if err != nil { + serrs = append(serrs, err) + } else { + statuses = append(statuses, utilsStatus) + } + } + var ( res reconcile.Result rerr error @@ -275,7 +315,7 @@ func (r *Reconciler) doStatus( Message: "Infrastructure objects not found", }) - if err := errors.Join(errs...); err != nil { + if err := errors.Join(serrs...); err != nil { log.Error(err, "Errors were encountered while checking object status") } case statuses.AnyTerminal(): @@ -289,7 +329,7 @@ func (r *Reconciler) doStatus( meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ Type: v1alpha1.MiniServiceConditionObjectsHealthy, Status: metav1.ConditionFalse, - Reason: v1alpha1.MiniServiceStatusReasonDegradedWorkerPods, + Reason: v1alpha1.MiniServiceStatusReasonDegradedWorker, Message: writeObjectStatuses(ctx, r.Client.Scheme(), statuses, filterTerminal), }) rerr = reconcile.TerminalError(fmt.Errorf("at least one object is degraded")) @@ -331,7 +371,7 @@ func (r *Reconciler) doStatus( } } - if err := errors.Join(errs...); err != nil { + if err := errors.Join(serrs...); err != nil { log.Error(err, "Errors were encountered while checking object status") } case statuses.AnyPending(): @@ -372,11 +412,11 @@ func (r *Reconciler) doStatus( // Requeue at least once after the scheduling timeout has passed in case no progress is made. res.RequeueAfter = getPodSchedulingTimeout(ctx, icmsReq, r.K8sTimeConfig) } - if err := errors.Join(errs...); err != nil { + if err := errors.Join(serrs...); err != nil { log.Error(err, "Errors were encountered while checking object status") } default: - if rerr = errors.Join(errs...); rerr != nil { + if rerr = errors.Join(serrs...); rerr != nil { log.Error(rerr, "Found errors while collecting object statuses") meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ Type: v1alpha1.MiniServiceConditionObjectsHealthy, @@ -398,6 +438,174 @@ func (r *Reconciler) doStatus( return res, rerr } +func (r *Reconciler) doStatusByWorkerReadiness( + ctx context.Context, + ms *v1alpha1.MiniService, + icmsReq *nvcav2beta1.ICMSRequest, +) (res reconcile.Result, rerr error) { + log := logf.FromContext(ctx) + + statuses, serrs, err := r.collectObjectStatuses(ctx, ms, icmsReq) + if err != nil { + return reconcile.Result{}, err + } + + // Handle all non-worker objects separately from the worker pod, in all cases. + // Health does not depend on these object's statuses, so they are purely informational. + // NVCA will report the condition's message to ICMS for users to debug. + { + anyObjectsNotReady := false + objectStatusMsg := writeObjectStatuses(ctx, r.Client.Scheme(), statuses, func(os ObjectStatus) bool { + v := os.TerminalBad || os.Scheduling || os.Pending + anyObjectsNotReady = anyObjectsNotReady || v + return v + }) + if anyObjectsNotReady { + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionObjectsHealthy, + Status: metav1.ConditionFalse, + Reason: v1alpha1.MiniServiceStatusReasonWaitingObjectReadiness, + Message: objectStatusMsg, + }) + } else { + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionObjectsHealthy, + Status: metav1.ConditionTrue, + Reason: "ObjectsReady", + }) + } + } + + // Below here, only the worker pod's status is checked for MiniService health. + + utilsPod := &corev1.Pod{} + if err := r.Client.Get(ctx, client.ObjectKey{Namespace: ms.Spec.Namespace, Name: common.UtilsPodName}, utilsPod); err != nil { + if !apierrors.IsNotFound(err) { + return reconcile.Result{}, err + } + // By the time doStatus* functions are called, the utils pod must exist. + rerr = reconcile.TerminalError(fmt.Errorf("utils pod not found")) + log.Error(rerr, "Workload is unhealthy") + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionWorkersHealthy, + Status: metav1.ConditionFalse, + Reason: "WorkerNotFound", + Message: "Worker pod not found", + }) + + if err := errors.Join(serrs...); err != nil { + log.Error(err, "Errors were encountered while checking object statuses") + } + return reconcile.Result{}, rerr + } + + utilsStatus, err := r.checkPodStatus(ctx, utilsPod, icmsReq) + if err != nil { + return reconcile.Result{}, errors.Join(append(serrs, err)...) + } + + workerContainerStatus, statusPresent, err := getFunctionWorkerContainerStatus(utilsPod) + if err != nil || !statusPresent { + // Utils pod events will requeue this once container statuses are present. + return reconcile.Result{}, err + } + + // If the worker container is ready, then the miniservice is running. + /* + TODO: + * only use utils container readiness + * fail on utils container restart count > 3 + */ + if workerContainerStatus.Ready { + ms.Status.Phase = v1alpha1.MiniServiceRunning + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionWorkersHealthy, + Status: metav1.ConditionTrue, + Reason: "WorkersReady", + }) + return reconcile.Result{}, nil + } + + // The worker container is either progressing towards a ready state, or is in a terminal bad state. + + var workerIssueReasons []string + if utilsStatus.AdditionalPodStatus != nil { + aps := utilsStatus.AdditionalPodStatus + // Check if the worker container is degraded. + if aps.Degraded != nil && slices.ContainsFunc(aps.Degraded.Containers, func(cd nvcak8sutil.ContainerDegraded) bool { + return isWorkerContainerName(cd.Name) + }) { + workerIssueReasons = append(workerIssueReasons, v1alpha1.MiniServiceStatusReasonDegradedWorker) + } + // Check if the worker container is stuck initializing. + if aps.StuckInitializing != nil && slices.ContainsFunc(aps.StuckInitializing.Containers, func(csi nvcak8sutil.ContainerStuckInitializing) bool { + return isWorkerContainerName(csi.Name) + }) { + workerIssueReasons = append(workerIssueReasons, "WorkerStuckInitializing") + } + // Check if the worker container has image pull issues. + if aps.ImagePullIssues != nil && slices.ContainsFunc(aps.ImagePullIssues, func(cips nvcak8sutil.ContainerImagePullIssues) bool { + return isWorkerContainerName(cips.Name) + }) { + workerIssueReasons = append(workerIssueReasons, "WorkerImagePullIssues") + } + } + + workersHealthyCond := meta.FindStatusCondition(ms.Status.Conditions, v1alpha1.MiniServiceConditionWorkersHealthy) + workerStatusMsg := writeObjectStatuses(ctx, r.Client.Scheme(), []ObjectStatus{utilsStatus}, nil) + + if len(workerIssueReasons) > 0 { + rerr = reconcile.TerminalError(fmt.Errorf("worker container is in a terminal bad state")) + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionWorkersHealthy, + Status: metav1.ConditionFalse, + Reason: "WorkerTerminalBadState", + Message: workerStatusMsg, + }) + log.Error(rerr, "Worker container is in a terminal bad state", + "workerIssueReasons", workerIssueReasons, + "workerStatus", workerStatusMsg) + } else if workersHealthyCond != nil && workersHealthyCond.Status != metav1.ConditionTrue && + workersHealthyCond.LastTransitionTime.Add(r.K8sTimeConfig.MaxRunningTimeout).Before(r.now()) { + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionWorkersHealthy, + Status: metav1.ConditionFalse, + Reason: v1alpha1.MiniServiceStatusReasonPendingTimeout, + Message: workerStatusMsg, + }) + rerr = reconcile.TerminalError(fmt.Errorf("worker did not become ready within %s", r.K8sTimeConfig.MaxRunningTimeout)) + log.Error(rerr, "Pending worker timeout reached", + "workerStatus", workerStatusMsg) + } else { + var message string + if utilsStatus.Scheduling { + // Once the worker pod is scheduled after installation, the miniservice can transition to Starting. + if ms.Status.Phase != v1alpha1.MiniServiceRunning { + ms.Status.Phase = v1alpha1.MiniServiceStarting + } + message = "Worker pod is scheduled, waiting on readiness" + } else { + log.V(1).Info("Waiting on worker pod to schedule") + message = "Worker pod is not scheduled" + } + meta.SetStatusCondition(&ms.Status.Conditions, metav1.Condition{ + Type: v1alpha1.MiniServiceConditionWorkersHealthy, + Status: metav1.ConditionFalse, + Reason: v1alpha1.MiniServiceStatusReasonWaitingObjectReadiness, + Message: fmt.Sprintf("%s\n%s", message, workerStatusMsg), + }) + log.V(1).Info(message) + + // Requeue at least once after the scheduling timeout has passed in case no progress is made. + res.RequeueAfter = getPodSchedulingTimeout(ctx, icmsReq, r.K8sTimeConfig) + } + if err := errors.Join(serrs...); err != nil { + log.Error(err, "Errors were encountered while checking object status") + } + + return res, rerr +} + func (r *Reconciler) doTerminalTaskStatus( ctx context.Context, ms *v1alpha1.MiniService, @@ -450,7 +658,7 @@ func (r *Reconciler) doTerminalTaskStatus( message = fmt.Sprintf("max runtime duration %s has been exceeded", mrd) case statuses.OnlyTerminalDegraded(): // If the only reason is worker degradation, set a condition and let the agent handle cleanup. - reason = v1alpha1.MiniServiceStatusReasonDegradedWorkerPods + reason = v1alpha1.MiniServiceStatusReasonDegradedWorker rerr = reconcile.TerminalError(fmt.Errorf("at least one object is degraded")) default: // Check if there is an existing MiniServiceConditionObjectsHealthy to @@ -497,6 +705,27 @@ func (r *Reconciler) doTerminalTaskStatus( return res, rerr } +// getFunctionWorkerContainerStatus returns the worker container's status from the utils pod for the given MiniService, +// either utils for normal functions/tasks or the llm worker (pylon) for LLM-type functions. +func getFunctionWorkerContainerStatus(utilsPod *corev1.Pod) (corev1.ContainerStatus, bool, error) { + for _, cs := range utilsPod.Status.ContainerStatuses { + if isWorkerContainerName(cs.Name) { + return cs, true, nil + } + } + // The container statuses are not yet available, indicating the pod hasn't been initialized by kubelet yet. + for _, c := range utilsPod.Spec.Containers { + if isWorkerContainerName(c.Name) { + return corev1.ContainerStatus{}, false, nil + } + } + return corev1.ContainerStatus{}, false, reconcile.TerminalError(fmt.Errorf("worker container not found")) +} + +func isWorkerContainerName(name string) bool { + return name == common.UtilsContainerName || name == function.LLMWorkerContainerName +} + func (r *Reconciler) makeStatusCheckers() map[schema.GroupVersionKind]checkStatusFunc { return map[schema.GroupVersionKind]checkStatusFunc{ podGVK: r.checkPodStatus, @@ -534,6 +763,14 @@ type ObjectStatus struct { // Some of these event types indicate a terminal status, based on truthiness of parseErrorEventMessage; // all others should only be reported when the object is terminal from conditions/phase. AbnormalEvents []corev1.Event + // AdditionalPodStatus contains additional status data for the pod, if any. + AdditionalPodStatus *AdditionalPodStatus +} + +type AdditionalPodStatus struct { + Degraded *nvcak8sutil.PodDegraded + StuckInitializing *nvcak8sutil.PodStuckInitializing + ImagePullIssues []nvcak8sutil.ContainerImagePullIssues } func (s ObjectStatus) toTerminalEventStatus() (cs ObjectStatus) { @@ -623,19 +860,25 @@ func getPodStatus(pod *corev1.Pod, k8sTimeConfig *nvcak8sutil.TimeConfig, schedu case corev1.PodPending, corev1.PodUnknown, "": isScheduled := nvcak8sutil.IsPodScheduled(ps) if isScheduled { - if _, state, imgIssues := nvcak8sutil.ImagePullIssuesReported(ps); imgIssues && + if imgPullIssues, imgIssues := nvcak8sutil.ImagePullIssuesReported(ps); imgIssues && nvcak8sutil.IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.MaxImagePullErrorThreshold) { return ObjectStatus{ Status: podFailedImagePullIssues, - Reason: state.Reason, + Reason: "ImagePullIssues", TerminalBad: true, + AdditionalPodStatus: &AdditionalPodStatus{ + ImagePullIssues: imgPullIssues, + }, } } - if stuck, reason := nvcak8sutil.IsPodStuckInitializing(pod, k8sTimeConfig); stuck { + if podStuckInitializing, stuck := nvcak8sutil.IsPodStuckInitializing(pod, k8sTimeConfig); stuck { return ObjectStatus{ Status: podFailedContainerStuck, - Reason: reason, + Reason: podStuckInitializing.Reason, TerminalBad: true, + AdditionalPodStatus: &AdditionalPodStatus{ + StuckInitializing: &podStuckInitializing, + }, } } if nvcak8sutil.IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.PodLaunchThresholdMinutesOnInitFailure) { @@ -671,18 +914,24 @@ func getPodStatus(pod *corev1.Pod, k8sTimeConfig *nvcak8sutil.TimeConfig, schedu Status: statusRunning, } } - if stuck, reason := nvcak8sutil.IsPodStuckInitializing(pod, k8sTimeConfig); stuck { + if podStuckInitializing, stuck := nvcak8sutil.IsPodStuckInitializing(pod, k8sTimeConfig); stuck { return ObjectStatus{ Status: podFailedContainerStuck, - Reason: reason, + Reason: podStuckInitializing.Reason, TerminalBad: true, + AdditionalPodStatus: &AdditionalPodStatus{ + StuckInitializing: &podStuckInitializing, + }, } } - if degraded, reason := nvcak8sutil.IsPodDegraded(pod, k8sTimeConfig); degraded { + if degradedStatus, degraded := nvcak8sutil.IsPodDegraded(pod, k8sTimeConfig); degraded { return ObjectStatus{ Status: statusDegradedWorker, - Reason: reason, + Reason: degradedStatus.Reason, TerminalBad: true, + AdditionalPodStatus: &AdditionalPodStatus{ + Degraded: °radedStatus, + }, } } return ObjectStatus{ @@ -1177,11 +1426,11 @@ func writeObjectStatuses(ctx context.Context, sch *runtime.Scheme, objStatuses O messageBuilder := &strings.Builder{} for _, objStatus := range objStatuses { // Let top-level object determine whether children may be in filtered state. - if !filter(objStatus) { + if filter != nil && !filter(objStatus) { continue } objStatus.Visit(func(os *ObjectStatus) { - if filter(*os) { + if filter == nil || filter(*os) { writeObjectStatus(ctx, sch, messageBuilder, *os) messageBuilder.WriteByte('\n') } diff --git a/src/compute-plane-services/nvca/internal/miniservice/status_karta_test.go b/src/compute-plane-services/nvca/internal/miniservice/status_karta_test.go index 12aac002a..1ad27a260 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/status_karta_test.go +++ b/src/compute-plane-services/nvca/internal/miniservice/status_karta_test.go @@ -34,14 +34,6 @@ import ( nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" ) -func newKartaScheme(t *testing.T) *runtime.Scheme { - t.Helper() - scheme := runtime.NewScheme() - require.NoError(t, kartav1alpha1.AddToScheme(scheme)) - require.NoError(t, corev1.AddToScheme(scheme)) - return scheme -} - func TestGetEmbeddedKartaDefinitions(t *testing.T) { kartas, err := getEmbeddedKartaDefinitions() require.NoError(t, err) @@ -858,7 +850,7 @@ func TestNewKartaDefinedObjectStatusChecker_EmbeddedDynamoRunning(t *testing.T) func TestLoadKartaDefinitions_WithLiveKartaFromFakeClient(t *testing.T) { liveKartas := []*kartav1alpha1.Karta{ - &kartav1alpha1.Karta{ + { ObjectMeta: metav1.ObjectMeta{ Name: "live-dynamo-v2", }, diff --git a/src/compute-plane-services/nvca/internal/miniservice/status_test.go b/src/compute-plane-services/nvca/internal/miniservice/status_test.go index 4993212d4..49cf59c73 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/status_test.go +++ b/src/compute-plane-services/nvca/internal/miniservice/status_test.go @@ -34,6 +34,8 @@ import ( nvcak8sutil "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/internal/util/k8sutil" "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1" + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common" + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/function" ) // testTimeConfig creates a TimeConfig with tighter intervals for faster testing. @@ -676,7 +678,7 @@ func Test_doTerminalTaskStatus(t *testing.T) { currentTime: baseTime.Add(10 * time.Minute), expectedPhase: "", expectedConditionType: v1alpha1.MiniServiceConditionObjectsHealthy, - expectedConditionReason: v1alpha1.MiniServiceStatusReasonDegradedWorkerPods, + expectedConditionReason: v1alpha1.MiniServiceStatusReasonDegradedWorker, expectedRequeue: false, expectedError: true, expectedTerminalError: true, @@ -1423,3 +1425,147 @@ func Test_collectObjectIDs(t *testing.T) { }) } } + +func TestGetFunctionWorkerContainerStatus(t *testing.T) { + utilsStatus := corev1.ContainerStatus{ + Name: common.UtilsContainerName, + Ready: true, + RestartCount: 2, + Image: "utils:latest", + } + llmStatus := corev1.ContainerStatus{ + Name: function.LLMWorkerContainerName, + Ready: true, + } + + tests := []struct { + name string + pod *corev1.Pod + wantStatus corev1.ContainerStatus + wantPresent bool + wantErr bool + wantErrMessage string + }{ + { + name: "utils worker container status present", + pod: &corev1.Pod{ + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{utilsStatus}, + }, + }, + wantStatus: utilsStatus, + wantPresent: true, + }, + { + name: "llm worker container status present", + pod: &corev1.Pod{ + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{llmStatus}, + }, + }, + wantStatus: llmStatus, + wantPresent: true, + }, + { + name: "worker container status present among sidecars", + pod: &corev1.Pod{ + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{ + {Name: "sidecar-a", Ready: true}, + utilsStatus, + {Name: "sidecar-b", Ready: false}, + }, + }, + }, + wantStatus: utilsStatus, + wantPresent: true, + }, + { + name: "status takes precedence over spec when both present", + pod: &corev1.Pod{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: common.UtilsContainerName}}, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{utilsStatus}, + }, + }, + wantStatus: utilsStatus, + wantPresent: true, + }, + { + name: "utils worker in spec but status not yet available", + pod: &corev1.Pod{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + {Name: "sidecar-a"}, + {Name: common.UtilsContainerName}, + }, + }, + }, + wantStatus: corev1.ContainerStatus{}, + wantPresent: false, + }, + { + name: "llm worker in spec but status not yet available", + pod: &corev1.Pod{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: function.LLMWorkerContainerName}}, + }, + }, + wantStatus: corev1.ContainerStatus{}, + wantPresent: false, + }, + { + name: "non-worker status present but worker still only in spec", + pod: &corev1.Pod{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: common.UtilsContainerName}}, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{{Name: "sidecar-a", Ready: true}}, + }, + }, + wantStatus: corev1.ContainerStatus{}, + wantPresent: false, + }, + { + name: "worker container not found in spec or status", + pod: &corev1.Pod{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "sidecar-a"}}, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{{Name: "sidecar-a"}}, + }, + }, + wantStatus: corev1.ContainerStatus{}, + wantPresent: false, + wantErr: true, + wantErrMessage: "terminal error: worker container not found", + }, + { + name: "empty pod returns terminal error", + pod: &corev1.Pod{}, + wantStatus: corev1.ContainerStatus{}, + wantPresent: false, + wantErr: true, + wantErrMessage: "terminal error: worker container not found", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, present, err := getFunctionWorkerContainerStatus(tt.pod) + if tt.wantErr { + // The error must be terminal so the reconciler does not requeue; the + // "terminal error: " prefix is added by reconcile.TerminalError. + assert.EqualError(t, err, tt.wantErrMessage) + } else { + require.NoError(t, err) + } + assert.Equal(t, tt.wantPresent, present) + assert.Equal(t, tt.wantStatus, got) + }) + } +} diff --git a/src/compute-plane-services/nvca/internal/util/k8sutil/pod.go b/src/compute-plane-services/nvca/internal/util/k8sutil/pod.go index c48905ef0..f1610c3ea 100644 --- a/src/compute-plane-services/nvca/internal/util/k8sutil/pod.go +++ b/src/compute-plane-services/nvca/internal/util/k8sutil/pod.go @@ -116,14 +116,27 @@ func IsPodAdmissionRejected(ps corev1.PodStatus) (bool, string) { return false, "" } -func ImagePullIssuesReported(ps corev1.PodStatus) (string, corev1.ContainerStateWaiting, bool) { +type ContainerImagePullIssues struct { + Name string + Image string + Message string + Reason string +} + +func ImagePullIssuesReported(ps corev1.PodStatus) ([]ContainerImagePullIssues, bool) { + imagePullIssues := []ContainerImagePullIssues{} for _, cs := range append(ps.InitContainerStatuses, ps.ContainerStatuses...) { if cs.State.Waiting != nil && (strings.EqualFold(cs.State.Waiting.Reason, ImagePullIssueReason) || strings.EqualFold(cs.State.Waiting.Reason, ImagePullIssueAlternateReason)) { - return cs.Image, *cs.State.Waiting, true + imagePullIssues = append(imagePullIssues, ContainerImagePullIssues{ + Name: cs.Name, + Image: cs.Image, + Message: cs.State.Waiting.Message, + Reason: cs.State.Waiting.Reason, + }) } } - return "", corev1.ContainerStateWaiting{}, false + return imagePullIssues, len(imagePullIssues) > 0 } func ParseImageRegistry(imageTag string) (reg string) { @@ -140,7 +153,18 @@ func ParseImageRegistry(imageTag string) (reg string) { return reg } -func IsPodDegraded(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, string) { +type PodDegraded struct { + Containers []ContainerDegraded + Reason string +} + +type ContainerDegraded struct { + Name string + Reason string + Message string +} + +func IsPodDegraded(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (PodDegraded, bool) { status := pod.Status containersNotReady := false podNotReady := false @@ -169,16 +193,28 @@ func IsPodDegraded(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, string) { } if containersNotReady && podNotReady && podInitialized { + degradedContainers := []ContainerDegraded{} isNonRestartableContainerTerminated := pod.Spec.RestartPolicy == corev1.RestartPolicyNever && slices.IndexFunc(status.ContainerStatuses, func(cs corev1.ContainerStatus) bool { - return cs.State.Terminated != nil && cs.State.Terminated.ExitCode != 0 + if cs.State.Terminated != nil && cs.State.Terminated.ExitCode != 0 { + degradedContainers = append(degradedContainers, ContainerDegraded{ + Name: cs.Name, + Reason: cs.State.Terminated.Reason, + Message: cs.State.Terminated.Message, + }) + return true + } + return false }) != -1 isWDPPassed := !podNotReadyCond.LastTransitionTime.IsZero() && time.Since(podNotReadyCond.LastTransitionTime.Time) > wdp if isNonRestartableContainerTerminated || isWDPPassed { - return true, containersNotReadyReason + return PodDegraded{ + Containers: degradedContainers, + Reason: containersNotReadyReason, + }, true } } - return false, "" + return PodDegraded{}, false } func IsPodInInitialStartup(status corev1.PodStatus) bool { @@ -196,7 +232,20 @@ func IsPodInInitialStartup(status corev1.PodStatus) bool { return false } -func IsPodStuckInitializing(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, string) { +const PodKey = "__pod" + +type PodStuckInitializing struct { + Containers []ContainerStuckInitializing + Reason string +} + +type ContainerStuckInitializing struct { + Name string + Reason string + Message string +} + +func IsPodStuckInitializing(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (PodStuckInitializing, bool) { podStatus := pod.Status initConditionTrue := false for _, cond := range podStatus.Conditions { @@ -205,18 +254,32 @@ func IsPodStuckInitializing(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, s break } } + stuckContainers := []ContainerStuckInitializing{} if initConditionTrue { for _, containerStatus := range podStatus.ContainerStatuses { if containerStatus.RestartCount >= RestartCountToFailInstance && IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.PodLaunchThresholdSecondsOnFailedRestarts) { reason := "ContainerInRestartLoop" + message := "" switch { case containerStatus.LastTerminationState.Terminated != nil: reason = containerStatus.LastTerminationState.Terminated.Reason + message = containerStatus.LastTerminationState.Terminated.Message case containerStatus.LastTerminationState.Waiting != nil: reason = containerStatus.LastTerminationState.Waiting.Reason + message = containerStatus.LastTerminationState.Waiting.Message } - return true, reason + stuckContainers = append(stuckContainers, ContainerStuckInitializing{ + Name: containerStatus.Name, + Reason: reason, + Message: message, + }) + } + if len(stuckContainers) > 0 { + return PodStuckInitializing{ + Containers: stuckContainers, + Reason: "ContainersInRestartLoop", + }, true } } } else { @@ -224,13 +287,26 @@ func IsPodStuckInitializing(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, s if containerStatus.RestartCount >= RestartCountToFailInstance && IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.PodLaunchThresholdSecondsOnFailedRestarts) { reason := "InitContainerInRestartLoop" + message := "" switch { case containerStatus.LastTerminationState.Terminated != nil: reason = containerStatus.LastTerminationState.Terminated.Reason + message = containerStatus.LastTerminationState.Terminated.Message case containerStatus.LastTerminationState.Waiting != nil: reason = containerStatus.LastTerminationState.Waiting.Reason + message = containerStatus.LastTerminationState.Waiting.Message } - return true, reason + stuckContainers = append(stuckContainers, ContainerStuckInitializing{ + Name: containerStatus.Name, + Reason: reason, + Message: message, + }) + } + if len(stuckContainers) > 0 { + return PodStuckInitializing{ + Containers: stuckContainers, + Reason: "InitContainersInRestartLoop", + }, true } } } @@ -238,10 +314,13 @@ func IsPodStuckInitializing(pod *corev1.Pod, k8sTimeConfig *TimeConfig) (bool, s // Just check if the Pod's init container is stuck for > 120minutes if !initConditionTrue { if IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.PodLaunchThresholdMinutesOnInitFailure) { - return true, "ContainerStuckAfterThreshold" + return PodStuckInitializing{ + Containers: stuckContainers, + Reason: "ContainerStuckAfterThreshold", + }, true } } - return false, "" + return PodStuckInitializing{}, false } func IsTimeSincePodLaunchedLaterThan(pod *corev1.Pod, dtc time.Duration) bool { diff --git a/src/compute-plane-services/nvca/internal/util/k8sutil/pod_test.go b/src/compute-plane-services/nvca/internal/util/k8sutil/pod_test.go index ce7a4c29a..ea8d7dbd2 100644 --- a/src/compute-plane-services/nvca/internal/util/k8sutil/pod_test.go +++ b/src/compute-plane-services/nvca/internal/util/k8sutil/pod_test.go @@ -482,8 +482,8 @@ func TestIsPodDegraded(t *testing.T) { workerDegradationPeriod time.Duration startupTimeoutPeriod time.Duration restartPolicy corev1.RestartPolicy + want PodDegraded wantDegraded bool - wantReason string }{ { name: "not degraded - all conditions healthy", @@ -510,8 +510,8 @@ func TestIsPodDegraded(t *testing.T) { }, workerDegradationPeriod: time.Minute, startupTimeoutPeriod: time.Minute, + want: PodDegraded{}, wantDegraded: false, - wantReason: "", }, { name: "degraded - containers not ready and pod not ready beyond threshold", @@ -549,8 +549,11 @@ func TestIsPodDegraded(t *testing.T) { }, workerDegradationPeriod: time.Hour, startupTimeoutPeriod: time.Hour, - wantDegraded: true, - wantReason: "ContainersNotReady", + want: PodDegraded{ + Containers: []ContainerDegraded{}, + Reason: "ContainersNotReady", + }, + wantDegraded: true, }, { name: "not degraded - within startup timeout period", @@ -588,8 +591,8 @@ func TestIsPodDegraded(t *testing.T) { }, workerDegradationPeriod: time.Hour, startupTimeoutPeriod: time.Hour, + want: PodDegraded{}, wantDegraded: false, - wantReason: "", }, { name: "degraded - within startup timeout period but restart policy is never", @@ -628,8 +631,11 @@ func TestIsPodDegraded(t *testing.T) { workerDegradationPeriod: time.Hour, startupTimeoutPeriod: time.Hour, restartPolicy: corev1.RestartPolicyNever, - wantDegraded: true, - wantReason: "ContainersNotReady", + want: PodDegraded{ + Containers: []ContainerDegraded{{Name: "foo"}}, + Reason: "ContainersNotReady", + }, + wantDegraded: true, }, } @@ -639,12 +645,12 @@ func TestIsPodDegraded(t *testing.T) { WorkerDegradationTimeout: tt.workerDegradationPeriod, WorkerStartupTimeout: tt.startupTimeoutPeriod, } - degraded, reason := IsPodDegraded(&corev1.Pod{ + got, degraded := IsPodDegraded(&corev1.Pod{ Spec: corev1.PodSpec{RestartPolicy: tt.restartPolicy}, Status: tt.status, }, k8sTimeConfig) assert.Equal(t, tt.wantDegraded, degraded) - assert.Equal(t, tt.wantReason, reason) + assert.Equal(t, tt.want, got) }) } } @@ -706,9 +712,15 @@ func TestImagePullIssuesReported(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - _, state, hasIssue := ImagePullIssuesReported(tt.status) + imagePullIssues, hasIssue := ImagePullIssuesReported(tt.status) assert.Equal(t, tt.wantIssue, hasIssue) - assert.Equal(t, tt.wantReason, state.Reason) + if tt.wantIssue { + if assert.Len(t, imagePullIssues, 1) { + assert.Equal(t, tt.wantReason, imagePullIssues[0].Reason) + } + } else { + assert.Empty(t, imagePullIssues) + } }) } } @@ -836,10 +848,10 @@ func TestIsPodStuckInitializing(t *testing.T) { defaultTimeConfig := (&TimeConfig{}).Complete() now := time.Now() tests := []struct { - name string - podStatus corev1.PodStatus - wantStuck bool - wantReason string + name string + podStatus corev1.PodStatus + want PodStuckInitializing + wantStuck bool }{ { name: "pod successfully initialized", @@ -857,8 +869,8 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: false, - wantReason: "", + want: PodStuckInitializing{}, + wantStuck: false, }, { name: "container in restart loop", @@ -881,8 +893,11 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: true, - wantReason: "CrashLoopBackOff", + want: PodStuckInitializing{ + Containers: []ContainerStuckInitializing{{Reason: "CrashLoopBackOff"}}, + Reason: "ContainersInRestartLoop", + }, + wantStuck: true, }, { name: "init container in restart loop", @@ -905,8 +920,11 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: true, - wantReason: "Error", + want: PodStuckInitializing{ + Containers: []ContainerStuckInitializing{{Reason: "Error"}}, + Reason: "InitContainersInRestartLoop", + }, + wantStuck: true, }, { name: "init container with waiting state", @@ -929,8 +947,11 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: true, - wantReason: "ImagePullBackOff", + want: PodStuckInitializing{ + Containers: []ContainerStuckInitializing{{Reason: "ImagePullBackOff"}}, + Reason: "InitContainersInRestartLoop", + }, + wantStuck: true, }, { name: "pod stuck in initialization beyond threshold", @@ -948,8 +969,11 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: true, - wantReason: "ContainerStuckAfterThreshold", + want: PodStuckInitializing{ + Containers: []ContainerStuckInitializing{}, + Reason: "ContainerStuckAfterThreshold", + }, + wantStuck: true, }, { name: "pod still initializing within threshold", @@ -967,8 +991,8 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: false, - wantReason: "", + want: PodStuckInitializing{}, + wantStuck: false, }, { name: "container restarts within threshold", @@ -986,8 +1010,8 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: false, - wantReason: "", + want: PodStuckInitializing{}, + wantStuck: false, }, { name: "no start time", @@ -1004,16 +1028,16 @@ func TestIsPodStuckInitializing(t *testing.T) { }, }, }, - wantStuck: false, - wantReason: "", + want: PodStuckInitializing{}, + wantStuck: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - stuck, reason := IsPodStuckInitializing(&corev1.Pod{Status: tt.podStatus}, defaultTimeConfig) + got, stuck := IsPodStuckInitializing(&corev1.Pod{Status: tt.podStatus}, defaultTimeConfig) assert.Equal(t, tt.wantStuck, stuck, "stuck status mismatch") - assert.Equal(t, tt.wantReason, reason, "reason mismatch") + assert.Equal(t, tt.want, got, "result mismatch") }) } } diff --git a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/generated.openapi.go b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/generated.openapi.go index 5c627e889..f46676e94 100644 --- a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/generated.openapi.go +++ b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/generated.openapi.go @@ -208,12 +208,18 @@ func schema_pkg_apis_nvca_v1alpha1_MiniServiceSpec(ref common.ReferenceCallback) Ref: ref("github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common.HelmConfig"), }, }, + "workloadConfig": { + SchemaProps: spec.SchemaProps{ + Description: "WorkloadConfig holds workload-specific configuration decoded from the user-provided workload config ConfigMap in the rendered Helm chart. It is populated by the controller during installation and read on subsequent reconciles.", + Ref: ref("github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1.WorkloadConfig"), + }, + }, }, Required: []string{"namespace", "icmsRequestName", "helmChartConfig"}, }, }, Dependencies: []string{ - "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common.HelmConfig"}, + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1.WorkloadConfig", "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common.HelmConfig"}, } } diff --git a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_status_types.go b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_status_types.go index d426590eb..ed9bed9a3 100644 --- a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_status_types.go +++ b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_status_types.go @@ -35,6 +35,7 @@ const ( MiniServiceConditionCacheSuccessful = "CacheSuccessful" MiniServiceConditionInstallSuccessful = "InstallSuccessful" MiniServiceConditionObjectsHealthy = "ObjectsHealthy" + MiniServiceConditionWorkersHealthy = "WorkersHealthy" MiniServiceConditionCleanupSuccessful = "CleanupSuccessful" ) @@ -51,8 +52,10 @@ const ( MiniServiceStatusReasonObjectsFailed = "ObjectsFailed" // When some objects failed to deploy with backoff. MiniServiceStatusReasonObjectsFailedWithinBackoffTimeout = "ObjectsFailedWithinBackoffTimeout" - // When some pods are degraded. - MiniServiceStatusReasonDegradedWorkerPods = "DegradedWorkerPods" + // When some workload pods are degraded. + MiniServiceStatusReasonDegradedWorkload = "DegradedWorkload" + // When some worker pods are degraded. + MiniServiceStatusReasonDegradedWorker = "DegradedWorker" // When some objects were pending for too long. MiniServiceStatusReasonPendingTimeout = "ObjectsTimedOutPending" // Waiting on objects to be ready. @@ -67,7 +70,8 @@ var MiniServiceStatusBadReasons = map[string]bool{ MiniServiceStatusReasonCachingFailed: true, MiniServiceStatusReasonReValResultInvalid: true, MiniServiceStatusReasonObjectsFailed: true, - MiniServiceStatusReasonDegradedWorkerPods: true, + MiniServiceStatusReasonDegradedWorkload: true, + MiniServiceStatusReasonDegradedWorker: true, MiniServiceStatusReasonPendingTimeout: true, MiniServiceStatusReasonObjectStatusErrors: true, MiniServiceStatusReasonUnexpectedInstallError: true, diff --git a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_types.go b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_types.go index e9ac5eda0..0b6fbafca 100644 --- a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_types.go +++ b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_types.go @@ -43,6 +43,29 @@ type MiniServiceSpec struct { ICMSRequestName string `json:"icmsRequestName"` HelmChartConfig common.HelmConfig `json:"helmChartConfig"` + + // WorkloadConfig holds workload-specific configuration decoded from the user-provided + // workload config ConfigMap in the rendered Helm chart. It is populated by the controller + // during installation and read on subsequent reconciles. + // +optional + WorkloadConfig *WorkloadConfig `json:"workloadConfig,omitempty"` +} + +// WorkloadConfig is workload-specific configuration a chart author may supply through the +// workload config ConfigMap. It is intended to grow with additional workload-scoped settings +// beyond feature flags. +type WorkloadConfig struct { + // FeatureFlags toggles per-workload behaviors, keyed by feature flag name. + // +optional + FeatureFlags map[string]bool `json:"featureFlags,omitempty"` +} + +// IsFeatureFlagEnabled reports whether the named feature flag is enabled. +func (c *WorkloadConfig) IsFeatureFlagEnabled(key string) bool { + if c == nil || c.FeatureFlags == nil { + return false + } + return c.FeatureFlags[key] } type LocalObjectReference struct { diff --git a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/zz_generated.deepcopy.go b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/zz_generated.deepcopy.go index 60da190a2..9c9ca2eba 100644 --- a/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/zz_generated.deepcopy.go +++ b/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/zz_generated.deepcopy.go @@ -108,6 +108,11 @@ func (in *MiniServiceList) DeepCopyObject() runtime.Object { func (in *MiniServiceSpec) DeepCopyInto(out *MiniServiceSpec) { *out = *in in.HelmChartConfig.DeepCopyInto(&out.HelmChartConfig) + if in.WorkloadConfig != nil { + in, out := &in.WorkloadConfig, &out.WorkloadConfig + *out = new(WorkloadConfig) + (*in).DeepCopyInto(*out) + } return } @@ -196,3 +201,26 @@ func (in *ResourceStatus) DeepCopy() *ResourceStatus { in.DeepCopyInto(out) return out } + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *WorkloadConfig) DeepCopyInto(out *WorkloadConfig) { + *out = *in + if in.FeatureFlags != nil { + in, out := &in.FeatureFlags, &out.FeatureFlags + *out = make(map[string]bool, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkloadConfig. +func (in *WorkloadConfig) DeepCopy() *WorkloadConfig { + if in == nil { + return nil + } + out := new(WorkloadConfig) + in.DeepCopyInto(out) + return out +} diff --git a/src/compute-plane-services/nvca/pkg/featureflag/BUILD.bazel b/src/compute-plane-services/nvca/pkg/featureflag/BUILD.bazel index 1dd605b66..12257fbae 100644 --- a/src/compute-plane-services/nvca/pkg/featureflag/BUILD.bazel +++ b/src/compute-plane-services/nvca/pkg/featureflag/BUILD.bazel @@ -8,6 +8,7 @@ go_library( srcs = [ "attribute.go", "featureflag.go", + "featureflag_workload.go", "fetcher.go", "persistentstorage.go", "sharedstorage.go", @@ -16,10 +17,13 @@ go_library( visibility = ["//visibility:public"], deps = [ "//pkg/apis/nvca/v1:nvca", + "//pkg/apis/nvca/v1alpha1", "//vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/core", + "//vendor/github.com/go-logr/logr", "//vendor/github.com/urfave/cli/v2:cli", "//vendor/k8s.io/api/core/v1:core", "//vendor/k8s.io/apimachinery/pkg/api/resource", + "//vendor/sigs.k8s.io/yaml", ], ) @@ -34,6 +38,7 @@ go_test( srcs = [ "attribute_test.go", "featureflag_test.go", + "featureflag_workload_test.go", "fetcher_test.go", "persistentstorage_test.go", "sharedstorage_test.go", @@ -41,6 +46,8 @@ go_test( embed = [":featureflag"], deps = [ "//pkg/apis/nvca/v1:nvca", + "//pkg/apis/nvca/v1alpha1", + "//vendor/github.com/go-logr/logr", "//vendor/github.com/stretchr/testify/assert", "//vendor/github.com/stretchr/testify/require", "//vendor/k8s.io/api/core/v1:core", diff --git a/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload.go b/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload.go new file mode 100644 index 000000000..5a576ec68 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload.go @@ -0,0 +1,79 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package featureflag + +import ( + "context" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1" + "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/yaml" +) + +const ( + // WorkloadConfigConfigMapName is the fixed name of the ConfigMap a chart author may include + // to supply workload-specific configuration. The ConfigMap is never created on-cluster; it is + // read from the ReVal-rendered objects by the MiniService controller and then dropped from the + // set of objects that are applied. + WorkloadConfigConfigMapName = "nvcf-workload-config" + + // WorkloadConfigDataKey is the ConfigMap data key holding the workload config YAML document. + WorkloadConfigDataKey = "config.yaml" +) + +// Workload feature flag keys recognized under the workload config featureFlags map. +const ( + // StatusByWorkerReadiness directs the MiniService controller to only consider worker + // container readiness when determining MiniService readiness, rather than aggressively + // accounting for the health of all workload objects. + StatusByWorkerReadiness = "StatusByWorkerReadiness" +) + +// workloadFeatureFlagKeys is the set of recognized workload feature flag keys. Unrecognized +// keys are ignored (with a warning) by DecodeWorkloadConfig. +var workloadFeatureFlagKeys = map[string]struct{}{ + StatusByWorkerReadiness: {}, +} + +// DecodeWorkloadConfig decodes the workload config ConfigMap into a WorkloadConfig. The +// config is read from the WorkloadConfigDataKey data key as a YAML document. Unrecognized +// feature flags are dropped and logged as a warning. A nil ConfigMap yields a zero config. +func DecodeWorkloadConfig(ctx context.Context, log logr.Logger, cm *corev1.ConfigMap) (*v1alpha1.WorkloadConfig, error) { + if cm == nil { + return nil, nil + } + + raw, ok := cm.Data[WorkloadConfigDataKey] + if !ok || raw == "" { + log.Info("Ignoring empty workload config in ConfigMap %q", WorkloadConfigConfigMapName) + return nil, nil + } + var cfg v1alpha1.WorkloadConfig + if err := yaml.Unmarshal([]byte(raw), &cfg); err != nil { + return nil, err + } + + for key := range cfg.FeatureFlags { + if _, known := workloadFeatureFlagKeys[key]; !known { + log.Info("Ignoring unknown workload feature flag %q in ConfigMap %q", key, WorkloadConfigConfigMapName) + delete(cfg.FeatureFlags, key) + } + } + return &cfg, nil +} diff --git a/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload_test.go b/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload_test.go new file mode 100644 index 000000000..bd5a9d40a --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload_test.go @@ -0,0 +1,128 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package featureflag + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1" +) + +func workloadConfigCM(configYAML string) *corev1.ConfigMap { + return &corev1.ConfigMap{ + Data: map[string]string{WorkloadConfigDataKey: configYAML}, + } +} + +func TestDecodeWorkloadConfig(t *testing.T) { + tests := []struct { + name string + cm *corev1.ConfigMap + want *v1alpha1.WorkloadConfig + wantErr bool + }{ + { + name: "nil ConfigMap yields nil config", + cm: nil, + want: nil, + }, + { + name: "ConfigMap without config.yaml key yields nil config", + cm: &corev1.ConfigMap{Data: map[string]string{"other": "x"}}, + want: nil, + }, + { + name: "empty config.yaml yields nil config", + cm: workloadConfigCM(""), + want: nil, + }, + { + name: "null config.yaml yields nil config", + cm: workloadConfigCM("null"), + want: &v1alpha1.WorkloadConfig{}, + }, + { + name: "feature flag enabled", + cm: workloadConfigCM("featureFlags:\n " + StatusByWorkerReadiness + ": true\n"), + want: &v1alpha1.WorkloadConfig{ + FeatureFlags: map[string]bool{StatusByWorkerReadiness: true}, + }, + }, + { + name: "feature flag disabled", + cm: workloadConfigCM("featureFlags:\n " + StatusByWorkerReadiness + ": false\n"), + want: &v1alpha1.WorkloadConfig{ + FeatureFlags: map[string]bool{StatusByWorkerReadiness: false}, + }, + }, + { + name: "unknown feature flag is dropped", + cm: workloadConfigCM("featureFlags:\n " + StatusByWorkerReadiness + ": true\n SomeUnknownFlag: true\n"), + want: &v1alpha1.WorkloadConfig{ + FeatureFlags: map[string]bool{StatusByWorkerReadiness: true}, + }, + }, + { + name: "invalid yaml returns error", + cm: workloadConfigCM("featureFlags: [not-a-map"), + wantErr: true, + }, + { + name: "wrong type for feature flag value returns error", + cm: workloadConfigCM("featureFlags:\n " + StatusByWorkerReadiness + ": notabool\n"), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := DecodeWorkloadConfig(context.Background(), logr.Discard(), tt.cm) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestWorkloadConfigConfigMapName(t *testing.T) { + assert.Equal(t, "nvcf-workload-config", WorkloadConfigConfigMapName) + assert.Equal(t, "config.yaml", WorkloadConfigDataKey) +} + +func TestWorkloadConfigIsFeatureFlagEnabled(t *testing.T) { + var nilCfg *v1alpha1.WorkloadConfig + assert.False(t, nilCfg.IsFeatureFlagEnabled(StatusByWorkerReadiness)) + + empty := &v1alpha1.WorkloadConfig{} + assert.False(t, empty.IsFeatureFlagEnabled(StatusByWorkerReadiness)) + + cfg := &v1alpha1.WorkloadConfig{ + FeatureFlags: map[string]bool{StatusByWorkerReadiness: true}, + } + assert.True(t, cfg.IsFeatureFlagEnabled(StatusByWorkerReadiness)) + assert.False(t, cfg.IsFeatureFlagEnabled("SomeOtherFlag")) +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go index b07d2e24f..d6d966208 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go @@ -1346,8 +1346,10 @@ func (c K8sComputeBackend) GetErroredPodLogs(ctx context.Context, pod *corev1.Po log := core.GetLogger(ctx) // Image pull issues should be directly reported to users for debugging. - if _, state, ok := k8sutil.ImagePullIssuesReported(pod.Status); ok { - prepend = fmt.Sprintf("%s (%s)", prepend, state.Message) + if imagePullIssues, ok := k8sutil.ImagePullIssuesReported(pod.Status); ok { + for _, imagePullIssue := range imagePullIssues { + prepend = fmt.Sprintf("%s (%s)\n", prepend, imagePullIssue.Message) + } } totalBytesWritten := int64(0) @@ -1696,9 +1698,11 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForCreatePodRequest(ctx context. string(is), st.LastReportedStatus, st.LastReportedTimestamp) { // Only increment metric once. - if imgTag, _, ok := k8sutil.ImagePullIssuesReported(p.Status); ok { - reg := k8sutil.ParseImageRegistry(imgTag) - metrics.ImagePullIssueTotal.WithLabelValues(metrics.WithDefaultLabelValues(reg)...).Inc() + if imagePullIssues, ok := k8sutil.ImagePullIssuesReported(p.Status); ok { + for _, imagePullIssue := range imagePullIssues { + reg := k8sutil.ParseImageRegistry(imagePullIssue.Image) + metrics.ImagePullIssueTotal.WithLabelValues(metrics.WithDefaultLabelValues(reg)...).Inc() + } } // Record workload result metric on terminal state transitions. @@ -1810,7 +1814,7 @@ func podPhaseToInstanceState(pod *corev1.Pod, k8sTimeConfig *k8sutil.TimeConfig) return types.ICMSInstanceFailedCreateContainerError, msg } if k8sutil.IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.MaxImagePullErrorThreshold) { - if _, _, ok := nvcak8sutil.ImagePullIssuesReported(ps); ok { + if _, ok := nvcak8sutil.ImagePullIssuesReported(ps); ok { return types.ICMSInstanceFailedImagePullIssues, "" } } @@ -1840,9 +1844,9 @@ func podPhaseToInstanceState(pod *corev1.Pod, k8sTimeConfig *k8sutil.TimeConfig) if stuck { return reason, "" } - degraded, msg := nvcak8sutil.IsPodDegraded(pod, k8sTimeConfig) + degradedStatus, degraded := nvcak8sutil.IsPodDegraded(pod, k8sTimeConfig) if degraded { - return types.ICMSInstanceDegradedWorker, msg + return types.ICMSInstanceDegradedWorker, degradedStatus.Reason } return types.ICMSInstanceStarted, "" case corev1.PodSucceeded: @@ -2083,7 +2087,7 @@ func podPhaseToAggregatedInstanceStatus(p *corev1.Pod, k8sTimeConfig *k8sutil.Ti if stuck, _ := IsPodStuckInitializing(p, k8sTimeConfig); stuck { return AggregatedInstanceStatusFailed } - if degraded, _ := nvcak8sutil.IsPodDegraded(p, k8sTimeConfig); degraded { + if _, degraded := nvcak8sutil.IsPodDegraded(p, k8sTimeConfig); degraded { return AggregatedInstanceStatusFailed } return AggregatedInstanceStatusPending @@ -2102,7 +2106,7 @@ func podPhaseToAggregatedInstanceStatus(p *corev1.Pod, k8sTimeConfig *k8sutil.Ti default: if nvcak8sutil.IsPodScheduled(ps) { overImagePullTimeout := k8sutil.IsTimeSincePodLaunchedLaterThan(p, k8sTimeConfig.MaxImagePullErrorThreshold) - if _, _, ok := nvcak8sutil.ImagePullIssuesReported(ps); overImagePullTimeout && ok { + if _, ok := nvcak8sutil.ImagePullIssuesReported(ps); overImagePullTimeout && ok { return AggregatedInstanceStatusFailed } if stuck, _ := IsPodStuckInitializing(p, k8sTimeConfig); stuck { diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go index 1084e9fec..36ffc709a 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go @@ -268,7 +268,8 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex if !v1alpha1.MiniServiceStatusBadReasons[relCond.Reason] { continue } - if relCond.Reason == v1alpha1.MiniServiceStatusReasonDegradedWorkerPods && + if (relCond.Reason == v1alpha1.MiniServiceStatusReasonDegradedWorker || + relCond.Reason == v1alpha1.MiniServiceStatusReasonDegradedWorkload) && !c.bk8s.autoPurgeDegradedWorkers { purgeInstance = false } @@ -353,13 +354,15 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex // Only increment metric once for each image registry with issues. imgRegSet := sets.New[string]() for _, pod := range pods { - if imgTag, _, ok := k8sutil.ImagePullIssuesReported(pod.Status); ok { - reg := k8sutil.ParseImageRegistry(imgTag) - if imgRegSet.Has(reg) { - continue + if imagePullIssues, ok := k8sutil.ImagePullIssuesReported(pod.Status); ok { + for _, imagePullIssue := range imagePullIssues { + reg := k8sutil.ParseImageRegistry(imagePullIssue.Image) + if imgRegSet.Has(reg) { + continue + } + metrics.ImagePullIssueTotal.WithLabelValues(metrics.WithDefaultLabelValues(reg)...).Inc() + imgRegSet.Insert(reg) } - metrics.ImagePullIssueTotal.WithLabelValues(metrics.WithDefaultLabelValues(reg)...).Inc() - imgRegSet.Insert(reg) } } diff --git a/src/compute-plane-services/nvca/pkg/storage/modelcache.go b/src/compute-plane-services/nvca/pkg/storage/modelcache.go index dea6329f9..b43e7654a 100644 --- a/src/compute-plane-services/nvca/pkg/storage/modelcache.go +++ b/src/compute-plane-services/nvca/pkg/storage/modelcache.go @@ -1458,11 +1458,11 @@ func isInitJobPodProgressing(pod *corev1.Pod, k8sTimeConfig *k8sutil.TimeConfig) case corev1.PodPending, corev1.PodUnknown: if k8sutil.IsPodScheduled(ps) { if k8sutil.IsTimeSincePodLaunchedLaterThan(pod, k8sTimeConfig.MaxImagePullErrorThreshold) { - if _, _, hasIssues := k8sutil.ImagePullIssuesReported(ps); hasIssues { + if _, hasIssues := k8sutil.ImagePullIssuesReported(ps); hasIssues { return "image pull issues", false } } - stuck, _ := k8sutil.IsPodStuckInitializing(pod, k8sTimeConfig) + _, stuck := k8sutil.IsPodStuckInitializing(pod, k8sTimeConfig) if stuck { return "init stuck initializing", false }