Skip to content

fix(operator): bind workloads to verified checkpoints #12873

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

Open
galletas1712 wants to merge 1 commit into from
Open

fix(operator): bind workloads to verified checkpoints #12873

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
133 changes: 132 additions & 1 deletion deploy/operator/internal/checkpoint/checkpoint_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,126 @@ func testInfo() *CheckpointInfo {
return &CheckpointInfo{Enabled: true, Ready: true, Hash: testHash}
}

func TestAutomaticCheckpointBinding(t *testing.T) {
owner := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{
Name: "test-dgd", Namespace: testNamespace, UID: types.UID("dgd-uid"),
}}
ckpt := mustExpectedAutoCheckpoint(
t,
testScheme(),
testIdentity(),
corev1.PodTemplateSpec{Spec: corev1.PodSpec{
Containers: []corev1.Container{{
Name: consts.MainContainerName,
Image: "worker:expected",
}},
}},
consts.MainContainerName,
nvidiacomv1alpha1.CheckpointDeletionPolicyDelete,
nil,
owner,
)
ckpt.UID = types.UID("checkpoint-uid")
ckpt.Generation = 7
ckpt.Status.CheckpointID = testHash

binding, err := AutomaticCheckpointBinding(ckpt)
require.NoError(t, err)
assert.Regexp(t, `^v1/checkpoint-uid/7/[0-9a-f]{64}$`, binding)
recomputed, err := AutomaticCheckpointBinding(ckpt.DeepCopy())
require.NoError(t, err)
assert.Equal(t, binding, recomputed)
Comment on lines +125 to +153

@devin-ai-integration devin-ai-integration Bot Aug 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 New checkpoint binding test does not narrate its steps as required by the repository test style

The new binding test contains no t.Log step headings (TestAutomaticCheckpointBinding at deploy/operator/internal/checkpoint/checkpoint_test.go:125-243), violating the mandatory Go test style for this module.
Impact: Test output does not tell the scenario's story, which the repository requires for reviewability.

Rule reference

deploy/operator/AGENTS.md, "Go Test Style": "Use t.Log to tell the test's story, with one heading before each block that implements a test step." The test has distinct steps (build the expected automatic checkpoint, compute and re-compute the binding, run the mutation table) with no headings. Other new tests added in this PR (for example TestValidateGroveCheckpointBindings) do follow the rule.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


tests := []struct {
name string
binding string
mutate func(*nvidiacomv1alpha1.DynamoCheckpoint)
wantErr bool
errorText string
}{
{name: "matching"},
{
name: "artifact version changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.Annotations[snapshotprotocol.CheckpointArtifactVersionAnnotation] = "changed"
},
wantErr: true,
errorText: "provenance differs",
},
{
name: "automatic marker changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.Annotations[consts.CheckpointAutoAnnotation] = "false"
},
wantErr: true,
errorText: "provenance differs",
},
{
name: "checkpoint ID label changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.Labels[snapshotprotocol.CheckpointIDLabel] = "changed"
},
wantErr: true,
errorText: "checkpoint ID label differs",
},
{
name: "owner reference changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.OwnerReferences[0].UID = types.UID("replacement-owner")
},
wantErr: true,
errorText: "provenance differs",
},
{
name: "deletion policy changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.Annotations[consts.CheckpointDeletionPolicyAnnotation] =
string(nvidiacomv1alpha1.CheckpointDeletionPolicyRetain)
},
wantErr: true,
errorText: "provenance differs",
},
{
name: "capture spec changed",
mutate: func(actual *nvidiacomv1alpha1.DynamoCheckpoint) {
actual.Spec.Job.PodTemplateSpec.Spec.Containers[0].Image = "worker:changed"
},
wantErr: true,
errorText: "provenance differs",
},
{
name: "old binding rejected",
binding: "checkpoint-uid/7",
wantErr: true,
errorText: "invalid format",
},
{
name: "malformed digest rejected",
binding: "v1/checkpoint-uid/7/not-a-digest",
wantErr: true,
errorText: "SHA-256 digest",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
actual := ckpt.DeepCopy()
if tt.mutate != nil {
tt.mutate(actual)
}
expectedBinding := binding
if tt.binding != "" {
expectedBinding = tt.binding
}
err := VerifyAutomaticCheckpointBinding(actual, expectedBinding)
if tt.wantErr {
require.ErrorContains(t, err, tt.errorText)
} else {
require.NoError(t, err)
}
})
}
}

func testSnapshotAgentDaemonSet() *appsv1.DaemonSet {
return &appsv1.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
Expand Down Expand Up @@ -358,12 +478,21 @@ func TestApplyRestorePodMetadataWithStorageConfig(t *testing.T) {
require.NoError(t, ApplyRestorePodMetadataWithStorageConfig(
labels,
annotations,
&CheckpointInfo{Enabled: true, Ready: true, Hash: testHash},
&CheckpointInfo{
Enabled: true,
Exists: true,
Ready: true,
Hash: testHash,
CheckpointName: "automatic-checkpoint",
AutoBinding: "checkpoint-uid/1",
},
storageConfig,
))

assert.Equal(t, "true", labels[snapshotprotocol.RestoreTargetLabel])
assert.Equal(t, testHash, labels[snapshotprotocol.CheckpointIDLabel])
assert.Equal(t, "automatic-checkpoint", annotations[consts.CheckpointNameAnnotation])
assert.Equal(t, "checkpoint-uid/1", annotations[consts.CheckpointBindingAnnotation])
assert.Equal(t, snapshotprotocol.StorageTypePVC, annotations[snapshotprotocol.CheckpointStorageTypeAnnotation])
assert.Equal(t, "/snapshots", annotations[snapshotprotocol.CheckpointStorageBasePathAnnotation])

Expand Down Expand Up @@ -1168,6 +1297,7 @@ func TestApplyRestoreCandidateMetadata(t *testing.T) {
Exists: true,
Ready: true,
CheckpointName: "worker-checkpoint",
AutoBinding: "checkpoint-uid/1",
StartupPolicy: nvidiacomv1alpha1.CheckpointStartupPolicyWaitForCheckpoint,
RestoreTargetContainers: []string{"engine-0", "engine-1"},
})
Expand All @@ -1178,6 +1308,7 @@ func TestApplyRestoreCandidateMetadata(t *testing.T) {
assert.Empty(t, annotations[snapshotprotocol.CheckpointArtifactVersionAnnotation])
assert.Equal(t, consts.KubeLabelValueTrue, annotations[consts.CheckpointRestoreCandidateAnnotation])
assert.Equal(t, "worker-checkpoint", annotations[consts.CheckpointNameAnnotation])
assert.Equal(t, "checkpoint-uid/1", annotations[consts.CheckpointBindingAnnotation])
assert.Equal(t, string(nvidiacomv1alpha1.CheckpointStartupPolicyWaitForCheckpoint), annotations[consts.CheckpointStartupPolicyAnnotation])
assert.Equal(t, "engine-0,engine-1", annotations[snapshotprotocol.TargetContainersAnnotation])
})
Expand Down
15 changes: 15 additions & 0 deletions deploy/operator/internal/checkpoint/podspec.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,17 @@ func ApplyRestorePodMetadataWithStorageConfig(
delete(annotations, commonconsts.CheckpointRestoreCandidateAnnotation)
delete(annotations, commonconsts.CheckpointNameAnnotation)
delete(annotations, commonconsts.CheckpointStartupPolicyAnnotation)
delete(annotations, commonconsts.CheckpointBindingAnnotation)

// Preserve the verified automatic checkpoint identity on shaped restore targets.
if checkpointInfo != nil &&
checkpointInfo.Enabled &&
checkpointInfo.Exists &&
checkpointInfo.CheckpointName != "" &&
checkpointInfo.AutoBinding != "" {
annotations[commonconsts.CheckpointNameAnnotation] = checkpointInfo.CheckpointName
annotations[commonconsts.CheckpointBindingAnnotation] = checkpointInfo.AutoBinding
}
}
if !enabled {
return nil
Expand Down Expand Up @@ -120,6 +131,7 @@ func ApplyRestoreCandidateMetadata(labels map[string]string, annotations map[str
delete(annotations, commonconsts.CheckpointRestoreCandidateAnnotation)
delete(annotations, commonconsts.CheckpointNameAnnotation)
delete(annotations, commonconsts.CheckpointStartupPolicyAnnotation)
delete(annotations, commonconsts.CheckpointBindingAnnotation)
delete(annotations, snapshotprotocol.TargetContainersAnnotation)
if checkpointInfo == nil || !checkpointInfo.Enabled || !checkpointInfo.Exists || checkpointInfo.CheckpointName == "" {
return nil
Expand All @@ -131,6 +143,9 @@ func ApplyRestoreCandidateMetadata(labels map[string]string, annotations map[str
}
annotations[commonconsts.CheckpointRestoreCandidateAnnotation] = commonconsts.KubeLabelValueTrue
annotations[commonconsts.CheckpointNameAnnotation] = checkpointInfo.CheckpointName
if checkpointInfo.AutoBinding != "" {
annotations[commonconsts.CheckpointBindingAnnotation] = checkpointInfo.AutoBinding
}
startupPolicy := checkpointInfo.StartupPolicy
if startupPolicy == "" {
startupPolicy = nvidiacomv1alpha1.CheckpointStartupPolicyImmediate
Expand Down
16 changes: 16 additions & 0 deletions deploy/operator/internal/checkpoint/resolve.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"fmt"

nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1alpha1"
"github.com/ai-dynamo/dynamo/deploy/operator/internal/consts"
snapshotprotocol "github.com/ai-dynamo/dynamo/deploy/snapshot/protocol"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
Expand All @@ -38,6 +39,11 @@ type CheckpointInfo struct {
StartupPolicy nvidiacomv1alpha1.CheckpointStartupPolicy
// Empty means the restore pod targets the default main container.
RestoreTargetContainers []string
// AutoBinding binds a generated workload to this checkpoint's exact object.
AutoBinding string
// Automatic marks a DGD-managed checkpoint even before the API server has
// assigned the UID and generation needed to construct AutoBinding.
Automatic bool
// RestorePaused keeps restored targets paused for owner election.
RestorePaused bool
}
Expand All @@ -47,6 +53,14 @@ func checkpointInfoFromObject(ckpt *nvidiacomv1alpha1.DynamoCheckpoint) (*Checkp
if err != nil {
return nil, err
}
autoBinding := ""
automatic := ckpt.Annotations[consts.CheckpointAutoAnnotation] == consts.KubeLabelValueTrue
if automatic && ckpt.UID != "" && ckpt.Generation > 0 {
autoBinding, err = AutomaticCheckpointBinding(ckpt)
if err != nil {
return nil, err
}
}
Comment on lines +56 to +63

@devin-ai-integration devin-ai-integration Bot Aug 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 Binding computation now fails resolution for any automatic checkpoint whose ID label diverges

checkpointInfoFromObject now calls AutomaticCheckpointBinding for every checkpoint carrying the automatic marker, and automaticCheckpointBindingFor (deploy/operator/internal/checkpoint/resource.go:399-403) hard-errors when labels[CheckpointIDLabel] differs from the canonical CheckpointID(ckpt) (which prefers status.checkpointID, then status.identityHash, then the label). Any legacy automatic checkpoint whose status.identityHash was written from the identity hash rather than the DGD-scoped checkpoint ID will therefore make ResolveCheckpointForService return an error, failing the whole DGD/DCD reconcile rather than degrading to "no binding". The current controller keeps label and status in sync (internal/controller/dynamocheckpoint_controller.go:132-160), so this only affects objects written by older versions; still, an error here is unrecoverable without manual edits.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


return &CheckpointInfo{
Enabled: true,
Expand All @@ -56,6 +70,8 @@ func checkpointInfoFromObject(ckpt *nvidiacomv1alpha1.DynamoCheckpoint) (*Checkp
ArtifactVersion: checkpointArtifactVersion(ckpt),
CheckpointName: ckpt.Name,
Ready: ckpt.Status.Phase == nvidiacomv1alpha1.DynamoCheckpointPhaseReady,
AutoBinding: autoBinding,
Automatic: automatic,
}, nil
}

Expand Down
Loading
Loading