Improve versionned resources deletion in a single apply

pkg/kapp/clusterapply/cluster_change_set.go

@@ -68,6 +68,8 @@ func (c ClusterChangeSet) Calculate() ([]*ClusterChange, *ctldgraph.ChangeGraph,
 			clusterChange.WaitOp() == ClusterChangeWaitOpNoop
 	})
 
+	c.configVersionedDeleteOrdering(changesGraph)
+
 	var clusterChanges []*ClusterChange
 
 	for _, change := range changesGraph.All() {
@@ -78,6 +80,31 @@ func (c ClusterChangeSet) Calculate() ([]*ClusterChange, *ctldgraph.ChangeGraph,
 	return clusterChanges, changesGraph, nil
 }
 
+// configures obsolete versioned resources to wait for all operations to complete before deleting
+func (c ClusterChangeSet) configVersionedDeleteOrdering(graph *ctldgraph.ChangeGraph) {
+	var versionedDeletes []*ctldgraph.Change
+	var upsertOperations []*ctldgraph.Change
+
+	for _, change := range graph.All() {
+		cc := change.Change.(wrappedClusterChange).ClusterChange
+		_, hasVersionedAnn := cc.Resource().Annotations()[ctldiff.VersionedResAnnKey]
+		if hasVersionedAnn && cc.ApplyOp() == ClusterChangeApplyOpDelete {
+			versionedDeletes = append(versionedDeletes, change)
+		} else if !hasVersionedAnn && cc.ApplyOp() != ClusterChangeApplyOpDelete && cc.ApplyOp() != ClusterChangeApplyOpNoop {
+			upsertOperations = append(upsertOperations, change)
+		}
+	}
+
+	for _, del := range versionedDeletes {
+		for _, op := range upsertOperations {
+			if op.IsTransitivelyWaitingFor(del) {
+				continue
+			}
+			del.WaitingFor = append(del.WaitingFor, op)
+		}
+	}
+}
+
 func (c ClusterChangeSet) markChangesToWait(change *ctldgraph.Change) bool {
 	var needsWaiting bool
 	for _, ch := range change.WaitingFor {

pkg/kapp/diff/change_set_with_versioned_rs.go

@@ -13,7 +13,7 @@
 )
 
 const (
-	versionedResAnnKey        = "kapp.k14s.io/versioned"               // Value is ignored
+	VersionedResAnnKey        = "kapp.k14s.io/versioned"               // Value is ignored
 	versionedResOrigAnnKey    = "kapp.k14s.io/versioned-keep-original" // Value is ignored
 	versionedResNumVersAnnKey = "kapp.k14s.io/num-versions"
 )
@@ -42,21 +42,21 @@
 
 	// First try to calculate changes will update references on all resources
 	// (which includes versioned and non-versioned resources)
-	_, _, err := d.addAndKeepChanges(newRs, existingRsGrouped)
+	_, _, _, err := d.addAndKeepChanges(newRs, existingRsGrouped)
 	if err != nil {
 		return nil, err
 	}
 
 	// Since there might have been circular dependencies;
 	// second try catches ones that werent changed during first run
-	addChanges, alreadyAdded, err := d.addAndKeepChanges(newRs, existingRsGrouped)
+	addChanges, alreadyAdded, newVersionAdded, err := d.addAndKeepChanges(newRs, existingRsGrouped)
 	if err != nil {
 		return nil, err
 	}
 
 	allChanges = append(allChanges, addChanges...)
 
-	keepAndDeleteChanges, err := d.noopAndDeleteChanges(existingRsGrouped, alreadyAdded)
+	keepAndDeleteChanges, err := d.noopAndDeleteChanges(existingRsGrouped, alreadyAdded, newVersionAdded)
 	if err != nil {
 		return nil, err
 	}
@@ -95,10 +95,11 @@
 
 func (d ChangeSetWithVersionedRs) addAndKeepChanges(
 	newRs versionedResources, existingRsGrouped map[string][]ctlres.Resource) (
-	[]Change, map[string]ctlres.Resource, error) {
+	[]Change, map[string]ctlres.Resource, map[string]struct{}, error) {
 
 	changes := []Change{}
 	alreadyAdded := map[string]ctlres.Resource{}
+	newVersionAdded := map[string]struct{}{} // keys where a genuinely new version was created
 
 	for _, newRes := range newRs.Versioned {
 		newResKey := VersionedResource{newRes, nil}.UniqVersionedKey().String()
@@ -110,24 +111,26 @@
 			// Calculate update change to determine if anything changed
 			updateChange, err := d.newChange(existingRes, newRes)
 			if err != nil {
-				return nil, nil, err
+				return nil, nil, nil, err
 			}
 
 			switch updateChange.Op() {
 			case ChangeOpUpdate:
 				changes = append(changes, d.newAddChangeFromUpdateChange(newRes, updateChange))
+				newVersionAdded[newResKey] = struct{}{}
 			case ChangeOpKeep:
 				// Use latest copy of resource to update affected resources
 				usedRes = existingRes
 				changes = append(changes, d.newKeepChange(existingRes))
+				// Not added to newVersionAdded: content unchanged, no new version created
 			default:
 				panic(fmt.Sprintf("Unexpected change op %s", updateChange.Op()))
 			}
 		} else {
 			// Since there no existing resource, create change for new resource
 			addChange, err := d.newChange(nil, newRes)
 			if err != nil {
-				return nil, nil, err
+				return nil, nil, nil, err
 			}
 			changes = append(changes, addChange)
 		}
@@ -137,18 +140,18 @@
 
 		err := verRes.UpdateAffected(newRs.NonVersioned)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 
 		err = verRes.UpdateAffected(newRs.Versioned)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 
 		alreadyAdded[newResKey] = newRes
 	}
 
-	return changes, alreadyAdded, nil
+	return changes, alreadyAdded, newVersionAdded, nil
 }
 
 func (d ChangeSetWithVersionedRs) newAddChangeFromUpdateChange(
@@ -160,7 +163,8 @@
 
 func (d ChangeSetWithVersionedRs) noopAndDeleteChanges(
 	existingRsGrouped map[string][]ctlres.Resource,
-	alreadyAdded map[string]ctlres.Resource) ([]Change, error) {
+	alreadyAdded map[string]ctlres.Resource,
+	newVersionAdded map[string]struct{}) ([]Change, error) {
 
 	changes := []Change{}
 
@@ -174,6 +178,11 @@
 			if err != nil {
 				return nil, err
 			}
+			// A new version is being added, we will have len(existingRs)+1 versions after apply
+			// reducing numToKeep by 1 allows to delete the obsolete version immediately rather than on the following apply.
+			if _, isNewVersion := newVersionAdded[existingResKey]; isNewVersion {
+				numToKeep = numToKeep - 1
+			}
 		}
 		if numToKeep > len(existingRs) {
 			numToKeep = len(existingRs)
@@ -241,7 +250,7 @@
 func newVersionedResources(rs []ctlres.Resource) versionedResources {
 	var result versionedResources
 	for _, res := range rs {
-		_, hasVersionedAnn := res.Annotations()[versionedResAnnKey]
+		_, hasVersionedAnn := res.Annotations()[VersionedResAnnKey]
 		_, hasVersionedOrigAnn := res.Annotations()[versionedResOrigAnnKey]
 
 		if hasVersionedAnn {
@@ -262,7 +271,7 @@
 		// Expect that versioned resources should not be transient
 		// (Annotations may have been copied from versioned resources
 		// onto transient resources for non-versioning related purposes).
-		_, hasVersionedAnn := res.Annotations()[versionedResAnnKey]
+		_, hasVersionedAnn := res.Annotations()[VersionedResAnnKey]
 
 		versionedRs := VersionedResource{res: res}
 		_, version := versionedRs.BaseNameAndVersion()
@@ -280,7 +289,7 @@
 	result := map[string][]ctlres.Resource{}
 
 	groupByFunc := func(res ctlres.Resource) string {
-		_, found := res.Annotations()[versionedResAnnKey]
+		_, found := res.Annotations()[VersionedResAnnKey]
 		if found {
 			return VersionedResource{res, nil}.UniqVersionedKey().String()
 		}

pkg/kapp/diff/change_set_with_versioned_rs_test.go

@@ -164,3 +164,73 @@ func checkChangeDiff(t *testing.T, change Change, expectedDiff string) {
 
 	require.Equal(t, expectedDiff, actualDiffString, "Expected diff to match")
 }
+
+func TestChangeSet_DeleteObsoleteVersions_InSameApply(t *testing.T) {
+	// Test with custom num-versions annotation set to 3
+
+	existingV1 := ctlres.MustNewResourceFromBytes([]byte(`
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-ver-1
+  annotations:
+    kapp.k14s.io/versioned: ""
+data:
+  key: value1
+`))
+
+	existingV2 := ctlres.MustNewResourceFromBytes([]byte(`
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-ver-2
+  annotations:
+    kapp.k14s.io/versioned: ""
+data:
+  key: value2
+`))
+
+	existingV3 := ctlres.MustNewResourceFromBytes([]byte(`
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-ver-3
+  annotations:
+    kapp.k14s.io/versioned: ""
+data:
+  key: value3
+`))
+
+	// Deploy a new version with num-versions: 3
+	newConfig := ctlres.MustNewResourceFromBytes([]byte(`
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config
+  annotations:
+    kapp.k14s.io/versioned: ""
+    kapp.k14s.io/num-versions: "3"
+data:
+  key: value4
+`))
+
+	existingResources := []ctlres.Resource{existingV1, existingV2, existingV3}
+	changeSetWithVerRes := NewChangeSetWithVersionedRs(existingResources, []ctlres.Resource{newConfig}, nil,
+		ChangeSetOpts{}, ChangeFactory{})
+
+	changes, err := changeSetWithVerRes.Calculate()
+	require.NoError(t, err)
+
+	// Expect: 1 add (v4), 1 delete (v1), 2 noop (v2-v3)
+	require.Len(t, changes, 4, "Expected 4 changes: 1 add, 1 delete, 2 noop")
+
+	var deleteCount int
+	for _, change := range changes {
+		if change.Op() == ChangeOpDelete {
+			deleteCount++
+			require.Equal(t, "config-ver-1", change.ExistingResource().Name(), "Expected v1 to be deleted")
+		}
+	}
+
+	require.Equal(t, 1, deleteCount, "Expected 1 delete operation (v1 should be deleted immediately)")
+}