Skip to content

refactor(operator): extract Grove workload rendering from graph controller #12263

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

Closed
julienmancuso wants to merge 2 commits into from
Closed

refactor(operator): extract Grove workload rendering from graph controller #12263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8b4eca1
refactor(operator): extract Grove workload rendering from graph contr…
julienmancuso Jul 28, 2026
d7f86f1
refactor(operator): tighten workload program ownership boundaries
julienmancuso Jul 31, 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
334 changes: 0 additions & 334 deletions deploy/operator/internal/controller/dynamographdeployment_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -544,329 +533,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,
Expand Down
Loading
Loading