From 8b4eca1ad4158abf10b8c4792dcfa1eeb6047475 Mon Sep 17 00:00:00 2001 From: Julien Mancuso Date: Tue, 28 Jul 2026 08:25:21 -0600 Subject: [PATCH 1/2] refactor(operator): extract Grove workload rendering from graph controller Signed-off-by: Julien Mancuso --- .../dynamographdeployment_controller.go | 323 ------------- .../dynamographdeployment_controller_test.go | 33 +- .../dynamographdeployment_grove_program.go | 82 +++- .../dynamographdeployment_grove_renderer.go | 428 ++++++++++++++++++ .../dynamographdeployment_program.go | 5 +- .../dynamographdeployment_program_test.go | 10 +- .../internal/controller/upgrade_test.go | 42 +- deploy/operator/internal/dynamo/graph.go | 16 +- 8 files changed, 546 insertions(+), 393 deletions(-) create mode 100644 deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go diff --git a/deploy/operator/internal/controller/dynamographdeployment_controller.go b/deploy/operator/internal/controller/dynamographdeployment_controller.go index fbbbe3395c22..fc3ed5ede489 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_controller.go +++ b/deploy/operator/internal/controller/dynamographdeployment_controller.go @@ -544,329 +544,6 @@ func (r *DynamoGraphDeploymentReconciler) scaleGroveResource(ctx context.Context return err } -func (r *DynamoGraphDeploymentReconciler) reconcileGrovePodCliqueSet( - ctx context.Context, - dynamoDeployment *nvidiacomv1beta1.DynamoGraphDeployment, - renderDeployment *nvidiacomv1beta1.DynamoGraphDeployment, - existingPodCliqueSet *grovev1alpha1.PodCliqueSet, - restartState *dynamo.RestartState, - checkpointInfos map[string]*checkpoint.CheckpointInfo, -) (*commoncontroller.Resource, error) { - logger := log.FromContext(ctx) - if renderDeployment == nil { - renderDeployment = dynamoDeployment - } - - existingRestartAnnotations := restartAnnotationsFromPodCliqueSet(existingPodCliqueSet) - - // generate the dynamoComponentsDeployments from the config - grovePodCliqueSet, err := dynamo.GenerateGrovePodCliqueSet(ctx, renderDeployment, r.Config, r.RuntimeConfig, r.Client, r.DockerSecretRetriever, restartState, existingRestartAnnotations, checkpointInfos) - if err != nil { - logger.Error(err, "failed to generate the Grove GangSet") - return nil, fmt.Errorf("failed to generate the Grove GangSet: %w", err) - } - prepareGroveTopologyConstraintUpgrade(grovePodCliqueSet, existingPodCliqueSet) - preserveGrovePodCliqueSetOrder(grovePodCliqueSet, existingPodCliqueSet) - preserveGrovePodCliqueSetReplicas(grovePodCliqueSet, existingPodCliqueSet, checkpointInfos) - _, syncedGrovePodCliqueSet, err := commoncontroller.SyncResource(ctx, r, dynamoDeployment, func(ctx context.Context) (*grovev1alpha1.PodCliqueSet, bool, error) { - return grovePodCliqueSet, false, nil - }) - if err != nil { - logger.Error(err, "failed to sync the Grove GangSet") - return nil, fmt.Errorf("failed to sync the Grove GangSet: %w", err) - } - syncedGrovePodCliqueSetAsResource, err := commoncontroller.NewResourceWithComponentStatuses( - syncedGrovePodCliqueSet, - func() (bool, string, map[string]nvidiacomv1beta1.ComponentReplicaStatus) { - // Grove readiness: all underlying PodCliques and PodCliqueScalingGroups have replicas == availableReplicas. - // A transient (non-NotFound) read error is handled authoritatively by - // groveProgram, which re-evaluates and returns the error so - // the reconcile retries; here we defensively treat it as not-ready so a - // read blip can never surface as "ready". - allComponentsReady, reason, componentStatuses, readErr := dynamo.GetComponentReadinessAndServiceReplicaStatuses(ctx, r.Client, dynamoDeployment) - if readErr != nil { - return false, nvidiacomv1beta1.DGDReadyReasonSomeResourcesNotReady, nil - } - if !allComponentsReady { - return false, reason, componentStatuses - } - return true, "", componentStatuses - }, - ) - if err != nil { - logger.Error(err, "failed to create the Grove PodClique Set resource") - return nil, fmt.Errorf("failed to create the Grove PodClique Set resource: %w", err) - } - return syncedGrovePodCliqueSetAsResource, nil -} - -func (r *DynamoGraphDeploymentReconciler) getExistingGrovePodCliqueSet(ctx context.Context, dgd *nvidiacomv1beta1.DynamoGraphDeployment) (*grovev1alpha1.PodCliqueSet, error) { - pcs := &grovev1alpha1.PodCliqueSet{} - err := r.Client.Get(ctx, types.NamespacedName{Name: dynamo.PCSNameForDGD(dgd.Name, dgd.Spec.Components), Namespace: dgd.Namespace}, pcs) - if err != nil && !errors.IsNotFound(err) { - return nil, fmt.Errorf("failed to get PodCliqueSet: %w", err) - } - if errors.IsNotFound(err) { - return nil, nil - } - return pcs, nil -} - -func restartAnnotationsFromPodCliqueSet(pcs *grovev1alpha1.PodCliqueSet) map[string]string { - restartAnnotations := make(map[string]string) - if pcs == nil { - return restartAnnotations - } - for _, clique := range pcs.Spec.Template.Cliques { - if clique.Annotations != nil { - if timestamp, ok := clique.Annotations[consts.RestartAnnotation]; ok { - if componentName, ok := clique.Labels[consts.KubeLabelDynamoComponent]; ok { - restartAnnotations[componentName] = timestamp - } - } - } - } - return restartAnnotations -} - -func preserveGrovePodCliqueSetOrder(desired *grovev1alpha1.PodCliqueSet, existing *grovev1alpha1.PodCliqueSet) { - if desired == nil || existing == nil { - return - } - desired.Spec.Template.Cliques = orderLikeExisting(existing.Spec.Template.Cliques, desired.Spec.Template.Cliques, podCliqueTemplateName) - desired.Spec.Template.PodCliqueScalingGroupConfigs = orderLikeExisting(existing.Spec.Template.PodCliqueScalingGroupConfigs, desired.Spec.Template.PodCliqueScalingGroupConfigs, podCliqueScalingGroupConfigName) - desired.Spec.Template.ResourceClaimTemplates = orderLikeExisting(existing.Spec.Template.ResourceClaimTemplates, desired.Spec.Template.ResourceClaimTemplates, resourceClaimTemplateConfigName) -} - -// prepareGroveTopologyConstraintUpgrade performs the first half of Grove's -// supported legacy topology migration. A pre-alpha.9 constraint has -// packDomain but no topologyName; Grove requires that object to be repaired by -// adding topologyName before packDomain can be migrated to pack.required. -// Keeping the legacy packing shape for this reconciliation lets the next -// reconciliation apply the generated modern shape without recreating the PCS. -func prepareGroveTopologyConstraintUpgrade(desired *grovev1alpha1.PodCliqueSet, existing *grovev1alpha1.PodCliqueSet) { - if desired == nil || existing == nil { - return - } - - prepareLegacyGroveTopologyConstraintRepair( - desired.Spec.Template.TopologyConstraint, - existing.Spec.Template.TopologyConstraint, - ) - - existingCliqueConstraints := make(map[string]*grovev1alpha1.TopologyConstraint, len(existing.Spec.Template.Cliques)) - for _, clique := range existing.Spec.Template.Cliques { - if clique != nil { - existingCliqueConstraints[clique.Name] = clique.TopologyConstraint - } - } - for _, clique := range desired.Spec.Template.Cliques { - if clique != nil { - prepareLegacyGroveTopologyConstraintRepair(clique.TopologyConstraint, existingCliqueConstraints[clique.Name]) - } - } - - existingScalingGroupConstraints := make(map[string]*grovev1alpha1.TopologyConstraint, len(existing.Spec.Template.PodCliqueScalingGroupConfigs)) - for i := range existing.Spec.Template.PodCliqueScalingGroupConfigs { - config := &existing.Spec.Template.PodCliqueScalingGroupConfigs[i] - existingScalingGroupConstraints[config.Name] = config.TopologyConstraint - } - for i := range desired.Spec.Template.PodCliqueScalingGroupConfigs { - config := &desired.Spec.Template.PodCliqueScalingGroupConfigs[i] - prepareLegacyGroveTopologyConstraintRepair(config.TopologyConstraint, existingScalingGroupConstraints[config.Name]) - } -} - -func prepareLegacyGroveTopologyConstraintRepair(desired *grovev1alpha1.TopologyConstraint, existing *grovev1alpha1.TopologyConstraint) { - if desired == nil || existing == nil { - return - } - // A constraint without an explicit desired name inherits from its repaired - // parent and can migrate packDomain directly in the same update. - if desired.TopologyName == "" || existing.TopologyName != "" || existing.PackDomain == "" { - return - } - - desired.PackDomain = existing.PackDomain - if existing.Pack == nil { - desired.Pack = nil - return - } - pack := *existing.Pack - desired.Pack = &pack -} - -// Grove horizontal replicas are driven through scale subresources after creation; -// keep existing template values so DGD replica changes do not update the PCS spec. -func preserveGrovePodCliqueSetReplicas( - desired *grovev1alpha1.PodCliqueSet, - existing *grovev1alpha1.PodCliqueSet, - checkpointInfoByComponent ...map[string]*checkpoint.CheckpointInfo, -) { - if desired == nil || existing == nil { - return - } - replicaPreserveSkips := map[string]struct{}{} - if len(checkpointInfoByComponent) > 0 { - for componentName, info := range checkpointInfoByComponent[0] { - if info != nil && - info.Enabled && - info.StartupPolicy == nvidiacomv1alpha1.CheckpointStartupPolicyWaitForCheckpoint && - !info.Ready { - replicaPreserveSkips[strings.ToLower(componentName)] = struct{}{} - } - } - } - - cliquesInScalingGroups := make(map[string]struct{}) - for _, config := range desired.Spec.Template.PodCliqueScalingGroupConfigs { - for _, cliqueName := range config.CliqueNames { - cliquesInScalingGroups[cliqueName] = struct{}{} - } - } - - cliqueReplicasByName := make(map[string]int32, len(existing.Spec.Template.Cliques)) - for _, clique := range existing.Spec.Template.Cliques { - if clique == nil || clique.Name == "" { - continue - } - cliqueReplicasByName[clique.Name] = clique.Spec.Replicas - } - for _, clique := range desired.Spec.Template.Cliques { - if clique == nil { - continue - } - if _, inScalingGroup := cliquesInScalingGroups[clique.Name]; inScalingGroup { - continue - } - if componentName := clique.Labels[consts.KubeLabelDynamoComponent]; componentName != "" { - if _, skip := replicaPreserveSkips[strings.ToLower(componentName)]; skip { - continue - } - } - if replicas, ok := cliqueReplicasByName[clique.Name]; ok { - clique.Spec.Replicas = replicas - } - } - - scalingGroupReplicasByName := make(map[string]*int32, len(existing.Spec.Template.PodCliqueScalingGroupConfigs)) - for _, config := range existing.Spec.Template.PodCliqueScalingGroupConfigs { - if config.Name == "" { - // Defensive only; generated PCSG configs always have names. - continue - } - scalingGroupReplicasByName[config.Name] = config.Replicas - } - for i := range desired.Spec.Template.PodCliqueScalingGroupConfigs { - config := &desired.Spec.Template.PodCliqueScalingGroupConfigs[i] - if _, skip := replicaPreserveSkips[strings.ToLower(config.Name)]; skip { - continue - } - if replicas, ok := scalingGroupReplicasByName[config.Name]; ok { - config.Replicas = replicas - } - } -} - -func orderLikeExisting[T any](existing []T, desired []T, nameOf func(T) string) []T { - if len(existing) == 0 || len(desired) < 2 { - return desired - } - desiredByName := make(map[string]T, len(desired)) - for _, item := range desired { - if name := nameOf(item); name != "" { - desiredByName[name] = item - } - } - ordered := make([]T, 0, len(desired)) - used := make(map[string]struct{}, len(desired)) - for _, existingItem := range existing { - name := nameOf(existingItem) - if desiredItem, ok := desiredByName[name]; ok { - ordered = append(ordered, desiredItem) - used[name] = struct{}{} - } - } - for _, item := range desired { - name := nameOf(item) - if name == "" { - ordered = append(ordered, item) - continue - } - if _, ok := used[name]; !ok { - ordered = append(ordered, item) - } - } - return ordered -} - -func podCliqueTemplateName(clique *grovev1alpha1.PodCliqueTemplateSpec) string { - if clique == nil { - return "" - } - return clique.Name -} - -func podCliqueScalingGroupConfigName(config grovev1alpha1.PodCliqueScalingGroupConfig) string { - return config.Name -} - -func resourceClaimTemplateConfigName(config grovev1alpha1.ResourceClaimTemplateConfig) string { - return config.Name -} - -func (r *DynamoGraphDeploymentReconciler) prepareGroveRenderDeployment(ctx context.Context, dgd *nvidiacomv1beta1.DynamoGraphDeployment) (*nvidiacomv1beta1.DynamoGraphDeployment, *grovev1alpha1.PodCliqueSet, error) { - existingPodCliqueSet, err := r.getExistingGrovePodCliqueSet(ctx, dgd) - if err != nil { - return nil, nil, err - } - - renderDeployment := dgd.DeepCopy() - for i := range renderDeployment.Spec.Components { - component := &renderDeployment.Spec.Components[i] - componentType := string(component.ComponentType) - if !groveComponentTypeCanUseLegacyWorkerSelector(componentType) { - continue - } - if podCliqueSetHasLegacyWorkerSelector(existingPodCliqueSet, component.ComponentName, componentType) { - applyLegacyGroveWorkerComponentType(component, componentType) - } - } - return renderDeployment, existingPodCliqueSet, nil -} - -func groveComponentTypeCanUseLegacyWorkerSelector(componentType string) bool { - return componentType == consts.ComponentTypePrefill || componentType == consts.ComponentTypeDecode -} - -func podCliqueSetHasLegacyWorkerSelector(pcs *grovev1alpha1.PodCliqueSet, componentName string, componentType string) bool { - if pcs == nil { - return false - } - for _, clique := range pcs.Spec.Template.Cliques { - if clique == nil || clique.Labels[consts.KubeLabelDynamoComponent] != componentName { - continue - } - if hasLegacyWorkerSelector(clique.Labels, componentType) { - return true - } - } - return false -} - -func applyLegacyGroveWorkerComponentType(component *nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec, subComponentType string) { - component.ComponentType = nvidiacomv1beta1.ComponentTypeWorker - if component.PodTemplate == nil { - component.PodTemplate = &corev1.PodTemplateSpec{} - } - if component.PodTemplate.Labels == nil { - component.PodTemplate.Labels = map[string]string{} - } - if _, ok := component.PodTemplate.Labels[consts.KubeLabelDynamoSubComponentType]; !ok { - component.PodTemplate.Labels[consts.KubeLabelDynamoSubComponentType] = subComponentType - } -} - // reconcileGroveScaling handles scaling operations for Grove resources based on component replica changes. func (r *DynamoGraphDeploymentReconciler) reconcileGroveScaling( ctx context.Context, diff --git a/deploy/operator/internal/controller/dynamographdeployment_controller_test.go b/deploy/operator/internal/controller/dynamographdeployment_controller_test.go index 759d92f83cd8..1bf040bfb549 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_controller_test.go +++ b/deploy/operator/internal/controller/dynamographdeployment_controller_test.go @@ -2639,7 +2639,7 @@ func TestGroveProgram_ReconcileWorkloads(t *testing.T) { }, } - result, err := (&groveProgram{reconciler: reconciler}).reconcileWorkloads( + result, err := newGroveProgram(reconciler).reconcileWorkloads( ctx, workloadReconcileRequest{DGD: dgd}, ) @@ -2712,7 +2712,7 @@ func TestGroveProgram_ReconcileWorkloadsUsesPreservedAlphaServiceIngress(t *test }, } - _, err := (&groveProgram{reconciler: reconciler}).reconcileWorkloads( + _, err := newGroveProgram(reconciler).reconcileWorkloads( ctx, workloadReconcileRequest{DGD: dgd}, ) @@ -2733,7 +2733,7 @@ func TestGroveProgram_ReconcileWorkloadsUsesPreservedAlphaServiceIngress(t *test g.Expect(service.Annotations["legacy-annotation"]).To(gomega.Equal("kept")) } -func TestDynamoGraphDeploymentReconciler_prepareGroveRenderDeployment_PreservesLegacyWorkerSelectors(t *testing.T) { +func TestGroveWorkloadRendererRenderPreservesLegacyWorkerSelectors(t *testing.T) { ctx := context.Background() g := gomega.NewGomegaWithT(t) @@ -2792,11 +2792,16 @@ func TestDynamoGraphDeploymentReconciler_prepareGroveRenderDeployment_PreservesL WithScheme(newDynamoGraphDeploymentControllerTestScheme(t)). WithObjects(dgd, existingPCS). Build() - reconciler := &DynamoGraphDeploymentReconciler{Client: fakeKubeClient} + renderer := newGroveWorkloadRenderer( + fakeKubeClient, + &configv1alpha1.OperatorConfiguration{}, + &controller_common.RuntimeConfig{}, + nil, + ) - renderDGD, existing, err := reconciler.prepareGroveRenderDeployment(ctx, dgd) + generatedPCS, err := renderer.Render(ctx, dgd, nil, nil) g.Expect(err).NotTo(gomega.HaveOccurred()) - g.Expect(existing).NotTo(gomega.BeNil()) + renderDGD := groveRenderDeployment(dgd, generatedPCS) g.Expect(dgd.GetComponentByName("VllmDecodeWorker").ComponentType).To(gomega.Equal(v1beta1.ComponentTypeDecode)) prefill := renderDGD.GetComponentByName("VllmPrefillWorker") @@ -2813,9 +2818,6 @@ func TestDynamoGraphDeploymentReconciler_prepareGroveRenderDeployment_PreservesL g.Expect(decode.ComponentType).To(gomega.Equal(v1beta1.ComponentTypeWorker)) g.Expect(decode.PodTemplate.Labels[commonconsts.KubeLabelDynamoSubComponentType]).To(gomega.Equal(commonconsts.ComponentTypeDecode)) - generatedPCS, err := dynamo.GenerateGrovePodCliqueSet(ctx, renderDGD, &configv1alpha1.OperatorConfiguration{}, &controller_common.RuntimeConfig{}, fakeKubeClient, nil, nil, nil, nil) - g.Expect(err).NotTo(gomega.HaveOccurred()) - preserveGrovePodCliqueSetOrder(generatedPCS, existing) g.Expect(generatedPCS.Spec.Template.Cliques[0].Name).To(gomega.Equal("vllmprefillworker")) var prefillClique *grovev1alpha1.PodCliqueTemplateSpec @@ -3031,7 +3033,7 @@ func TestPreserveGrovePodCliqueSetReplicasSkipsCheckpointGatedComponents(t *test g.Expect(*desired.Spec.Template.PodCliqueScalingGroupConfigs[0].Replicas).To(gomega.Equal(int32(7))) } -func TestDynamoGraphDeploymentReconciler_prepareGroveRenderDeployment_KeepsNativeWorkerSelectors(t *testing.T) { +func TestGroveWorkloadRendererRenderKeepsNativeWorkerSelectors(t *testing.T) { ctx := context.Background() g := gomega.NewGomegaWithT(t) @@ -3064,10 +3066,15 @@ func TestDynamoGraphDeploymentReconciler_prepareGroveRenderDeployment_KeepsNativ WithScheme(newDynamoGraphDeploymentControllerTestScheme(t)). WithObjects(dgd, existingPCS). Build() - reconciler := &DynamoGraphDeploymentReconciler{Client: fakeKubeClient} - - renderDGD, _, err := reconciler.prepareGroveRenderDeployment(ctx, dgd) + renderer := newGroveWorkloadRenderer( + fakeKubeClient, + &configv1alpha1.OperatorConfiguration{}, + &controller_common.RuntimeConfig{}, + nil, + ) + desired, err := renderer.Render(ctx, dgd, nil, nil) g.Expect(err).NotTo(gomega.HaveOccurred()) + renderDGD := groveRenderDeployment(dgd, desired) prefill := renderDGD.GetComponentByName("prefill") if prefill == nil { t.Fatal("expected rendered prefill component") diff --git a/deploy/operator/internal/controller/dynamographdeployment_grove_program.go b/deploy/operator/internal/controller/dynamographdeployment_grove_program.go index cf60ca863aba..edc1c4d4dc20 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_grove_program.go +++ b/deploy/operator/internal/controller/dynamographdeployment_grove_program.go @@ -25,6 +25,8 @@ import ( commonconsts "github.com/ai-dynamo/dynamo/deploy/operator/internal/consts" commoncontroller "github.com/ai-dynamo/dynamo/deploy/operator/internal/controller_common" "github.com/ai-dynamo/dynamo/deploy/operator/internal/dynamo" + "github.com/ai-dynamo/dynamo/deploy/operator/internal/features" + grovev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1" networkingv1beta1 "istio.io/client-go/pkg/apis/networking/v1beta1" corev1 "k8s.io/api/core/v1" networkingv1 "k8s.io/api/networking/v1" @@ -36,9 +38,25 @@ type groveProgram struct { // Grove rendering, scaling, and persistence helpers. Later extractions can // narrow this without moving Grove orchestration back into the common flow. reconciler *DynamoGraphDeploymentReconciler + renderer *groveWorkloadRenderer lwsEnabled bool } +func newGroveProgram(reconciler *DynamoGraphDeploymentReconciler) *groveProgram { + lwsEnabled := reconciler.RuntimeConfig != nil && + reconciler.RuntimeConfig.Gate.Enabled(features.LWS) + return &groveProgram{ + reconciler: reconciler, + renderer: newGroveWorkloadRenderer( + reconciler.Client, + reconciler.Config, + reconciler.RuntimeConfig, + reconciler.DockerSecretRetriever, + ), + lwsEnabled: lwsEnabled, + } +} + // Reconcile composes the complete Grove pathway. Each earlier operation // returns a typed value consumed by later operations. Non-status DGD changes // are persisted through req.DGD; status accumulates in the returned result. @@ -108,18 +126,22 @@ func (p *groveProgram) reconcileWorkloads( dynamoDeployment := req.DGD logger := log.FromContext(ctx) - renderDeployment, existingPodCliqueSet, err := r.prepareGroveRenderDeployment(ctx, dynamoDeployment) + desiredPodCliqueSet, err := p.renderer.Render( + ctx, + dynamoDeployment, + req.RestartState, + req.CheckpointInfos, + ) if err != nil { - return ReconcileResult{}, err + logger.Error(err, "failed to generate the Grove GangSet") + return ReconcileResult{}, fmt.Errorf("failed to generate the Grove GangSet: %w", err) } + renderDeployment := groveRenderDeployment(dynamoDeployment, desiredPodCliqueSet) - grovePodCliqueSetAsResource, err := r.reconcileGrovePodCliqueSet( + grovePodCliqueSetAsResource, err := p.reconcilePodCliqueSet( ctx, dynamoDeployment, - renderDeployment, - existingPodCliqueSet, - req.RestartState, - req.CheckpointInfos, + desiredPodCliqueSet, ) if err != nil { logger.Error(err, "failed to reconcile the Grove PodClique Set") @@ -308,6 +330,52 @@ func (p *groveProgram) reconcileWorkloads( return p.checkResourcesReadiness(ctx, dynamoDeployment, resources) } +func (p *groveProgram) reconcilePodCliqueSet( + ctx context.Context, + dynamoDeployment *nvidiacomv1beta1.DynamoGraphDeployment, + desired *grovev1alpha1.PodCliqueSet, +) (*commoncontroller.Resource, error) { + r := p.reconciler + logger := log.FromContext(ctx) + + _, synced, err := commoncontroller.SyncResource( + ctx, + r, + dynamoDeployment, + func(context.Context) (*grovev1alpha1.PodCliqueSet, bool, error) { + return desired, false, nil + }, + ) + if err != nil { + logger.Error(err, "failed to sync the Grove GangSet") + return nil, fmt.Errorf("failed to sync the Grove GangSet: %w", err) + } + + resource, err := commoncontroller.NewResourceWithComponentStatuses( + synced, + func() (bool, string, map[string]nvidiacomv1beta1.ComponentReplicaStatus) { + // Grove readiness: all underlying PodCliques and + // PodCliqueScalingGroups have replicas == availableReplicas. A + // transient read error is handled authoritatively by groveProgram, + // which re-evaluates and returns the error so reconciliation retries. + allComponentsReady, reason, componentStatuses, readErr := + dynamo.GetComponentReadinessAndServiceReplicaStatuses(ctx, r.Client, dynamoDeployment) + if readErr != nil { + return false, nvidiacomv1beta1.DGDReadyReasonSomeResourcesNotReady, nil + } + if !allComponentsReady { + return false, reason, componentStatuses + } + return true, "", componentStatuses + }, + ) + if err != nil { + logger.Error(err, "failed to create the Grove PodClique Set resource") + return nil, fmt.Errorf("failed to create the Grove PodClique Set resource: %w", err) + } + return resource, nil +} + // checkResourcesReadiness computes the readiness result for the synced Grove // resources and overlays the Grove-specific Ready reason classification on a // not-ready result. A transient Grove read error is returned so reconciliation diff --git a/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go b/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go new file mode 100644 index 000000000000..17c0aead5deb --- /dev/null +++ b/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go @@ -0,0 +1,428 @@ +/* + * 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 controller + +import ( + "context" + "fmt" + "strings" + + configv1alpha1 "github.com/ai-dynamo/dynamo/deploy/operator/api/config/v1alpha1" + nvidiacomv1alpha1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1alpha1" + nvidiacomv1beta1 "github.com/ai-dynamo/dynamo/deploy/operator/api/v1beta1" + "github.com/ai-dynamo/dynamo/deploy/operator/internal/checkpoint" + commonconsts "github.com/ai-dynamo/dynamo/deploy/operator/internal/consts" + commoncontroller "github.com/ai-dynamo/dynamo/deploy/operator/internal/controller_common" + "github.com/ai-dynamo/dynamo/deploy/operator/internal/dynamo" + grovev1alpha1 "github.com/ai-dynamo/grove/operator/api/core/v1alpha1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// groveWorkloadRenderer renders the desired Grove PodCliqueSet using read-only +// Kubernetes access. It does not reconcile resources, register watches, own +// finalizers, or write status. +type groveWorkloadRenderer struct { + reader client.Reader + config *configv1alpha1.OperatorConfiguration + runtimeConfig *commoncontroller.RuntimeConfig + dockerSecretRetriever dockerSecretRetriever +} + +func newGroveWorkloadRenderer( + reader client.Reader, + config *configv1alpha1.OperatorConfiguration, + runtimeConfig *commoncontroller.RuntimeConfig, + dockerSecretRetriever dockerSecretRetriever, +) *groveWorkloadRenderer { + return &groveWorkloadRenderer{ + reader: reader, + config: config, + runtimeConfig: runtimeConfig, + dockerSecretRetriever: dockerSecretRetriever, + } +} + +func (r *groveWorkloadRenderer) Render( + ctx context.Context, + dgd *nvidiacomv1beta1.DynamoGraphDeployment, + restartState *dynamo.RestartState, + checkpointInfos map[string]*checkpoint.CheckpointInfo, +) (*grovev1alpha1.PodCliqueSet, error) { + if dgd == nil { + return nil, fmt.Errorf("cannot render Grove PodCliqueSet without a DynamoGraphDeployment") + } + if r.reader == nil { + return nil, fmt.Errorf("cannot render Grove PodCliqueSet without a Kubernetes reader") + } + existingPodCliqueSet, err := r.getExistingPodCliqueSet(ctx, dgd) + if err != nil { + return nil, err + } + + renderDeployment := groveRenderDeployment(dgd, existingPodCliqueSet) + existingRestartAnnotations := restartAnnotationsFromPodCliqueSet(existingPodCliqueSet) + desired, err := dynamo.GenerateGrovePodCliqueSet( + ctx, + renderDeployment, + r.config, + r.runtimeConfig, + r.reader, + r.dockerSecretRetriever, + restartState, + existingRestartAnnotations, + checkpointInfos, + ) + if err != nil { + return nil, err + } + + prepareGroveTopologyConstraintUpgrade(desired, existingPodCliqueSet) + preserveGrovePodCliqueSetOrder(desired, existingPodCliqueSet) + preserveGrovePodCliqueSetReplicas(desired, existingPodCliqueSet, checkpointInfos) + return desired, nil +} + +func groveRenderDeployment( + dgd *nvidiacomv1beta1.DynamoGraphDeployment, + pcs *grovev1alpha1.PodCliqueSet, +) *nvidiacomv1beta1.DynamoGraphDeployment { + renderDeployment := dgd.DeepCopy() + applyGroveCompatibility(renderDeployment, pcs) + return renderDeployment +} + +func applyGroveCompatibility( + dgd *nvidiacomv1beta1.DynamoGraphDeployment, + pcs *grovev1alpha1.PodCliqueSet, +) { + for i := range dgd.Spec.Components { + component := &dgd.Spec.Components[i] + componentType := string(component.ComponentType) + if !groveComponentTypeCanUseLegacyWorkerSelector(componentType) { + continue + } + if podCliqueSetHasLegacyWorkerSelector(pcs, component.ComponentName, componentType) { + applyLegacyGroveWorkerComponentType(component, componentType) + } + } +} + +func (r *groveWorkloadRenderer) getExistingPodCliqueSet( + ctx context.Context, + dgd *nvidiacomv1beta1.DynamoGraphDeployment, +) (*grovev1alpha1.PodCliqueSet, error) { + pcs := &grovev1alpha1.PodCliqueSet{} + err := r.reader.Get( + ctx, + types.NamespacedName{ + Name: dynamo.PCSNameForDGD(dgd.Name, dgd.Spec.Components), + Namespace: dgd.Namespace, + }, + pcs, + ) + if err != nil && !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("failed to get PodCliqueSet: %w", err) + } + if apierrors.IsNotFound(err) { + return nil, nil + } + return pcs, nil +} + +func restartAnnotationsFromPodCliqueSet(pcs *grovev1alpha1.PodCliqueSet) map[string]string { + restartAnnotations := make(map[string]string) + if pcs == nil { + return restartAnnotations + } + for _, clique := range pcs.Spec.Template.Cliques { + if clique.Annotations != nil { + if timestamp, ok := clique.Annotations[commonconsts.RestartAnnotation]; ok { + if componentName, ok := clique.Labels[commonconsts.KubeLabelDynamoComponent]; ok { + restartAnnotations[componentName] = timestamp + } + } + } + } + return restartAnnotations +} + +func preserveGrovePodCliqueSetOrder( + desired *grovev1alpha1.PodCliqueSet, + existing *grovev1alpha1.PodCliqueSet, +) { + if desired == nil || existing == nil { + return + } + desired.Spec.Template.Cliques = orderLikeExisting( + existing.Spec.Template.Cliques, + desired.Spec.Template.Cliques, + podCliqueTemplateName, + ) + desired.Spec.Template.PodCliqueScalingGroupConfigs = orderLikeExisting( + existing.Spec.Template.PodCliqueScalingGroupConfigs, + desired.Spec.Template.PodCliqueScalingGroupConfigs, + podCliqueScalingGroupConfigName, + ) + desired.Spec.Template.ResourceClaimTemplates = orderLikeExisting( + existing.Spec.Template.ResourceClaimTemplates, + desired.Spec.Template.ResourceClaimTemplates, + resourceClaimTemplateConfigName, + ) +} + +// prepareGroveTopologyConstraintUpgrade performs the first half of Grove's +// supported legacy topology migration. A pre-alpha.9 constraint has +// packDomain but no topologyName; Grove requires that object to be repaired by +// adding topologyName before packDomain can be migrated to pack.required. +// Keeping the legacy packing shape for this reconciliation lets the next +// reconciliation apply the generated modern shape without recreating the PCS. +func prepareGroveTopologyConstraintUpgrade( + desired *grovev1alpha1.PodCliqueSet, + existing *grovev1alpha1.PodCliqueSet, +) { + if desired == nil || existing == nil { + return + } + + prepareLegacyGroveTopologyConstraintRepair( + desired.Spec.Template.TopologyConstraint, + existing.Spec.Template.TopologyConstraint, + ) + + existingCliqueConstraints := make( + map[string]*grovev1alpha1.TopologyConstraint, + len(existing.Spec.Template.Cliques), + ) + for _, clique := range existing.Spec.Template.Cliques { + if clique != nil { + existingCliqueConstraints[clique.Name] = clique.TopologyConstraint + } + } + for _, clique := range desired.Spec.Template.Cliques { + if clique != nil { + prepareLegacyGroveTopologyConstraintRepair( + clique.TopologyConstraint, + existingCliqueConstraints[clique.Name], + ) + } + } + + existingScalingGroupConstraints := make( + map[string]*grovev1alpha1.TopologyConstraint, + len(existing.Spec.Template.PodCliqueScalingGroupConfigs), + ) + for i := range existing.Spec.Template.PodCliqueScalingGroupConfigs { + config := &existing.Spec.Template.PodCliqueScalingGroupConfigs[i] + existingScalingGroupConstraints[config.Name] = config.TopologyConstraint + } + for i := range desired.Spec.Template.PodCliqueScalingGroupConfigs { + config := &desired.Spec.Template.PodCliqueScalingGroupConfigs[i] + prepareLegacyGroveTopologyConstraintRepair( + config.TopologyConstraint, + existingScalingGroupConstraints[config.Name], + ) + } +} + +func prepareLegacyGroveTopologyConstraintRepair( + desired *grovev1alpha1.TopologyConstraint, + existing *grovev1alpha1.TopologyConstraint, +) { + if desired == nil || existing == nil { + return + } + // A constraint without an explicit desired name inherits from its repaired + // parent and can migrate packDomain directly in the same update. + if desired.TopologyName == "" || existing.TopologyName != "" || existing.PackDomain == "" { + return + } + + desired.PackDomain = existing.PackDomain + if existing.Pack == nil { + desired.Pack = nil + return + } + pack := *existing.Pack + desired.Pack = &pack +} + +// Grove horizontal replicas are driven through scale subresources after +// creation; keep existing template values so DGD replica changes do not update +// the PodCliqueSet spec. +func preserveGrovePodCliqueSetReplicas( + desired *grovev1alpha1.PodCliqueSet, + existing *grovev1alpha1.PodCliqueSet, + checkpointInfoByComponent ...map[string]*checkpoint.CheckpointInfo, +) { + if desired == nil || existing == nil { + return + } + replicaPreserveSkips := map[string]struct{}{} + if len(checkpointInfoByComponent) > 0 { + for componentName, info := range checkpointInfoByComponent[0] { + if info != nil && + info.Enabled && + info.StartupPolicy == nvidiacomv1alpha1.CheckpointStartupPolicyWaitForCheckpoint && + !info.Ready { + replicaPreserveSkips[strings.ToLower(componentName)] = struct{}{} + } + } + } + + cliquesInScalingGroups := make(map[string]struct{}) + for _, config := range desired.Spec.Template.PodCliqueScalingGroupConfigs { + for _, cliqueName := range config.CliqueNames { + cliquesInScalingGroups[cliqueName] = struct{}{} + } + } + + cliqueReplicasByName := make(map[string]int32, len(existing.Spec.Template.Cliques)) + for _, clique := range existing.Spec.Template.Cliques { + if clique == nil || clique.Name == "" { + continue + } + cliqueReplicasByName[clique.Name] = clique.Spec.Replicas + } + for _, clique := range desired.Spec.Template.Cliques { + if clique == nil { + continue + } + if _, inScalingGroup := cliquesInScalingGroups[clique.Name]; inScalingGroup { + continue + } + if componentName := clique.Labels[commonconsts.KubeLabelDynamoComponent]; componentName != "" { + if _, skip := replicaPreserveSkips[strings.ToLower(componentName)]; skip { + continue + } + } + if replicas, ok := cliqueReplicasByName[clique.Name]; ok { + clique.Spec.Replicas = replicas + } + } + + scalingGroupReplicasByName := make( + map[string]*int32, + len(existing.Spec.Template.PodCliqueScalingGroupConfigs), + ) + for _, config := range existing.Spec.Template.PodCliqueScalingGroupConfigs { + if config.Name == "" { + // Defensive only; generated PCSG configs always have names. + continue + } + scalingGroupReplicasByName[config.Name] = config.Replicas + } + for i := range desired.Spec.Template.PodCliqueScalingGroupConfigs { + config := &desired.Spec.Template.PodCliqueScalingGroupConfigs[i] + if _, skip := replicaPreserveSkips[strings.ToLower(config.Name)]; skip { + continue + } + if replicas, ok := scalingGroupReplicasByName[config.Name]; ok { + config.Replicas = replicas + } + } +} + +func orderLikeExisting[T any](existing []T, desired []T, nameOf func(T) string) []T { + if len(existing) == 0 || len(desired) < 2 { + return desired + } + desiredByName := make(map[string]T, len(desired)) + for _, item := range desired { + if name := nameOf(item); name != "" { + desiredByName[name] = item + } + } + ordered := make([]T, 0, len(desired)) + used := make(map[string]struct{}, len(desired)) + for _, existingItem := range existing { + name := nameOf(existingItem) + if desiredItem, ok := desiredByName[name]; ok { + ordered = append(ordered, desiredItem) + used[name] = struct{}{} + } + } + for _, item := range desired { + name := nameOf(item) + if name == "" { + ordered = append(ordered, item) + continue + } + if _, ok := used[name]; !ok { + ordered = append(ordered, item) + } + } + return ordered +} + +func podCliqueTemplateName(clique *grovev1alpha1.PodCliqueTemplateSpec) string { + if clique == nil { + return "" + } + return clique.Name +} + +func podCliqueScalingGroupConfigName(config grovev1alpha1.PodCliqueScalingGroupConfig) string { + return config.Name +} + +func resourceClaimTemplateConfigName(config grovev1alpha1.ResourceClaimTemplateConfig) string { + return config.Name +} + +func groveComponentTypeCanUseLegacyWorkerSelector(componentType string) bool { + return componentType == commonconsts.ComponentTypePrefill || + componentType == commonconsts.ComponentTypeDecode +} + +func podCliqueSetHasLegacyWorkerSelector( + pcs *grovev1alpha1.PodCliqueSet, + componentName string, + componentType string, +) bool { + if pcs == nil { + return false + } + for _, clique := range pcs.Spec.Template.Cliques { + if clique == nil || clique.Labels[commonconsts.KubeLabelDynamoComponent] != componentName { + continue + } + if hasLegacyWorkerSelector(clique.Labels, componentType) { + return true + } + } + return false +} + +func applyLegacyGroveWorkerComponentType( + component *nvidiacomv1beta1.DynamoComponentDeploymentSharedSpec, + subComponentType string, +) { + component.ComponentType = nvidiacomv1beta1.ComponentTypeWorker + if component.PodTemplate == nil { + component.PodTemplate = &corev1.PodTemplateSpec{} + } + if component.PodTemplate.Labels == nil { + component.PodTemplate.Labels = map[string]string{} + } + if _, ok := component.PodTemplate.Labels[commonconsts.KubeLabelDynamoSubComponentType]; !ok { + component.PodTemplate.Labels[commonconsts.KubeLabelDynamoSubComponentType] = subComponentType + } +} diff --git a/deploy/operator/internal/controller/dynamographdeployment_program.go b/deploy/operator/internal/controller/dynamographdeployment_program.go index 22a87a443762..b14447d8dede 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_program.go +++ b/deploy/operator/internal/controller/dynamographdeployment_program.go @@ -589,10 +589,7 @@ func (r *DynamoGraphDeploymentReconciler) selectWorkloadProgram( dgd *nvidiacomv1beta1.DynamoGraphDeployment, ) workloadProgram { if r.isGrovePathway(dgd) { - return &groveProgram{ - reconciler: r, - lwsEnabled: r.RuntimeConfig.Gate.Enabled(features.LWS), - } + return newGroveProgram(r) } return &componentProgram{ reconciler: r, diff --git a/deploy/operator/internal/controller/dynamographdeployment_program_test.go b/deploy/operator/internal/controller/dynamographdeployment_program_test.go index 6e8deb6d19c3..be3e586c2a31 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_program_test.go +++ b/deploy/operator/internal/controller/dynamographdeployment_program_test.go @@ -302,12 +302,10 @@ func TestGroveProgram_ReconcilePreservesResultOnError(t *testing.T) { }, }). Build() - program := &groveProgram{ - reconciler: &DynamoGraphDeploymentReconciler{ - Client: kubeClient, - Recorder: record.NewFakeRecorder(10), - }, - } + program := newGroveProgram(&DynamoGraphDeploymentReconciler{ + Client: kubeClient, + Recorder: record.NewFakeRecorder(10), + }) dgd.Status = nvidiacomv1beta1.DynamoGraphDeploymentStatus{ State: nvidiacomv1beta1.DGDStatePending, Components: map[string]nvidiacomv1beta1.ComponentReplicaStatus{ diff --git a/deploy/operator/internal/controller/upgrade_test.go b/deploy/operator/internal/controller/upgrade_test.go index 38592a41004a..95cefb437446 100644 --- a/deploy/operator/internal/controller/upgrade_test.go +++ b/deploy/operator/internal/controller/upgrade_test.go @@ -796,29 +796,18 @@ spec: render: func(ctx context.Context, t *testing.T, parent, child client.Object) (client.Object, map[string]string) { t.Helper() - t.Log("prepare Grove render deployment from the converted DGD and existing PodCliqueSet") + t.Log("render Grove workloads from the converted DGD and existing PodCliqueSet") dgd := parent.(*v1beta1.DynamoGraphDeployment) reconciler := newUpgradeDGDReconciler(t, dgd, child) - renderDGD, existing, err := reconciler.prepareGroveRenderDeployment(ctx, dgd) - require.NoError(t, err) - require.NotNil(t, existing) - - t.Log("generate the desired Grove PodCliqueSet from the prepared render deployment") - pcs, err := dynamo.GenerateGrovePodCliqueSet( - ctx, - renderDGD, + renderer := newGroveWorkloadRenderer( + reconciler.Client, &configv1alpha1.OperatorConfiguration{}, &controller_common.RuntimeConfig{}, - reconciler.Client, - nil, - nil, - nil, nil, ) + pcs, err := renderer.Render(ctx, dgd, nil, nil) require.NoError(t, err) - - t.Log("preserve the existing PodCliqueSet clique order before comparing specs") - preserveGrovePodCliqueSetOrder(pcs, existing) + renderDGD := groveRenderDeployment(dgd, pcs) t.Log("generate the decode service selector from the same prepared Grove component") decodeComponent := renderDGD.GetComponentByName("VllmDecodeWorker") @@ -934,27 +923,16 @@ func TestGroveNativeWorkerIdentityLabelsStayNative(t *testing.T) { t.Log("seed the fake client with a native v1beta1 DGD and existing PodCliqueSet") reconciler := newUpgradeDGDReconciler(t, dgd, existingPCS) - t.Log("prepare the Grove render deployment without legacy worker selector migration") - renderDGD, existing, err := reconciler.prepareGroveRenderDeployment(ctx, dgd) - require.NoError(t, err) - require.NotNil(t, existing) - - t.Log("generate the desired PodCliqueSet from the prepared native render deployment") - desired, err := dynamo.GenerateGrovePodCliqueSet( - ctx, - renderDGD, + t.Log("render Grove workloads without legacy worker selector migration") + renderer := newGroveWorkloadRenderer( + reconciler.Client, &configv1alpha1.OperatorConfiguration{}, &controller_common.RuntimeConfig{}, - reconciler.Client, - nil, - nil, - nil, nil, ) + desired, err := renderer.Render(ctx, dgd, nil, nil) require.NoError(t, err) - - t.Log("preserve existing clique order before checking native labels") - preserveGrovePodCliqueSetOrder(desired, existing) + renderDGD := groveRenderDeployment(dgd, desired) t.Log("assert the native prefill component stays prefill instead of legacy worker") prefillComponent := renderDGD.GetComponentByName("prefill") diff --git a/deploy/operator/internal/dynamo/graph.go b/deploy/operator/internal/dynamo/graph.go index 6ebaebe15d23..3c58b541ed4a 100644 --- a/deploy/operator/internal/dynamo/graph.go +++ b/deploy/operator/internal/dynamo/graph.go @@ -2137,7 +2137,7 @@ type cliqueParams struct { restartState *RestartState existingRestartAnnotations map[string]string validatedQueueName string - kubeClient ctrlclient.Client + reader ctrlclient.Reader ctx context.Context groveClusterTopologyDomains []v1beta1.TopologyDomain } @@ -2165,7 +2165,7 @@ func buildCliqueForRole(p cliqueParams) (*grovev1alpha1.PodCliqueTemplateSpec, e p.checkpointInfo.StartupPolicy == v1alpha1.CheckpointStartupPolicyImmediate) if checkpointEnabled && p.r.Role != RoleGMS && !shouldUseAdmissionRestore { if err := checkpoint.InjectCheckpointIntoPodSpecWithStorageConfig( - p.ctx, p.kubeClient, p.dynamoDeployment.Namespace, podSpec, p.checkpointInfo, + p.ctx, p.reader, p.dynamoDeployment.Namespace, podSpec, p.checkpointInfo, p.operatorConfig.Checkpoint.Storage, p.operatorConfig.Checkpoint.EffectiveSeccompProfile(), ); err != nil { @@ -2285,16 +2285,16 @@ func applyRestartAnnotation(annotations map[string]string, componentName string, return annotations } -func resolveGroveClusterTopologyDomains(ctx context.Context, kubeClient ctrlclient.Client, kvt *v1beta1.KvTransferPolicy) ([]v1beta1.TopologyDomain, error) { +func resolveGroveClusterTopologyDomains(ctx context.Context, reader ctrlclient.Reader, kvt *v1beta1.KvTransferPolicy) ([]v1beta1.TopologyDomain, error) { if kvt == nil || kvt.ClusterTopologyName == "" { return nil, nil } - if kubeClient == nil { + if reader == nil { return nil, fmt.Errorf("spec.experimental.kvTransferPolicy.clusterTopologyName %q requires a Kubernetes client to read ClusterTopologyBinding", kvt.ClusterTopologyName) } ct := &grovev1alpha1.ClusterTopologyBinding{} - if err := kubeClient.Get(ctx, types.NamespacedName{Name: kvt.ClusterTopologyName}, ct); err != nil { + if err := reader.Get(ctx, types.NamespacedName{Name: kvt.ClusterTopologyName}, ct); err != nil { if k8serrors.IsNotFound(err) { return nil, fmt.Errorf("spec.experimental.kvTransferPolicy.clusterTopologyName %q references a ClusterTopologyBinding resource that was not found", kvt.ClusterTopologyName) } @@ -2349,7 +2349,7 @@ func GenerateGrovePodCliqueSet( dynamoDeployment *v1beta1.DynamoGraphDeployment, operatorConfig *configv1alpha1.OperatorConfiguration, runtimeConfig *controller_common.RuntimeConfig, - kubeClient ctrlclient.Client, + reader ctrlclient.Reader, secretsRetriever SecretsRetriever, restartState *RestartState, existingRestartAnnotations map[string]string, @@ -2401,7 +2401,7 @@ func GenerateGrovePodCliqueSet( var groveClusterTopologyDomains []v1beta1.TopologyDomain if dynamoDeployment.Spec.Experimental != nil { var err error - groveClusterTopologyDomains, err = resolveGroveClusterTopologyDomains(ctx, kubeClient, dynamoDeployment.Spec.Experimental.KvTransferPolicy) + groveClusterTopologyDomains, err = resolveGroveClusterTopologyDomains(ctx, reader, dynamoDeployment.Spec.Experimental.KvTransferPolicy) if err != nil { return nil, err } @@ -2463,7 +2463,7 @@ func GenerateGrovePodCliqueSet( restartState: restartState, existingRestartAnnotations: existingRestartAnnotations, validatedQueueName: validatedQueueName, - kubeClient: kubeClient, + reader: reader, ctx: ctx, groveClusterTopologyDomains: groveClusterTopologyDomains, }) From d7f86f1be904e9403bdb26dc2078c7509a5a4927 Mon Sep 17 00:00:00 2001 From: Julien Mancuso Date: Fri, 31 Jul 2026 10:41:17 -0600 Subject: [PATCH 2/2] refactor(operator): tighten workload program ownership boundaries Signed-off-by: Julien Mancuso --- .../dynamographdeployment_controller.go | 11 ------ .../dynamographdeployment_grove_program.go | 4 ++- .../dynamographdeployment_grove_renderer.go | 35 ++++++------------- .../dynamographdeployment_program.go | 28 ++++++++++++++- .../dynamographdeployment_program_test.go | 4 +-- 5 files changed, 42 insertions(+), 40 deletions(-) diff --git a/deploy/operator/internal/controller/dynamographdeployment_controller.go b/deploy/operator/internal/controller/dynamographdeployment_controller.go index fc3ed5ede489..ae43598e7d92 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_controller.go +++ b/deploy/operator/internal/controller/dynamographdeployment_controller.go @@ -310,21 +310,10 @@ func (r *DynamoGraphDeploymentReconciler) resolveProgramRestartState( ctx context.Context, dgd *nvidiacomv1beta1.DynamoGraphDeployment, status *nvidiacomv1beta1.DynamoGraphDeploymentStatus, - result *workloadProgramResult, ) programRestart { statusView := dgd.DeepCopy() statusView.Status = *status restartStatus := r.computeRestartStatus(ctx, statusView) - if restartStatus != nil && restartStatus.Phase == nvidiacomv1beta1.RestartPhaseSuperseded && - (status.Restart == nil || status.Restart.ObservedID != restartStatus.ObservedID || - status.Restart.Phase != restartStatus.Phase) { - result.Eventf( - corev1.EventTypeWarning, - "RestartSuperseded", - "Restart %s superseded by rolling update", - restartStatus.ObservedID, - ) - } return programRestart{ State: dynamo.DetermineRestartState(statusView, restartStatus), Status: restartStatus, diff --git a/deploy/operator/internal/controller/dynamographdeployment_grove_program.go b/deploy/operator/internal/controller/dynamographdeployment_grove_program.go index edc1c4d4dc20..60a5545b9421 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_grove_program.go +++ b/deploy/operator/internal/controller/dynamographdeployment_grove_program.go @@ -95,7 +95,9 @@ func (p *groveProgram) Reconcile( if err != nil { return programResult, err } - restart := p.reconciler.resolveProgramRestartState(ctx, req.DGD, &programResult.Status, &programResult) + previousRestart := programResult.Status.Restart + restart := p.reconciler.resolveProgramRestartState(ctx, req.DGD, &programResult.Status) + recordRestartTransition(previousRestart, restart.Status, &programResult) programResult.Status.Restart = restart.Status result, err := p.reconcileWorkloads(ctx, workloadReconcileRequest{ diff --git a/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go b/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go index 17c0aead5deb..23ff61e37e30 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go +++ b/deploy/operator/internal/controller/dynamographdeployment_grove_renderer.go @@ -72,9 +72,16 @@ func (r *groveWorkloadRenderer) Render( if r.reader == nil { return nil, fmt.Errorf("cannot render Grove PodCliqueSet without a Kubernetes reader") } - existingPodCliqueSet, err := r.getExistingPodCliqueSet(ctx, dgd) - if err != nil { - return nil, err + existingPodCliqueSet := &grovev1alpha1.PodCliqueSet{} + key := types.NamespacedName{ + Name: dynamo.PCSNameForDGD(dgd.Name, dgd.Spec.Components), + Namespace: dgd.Namespace, + } + if err := r.reader.Get(ctx, key, existingPodCliqueSet); err != nil { + if !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("get PodCliqueSet %s: %w", key, err) + } + existingPodCliqueSet = nil } renderDeployment := groveRenderDeployment(dgd, existingPodCliqueSet) @@ -125,28 +132,6 @@ func applyGroveCompatibility( } } -func (r *groveWorkloadRenderer) getExistingPodCliqueSet( - ctx context.Context, - dgd *nvidiacomv1beta1.DynamoGraphDeployment, -) (*grovev1alpha1.PodCliqueSet, error) { - pcs := &grovev1alpha1.PodCliqueSet{} - err := r.reader.Get( - ctx, - types.NamespacedName{ - Name: dynamo.PCSNameForDGD(dgd.Name, dgd.Spec.Components), - Namespace: dgd.Namespace, - }, - pcs, - ) - if err != nil && !apierrors.IsNotFound(err) { - return nil, fmt.Errorf("failed to get PodCliqueSet: %w", err) - } - if apierrors.IsNotFound(err) { - return nil, nil - } - return pcs, nil -} - func restartAnnotationsFromPodCliqueSet(pcs *grovev1alpha1.PodCliqueSet) map[string]string { restartAnnotations := make(map[string]string) if pcs == nil { diff --git a/deploy/operator/internal/controller/dynamographdeployment_program.go b/deploy/operator/internal/controller/dynamographdeployment_program.go index b14447d8dede..5fe8c3c58b16 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_program.go +++ b/deploy/operator/internal/controller/dynamographdeployment_program.go @@ -52,7 +52,12 @@ type workloadProgramEvent struct { type workloadProgramResult struct { ctrl.Result + // Status is the complete desired DGD status after this attempt. It remains + // meaningful when Reconcile returns an error so the outer reconciler can + // persist any partial progress before returning that error. Status nvidiacomv1beta1.DynamoGraphDeploymentStatus + // Events contains status-transition events that become real only after + // Status has been persisted successfully. Events []workloadProgramEvent } @@ -101,6 +106,25 @@ func (r *workloadProgramResult) Eventf( }) } +func recordRestartTransition( + previous *nvidiacomv1beta1.RestartStatus, + next *nvidiacomv1beta1.RestartStatus, + result *workloadProgramResult, +) { + if next == nil || next.Phase != nvidiacomv1beta1.RestartPhaseSuperseded { + return + } + if previous != nil && previous.ObservedID == next.ObservedID && previous.Phase == next.Phase { + return + } + result.Eventf( + corev1.EventTypeWarning, + "RestartSuperseded", + "Restart %s superseded by rolling update", + next.ObservedID, + ) +} + func (r *workloadProgramResult) Fail(generation int64, reason Reason, err error) { r.Status.State = nvidiacomv1beta1.DGDStateFailed meta.SetStatusCondition(&r.Status.Conditions, metav1.Condition{ @@ -248,7 +272,9 @@ func (p *componentProgram) Reconcile( ) return programResult, failWorkloadProgram(reasonNoMultinodeOrchestrator, err) } - restart := p.reconciler.resolveProgramRestartState(ctx, req.DGD, &programResult.Status, &programResult) + previousRestart := programResult.Status.Restart + restart := p.reconciler.resolveProgramRestartState(ctx, req.DGD, &programResult.Status) + recordRestartTransition(previousRestart, restart.Status, &programResult) programResult.Status.Restart = restart.Status result, err := p.reconcileWorkloads(ctx, workloadReconcileRequest{ diff --git a/deploy/operator/internal/controller/dynamographdeployment_program_test.go b/deploy/operator/internal/controller/dynamographdeployment_program_test.go index be3e586c2a31..7c26c6136c4b 100644 --- a/deploy/operator/internal/controller/dynamographdeployment_program_test.go +++ b/deploy/operator/internal/controller/dynamographdeployment_program_test.go @@ -419,7 +419,7 @@ func TestUnsupportedWorkerRolloutEmitsWarningOnlyAfterHashUpdate(t *testing.T) { } } -func TestResolveProgramRestartStateQueuesSupersededTransition(t *testing.T) { +func TestRecordRestartTransitionQueuesSupersededTransition(t *testing.T) { t.Log("Build an active rolling update that supersedes a new restart request") dgd := &nvidiacomv1beta1.DynamoGraphDeployment{ Spec: nvidiacomv1beta1.DynamoGraphDeploymentSpec{ @@ -437,8 +437,8 @@ func TestResolveProgramRestartStateQueuesSupersededTransition(t *testing.T) { context.Background(), dgd, &result.Status, - &result, ) + recordRestartTransition(result.Status.Restart, restart.Status, &result) result.Status.Restart = restart.Status t.Log("Verify status and its transition event remain coupled in the result")