From 4c7f2177680b7052ff22b1e0f21596f7792c64b9 Mon Sep 17 00:00:00 2001 From: Oleg Kushniriov Date: Sun, 30 Aug 2026 09:30:29 +0300 Subject: [PATCH] refactor(operator): drop CheckpointJob naming from the snapshot protocol SnapshotJob creates a plain batch/v1 Job whose pod is the capture source; the CheckpointJob identifiers are Dynamo-era naming left in the snapshot protocol. Rename them to match the source-Job vocabulary: - NewCheckpointJob -> NewSourceJob - CheckpointJobOptions -> SourceJobOptions - DisableCheckpointJobSidecarInjection -> DisableSidecarInjection protocol/checkpoint.go and its tests move to source_job*.go, and the "checkpoint job" prose in errors and comments now says "source job" where it names the Job (the CRIU capture itself is still a checkpoint). The dead-code half of the planned cleanup (GetCheckpointJobName, DefaultCheckpointJobTTLSeconds, ApplyCheckpointStorageMetadata, DiscoverAndResolveStorage, PrepareRestorePodSpecForCheckpoint, CheckpointStatus*) was already removed by earlier PRs, so this is renames only. No behavior change: the snapshotctl CLI surface (checkpoint subcommand, flags, output keys) is untouched, and no api/v1alpha1 type, constant, or CRD schema changes. Signed-off-by: Oleg Kushniriov --- api/v1alpha1/constants.go | 2 +- operator/cmd/snapshotctl/checkpoint.go | 4 +- .../internal/controller/snapshotjob_job.go | 4 +- .../controller/snapshotjob_job_test.go | 2 +- operator/internal/protocol/control_volume.go | 2 +- .../protocol/{checkpoint.go => source_job.go} | 26 +++++----- ...ty_test.go => source_job_identity_test.go} | 0 ...ckpoint_job_test.go => source_job_test.go} | 52 +++++++++---------- 8 files changed, 46 insertions(+), 46 deletions(-) rename operator/internal/protocol/{checkpoint.go => source_job.go} (83%) rename operator/internal/protocol/{checkpoint_job_identity_test.go => source_job_identity_test.go} (100%) rename operator/internal/protocol/{checkpoint_job_test.go => source_job_test.go} (89%) diff --git a/api/v1alpha1/constants.go b/api/v1alpha1/constants.go index 2d2c32ac..1c55ecb0 100644 --- a/api/v1alpha1/constants.go +++ b/api/v1alpha1/constants.go @@ -82,7 +82,7 @@ const ( // ReadyForSnapshotFile is written by the workload inside the control // volume when the model is loaded and the workload is ready for a - // checkpoint. Observed by the checkpoint job's kubelet readiness probe + // checkpoint. Observed by the source job's kubelet readiness probe // on the worker container. ReadyForSnapshotFile = "ready-for-snapshot" diff --git a/operator/cmd/snapshotctl/checkpoint.go b/operator/cmd/snapshotctl/checkpoint.go index d6a45c62..f2819d3e 100644 --- a/operator/cmd/snapshotctl/checkpoint.go +++ b/operator/cmd/snapshotctl/checkpoint.go @@ -64,13 +64,13 @@ func runCheckpointFlow(ctx context.Context, opts checkpointOptions) (_ *result, } checkpointJobName := captureJobName(snapshotName) - job, err := snapshotprotocol.NewCheckpointJob(&corev1.PodTemplateSpec{ + job, err := snapshotprotocol.NewSourceJob(&corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Labels: pod.Labels, Annotations: pod.Annotations, }, Spec: *pod.Spec.DeepCopy(), - }, snapshotprotocol.CheckpointJobOptions{ + }, snapshotprotocol.SourceJobOptions{ Namespace: namespace, TargetContainer: containerName, SeccompProfile: snapshotv1alpha1.DefaultSeccompLocalhostProfile, diff --git a/operator/internal/controller/snapshotjob_job.go b/operator/internal/controller/snapshotjob_job.go index 7ae0675c..df2e58c7 100644 --- a/operator/internal/controller/snapshotjob_job.go +++ b/operator/internal/controller/snapshotjob_job.go @@ -15,7 +15,7 @@ import ( ) // buildSourceJob constructs the desired batch/v1 Job for a SnapshotJob's source pod. -// It reuses protocol.NewCheckpointJob unchanged — that function's body is the agent +// It reuses protocol.NewSourceJob unchanged — that function's body is the agent // contract (control volume, readiness probe, labels, seccomp, sidecar opt-outs), not // Dynamo-specific code — and adds only the owner label so the PodSnapshot created // later (PR 4) can be mapped back to this SnapshotJob without an ownerReference. @@ -47,7 +47,7 @@ func buildSourceJob(sj *snapshotv1alpha1.SnapshotJob) (*batchv1.Job, error) { podTemplate.Labels[snapshotv1alpha1.SnapshotJobOwnerLabel] = sj.Name podTemplate.Labels[snapshotv1alpha1.SnapshotJobOwnerUIDLabel] = string(sj.UID) - return protocol.NewCheckpointJob(podTemplate, protocol.CheckpointJobOptions{ + return protocol.NewSourceJob(podTemplate, protocol.SourceJobOptions{ Namespace: sj.Namespace, Name: sj.Name, TargetContainer: targetContainer, diff --git a/operator/internal/controller/snapshotjob_job_test.go b/operator/internal/controller/snapshotjob_job_test.go index b5e8f85d..a06e867b 100644 --- a/operator/internal/controller/snapshotjob_job_test.go +++ b/operator/internal/controller/snapshotjob_job_test.go @@ -35,7 +35,7 @@ func minimalSnapshotJob() *snapshotv1alpha1.SnapshotJob { } func TestBuildSourceJob(t *testing.T) { - t.Run("wires identity, target, and options through to NewCheckpointJob", func(t *testing.T) { + t.Run("wires identity, target, and options through to NewSourceJob", func(t *testing.T) { sj := minimalSnapshotJob() job, err := buildSourceJob(sj) diff --git a/operator/internal/protocol/control_volume.go b/operator/internal/protocol/control_volume.go index 21703197..5578192c 100644 --- a/operator/internal/protocol/control_volume.go +++ b/operator/internal/protocol/control_volume.go @@ -16,7 +16,7 @@ import ( // and LegacySnapshotControlDirEnv (deprecated) on the container's env, so // workload images can migrate off the legacy name independently of the // operator release. Idempotent — safe to call from multiple code paths -// (operator checkpoint job, restore pod shaping, etc.); each env var is +// (operator source job, restore pod shaping, etc.); each env var is // guarded independently so a pod that already carries one (e.g. a // hand-crafted template with only the legacy name) still gets the other // injected without duplicating either. diff --git a/operator/internal/protocol/checkpoint.go b/operator/internal/protocol/source_job.go similarity index 83% rename from operator/internal/protocol/checkpoint.go rename to operator/internal/protocol/source_job.go index e7add969..1f68df98 100644 --- a/operator/internal/protocol/checkpoint.go +++ b/operator/internal/protocol/source_job.go @@ -15,7 +15,7 @@ import ( snapshotv1alpha1 "github.com/ai-dynamo/snapshot/api/v1alpha1" ) -type CheckpointJobOptions struct { +type SourceJobOptions struct { Namespace string TargetContainer string SeccompProfile string @@ -25,14 +25,14 @@ type CheckpointJobOptions struct { WrapLaunchJob bool } -func NewCheckpointJob(podTemplate *corev1.PodTemplateSpec, opts CheckpointJobOptions) (*batchv1.Job, error) { +func NewSourceJob(podTemplate *corev1.PodTemplateSpec, opts SourceJobOptions) (*batchv1.Job, error) { podTemplate = podTemplate.DeepCopy() for _, annotation := range []string{ snapshotv1alpha1.RestoreFromAnnotation, snapshotv1alpha1.RestoreContainerMapAnnotation, } { if _, restoreRequested := podTemplate.Annotations[annotation]; restoreRequested { - return nil, fmt.Errorf("checkpoint job pod template must not set %s", annotation) + return nil, fmt.Errorf("source job pod template must not set %s", annotation) } } if podTemplate.Labels == nil { @@ -41,21 +41,21 @@ func NewCheckpointJob(podTemplate *corev1.PodTemplateSpec, opts CheckpointJobOpt if podTemplate.Annotations == nil { podTemplate.Annotations = map[string]string{} } - podTemplate.Annotations = DisableCheckpointJobSidecarInjection(podTemplate.Annotations) + podTemplate.Annotations = DisableSidecarInjection(podTemplate.Annotations) podTemplate.Spec.RestartPolicy = corev1.RestartPolicyNever if opts.SeccompProfile != "" { EnsureLocalhostSeccompProfile(&podTemplate.Spec, opts.SeccompProfile) } if len(podTemplate.Spec.Containers) == 0 { - return nil, fmt.Errorf("checkpoint job requires at least one container") + return nil, fmt.Errorf("source job requires at least one container") } - // Checkpoint contract: exactly one target container per Job. The caller (the operator, + // Snapshot contract: exactly one target container per Job. The caller (the operator, // snapshotctl) resolves the single target and passes it in opts so there is no // Containers[0]-vs-"main" ambiguity. targetName := opts.TargetContainer if targetName == "" { - return nil, fmt.Errorf("checkpoint job pod template: opts.TargetContainer is required") + return nil, fmt.Errorf("source job pod template: opts.TargetContainer is required") } var targetContainer *corev1.Container for i := range podTemplate.Spec.Containers { @@ -65,14 +65,14 @@ func NewCheckpointJob(podTemplate *corev1.PodTemplateSpec, opts CheckpointJobOpt } } if targetContainer == nil { - return nil, fmt.Errorf("checkpoint job pod template has no container named %q (from opts.TargetContainer)", targetName) + return nil, fmt.Errorf("source job pod template has no container named %q (from opts.TargetContainer)", targetName) } // Snapshot contract: control volume + ready-file readiness probe. The // agent reads the pod's Ready condition before starting CRIU dump, so // the workload signals "model loaded, safe to checkpoint" by writing // $SNAPSHOT_CONTROL_DIR/ready-for-snapshot. Any per-container - // liveness/startup probes are cleared — a checkpoint job runs to a + // liveness/startup probes are cleared — a source job runs to a // quiesce-and-sit state, not a long-lived serving state. EnsureControlVolume(&podTemplate.Spec, targetContainer) targetContainer.ReadinessProbe = &corev1.Probe{ @@ -88,7 +88,7 @@ func NewCheckpointJob(podTemplate *corev1.PodTemplateSpec, opts CheckpointJobOpt if opts.WrapLaunchJob { if len(targetContainer.Command) == 0 { - return nil, fmt.Errorf("checkpoint job requires container.command when cuda-checkpoint launch-job wrapping is enabled") + return nil, fmt.Errorf("source job requires container.command when cuda-checkpoint launch-job wrapping is enabled") } targetContainer.Command, targetContainer.Args = wrapWithCudaCheckpointLaunchJob( targetContainer.Command, @@ -129,14 +129,14 @@ func EnsureLocalhostSeccompProfile(podSpec *corev1.PodSpec, profile string) { } } -// DisableCheckpointJobSidecarInjection stamps sidecar opt-out annotations on a -// pod annotation map. Checkpoint Jobs must complete when the target container +// DisableSidecarInjection stamps sidecar opt-out annotations on a +// pod annotation map. Source Jobs must complete when the target container // exits; an injected sidecar that outlives the checkpoint keeps the pod alive, // preventing Kubernetes from marking the Job complete. // // Mutates and returns the passed-in map. Allocates a new map when annotations // is nil; callers must use the returned value. -func DisableCheckpointJobSidecarInjection(annotations map[string]string) map[string]string { +func DisableSidecarInjection(annotations map[string]string) map[string]string { if annotations == nil { annotations = map[string]string{} } diff --git a/operator/internal/protocol/checkpoint_job_identity_test.go b/operator/internal/protocol/source_job_identity_test.go similarity index 100% rename from operator/internal/protocol/checkpoint_job_identity_test.go rename to operator/internal/protocol/source_job_identity_test.go diff --git a/operator/internal/protocol/checkpoint_job_test.go b/operator/internal/protocol/source_job_test.go similarity index 89% rename from operator/internal/protocol/checkpoint_job_test.go rename to operator/internal/protocol/source_job_test.go index abf31611..a08ea931 100644 --- a/operator/internal/protocol/checkpoint_job_test.go +++ b/operator/internal/protocol/source_job_test.go @@ -44,8 +44,8 @@ func requireStableLaunchJobWrapper(t *testing.T, container *corev1.Container, or } } -func TestNewCheckpointJob(t *testing.T) { - job, err := NewCheckpointJob(&corev1.PodTemplateSpec{ +func TestNewSourceJob(t *testing.T) { + job, err := NewSourceJob(&corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Labels: map[string]string{"existing": "label"}, Annotations: map[string]string{ @@ -61,7 +61,7 @@ func TestNewCheckpointJob(t *testing.T) { Args: []string{"--model", "Qwen"}, }}, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", TargetContainer: "main", SeccompProfile: snapshotv1alpha1.DefaultSeccompLocalhostProfile, @@ -71,7 +71,7 @@ func TestNewCheckpointJob(t *testing.T) { WrapLaunchJob: true, }) if err != nil { - t.Fatalf("expected checkpoint job, got error: %v", err) + t.Fatalf("expected source job, got error: %v", err) } if job.Name != "test-job" || job.Namespace != "test-ns" { @@ -115,8 +115,8 @@ func TestNewCheckpointJob(t *testing.T) { } } -func TestNewCheckpointJobWrapsTargetContainer(t *testing.T) { - job, err := NewCheckpointJob(&corev1.PodTemplateSpec{ +func TestNewSourceJobWrapsTargetContainer(t *testing.T) { + job, err := NewSourceJob(&corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{}, Spec: corev1.PodSpec{ Containers: []corev1.Container{ @@ -124,7 +124,7 @@ func TestNewCheckpointJobWrapsTargetContainer(t *testing.T) { {Name: "worker", Command: []string{"python3", "-m", "dynamo.vllm"}, Args: []string{"--model", "Qwen"}}, }, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", TargetContainer: "worker", Name: "test-job", @@ -132,7 +132,7 @@ func TestNewCheckpointJobWrapsTargetContainer(t *testing.T) { WrapLaunchJob: true, }) if err != nil { - t.Fatalf("expected checkpoint job, got error: %v", err) + t.Fatalf("expected source job, got error: %v", err) } worker := requireCheckpointContainer(t, job.Spec.Template.Spec.Containers, "worker") @@ -161,7 +161,7 @@ func TestNewCheckpointJobWrapsTargetContainer(t *testing.T) { } } -func TestNewCheckpointJobDisablesServiceMeshInjection(t *testing.T) { +func TestNewSourceJobDisablesServiceMeshInjection(t *testing.T) { cases := []struct { name string annotations map[string]string @@ -202,13 +202,13 @@ func TestNewCheckpointJobDisablesServiceMeshInjection(t *testing.T) { }, } - job, err := NewCheckpointJob(source, CheckpointJobOptions{ + job, err := NewSourceJob(source, SourceJobOptions{ Namespace: "test-ns", Name: "test-job", TargetContainer: "main", }) if err != nil { - t.Fatalf("NewCheckpointJob() error = %v", err) + t.Fatalf("NewSourceJob() error = %v", err) } got := job.Spec.Template.Annotations @@ -233,8 +233,8 @@ func TestNewCheckpointJobDisablesServiceMeshInjection(t *testing.T) { } } -func TestDisableCheckpointJobSidecarInjectionNilMap(t *testing.T) { - got := DisableCheckpointJobSidecarInjection(nil) +func TestDisableSidecarInjectionNilMap(t *testing.T) { + got := DisableSidecarInjection(nil) if got == nil { t.Fatal("expected non-nil map, got nil") } @@ -246,12 +246,12 @@ func TestDisableCheckpointJobSidecarInjectionNilMap(t *testing.T) { } } -func TestNewCheckpointJobRequiresTarget(t *testing.T) { - _, err := NewCheckpointJob(&corev1.PodTemplateSpec{ +func TestNewSourceJobRequiresTarget(t *testing.T) { + _, err := NewSourceJob(&corev1.PodTemplateSpec{ Spec: corev1.PodSpec{ Containers: []corev1.Container{{Name: "worker", Command: []string{"python3"}}}, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", Name: "test-job", }) @@ -260,7 +260,7 @@ func TestNewCheckpointJobRequiresTarget(t *testing.T) { } } -func TestNewCheckpointJobRejectsRestoreAnnotations(t *testing.T) { +func TestNewSourceJobRejectsRestoreAnnotations(t *testing.T) { for _, annotation := range []string{ snapshotv1alpha1.RestoreFromAnnotation, snapshotv1alpha1.RestoreContainerMapAnnotation, @@ -270,14 +270,14 @@ func TestNewCheckpointJobRejectsRestoreAnnotations(t *testing.T) { if annotation == snapshotv1alpha1.RestoreFromAnnotation { value = "snapshot-a" } - _, err := NewCheckpointJob(&corev1.PodTemplateSpec{ + _, err := NewSourceJob(&corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ Annotations: map[string]string{annotation: value}, }, Spec: corev1.PodSpec{ Containers: []corev1.Container{{Name: "main", Command: []string{"python3"}}}, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", TargetContainer: "main", Name: "test-job", @@ -290,12 +290,12 @@ func TestNewCheckpointJobRejectsRestoreAnnotations(t *testing.T) { } } -func TestNewCheckpointJobRejectsUnknownTarget(t *testing.T) { - _, err := NewCheckpointJob(&corev1.PodTemplateSpec{ +func TestNewSourceJobRejectsUnknownTarget(t *testing.T) { + _, err := NewSourceJob(&corev1.PodTemplateSpec{ Spec: corev1.PodSpec{ Containers: []corev1.Container{{Name: "worker", Command: []string{"python3"}}}, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", TargetContainer: "missing", Name: "test-job", @@ -305,16 +305,16 @@ func TestNewCheckpointJobRejectsUnknownTarget(t *testing.T) { } } -// TestNewCheckpointJobNoWrapByDefault verifies that the container command is +// TestNewSourceJobNoWrapByDefault verifies that the container command is // preserved unchanged when WrapLaunchJob is false (the default). This guards // against accidentally re-introducing cuda-checkpoint wrapping as the default, // which would require cuda-checkpoint to be present in the placeholder image // at the exact path CRIU checkpointed it from. -func TestNewCheckpointJobNoWrapByDefault(t *testing.T) { +func TestNewSourceJobNoWrapByDefault(t *testing.T) { originalCmd := []string{"python3", "-m", "dynamo.vllm"} originalArgs := []string{"--model", "Qwen"} - job, err := NewCheckpointJob(&corev1.PodTemplateSpec{ + job, err := NewSourceJob(&corev1.PodTemplateSpec{ Spec: corev1.PodSpec{ Containers: []corev1.Container{{ Name: "main", @@ -323,7 +323,7 @@ func TestNewCheckpointJobNoWrapByDefault(t *testing.T) { Args: originalArgs, }}, }, - }, CheckpointJobOptions{ + }, SourceJobOptions{ Namespace: "test-ns", TargetContainer: "main", Name: "test-job",