diff --git a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel b/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel index 0d554c8c2..23eeb1d7d 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel +++ b/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel @@ -22,6 +22,8 @@ go_library( "k8scomputebackend_miniservice.go", "k8scomputebackend_modelcache.go", "k8scomputebackend_task_container.go", + "ledger_event_correlator.go", + "ledger_events.go", "nvsnap_coldstart_gate.go", "nvsnap_coldstart_metrics.go", "nvsnap_controller_start.go", @@ -185,6 +187,8 @@ go_test( "k8scomputebackend_modelcache_test.go", "k8scomputebackend_task_container_test.go", "k8scomputebackend_test.go", + "ledger_event_correlator_test.go", + "ledger_events_test.go", "nvsnap_hook_lookup_test.go", "nvsnap_hook_test.go", "queue_manager_test.go", diff --git a/src/compute-plane-services/nvca/pkg/nvca/agent.go b/src/compute-plane-services/nvca/pkg/nvca/agent.go index 870c05adf..67df76601 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/agent.go +++ b/src/compute-plane-services/nvca/pkg/nvca/agent.go @@ -1670,8 +1670,8 @@ func (a *Agent) PutICMSRequestAcknowledgement(ctx context.Context) error { req.Spec.CreationMsgInfo.InstanceCount, req.Spec.GetTraceContext()) if err != nil { - a.backendk8scache.eventRecorder.Eventf(req, v1.EventTypeWarning, - string(types.EventCategoryInstanceStatusUpdate), "Acknowledgement failed: %v", err) + a.backendk8scache.EmitICMSEventf(req, v1.EventTypeWarning, + string(types.EventCategoryInstanceStatusUpdate), "Acknowledgement failed: %v", nil, err) log.WithError(err).Error("Failed to acknowledge request") // If it has only been five minutes since the request was created, and a 404 is return, retry @@ -1735,8 +1735,8 @@ func (a *Agent) PutICMSRequestAcknowledgement(ctx context.Context) error { if !ackSR(ctx, req) { return } - a.backendk8scache.eventRecorder.Event(req, v1.EventTypeNormal, string(types.EventCategoryInstanceStatusUpdate), - "Request accepted for processing") + a.backendk8scache.EmitICMSEvent(req, v1.EventTypeNormal, string(types.EventCategoryInstanceStatusUpdate), + "Request accepted for processing", nil) // If ACK is successful, purge the message now err = a.queueManager.DeleteCreationMessageV2(ctx, req.Spec.MessageReceipt, req.Spec.CreationMsgInfo.QueueURL) @@ -1796,8 +1796,8 @@ func (a *Agent) putTaskICMSRequestAcknowledgementAfterScheduled( if !ackSR(ctx, req) { return } - a.backendk8scache.eventRecorder.Event(req, v1.EventTypeNormal, string(types.EventCategoryInstanceStatusUpdate), - "Request accepted for processing") + a.backendk8scache.EmitICMSEvent(req, v1.EventTypeNormal, string(types.EventCategoryInstanceStatusUpdate), + "Request accepted for processing", nil) modify := func(ctx context.Context, sr *nvcav2beta1.ICMSRequest) { sr.Status.LastACKTimestamp = &metav1.Time{Time: core.GetCurrentTime(ctx)} @@ -1845,8 +1845,8 @@ func (a *Agent) putTaskICMSRequestAcknowledgementAfterScheduled( } return } - a.backendk8scache.eventRecorder.Event(req, v1.EventTypeNormal, string(types.EventCategoryInstanceCreation), - "Message visibility extended") + a.backendk8scache.EmitICMSEvent(req, v1.EventTypeNormal, string(types.EventCategoryInstanceCreation), + "Message visibility extended", nil) modify := func(ctx context.Context, sr *nvcav2beta1.ICMSRequest) { sr.Status.LastStatusUpdated = &metav1.Time{Time: core.GetCurrentTime(ctx)} @@ -2102,8 +2102,8 @@ func (a *Agent) PostICMSInstanceRequestStatusUpdates(ctx context.Context) error } continue } - a.backendk8scache.eventRecorder.Eventf(req, v1.EventTypeNormal, - string(types.EventCategoryInstanceStatusUpdate), "%v is %v", ru.InstanceID, ruPayload.InstanceState) + a.backendk8scache.EmitICMSEventf(req, v1.EventTypeNormal, + string(types.EventCategoryInstanceStatusUpdate), "%v is %v", &ru, ru.InstanceID, ruPayload.InstanceState) // successfully posted this update so this has to be updated to Status postedInstanceStatus[ru.InstanceID] = getPostedInstanceStatus(ctx, ru) } diff --git a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go index 7a6ffd961..328f73d4c 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go +++ b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go @@ -498,7 +498,11 @@ func (b *BackendK8sCacheBuilder) Start(ctx context.Context) (*BackendK8sCache, < return nil, nil, fmt.Errorf("addSharedClusterNodePublisher is required") } - eventBroadcaster := record.NewBroadcaster() + // Per-instance spam/aggregation keys so multi-instance heartbeats on one + // ICMSRequest keep ledger annotations (see NewLedgerEventCorrelatorOptions). + eventBroadcaster := record.NewBroadcasterWithCorrelatorOptions( + NewLedgerEventCorrelatorOptions(b.periodicInstanceStatusUpdateInterval), + ) // Certain features must be turned on for security in OVC environments. ovcSecEnforcementsEnabled := b.enabledAttrs.Enabled(featureflag.AttrOVCSecurityEnforcements) diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go index 8419c652e..48a8a36ed 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go @@ -475,7 +475,8 @@ func (c K8sComputeBackend) applyFunctionCreationMessage(ctx context.Context, req return c.bk8s.ApplyICMSRequestStatusChange(ctx, req) } - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceCreation), "Creating %v requested instances", instCount) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceCreation), + "Creating %v requested instances", nil, instCount) labelsForReq := nvcatypes.GetLabelsForRequest(req, c.bk8s.featureFlagFetcher) annosForReq := nvcatypes.GetAnnotationsForRequest(req) @@ -674,7 +675,7 @@ func (c K8sComputeBackend) setupContainerModelCaching(ctx context.Context, switch mc { case ModelCachingCompleted: log.Infof("model caching completed, starting worker creation") - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, string(types.EventCategoryModelCaching), "%v ready for instance", roPVCName) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(types.EventCategoryModelCaching), "%v ready for instance", nil, roPVCName) // Modify the pod volume to be that of the ROPVCName mf = func(pod *corev1.Pod) { for id := range pod.Spec.Volumes { @@ -702,8 +703,8 @@ func (c K8sComputeBackend) setupContainerModelCaching(ctx context.Context, } return nil, "", fmt.Errorf("model caching is still in progress") case ModelCachingFailed: - c.bk8s.eventRecorder.Event(req, corev1.EventTypeWarning, - string(types.EventCategoryModelCaching), "Caching setup failed, resort to non-cached workers") + c.bk8s.EmitICMSEvent(req, corev1.EventTypeWarning, + string(types.EventCategoryModelCaching), "Caching setup failed, resort to non-cached workers", nil) log.Warnf("model caching failed, NVCA will create non-cached workers") } return func(*corev1.Pod) {}, "", nil @@ -814,8 +815,8 @@ func (c K8sComputeBackend) doHelmChartStorageRequests(ctx context.Context, case nvcav1new.StorageFailed: switch st.Spec.Type { case nvcav1new.ModelCacheRequest: - c.bk8s.eventRecorder.Event(req, corev1.EventTypeWarning, - string(types.EventCategoryModelCaching), "Caching setup failed, resort to non-cached workers") + c.bk8s.EmitICMSEvent(req, corev1.EventTypeWarning, + string(types.EventCategoryModelCaching), "Caching setup failed, resort to non-cached workers", nil) log.Error("Model cache storage failed, model caching will be disabled") metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventPVCModelCachingError)...).Inc() metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventModelCachingFailed)...).Inc() @@ -1083,8 +1084,8 @@ func (c K8sComputeBackend) CreatePodArtifactInstances(ctx context.Context, pod * LastReportedTimestamp: nil, }) - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, - string(types.EventCategoryInstanceCreation), "Created %v Instance %v", nvcav2beta1.InstanceTypePod, pod.Name) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, + string(types.EventCategoryInstanceCreation), "Created %v Instance %v", instanceUpdate(pod.Name), nvcav2beta1.InstanceTypePod, pod.Name) } if len(newActiveInstances) != 0 { @@ -1122,8 +1123,8 @@ func (c K8sComputeBackend) purgeInstanceID(ctx context.Context, req *nvcav2beta1 ms.Name = id if err := c.clients.HelmV2.Get(ctx, client.ObjectKeyFromObject(ms), ms); err != nil { if !apierrors.IsNotFound(err) { - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeWarning, - string(types.EventCategoryInstanceTermination), "Failed to get instance %v", id) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeWarning, + string(types.EventCategoryInstanceTermination), "Failed to get instance %v", instanceUpdate(id), id) log.WithError(err).Errorf("failed to get miniservice instance %v, for request %v/%v", id, req.Namespace, req.Name) return false @@ -1132,8 +1133,8 @@ func (c K8sComputeBackend) purgeInstanceID(ctx context.Context, req *nvcav2beta1 } else if ms.DeletionTimestamp == nil { if err := c.clients.HelmV2.Delete(ctx, ms); err != nil { if !apierrors.IsNotFound(err) { - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeWarning, - string(types.EventCategoryInstanceTermination), "Failed to stop instance %v", id) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeWarning, + string(types.EventCategoryInstanceTermination), "Failed to stop instance %v", instanceUpdate(id), id) log.WithError(err).Errorf("failed to terminate miniservice instance %v, for request %v/%v", id, req.Namespace, req.Name) return false @@ -1141,8 +1142,8 @@ func (c K8sComputeBackend) purgeInstanceID(ctx context.Context, req *nvcav2beta1 log.Debug("Miniservice not found, report as terminated") } else { log.Debug("Terminated miniservice") - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceTermination), - "Stopped instance %v", id) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceTermination), + "Stopped instance %v", instanceUpdate(id), id) } } @@ -1160,15 +1161,15 @@ func (c K8sComputeBackend) purgeInstanceID(ctx context.Context, req *nvcav2beta1 err := c.clients.K8s.CoreV1().Pods(c.bk8s.podInstanceNamespace).Delete(ctx, id, metav1.DeleteOptions{}) if err != nil && !apierrors.IsNotFound(err) { log.WithError(err).Errorf("failed to terminate instance %v, for request %v/%v", id, req.Namespace, req.Name) - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeWarning, - string(types.EventCategoryInstanceTermination), "Failed to stop instance %v/%v", c.bk8s.podInstanceNamespace, id) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeWarning, + string(types.EventCategoryInstanceTermination), "Failed to stop instance %v/%v", instanceUpdate(id), c.bk8s.podInstanceNamespace, id) return false } else if err != nil && apierrors.IsNotFound(err) { log.Debug("Pod not found, report as terminated") } else { log.Debug("Terminated Pod") - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, - string(types.EventCategoryInstanceTermination), "Stopped instance %v/%v", c.bk8s.podInstanceNamespace, id) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, + string(types.EventCategoryInstanceTermination), "Stopped instance %v/%v", instanceUpdate(id), c.bk8s.podInstanceNamespace, id) } if _, ok := terminatedInstances[id]; !ok { @@ -1611,12 +1612,15 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForCreatePodRequest(ctx context. srUpdateInfo.Payload.HealthInfo.ErrorLog = "Container arguments are malformed: " + errMalformedArgsSubstring } + failureCategory := nvcametrics.ICMSInstanceStateToFailureCategory(srUpdateInfo.Payload.TerminationCause) + srUpdateInfo.Payload.FailureCategory = string(failureCategory) + if m := nvcametrics.FromContext(ctx); m != nil { m.RecordWorkloadStatus( workloadtypes.WorkloadTypeContainer, nvcametrics.ActionToWorkloadKind(req.Spec.Action), workloadtypes.WorkloadStatusFailure, - nvcametrics.ICMSInstanceStateToFailureCategory(srUpdateInfo.Payload.TerminationCause), + failureCategory, ) } @@ -1732,22 +1736,32 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForCreatePodRequest(ctx context. } // Record workload result metric on terminal state transitions. + // Default to the explicit success category so a running transition + // without a metrics provider still stamps failure_category, matching the + // MiniService path which always sets failureCategory before the metric + // call regardless of whether a metrics provider is present; needsPurge + // overrides it below. + failureCategory := workloadtypes.FailureCategoryNone if m := nvcametrics.FromContext(ctx); m != nil { if needsPurge { + failureCategory = nvcametrics.ICMSInstanceStateToFailureCategory(tc) m.RecordWorkloadStatus( workloadtypes.WorkloadTypeContainer, nvcametrics.ActionToWorkloadKind(req.Spec.Action), workloadtypes.WorkloadStatusFailure, - nvcametrics.ICMSInstanceStateToFailureCategory(tc), + failureCategory, ) } else if is == types.ICMSInstanceRunning && st.LastReportedStatus != string(types.ICMSInstanceRunning) { + failureCategory = workloadtypes.FailureCategoryNone m.RecordWorkloadStatus( workloadtypes.WorkloadTypeContainer, nvcametrics.ActionToWorkloadKind(req.Spec.Action), workloadtypes.WorkloadStatusSuccess, - workloadtypes.FailureCategoryNone, + failureCategory, ) } + } else if needsPurge { + failureCategory = nvcametrics.ICMSInstanceStateToFailureCategory(tc) } return types.ICMSRequestUpdateInfo{ @@ -1763,8 +1777,9 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForCreatePodRequest(ctx context. ErrorLog: fPL, ErrorSource: errSource, }, - SystemFailure: string(tc), - InstanceIPs: instanceIPs, + SystemFailure: string(tc), + InstanceIPs: instanceIPs, + FailureCategory: string(failureCategory), }, }, nil } @@ -1929,6 +1944,8 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForTerminationRequest(ctx contex updateInfo.Payload.Status = types.ICMSRequestInstanceTerminatedByService updateInfo.Payload.TerminationCause = types.ICMSInstanceTerminatedServiceMaintenance updateInfo.Payload.SystemFailure = string(types.ICMSInstanceTerminatedServiceMaintenance) + updateInfo.Payload.FailureCategory = string(nvcametrics.ICMSInstanceStateToFailureCategory( + types.ICMSInstanceTerminatedServiceMaintenance)) } icmsRequestUpdates = append(icmsRequestUpdates, updateInfo) 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 6255b69a7..bc0f8b4f5 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go @@ -125,8 +125,8 @@ func (c K8sComputeBackend) applyMiniServiceCreationMessage(ctx context.Context, return err } - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, string(nvcatypes.EventCategoryInstanceCreation), - "Creating %v requested instances", instCount) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(nvcatypes.EventCategoryInstanceCreation), + "Creating %v requested instances", nil, instCount) labelsForReq := nvcatypes.GetLabelsForRequest(req, c.bk8s.featureFlagFetcher) annosForReq := nvcatypes.GetAnnotationsForRequest(req) @@ -167,8 +167,8 @@ func (c K8sComputeBackend) applyMiniServiceCreationMessage(ctx context.Context, } log.Debugf("Successfully created MiniService instance %s", instanceID) - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, - string(nvcatypes.EventCategoryInstanceCreation), "Created %v Instance %v", instance.Type, instance.ID) + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, + string(nvcatypes.EventCategoryInstanceCreation), "Created %v Instance %v", instanceUpdate(instance.ID), instance.Type, instance.ID) // update timestamp only once for InProgress if req.Status.RequestStatus != nvcav2beta1.ICMSRequestStatusInProgress && @@ -221,6 +221,7 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex if c.bk8s.shouldReportInstanceStatusHeartbeat(ctx, req, st.ID, string(nvcatypes.ICMSInstanceTerminated), st.LastReportedStatus, st.LastReportedTimestamp) { log.WithError(err).Warnf("Instance is not running, report it as killed") + failureCategory := nvcametrics.ICMSInstanceStateToFailureCategory(nvcatypes.ICMSInstanceFailedNotFound) updateInfo.Payload = nvcatypes.ICMSInstanceStatusUpdateRequest{ Status: nvcatypes.ICMSRequestInstanceTerminatedByService, InstanceState: nvcatypes.ICMSInstanceTerminated, @@ -228,13 +229,14 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex RequestState: nvcatypes.ICMSInstanceRequestClosed, TerminationCause: nvcatypes.ICMSInstanceFailedNotFound, SystemFailure: string(nvcatypes.ICMSInstanceFailedNotFound), + FailureCategory: string(failureCategory), } if metrics != nil { metrics.RecordWorkloadStatus( workloadtypes.WorkloadTypeHelm, nvcametrics.ActionToWorkloadKind(req.Spec.Action), workloadtypes.WorkloadStatusFailure, - nvcametrics.ICMSInstanceStateToFailureCategory(nvcatypes.ICMSInstanceFailedNotFound), + failureCategory, ) } return updateInfo, nil @@ -397,6 +399,7 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex } } + updateInfo.Payload.FailureCategory = string(failureCategory) return updateInfo, nil } @@ -416,6 +419,7 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex updateInfo.Payload.Action = common.TerminationAction updateInfo.Payload.RequestState = nvcatypes.ICMSInstanceRequestClosed updateInfo.Payload.TerminationCause = storageReqState + updateInfo.Payload.FailureCategory = string(nvcametrics.ICMSInstanceStateToFailureCategory(storageReqState)) // Let the miniservice controller handle top-level resource deletion. if err := c.clients.HelmV2.Delete(ctx, ms); err != nil && !apierrors.IsNotFound(err) { diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_modelcache.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_modelcache.go index e6519914c..a36f84cf1 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_modelcache.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_modelcache.go @@ -175,8 +175,8 @@ func (c K8sComputeBackend) SetupModelCachingForRequest(ctx context.Context, if (err != nil && errors.IsNotFound(err)) || (pvObjList != nil && len(pvObjList.Items) == 0) { err = c.SetupInitCacheJobBlockDevice(ctx, rwPVC, initJob, req) if err != nil { - c.bk8s.eventRecorder.Event(req, v1.EventTypeWarning, - string(types.EventCategoryModelCaching), "failed caching setup, resort to non-caching") + c.bk8s.EmitICMSEvent(req, v1.EventTypeWarning, + string(types.EventCategoryModelCaching), "failed caching setup, resort to non-caching", nil) log.WithError(err).Error("failed SetupInitCacheJobBlockDevice, model caching will be disabled") return ModelCachingFailed, "" } @@ -191,8 +191,8 @@ func (c K8sComputeBackend) SetupModelCachingForRequest(ctx context.Context, if err != nil { log.WithError(err).Error("failed to cleanup ModelCaching resources, needs manual cleanup") } - c.bk8s.eventRecorder.Event(req, v1.EventTypeWarning, - string(types.EventCategoryModelCaching), "failed pvc setup, resort to non-caching") + c.bk8s.EmitICMSEvent(req, v1.EventTypeWarning, + string(types.EventCategoryModelCaching), "failed pvc setup, resort to non-caching", nil) metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventModelCachingFailed)...).Inc() mc = ModelCachingFailed } @@ -206,8 +206,8 @@ func (c K8sComputeBackend) SetupModelCachingForRequest(ctx context.Context, if err != nil { log.WithError(err).Error("failed to cleanup ModelCaching resources, needs manual cleanup") } - c.bk8s.eventRecorder.Eventf(req, v1.EventTypeWarning, - string(types.EventCategoryModelCaching), "%v failed, resort to non-caching", initJob.Name) + c.bk8s.EmitICMSEventf(req, v1.EventTypeWarning, + string(types.EventCategoryModelCaching), "%v failed, resort to non-caching", nil, initJob.Name) reason := c.getInitCacheJobFailureReason(ctx, initJob) metrics.RecordModelCacheResult(modelcachetypes.ResultFailure, reason, string(types.HelmCacheBackendNVMesh)) return ModelCachingFailed, "" @@ -220,8 +220,8 @@ func (c K8sComputeBackend) SetupModelCachingForRequest(ctx context.Context, if err != nil { log.WithError(err).Error("failed to cleanup ModelCaching resources, needs manual cleanup") } - c.bk8s.eventRecorder.Event(req, v1.EventTypeWarning, - string(types.EventCategoryModelCaching), "failed pvc setup, resort to non-caching") + c.bk8s.EmitICMSEvent(req, v1.EventTypeWarning, + string(types.EventCategoryModelCaching), "failed pvc setup, resort to non-caching", nil) metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventModelCachingFailed)...).Inc() metrics.RecordModelCacheResult(modelcachetypes.ResultFailure, modelcachetypes.ReasonPVCSetupFailed, string(types.HelmCacheBackendNVMesh)) mc = ModelCachingFailed @@ -246,8 +246,8 @@ func (c K8sComputeBackend) SetupModelCachingForRequest(ctx context.Context, // TODO: Perform Deeper Cleanup on reconciliation log.WithError(err).Errorf("failed to cleanup ModelCaching resources, needs manual cleanup") } - c.bk8s.eventRecorder.Eventf(req, v1.EventTypeWarning, - string(types.EventCategoryModelCaching), "%v bind failed, resort to non-caching", roPVCName) + c.bk8s.EmitICMSEventf(req, v1.EventTypeWarning, + string(types.EventCategoryModelCaching), "%v bind failed, resort to non-caching", nil, roPVCName) metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventPVCModelCachingError)...).Inc() metrics.EventErrorTotal.WithLabelValues(metrics.WithDefaultLabelValues(EventModelCachingFailed)...).Inc() metrics.RecordModelCacheResult(modelcachetypes.ResultFailure, modelcachetypes.ReasonPVCBindFailed, string(types.HelmCacheBackendNVMesh)) diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_task_container.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_task_container.go index 2b12f09ee..a9fbcb212 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_task_container.go +++ b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_task_container.go @@ -65,9 +65,9 @@ func (c K8sComputeBackend) applyContainerTaskCreationMessage(ctx context.Context metrics := nvcametrics.FromContext(ctx) - c.bk8s.eventRecorder.Eventf(req, + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceCreation), - "Creating %d remaining requested instances", int(instCount)-len(activeInstances), + "Creating %d remaining requested instances", nil, int(instCount)-len(activeInstances), ) labelsForReq := types.GetLabelsForRequest(req, c.bk8s.featureFlagFetcher) @@ -305,9 +305,9 @@ func (c K8sComputeBackend) applyContainerTaskCreationMessage(ctx context.Context }) if podCreated { - c.bk8s.eventRecorder.Eventf(req, corev1.EventTypeNormal, + c.bk8s.EmitICMSEventf(req, corev1.EventTypeNormal, string(types.EventCategoryInstanceCreation), "Created %v Instance %v", - nvcav2beta1.InstanceTypePod, instanceID, + instanceUpdate(instanceID), nvcav2beta1.InstanceTypePod, instanceID, ) } } diff --git a/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator.go b/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator.go new file mode 100644 index 000000000..806ed1c78 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator.go @@ -0,0 +1,117 @@ +/* +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 nvca + +import ( + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/tools/record" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" +) + +// defaultHeartbeatFallback matches config.defaultPeriodicInstanceStatusInterval. +// Used when the agent has not configured a periodic status interval yet. +const defaultHeartbeatFallback = 5 * time.Minute + +// ledgerAggregationDisabledMaxEvents is a MaxEvents value large enough that the +// client-go aggregator never reaches its unique-event threshold, so ledger +// Events are never collapsed into an aggregate. This matters because +// EventAggregate builds the aggregate Event without copying Annotations, which +// would strip the FnDs ledger context. It is used when the heartbeat interval +// is too small to derive a safe sub-heartbeat aggregation window. +const ledgerAggregationDisabledMaxEvents = 1 << 30 + +// NewLedgerEventCorrelatorOptions builds client-go Event correlator options so +// multi-instance ICMSRequest Events do not share one spam budget or collapse +// into annotation-less aggregates. +// +// For a usable heartbeat interval, aggregation MaxInterval is set just below +// the periodic status heartbeat (same config source) so each re-report starts +// a fresh window. When no such window exists (interval <=1s), aggregation is +// disabled outright via MaxEvents: client-go treats MaxIntervalInSeconds==0 as +// its 10m default, so a zero here would re-enable aggregation and drop ledger +// annotations. +func NewLedgerEventCorrelatorOptions(heartbeatInterval time.Duration) record.CorrelatorOptions { + opts := record.CorrelatorOptions{ + KeyFunc: ledgerEventAggregatorKey, + SpamKeyFunc: ledgerEventSpamKey, + } + if secs := ledgerEventAggregateMaxIntervalSeconds(heartbeatInterval); secs > 0 { + opts.MaxIntervalInSeconds = secs + } else { + opts.MaxEvents = ledgerAggregationDisabledMaxEvents + } + return opts +} + +// ledgerEventSpamKey mirrors client-go's default spam key (source + object + +// type) and appends the ledger instance-id annotation when present. +func ledgerEventSpamKey(event *corev1.Event) string { + if event == nil { + return "" + } + return strings.Join([]string{ + event.Source.Component, + event.Source.Host, + event.InvolvedObject.Kind, + event.InvolvedObject.Namespace, + event.InvolvedObject.Name, + string(event.InvolvedObject.UID), + event.InvolvedObject.APIVersion, + event.Type, + eventAnnotation(event, types.LedgerAnnotationInstanceID), + }, "") +} + +// ledgerEventAggregatorKey wraps EventAggregatorByReasonFunc and appends the +// ledger instance-id annotation to the aggregate group key so instances on the +// same ICMSRequest CR do not merge. +func ledgerEventAggregatorKey(event *corev1.Event) (string, string) { + if event == nil { + return "", "" + } + aggregateKey, localKey := record.EventAggregatorByReasonFunc(event) + return aggregateKey + eventAnnotation(event, types.LedgerAnnotationInstanceID), localKey +} + +func eventAnnotation(event *corev1.Event, key string) string { + if event == nil || event.Annotations == nil { + return "" + } + return event.Annotations[key] +} + +// ledgerEventAggregateMaxIntervalSeconds returns the aggregation window in +// whole seconds just below the heartbeat interval so the aggregator resets +// between periodic reports. It returns 0 when the interval is <=1s to signal +// the caller that no safe sub-heartbeat window exists and aggregation should be +// disabled instead (a 0 passed to client-go would fall back to its 10m +// default). A non-positive interval falls back to the default heartbeat. +func ledgerEventAggregateMaxIntervalSeconds(heartbeatInterval time.Duration) int { + if heartbeatInterval <= 0 { + heartbeatInterval = defaultHeartbeatFallback + } + secs := int(heartbeatInterval / time.Second) + if secs <= 1 { + return 0 + } + return secs - 1 +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator_test.go b/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator_test.go new file mode 100644 index 000000000..b4c66b984 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/nvca/ledger_event_correlator_test.go @@ -0,0 +1,186 @@ +/* +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 nvca + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/record" + + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" +) + +// fakePassiveClock is a minimal clock.PassiveClock whose time only advances +// when Step is called, so aggregation-window behavior is deterministic. +type fakePassiveClock struct{ now time.Time } + +func (c *fakePassiveClock) Now() time.Time { return c.now } +func (c *fakePassiveClock) Since(ts time.Time) time.Duration { return c.now.Sub(ts) } +func (c *fakePassiveClock) Step(d time.Duration) { c.now = c.now.Add(d) } + +func TestLedgerEventSpamKey_IncludesInstanceID(t *testing.T) { + base := &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns"}, + InvolvedObject: corev1.ObjectReference{ + Kind: "ICMSRequest", + Namespace: "ns", + Name: "req-1", + UID: "uid-1", + APIVersion: "nvca.nvcf.nvidia.io/v2beta1", + }, + Type: corev1.EventTypeNormal, + Source: corev1.EventSource{Component: "nvca"}, + } + a := base.DeepCopy() + a.Annotations = map[string]string{types.LedgerAnnotationInstanceID: "0-sr-a"} + b := base.DeepCopy() + b.Annotations = map[string]string{types.LedgerAnnotationInstanceID: "1-sr-a"} + none := base.DeepCopy() + + assert.NotEqual(t, ledgerEventSpamKey(a), ledgerEventSpamKey(b), + "different instance-ids must not share a spam budget") + assert.NotEqual(t, ledgerEventSpamKey(a), ledgerEventSpamKey(none), + "instance-level and request-level events must not share a spam budget") + assert.Equal(t, ledgerEventSpamKey(none), ledgerEventSpamKey(base.DeepCopy()), + "request-level events (no instance-id) share one key") +} + +func TestLedgerEventAggregatorKey_IncludesInstanceID(t *testing.T) { + base := &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns"}, + InvolvedObject: corev1.ObjectReference{ + Kind: "ICMSRequest", + Namespace: "ns", + Name: "req-1", + UID: "uid-1", + APIVersion: "nvca.nvcf.nvidia.io/v2beta1", + }, + Type: corev1.EventTypeNormal, + Reason: "InstanceStatusUpdate", + Message: "0-sr-a is running", + Source: corev1.EventSource{Component: "nvca"}, + } + a := base.DeepCopy() + a.Annotations = map[string]string{types.LedgerAnnotationInstanceID: "0-sr-a"} + b := base.DeepCopy() + b.Message = "1-sr-a is running" + b.Annotations = map[string]string{types.LedgerAnnotationInstanceID: "1-sr-a"} + + aggA, localA := ledgerEventAggregatorKey(a) + aggB, localB := ledgerEventAggregatorKey(b) + assert.NotEqual(t, aggA, aggB, "different instance-ids must not share an aggregate group") + assert.NotEqual(t, localA, localB, "local keys remain message-based") +} + +func TestLedgerEventAggregateMaxIntervalSeconds(t *testing.T) { + tests := []struct { + name string + interval time.Duration + want int + }{ + {name: "five minute heartbeat", interval: 5 * time.Minute, want: 299}, + {name: "zero falls back to default heartbeat - 1s", interval: 0, want: 299}, + {name: "one second has no safe window", interval: time.Second, want: 0}, + {name: "sub-second has no safe window", interval: 500 * time.Millisecond, want: 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, ledgerEventAggregateMaxIntervalSeconds(tt.interval)) + }) + } +} + +func TestNewLedgerEventCorrelatorOptions(t *testing.T) { + tests := []struct { + name string + interval time.Duration + wantMaxInterval int + wantMaxEvents int + }{ + { + name: "usable heartbeat sets a sub-heartbeat window", + interval: 5 * time.Minute, + wantMaxInterval: 299, + wantMaxEvents: 0, // client-go default (10) + }, + { + name: "sub-second heartbeat disables aggregation", + interval: 500 * time.Millisecond, + wantMaxInterval: 0, // client-go default (10m) - unused because aggregation is disabled + wantMaxEvents: ledgerAggregationDisabledMaxEvents, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + opts := NewLedgerEventCorrelatorOptions(tt.interval) + assert.Equal(t, tt.wantMaxInterval, opts.MaxIntervalInSeconds) + assert.Equal(t, tt.wantMaxEvents, opts.MaxEvents) + assert.NotNil(t, opts.KeyFunc) + assert.NotNil(t, opts.SpamKeyFunc) + }) + } +} + +// TestLedgerCorrelator_PreservesAnnotationsUnderRapidHeartbeats drives a real +// client-go correlator with a sub-second heartbeat and ten status Events for a +// single instance at 500ms cadence. Aggregation must stay disabled so the +// ledger instance-id annotation survives every Event (EventAggregate would +// otherwise drop annotations once the aggregate threshold is hit). +func TestLedgerCorrelator_PreservesAnnotationsUnderRapidHeartbeats(t *testing.T) { + clk := &fakePassiveClock{now: time.Unix(1700000000, 0)} + opts := NewLedgerEventCorrelatorOptions(500 * time.Millisecond) + opts.Clock = clk + correlator := record.NewEventCorrelatorWithOptions(opts) + + const instanceID = "0-sr-a" + for i := 0; i < 10; i++ { + ev := &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "ns", + Annotations: map[string]string{types.LedgerAnnotationInstanceID: instanceID}, + }, + InvolvedObject: corev1.ObjectReference{ + Kind: "ICMSRequest", + Namespace: "ns", + Name: "req-1", + UID: "uid-1", + APIVersion: "nvca.nvcf.nvidia.io/v2beta1", + }, + Type: corev1.EventTypeNormal, + Reason: "InstanceStatusUpdate", + Message: fmt.Sprintf("%s transition %d", instanceID, i), + Source: corev1.EventSource{Component: "nvca"}, + } + + res, err := correlator.EventCorrelate(ev) + require.NoError(t, err) + require.NotNil(t, res) + require.False(t, res.Skip, "distinct per-instance heartbeats must not be spam-filtered") + require.NotNil(t, res.Event) + assert.Equal(t, instanceID, res.Event.Annotations[types.LedgerAnnotationInstanceID], + "aggregation must not strip the ledger instance-id annotation") + + clk.Step(500 * time.Millisecond) + } +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/ledger_events.go b/src/compute-plane-services/nvca/pkg/nvca/ledger_events.go new file mode 100644 index 000000000..4ad9abcf1 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/nvca/ledger_events.go @@ -0,0 +1,56 @@ +/* +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 nvca + +import ( + nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" +) + +// EmitICMSEventf emits a Kubernetes Event on the ICMSRequest with FnDs ledger +// annotations. Pass update=nil for request-level events; pass an update with +// InstanceID (and status payload fields when applicable) for instance-level events. +func (c *BackendK8sCache) EmitICMSEventf( + req *nvcav2beta1.ICMSRequest, + eventType, reason, msgFmt string, + update *types.ICMSRequestUpdateInfo, + args ...any, +) { + if c == nil || c.eventRecorder == nil || req == nil { + return + } + annotations := types.LedgerEventAnnotations(req, c.clusterName, c.clusterRegion, update) + c.eventRecorder.AnnotatedEventf(req, annotations, eventType, reason, msgFmt, args...) +} + +// EmitICMSEvent is EmitICMSEventf without formatting args. +func (c *BackendK8sCache) EmitICMSEvent( + req *nvcav2beta1.ICMSRequest, + eventType, reason, message string, + update *types.ICMSRequestUpdateInfo, +) { + c.EmitICMSEventf(req, eventType, reason, "%s", update, message) +} + +// instanceUpdate is a convenience for instance-level Events that only need instance-id. +func instanceUpdate(instanceID string) *types.ICMSRequestUpdateInfo { + if instanceID == "" { + return nil + } + return &types.ICMSRequestUpdateInfo{InstanceID: instanceID} +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/ledger_events_test.go b/src/compute-plane-services/nvca/pkg/nvca/ledger_events_test.go new file mode 100644 index 000000000..cdd39bdf3 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/nvca/ledger_events_test.go @@ -0,0 +1,207 @@ +/* +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 nvca + +import ( + "testing" + "time" + + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/function" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/tools/record" + + nvcametrics "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/internal/metrics" + nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" +) + +func TestFailureCategoryAnnotationParity_ContainerImagePull(t *testing.T) { + // Same mapping used at RecordWorkloadStatus call sites for container failures. + cause := types.ICMSInstanceFailedImagePullIssues + metricCategory := nvcametrics.ICMSInstanceStateToFailureCategory(cause) + + req := &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-parity", + FunctionDetails: function.Details{ + FunctionVersionID: "fv-1", + }, + }, + } + update := &types.ICMSRequestUpdateInfo{ + InstanceID: "0-sr-parity", + Payload: types.ICMSInstanceStatusUpdateRequest{ + InstanceState: types.ICMSInstanceTerminated, + TerminationCause: cause, + FailureCategory: string(metricCategory), + }, + } + + annotations := types.LedgerEventAnnotations(req, "cluster-east", "us-east-1", update) + assert.Equal(t, string(metricCategory), annotations[types.LedgerAnnotationFailureCategory]) + assert.Equal(t, "image_pull", annotations[types.LedgerAnnotationFailureCategory]) +} + +func TestFailureCategoryAnnotationParity_HelmNotFound(t *testing.T) { + cause := types.ICMSInstanceFailedNotFound + metricCategory := nvcametrics.ICMSInstanceStateToFailureCategory(cause) + + req := &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-helm", + FunctionDetails: function.Details{ + FunctionVersionID: "fv-helm", + }, + }, + } + update := &types.ICMSRequestUpdateInfo{ + InstanceID: "sr-abc-miniservice", + Payload: types.ICMSInstanceStatusUpdateRequest{ + InstanceState: types.ICMSInstanceTerminated, + TerminationCause: cause, + FailureCategory: string(metricCategory), + }, + } + + annotations := types.LedgerEventAnnotations(req, "cluster-east", "us-east-1", update) + assert.Equal(t, string(metricCategory), annotations[types.LedgerAnnotationFailureCategory]) + assert.Equal(t, "not_found", annotations[types.LedgerAnnotationFailureCategory]) +} + +func newLedgerTestCache(rec record.EventRecorder) *BackendK8sCache { + return &BackendK8sCache{ + eventRecorder: rec, + clusterName: "cluster-east", + clusterRegion: "us-east-1", + } +} + +// receiveEvent reads one event from the fake recorder, failing the test rather +// than blocking indefinitely if event emission regressed. +func receiveEvent(t *testing.T, rec *record.FakeRecorder) string { + t.Helper() + select { + case ev := <-rec.Events: + return ev + case <-time.After(time.Second): + t.Fatal("timed out waiting for an event to be recorded") + return "" + } +} + +func functionLedgerRequest() *nvcav2beta1.ICMSRequest { + return &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-1", + NCAId: "nca-1", + FunctionDetails: function.Details{ + FunctionID: "func-1", + FunctionVersionID: "fv-1", + }, + }, + } +} + +func TestEmitICMSEventf_ForwardsAnnotationsAndArgs(t *testing.T) { + rec := record.NewFakeRecorder(4) + c := newLedgerTestCache(rec) + + c.EmitICMSEventf(functionLedgerRequest(), corev1.EventTypeNormal, + "InstanceStatusUpdate", "%v is %v", instanceUpdate("0-sr-a"), "0-sr-a", "running") + + got := receiveEvent(t, rec) + // Formatting args are forwarded to the recorder. + assert.Contains(t, got, "Normal InstanceStatusUpdate 0-sr-a is running") + // Instance-level annotations are stamped. + assert.Contains(t, got, types.LedgerAnnotationInstanceID+":0-sr-a") + assert.Contains(t, got, types.LedgerAnnotationICMSRequestID+":req-1") + assert.Contains(t, got, types.LedgerAnnotationClusterID+":cluster-east") + assert.Contains(t, got, types.LedgerAnnotationRegion+":us-east-1") + assert.Empty(t, rec.Events, "exactly one event should be emitted") +} + +func TestEmitICMSEvent_RequestLevelOmitsInstanceID(t *testing.T) { + rec := record.NewFakeRecorder(4) + c := newLedgerTestCache(rec) + + c.EmitICMSEvent(functionLedgerRequest(), corev1.EventTypeNormal, + "InstanceCreation", "Request accepted for processing", nil) + + got := receiveEvent(t, rec) + assert.Contains(t, got, "Normal InstanceCreation Request accepted for processing") + assert.Contains(t, got, types.LedgerAnnotationICMSRequestID+":req-1") + assert.NotContains(t, got, types.LedgerAnnotationInstanceID, + "request-level events must not carry instance-id") +} + +func TestEmitICMSEventf_NilGuards(t *testing.T) { + tests := []struct { + name string + // newCache builds the cache under test. rec is non-nil only when the + // case wires a fake recorder (so we can assert nothing was emitted). + newCache func(rec record.EventRecorder) *BackendK8sCache + withRec bool + req *nvcav2beta1.ICMSRequest + }{ + { + name: "nil cache", + newCache: func(record.EventRecorder) *BackendK8sCache { return nil }, + req: functionLedgerRequest(), + }, + { + name: "nil recorder", + newCache: func(record.EventRecorder) *BackendK8sCache { return &BackendK8sCache{clusterName: "cluster-east"} }, + req: functionLedgerRequest(), + }, + { + name: "nil request", + newCache: newLedgerTestCache, + withRec: true, + req: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var rec *record.FakeRecorder + var recorder record.EventRecorder + if tt.withRec { + rec = record.NewFakeRecorder(4) + recorder = rec + } + c := tt.newCache(recorder) + + assert.NotPanics(t, func() { + c.EmitICMSEventf(tt.req, corev1.EventTypeNormal, "R", "m", nil) + }) + if rec != nil { + assert.Empty(t, rec.Events, "no event should be emitted") + } + }) + } +} + +func TestInstanceUpdate(t *testing.T) { + assert.Nil(t, instanceUpdate(""), "empty instance-id yields a request-level (nil) update") + + got := instanceUpdate("0-sr-a") + require.NotNil(t, got) + assert.Equal(t, "0-sr-a", got.InstanceID) +} diff --git a/src/compute-plane-services/nvca/pkg/types/BUILD.bazel b/src/compute-plane-services/nvca/pkg/types/BUILD.bazel index 03f550f31..91a591521 100644 --- a/src/compute-plane-services/nvca/pkg/types/BUILD.bazel +++ b/src/compute-plane-services/nvca/pkg/types/BUILD.bazel @@ -7,6 +7,7 @@ go_library( name = "types", srcs = [ "common_labels.go", + "event_annotations.go", "gpu.go", "miniservice_types.go", "modelcache_types.go", @@ -43,6 +44,7 @@ go_test( name = "types_test", srcs = [ "common_labels_test.go", + "event_annotations_test.go", "gpu_test.go", "miniservice_types_test.go", "resource_types_test.go", diff --git a/src/compute-plane-services/nvca/pkg/types/event_annotations.go b/src/compute-plane-services/nvca/pkg/types/event_annotations.go new file mode 100644 index 000000000..1a783d0ff --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/types/event_annotations.go @@ -0,0 +1,99 @@ +/* +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 types //nolint:revive + +import ( + nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" +) + +const ( + ledgerAnnotationPrefix = "nvcf.nvidia.io/" + + // Ledger Event annotation keys stamped on ICMSRequest Kubernetes Events for FnDs. + LedgerAnnotationICMSRequestID = ledgerAnnotationPrefix + "icms-request-id" + LedgerAnnotationFunctionVersionID = ledgerAnnotationPrefix + "function-version-id" + LedgerAnnotationTaskID = ledgerAnnotationPrefix + "task-id" + LedgerAnnotationInstanceID = ledgerAnnotationPrefix + "instance-id" + LedgerAnnotationFunctionID = ledgerAnnotationPrefix + "function-id" + LedgerAnnotationNCAID = ledgerAnnotationPrefix + "nca-id" + LedgerAnnotationClusterID = ledgerAnnotationPrefix + "cluster-id" + LedgerAnnotationRegion = ledgerAnnotationPrefix + "region" + LedgerAnnotationInstanceState = ledgerAnnotationPrefix + "instance-state" + LedgerAnnotationStatus = ledgerAnnotationPrefix + "status" + LedgerAnnotationTerminationCause = ledgerAnnotationPrefix + "termination-cause" + LedgerAnnotationFailureCategory = ledgerAnnotationPrefix + "failure-category" +) + +// LedgerEventAnnotations builds FnDs ledger annotations for an ICMSRequest Event. +// Empty values are omitted. Instance-level fields come from update when non-nil. +// Status subset (instance-state, status, termination-cause, failure-category) is +// included only when the corresponding payload fields are set. +func LedgerEventAnnotations( + req *nvcav2beta1.ICMSRequest, + clusterID, region string, + update *ICMSRequestUpdateInfo, +) map[string]string { + if req == nil { + return nil + } + + annotations := make(map[string]string) + set := func(key, value string) { + if value != "" { + annotations[key] = value + } + } + + set(LedgerAnnotationICMSRequestID, req.Spec.RequestID) + set(LedgerAnnotationNCAID, req.Spec.NCAId) + set(LedgerAnnotationClusterID, clusterID) + set(LedgerAnnotationRegion, region) + + // FunctionDetails is authoritative; the flat Spec.Function*ID fields are + // deprecated. Fall back to them only for pre-FunctionDetails CRs still in + // flight during an upgrade (parity with common_labels.go). + functionID := req.Spec.FunctionDetails.FunctionID + if functionID == "" { + functionID = req.Spec.FunctionID + } + set(LedgerAnnotationFunctionID, functionID) + + functionVersionID := req.Spec.FunctionDetails.FunctionVersionID + if functionVersionID == "" { + functionVersionID = req.Spec.FunctionVersionID + } + taskID := req.Spec.TaskDetails.TaskID + // A request is either a function deployment or a task, never both, so TaskID + // and FunctionVersionID are mutually exclusive by design. Emit task-id for + // tasks; otherwise function-version-id. + if taskID != "" { + set(LedgerAnnotationTaskID, taskID) + } else { + set(LedgerAnnotationFunctionVersionID, functionVersionID) + } + + if update != nil { + set(LedgerAnnotationInstanceID, update.InstanceID) + set(LedgerAnnotationInstanceState, string(update.Payload.InstanceState)) + set(LedgerAnnotationStatus, string(update.Payload.Status)) + set(LedgerAnnotationTerminationCause, string(update.Payload.TerminationCause)) + set(LedgerAnnotationFailureCategory, update.Payload.FailureCategory) + } + + return annotations +} diff --git a/src/compute-plane-services/nvca/pkg/types/event_annotations_test.go b/src/compute-plane-services/nvca/pkg/types/event_annotations_test.go new file mode 100644 index 000000000..1b45e5cf1 --- /dev/null +++ b/src/compute-plane-services/nvca/pkg/types/event_annotations_test.go @@ -0,0 +1,119 @@ +/* +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 types + +import ( + "encoding/json" + "testing" + + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/function" + "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/task" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" +) + +func TestLedgerEventAnnotations_FunctionRequestLevel(t *testing.T) { + req := &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-123", + NCAId: "nca-001", + FunctionDetails: function.Details{ + FunctionID: "func-abc", + FunctionVersionID: "fv-xyz", + }, + }, + } + + got := LedgerEventAnnotations(req, "cluster-east", "us-east-1", nil) + assert.Equal(t, map[string]string{ + LedgerAnnotationICMSRequestID: "req-123", + LedgerAnnotationNCAID: "nca-001", + LedgerAnnotationClusterID: "cluster-east", + LedgerAnnotationRegion: "us-east-1", + LedgerAnnotationFunctionID: "func-abc", + LedgerAnnotationFunctionVersionID: "fv-xyz", + }, got) + _, hasInstance := got[LedgerAnnotationInstanceID] + assert.False(t, hasInstance, "request-level events must omit instance-id") + _, hasTask := got[LedgerAnnotationTaskID] + assert.False(t, hasTask) +} + +func TestLedgerEventAnnotations_TaskIdentity(t *testing.T) { + req := &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-task", + NCAId: "nca-001", + TaskDetails: task.Details{ + TaskID: "task-999", + }, + }, + } + + got := LedgerEventAnnotations(req, "cluster-east", "", nil) + assert.Equal(t, "task-999", got[LedgerAnnotationTaskID]) + _, hasFV := got[LedgerAnnotationFunctionVersionID] + assert.False(t, hasFV, "tasks use task-id, not function-version-id") + _, hasRegion := got[LedgerAnnotationRegion] + assert.False(t, hasRegion, "empty fields omitted") +} + +func TestLedgerEventAnnotations_InstanceStatusSubset(t *testing.T) { + req := &nvcav2beta1.ICMSRequest{ + Spec: nvcav2beta1.ICMSRequestSpec{ + RequestID: "req-123", + FunctionDetails: function.Details{ + FunctionVersionID: "fv-xyz", + }, + }, + } + update := &ICMSRequestUpdateInfo{ + InstanceID: "0-sr-abc", + Payload: ICMSInstanceStatusUpdateRequest{ + InstanceState: ICMSInstanceTerminated, + Status: ICMSRequestInstanceTerminatedByService, + TerminationCause: ICMSInstanceFailedImagePullIssues, + FailureCategory: "image_pull", + }, + } + + got := LedgerEventAnnotations(req, "cluster-east", "us-east-1", update) + assert.Equal(t, "0-sr-abc", got[LedgerAnnotationInstanceID]) + assert.Equal(t, string(ICMSInstanceTerminated), got[LedgerAnnotationInstanceState]) + assert.Equal(t, string(ICMSRequestInstanceTerminatedByService), got[LedgerAnnotationStatus]) + assert.Equal(t, string(ICMSInstanceFailedImagePullIssues), got[LedgerAnnotationTerminationCause]) + assert.Equal(t, "image_pull", got[LedgerAnnotationFailureCategory]) +} + +func TestLedgerEventAnnotations_NilRequest(t *testing.T) { + assert.Nil(t, LedgerEventAnnotations(nil, "c", "r", nil)) +} + +func TestICMSInstanceStatusUpdateRequest_FailureCategoryNotMarshaled(t *testing.T) { + payload := ICMSInstanceStatusUpdateRequest{ + InstanceState: ICMSInstanceTerminated, + FailureCategory: "image_pull", + } + data, err := json.Marshal(payload) + require.NoError(t, err) + assert.NotContains(t, string(data), "failureCategory") + assert.NotContains(t, string(data), "image_pull") + assert.Contains(t, string(data), "terminated") +} diff --git a/src/compute-plane-services/nvca/pkg/types/types.go b/src/compute-plane-services/nvca/pkg/types/types.go index 4a81ee52c..9bd0e8989 100644 --- a/src/compute-plane-services/nvca/pkg/types/types.go +++ b/src/compute-plane-services/nvca/pkg/types/types.go @@ -263,6 +263,9 @@ type ICMSInstanceStatusUpdateRequest struct { SystemFailure string `json:"systemFailure,omitempty"` MessageBatchID string `json:"messageBatchId,omitempty"` InstanceIPs []string `json:"instanceIps,omitempty"` + // FailureCategory is NVCA-only (not sent to ICMS). Same value as + // nvca_workload_result_total's failure_category; stamped on K8s Event annotations for FnDs. + FailureCategory string `json:"-"` } type ICMSRequestUpdateInfo struct {