fix: reduce spurious resource updates (#4868)

Signed-off-by: Julien Mancuso <jmancuso@nvidia.com>
Co-authored-by: tmontfort <tmontfort@nvidia.com>

deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller_test.go

@@ -1231,8 +1231,13 @@ func TestDynamoComponentDeploymentReconciler_createOrUpdateOrDeleteDeployments_R
 	g.Expect(*createdDeployment.Spec.Replicas).To(gomega.Equal(int32(1)), "Initial deployment should have 1 replica")
 
 	// Step 2: Manually update the deployment to 2 replicas (simulating manual edit)
+	// Note: Real Kubernetes API server increments generation on spec changes,
+	// but the fake client doesn't, so we simulate it here.
+	// The operator sets last-applied-generation=1 on create, so we need generation > 1
+	// to trigger manual change detection.
 	manualReplicaCount := int32(2)
 	createdDeployment.Spec.Replicas = &manualReplicaCount
+	createdDeployment.Generation = 2 // Simulate K8s incrementing generation on spec change
 	err = fakeKubeClient.Update(ctx, createdDeployment)
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 
@@ -1255,6 +1260,106 @@ func TestDynamoComponentDeploymentReconciler_createOrUpdateOrDeleteDeployments_R
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 	g.Expect(reconciledDeployment.Spec.Replicas).NotTo(gomega.BeNil())
 	g.Expect(*reconciledDeployment.Spec.Replicas).To(gomega.Equal(int32(1)), "Deployment should have been reconciled back to 1 replica")
+
+	// Step 5: Call createOrUpdateOrDeleteDeployments again - it should not be modified
+	modified3, deployment3, err := reconciler.createOrUpdateOrDeleteDeployments(ctx, opt)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(modified3).To(gomega.BeFalse(), "Deployment should have been not modified")
+	g.Expect(deployment3).NotTo(gomega.BeNil())
+}
+
+func Test_createOrUpdateOrDeleteDeployments_K8sAPIDefaults(t *testing.T) {
+	g := gomega.NewGomegaWithT(t)
+	ctx := context.Background()
+
+	// Set up scheme
+	s := scheme.Scheme
+	err := v1alpha1.AddToScheme(s)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	err = appsv1.AddToScheme(s)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	err = corev1.AddToScheme(s)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+
+	name := "test-component"
+	namespace := defaultNamespace
+
+	// Create DynamoComponentDeployment
+	replicaCount := int32(3)
+	dcd := &v1alpha1.DynamoComponentDeployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: v1alpha1.DynamoComponentDeploymentSpec{
+			BackendFramework: string(dynamo.BackendFrameworkVLLM),
+			DynamoComponentDeploymentSharedSpec: v1alpha1.DynamoComponentDeploymentSharedSpec{
+				ServiceName:     "test-service",
+				DynamoNamespace: ptr.To("default"),
+				ComponentType:   string(commonconsts.ComponentTypeDecode),
+				Replicas:        &replicaCount,
+			},
+		},
+	}
+
+	fakeKubeClient := fake.NewClientBuilder().
+		WithScheme(s).
+		WithObjects(dcd).
+		Build()
+
+	recorder := record.NewFakeRecorder(100)
+	reconciler := &DynamoComponentDeploymentReconciler{
+		Client:      fakeKubeClient,
+		Recorder:    recorder,
+		Config:      controller_common.Config{},
+		EtcdStorage: nil,
+		DockerSecretRetriever: &mockDockerSecretRetriever{
+			GetSecretsFunc: func(namespace, imageName string) ([]string, error) {
+				return []string{}, nil
+			},
+		},
+	}
+
+	opt := generateResourceOption{
+		dynamoComponentDeployment: dcd,
+	}
+
+	t.Log("=== Step 1: Create deployment (operator's first apply) ===")
+
+	modified1, deployment1, err := reconciler.createOrUpdateOrDeleteDeployments(ctx, opt)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(modified1).To(gomega.BeTrue(), "First create should report as modified")
+	g.Expect(deployment1).NotTo(gomega.BeNil())
+	g.Expect(deployment1.Spec.RevisionHistoryLimit).To(gomega.BeNil())
+
+	operatorCreatedDeployment := &appsv1.Deployment{}
+	err = fakeKubeClient.Get(ctx, client.ObjectKey{Name: name, Namespace: namespace}, operatorCreatedDeployment)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(*operatorCreatedDeployment.Spec.Replicas).To(gomega.Equal(replicaCount))
+
+	annotations := operatorCreatedDeployment.GetAnnotations()
+	g.Expect(annotations).NotTo(gomega.BeNil())
+	originalHash, hasHash := annotations[controller_common.NvidiaAnnotationHashKey]
+	g.Expect(hasHash).To(gomega.BeTrue(), "Hash annotation should be set")
+	t.Logf("Hash annotation after create: %s", originalHash)
+
+	t.Log("\n=== Step 2: Simulate K8s adding defaults ===")
+
+	// Operator does not set RevisionHistoryLimit but the k8s API defaults to 10
+	operatorCreatedDeployment.Spec.RevisionHistoryLimit = ptr.To(int32(10))
+	err = fakeKubeClient.Update(ctx, operatorCreatedDeployment)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+
+	// The deployment should not be modified because the spec is the same
+	modified2, deployment2, err := reconciler.createOrUpdateOrDeleteDeployments(ctx, opt)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(modified2).To(gomega.BeFalse(), "Second create should report as not modified")
+	g.Expect(deployment2).NotTo(gomega.BeNil())
+
+	modified3, deployment3, err := reconciler.createOrUpdateOrDeleteDeployments(ctx, opt)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(modified3).To(gomega.BeFalse(), "Third create should report as not modified")
+	g.Expect(deployment3).NotTo(gomega.BeNil())
 }
 
 func Test_reconcileLeaderWorkerSetResources(t *testing.T) {

deploy/cloud/operator/internal/controller_common/resource.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
+	"strconv"
 
 	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/api/v1alpha1"
 	"github.com/ai-dynamo/dynamo/deploy/cloud/operator/internal/consts"
@@ -44,6 +45,9 @@ import (
 const (
 	// NvidiaAnnotationHashKey indicates annotation name for last applied hash by the operator
 	NvidiaAnnotationHashKey = "nvidia.com/last-applied-hash"
+	// NvidiaAnnotationGenerationKey indicates annotation name for last applied generation by the operator
+	// This is used to detect manual changes to resources
+	NvidiaAnnotationGenerationKey = "nvidia.com/last-applied-generation"
 )
 
 type Reconciler interface {
@@ -120,7 +124,8 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 			return
 		}
 
-		updateHashAnnotation(resource, hash)
+		// On create, set generation to 1 (new resources start at generation 1)
+		updateAnnotations(resource, hash, 1)
 
 		r.GetRecorder().Eventf(parentResource, corev1.EventTypeNormal, fmt.Sprintf("Create%s", resourceType), "Creating a new %s %s", resourceType, resourceNamespace)
 		err = r.Create(ctx, resource)
@@ -136,7 +141,7 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 	} else {
 		logs.Info(fmt.Sprintf("%s found.", resourceType))
 		if toDelete {
-			logs.Info(fmt.Sprintf("%s not found. Deleting the existing one.", resourceType))
+			logs.Info(fmt.Sprintf("%s found. Deleting the existing one.", resourceType))
 			err = r.Delete(ctx, oldResource)
 			if err != nil {
 				logs.Error(err, fmt.Sprintf("Failed to delete %s.", resourceType))
@@ -150,13 +155,27 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 		}
 
 		// Check if the Spec has changed and update if necessary
-		var newHash *string
-		newHash, err = IsSpecChanged(oldResource, resource)
+		var changeResult SpecChangeResult
+		changeResult, err = GetSpecChangeResult(oldResource, resource)
 		if err != nil {
 			r.GetRecorder().Eventf(parentResource, corev1.EventTypeWarning, fmt.Sprintf("CalculatePatch%s", resourceType), "Failed to calculate patch for %s %s: %s", resourceType, resourceNamespace, err)
 			return false, resource, fmt.Errorf("failed to check if spec has changed: %w", err)
 		}
-		if newHash != nil {
+
+		if !changeResult.NeedsUpdate {
+			logs.Info(fmt.Sprintf("%s spec is the same. Skipping update.", resourceType))
+			r.GetRecorder().Eventf(parentResource, corev1.EventTypeNormal, fmt.Sprintf("Update%s", resourceType), "Skipping update %s %s", resourceType, resourceNamespace)
+			res = oldResource
+			return
+		}
+
+		// Log if manual changes were detected
+		if changeResult.ManualChangeDetected {
+			logs.Info(fmt.Sprintf("Manual changes detected on %s, will be overwritten", resourceType),
+				"currentGeneration", oldResource.GetGeneration(),
+				"lastAppliedGeneration", getAnnotation(oldResource, NvidiaAnnotationGenerationKey))
+		}
+
 		// Generate and log diff before updating
 		diff, diffErr := generateSpecDiff(oldResource, resource)
 		if diffErr != nil {
@@ -165,7 +184,7 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 			logs.Info(fmt.Sprintf("%s spec changes detected", resourceType), "diff", diff)
 		}
 
-			// update the spec of the current object with the desired spec
+		// Update the spec of the current object with the desired spec
 		err = CopySpec(resource, oldResource)
 		if err != nil {
 			logs.Error(err, fmt.Sprintf("Failed to copy spec for %s.", resourceType))
@@ -173,7 +192,7 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 			return
 		}
 
-			updateHashAnnotation(oldResource, *newHash)
+		updateAnnotations(oldResource, *changeResult.NewHash, changeResult.NewGeneration)
 
 		err = r.Update(ctx, oldResource)
 		if err != nil {
@@ -185,11 +204,6 @@ func SyncResource[T client.Object](ctx context.Context, r Reconciler, parentReso
 		r.GetRecorder().Eventf(parentResource, corev1.EventTypeNormal, fmt.Sprintf("Update%s", resourceType), "Updated %s %s", resourceType, resourceNamespace)
 		modified = true
 		res = oldResource
-		} else {
-			logs.Info(fmt.Sprintf("%s spec is the same. Skipping update.", resourceType))
-			r.GetRecorder().Eventf(parentResource, corev1.EventTypeNormal, fmt.Sprintf("Update%s", resourceType), "Skipping update %s %s", resourceType, resourceNamespace)
-			res = oldResource
-		}
 	}
 	return
 }
@@ -247,27 +261,102 @@ func getSpec(obj client.Object) (any, error) {
 	return spec, nil
 }
 
-// IsSpecChanged returns the new hash if the spec has changed between the existing one
-// It compares the actual current spec hash with the desired spec hash to detect manual edits
-func IsSpecChanged(current client.Object, desired client.Object) (*string, error) {
+// SpecChangeResult contains the result of spec change detection
+type SpecChangeResult struct {
+	// NewHash is the hash to set in the annotation (nil if no update needed)
+	NewHash *string
+	// NewGeneration is the generation to set in the annotation
+	NewGeneration int64
+	// NeedsUpdate indicates whether the resource needs to be updated
+	NeedsUpdate bool
+	// ManualChangeDetected indicates whether a manual change was detected
+	ManualChangeDetected bool
+}
+
+// GetSpecChangeResult determines if a resource needs to be updated by comparing the desired spec hash
+// with the last applied hash annotation. It also tracks generation to detect manual changes.
+//
+// Returns:
+//   - SpecChangeResult with update information
+//   - error if hash computation fails
+func GetSpecChangeResult(current client.Object, desired client.Object) (SpecChangeResult, error) {
 	desiredHash, err := GetSpecHash(desired)
 	if err != nil {
-		return nil, err
+		return SpecChangeResult{}, err
+	}
+
+	lastAppliedHash := getAnnotation(current, NvidiaAnnotationHashKey)
+	lastAppliedGenStr := getAnnotation(current, NvidiaAnnotationGenerationKey)
+	currentGen := current.GetGeneration()
+
+	// Case 1: Hash annotation missing (external create or pre-upgrade resource)
+	// Note: This is not first-time CREATE (handled separately in SyncResource with generation=1).
+	// This handles existing resources without our annotations - we're about to update them,
+	// so NewGeneration = currentGen + 1 is correct.
+	if lastAppliedHash == "" {
+		return SpecChangeResult{
+			NewHash:       &desiredHash,
+			NewGeneration: currentGen + 1,
+			NeedsUpdate:   true,
+		}, nil
+	}
+
+	// Case 2: Hash different (spec changed)
+	if desiredHash != lastAppliedHash {
+		return SpecChangeResult{
+			NewHash:       &desiredHash,
+			NewGeneration: currentGen + 1,
+			NeedsUpdate:   true,
+		}, nil
 	}
 
-	// Compute hash of the actual current spec (not just the annotation)
-	// This ensures we detect manual edits even if the annotation is stale
-	currentHash, err := GetSpecHash(current)
+	// Case 3: Hash same, but generation annotation missing (upgrade scenario)
+	// Do a full update to ensure spec is exactly what we want - there could have been
+	// manual edits before we added generation tracking. The cost is one extra Update
+	// per resource during upgrade, but on next reconcile generations will match.
+	if lastAppliedGenStr == "" {
+		return SpecChangeResult{
+			NewHash:       &desiredHash,
+			NewGeneration: currentGen + 1,
+			NeedsUpdate:   true,
+		}, nil
+	}
+
+	// Case 4: Both annotations exist, check for manual changes
+	lastAppliedGen, err := strconv.ParseInt(lastAppliedGenStr, 10, 64)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get current spec hash: %w", err)
+		// Corrupted annotation, force update to fix
+		return SpecChangeResult{
+			NewHash:       &desiredHash,
+			NewGeneration: currentGen + 1,
+			NeedsUpdate:   true,
+		}, nil
 	}
 
-	// Compare actual spec hashes
-	if currentHash == desiredHash {
-		return nil, nil
+	// Detect manual changes: if current generation > last applied generation,
+	// someone else modified the resource after our last update
+	if currentGen > 0 && currentGen > lastAppliedGen {
+		return SpecChangeResult{
+			NewHash:              &desiredHash,
+			NewGeneration:        currentGen + 1,
+			NeedsUpdate:          true,
+			ManualChangeDetected: true,
+		}, nil
 	}
 
-	return &desiredHash, nil
+	// No update needed
+	return SpecChangeResult{
+		NeedsUpdate: false,
+	}, nil
+}
+
+// getAnnotation safely retrieves an annotation value from an object
+func getAnnotation(obj client.Object, key string) string {
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		return ""
+	}
+	return annotations[key]
 }
 
 // generateSpecDiff creates a unified diff showing changes between old and new resource specs
@@ -299,12 +388,14 @@ func GetSpecHash(obj client.Object) (string, error) {
 	return GetResourceHash(spec)
 }
 
-func updateHashAnnotation(obj client.Object, hash string) {
+// updateAnnotations sets both hash and generation annotations on an object
+func updateAnnotations(obj client.Object, hash string, generation int64) {
 	annotations := obj.GetAnnotations()
 	if annotations == nil {
 		annotations = map[string]string{}
 	}
 	annotations[NvidiaAnnotationHashKey] = hash
+	annotations[NvidiaAnnotationGenerationKey] = strconv.FormatInt(generation, 10)
 	obj.SetAnnotations(annotations)
 }
 

deploy/cloud/operator/internal/controller_common/resource_test.go

@@ -31,7 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func TestIsSpecChanged2(t *testing.T) {
+func TestGetSpecChangeResult(t *testing.T) {
 	tests := []struct {
 		name          string
 		current       client.Object
@@ -407,16 +407,182 @@ func TestIsSpecChanged2(t *testing.T) {
 			if err != nil {
 				t.Errorf("failed to get spec hash in test for resource %s: %s", tt.current.GetName(), err)
 			}
-			updateHashAnnotation(tt.current, hash)
-			gotHash, err := IsSpecChanged(tt.current, tt.desired)
+			// Set both hash and generation annotations (generation=1 simulates initial state)
+			updateAnnotations(tt.current, hash, 1)
+			result, err := GetSpecChangeResult(tt.current, tt.desired)
 			if err != nil {
 				t.Errorf("failed to check if spec has changed in test for resource %s: %s", tt.current.GetName(), err)
 			}
-			if tt.expectedHash && gotHash == nil {
-				t.Errorf("IsSpecChanged() = %v, want %v", gotHash, tt.expectedHash)
+			if tt.expectedHash && !result.NeedsUpdate {
+				t.Errorf("GetSpecChangeResult() NeedsUpdate = %v, want %v", result.NeedsUpdate, tt.expectedHash)
 			}
-			if !tt.expectedHash && gotHash != nil {
-				t.Errorf("IsSpecChanged() = %v, want %v", gotHash, tt.expectedHash)
+			if !tt.expectedHash && result.NeedsUpdate {
+				t.Errorf("GetSpecChangeResult() NeedsUpdate = %v, want %v", result.NeedsUpdate, tt.expectedHash)
+			}
+		})
+	}
+}
+
+func TestGetSpecChangeResult_GenerationTracking(t *testing.T) {
+	tests := []struct {
+		name                       string
+		currentGeneration          int64
+		lastAppliedGeneration      string // empty string means annotation not set
+		lastAppliedHash            string // empty string means annotation not set, "match" means compute from desired
+		desiredReplicas            int64  // different from current (2) means hash will differ
+		expectNeedsUpdate          bool
+		expectManualChangeDetected bool
+		expectNewGeneration        int64 // 0 means don't check
+	}{
+		{
+			name:                  "no change - generations and hash match",
+			currentGeneration:     5,
+			lastAppliedGeneration: "5",
+			lastAppliedHash:       "match",
+			desiredReplicas:       2, // same as current
+			expectNeedsUpdate:     false,
+		},
+		{
+			name:                       "manual change detected - generation increased",
+			currentGeneration:          7,
+			lastAppliedGeneration:      "5",
+			lastAppliedHash:            "match",
+			desiredReplicas:            2,
+			expectNeedsUpdate:          true,
+			expectManualChangeDetected: true,
+			expectNewGeneration:        8, // current(7) + 1
+		},
+		{
+			// Upgrade scenario: hash matches but no generation annotation yet.
+			// We do a full update to ensure spec is correct (could have been manual edits
+			// before we added generation tracking).
+			name:                  "missing generation annotation - full update for safety",
+			currentGeneration:     5,
+			lastAppliedGeneration: "", // missing
+			lastAppliedHash:       "match",
+			desiredReplicas:       2,
+			expectNeedsUpdate:     true,
+			expectNewGeneration:   6, // current + 1
+		},
+		{
+			name:                  "missing hash annotation - needs full update",
+			currentGeneration:     5,
+			lastAppliedGeneration: "5",
+			lastAppliedHash:       "", // missing
+			desiredReplicas:       2,
+			expectNeedsUpdate:     true,
+			expectNewGeneration:   6, // current(5) + 1
+		},
+		{
+			name:                  "hash changed - needs full update",
+			currentGeneration:     5,
+			lastAppliedGeneration: "5",
+			lastAppliedHash:       "match",
+			desiredReplicas:       3, // different from current (2)
+			expectNeedsUpdate:     true,
+			expectNewGeneration:   6, // current(5) + 1
+		},
+		{
+			name:                  "corrupted generation annotation - needs full update",
+			currentGeneration:     5,
+			lastAppliedGeneration: "invalid",
+			lastAppliedHash:       "match",
+			desiredReplicas:       2,
+			expectNeedsUpdate:     true,
+			expectNewGeneration:   6, // current(5) + 1
+		},
+		{
+			name:                  "both annotations missing - needs full update",
+			currentGeneration:     5,
+			lastAppliedGeneration: "",
+			lastAppliedHash:       "",
+			desiredReplicas:       2,
+			expectNeedsUpdate:     true,
+			expectNewGeneration:   6, // current(5) + 1
+		},
+		{
+			name:                       "manual change with hash also changed",
+			currentGeneration:          7,
+			lastAppliedGeneration:      "5",
+			lastAppliedHash:            "match",
+			desiredReplicas:            3, // different
+			expectNeedsUpdate:          true,
+			expectManualChangeDetected: false, // hash change takes precedence
+			expectNewGeneration:        8,
+		},
+		{
+			// Generation=0 can occur with CRDs that don't have generation tracking enabled,
+			// or as a safety net for edge cases. When gen=0, we skip generation-based
+			// manual change detection and rely solely on hash comparison.
+			name:                  "generation zero - skip generation check",
+			currentGeneration:     0,
+			lastAppliedGeneration: "0",
+			lastAppliedHash:       "match",
+			desiredReplicas:       2,
+			expectNeedsUpdate:     false, // gen check skipped when gen=0, hash matches
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := gomega.NewGomegaWithT(t)
+
+			// Create current resource
+			current := &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"apiVersion": "apps/v1",
+					"kind":       "Deployment",
+					"metadata": map[string]interface{}{
+						"name":        "test-deployment",
+						"namespace":   "default",
+						"generation":  tt.currentGeneration,
+						"annotations": map[string]interface{}{},
+					},
+					"spec": map[string]interface{}{
+						"replicas": int64(2),
+					},
+				},
+			}
+
+			// Create desired resource
+			desired := &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"apiVersion": "apps/v1",
+					"kind":       "Deployment",
+					"metadata": map[string]interface{}{
+						"name":      "test-deployment",
+						"namespace": "default",
+					},
+					"spec": map[string]interface{}{
+						"replicas": tt.desiredReplicas,
+					},
+				},
+			}
+
+			// Set annotations based on test case
+			// "match" means the lastAppliedHash should match the CURRENT spec's hash
+			// (simulating that operator last applied what's currently in the cluster)
+			annotations := make(map[string]string)
+			if tt.lastAppliedHash == "match" {
+				hash, err := GetSpecHash(current)
+				g.Expect(err).To(gomega.BeNil())
+				annotations[NvidiaAnnotationHashKey] = hash
+			} else if tt.lastAppliedHash != "" {
+				annotations[NvidiaAnnotationHashKey] = tt.lastAppliedHash
+			}
+			if tt.lastAppliedGeneration != "" {
+				annotations[NvidiaAnnotationGenerationKey] = tt.lastAppliedGeneration
+			}
+			if len(annotations) > 0 {
+				current.SetAnnotations(annotations)
+			}
+
+			result, err := GetSpecChangeResult(current, desired)
+			g.Expect(err).To(gomega.BeNil())
+			g.Expect(result.NeedsUpdate).To(gomega.Equal(tt.expectNeedsUpdate), "NeedsUpdate mismatch")
+			g.Expect(result.ManualChangeDetected).To(gomega.Equal(tt.expectManualChangeDetected), "ManualChangeDetected mismatch")
+			if tt.expectNewGeneration != 0 {
+				g.Expect(result.NewGeneration).To(gomega.Equal(tt.expectNewGeneration), "NewGeneration mismatch")
 			}
 		})
 	}