Skip to content

feat(nvca): stamp FnDs ledger annotations and per-instance Event correlator #938

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shobham-nv merged 4 commits into from
Aug 21, 2026
Merged

feat(nvca): stamp FnDs ledger annotations and per-instance Event correlator #938

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3ca90b2
feat(nvca): stamp FnDs ledger annotations on ICMSRequest Events
shobham-nv Aug 18, 2026
537377d
feat(nvca): key Event spam/aggregation by instance-id
shobham-nv Aug 18, 2026
f217d40
fix(nvca): address review feedback on ledger Event correlator and tests
shobham-nv Aug 18, 2026
161675d
docs(nvca): clarify ledger annotation identity comments
shobham-nv Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
20 changes: 10 additions & 10 deletions src/compute-plane-services/nvca/pkg/nvca/agent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)}
Expand Down Expand Up @@ -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)}
Expand Down Expand Up @@ -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)
}
Expand Down
6 changes: 5 additions & 1 deletion src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
63 changes: 40 additions & 23 deletions src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand All @@ -1132,17 +1133,17 @@ 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
}
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)
}
}

Expand All @@ -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 {
Expand Down Expand Up @@ -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,
)
}

Expand Down Expand Up @@ -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{
Expand All @@ -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
}
Expand Down Expand Up @@ -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)
Expand Down
14 changes: 9 additions & 5 deletions src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend_miniservice.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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 &&
Expand Down Expand Up @@ -221,20 +221,22 @@ 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,
Action: common.TerminationAction,
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
Expand Down Expand Up @@ -397,6 +399,7 @@ func (c K8sComputeBackend) GetICMSRequestUpdatesForMiniServiceRequest(ctx contex
}
}

updateInfo.Payload.FailureCategory = string(failureCategory)
return updateInfo, nil
}

Expand All @@ -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) {
Expand Down
Loading
Loading