Add watch filtering by applyset label

- Add "applyset.kubernetes.io/part-of" label to managed objects
- Add "applyset.kubernetes.io/id" label to RootSyncs/RepoSyncs
- Generate ApplySet IDs using kubectl code for compatibility.
  Note: kubectl impl is slightly different than the KEP, in that
  it uses "applyset.kubernetes.io" instead of "applyset.k8s.io"

e2e/testcases/custom_resources_test.go

@@ -109,16 +109,12 @@ func TestCRDDeleteBeforeRemoveCustomResourceV1(t *testing.T) {
 		nt.T.Fatal(err)
 	}
 
-	// Wait for remediator to detect the drift (managed Anvil CR was deleted)
-	// and record it as a conflict.
-	// TODO: distinguish between management conflict (spec/generation drift) and concurrent status update conflict (resource version change)
-	err = nomostest.ValidateMetrics(nt,
+	// Reverting a manual change is NOT considered a "conflict", unless the new
+	// owner is a different reconciler.
+	nt.Must(nomostest.ValidateMetrics(nt,
 		nomostest.ReconcilerErrorMetrics(nt, rootSyncLabels, firstCommitHash, metrics.ErrorSummary{
-			Conflicts: 1,
-		}))
-	if err != nil {
-		nt.T.Fatal(err)
-	}
+			Conflicts: 0, // from the remediator
+		})))
 
 	// Reset discovery client to invalidate the cached Anvil CRD
 	nt.RenewClient()

e2e/testcases/managed_resources_test.go

@@ -37,19 +37,21 @@ import (
 
 // This file includes tests for drift correction and drift prevention.
 //
-// The drift correction in the mono-repo mode utilizes the following two annotations:
-//  * configmanagement.gke.io/managed
-//  * configsync.gke.io/resource-id
+// The drift correction in the mono-repo mode utilizes the following metadata:
+//  * the configmanagement.gke.io/managed annotation
+//  * the configsync.gke.io/resource-id annotation
 //
-// The drift correction in the multi-repo mode utilizes the following three annotations:
-//  * configmanagement.gke.io/managed
-//  * configsync.gke.io/resource-id
-//  * configsync.gke.io/manager
+// The drift correction in the multi-repo mode utilizes the following metadata:
+//  * the configmanagement.gke.io/managed annotation
+//  * the configsync.gke.io/resource-id annotation
+//  * the configsync.gke.io/manager annotation
+//  * the config.k8s.io/owning-inventory label
 //
 // The drift prevention is only supported in the multi-repo mode, and utilizes the following Config Sync metadata:
 //  * the configmanagement.gke.io/managed annotation
 //  * the configsync.gke.io/resource-id annotation
 //  * the configsync.gke.io/declared-version label
+//  * the config.k8s.io/owning-inventory label
 
 // The reason we have both TestKubectlCreatesManagedNamespaceResourceMonoRepo and
 // TestKubectlCreatesManagedNamespaceResourceMultiRepo is that the mono-repo mode and
@@ -79,6 +81,8 @@ metadata:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _namespace_test-ns1
     configsync.gke.io/manager: :root
+  labels:
+    configsync.gke.io/parent-package-id: root-sync.config-management-system.RootSync
 `)
 
 	if err := os.WriteFile(filepath.Join(nt.TmpDir, "test-ns1.yaml"), ns, 0644); err != nil {
@@ -106,7 +110,9 @@ metadata:
   annotations:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _namespace_test-ns2
-    configsync.gke.io/manager: :abcdef
+	configsync.gke.io/manager: config-management-system_abcdef
+  labels:
+    configsync.gke.io/parent-package-id: abcdef.config-management-system.RootSync
 `)
 
 	if err := os.WriteFile(filepath.Join(nt.TmpDir, "test-ns2.yaml"), ns, 0644); err != nil {
@@ -121,8 +127,10 @@ metadata:
 	// Wait 5 seconds so that the remediator can process the event.
 	time.Sleep(5 * time.Second)
 
-	// The `configsync.gke.io/manager` annotation of `test-ns2` suggests that its manager is ':abcdef'.
-	// The root reconciler does not manage `test-ns2`, therefore should not remove `test-ns2`.
+	// The `configsync.gke.io/manager` annotation of `test-ns2` suggests that
+	// its manager is 'config-management-system_abcdef' (RootSync named 'abcdef').
+	// The root reconciler (RootSync named 'root-sync') does not manage
+	// `test-ns2`, therefore should not remove `test-ns2`.
 	err = nt.Validate("test-ns2", "", &corev1.Namespace{},
 		testpredicates.HasExactlyAnnotationKeys(
 			metadata.ResourceManagementKey,
@@ -233,6 +241,8 @@ metadata:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _configmap_bookstore_test-cm1
     configsync.gke.io/manager: :root
+  labels:
+    configsync.gke.io/parent-package-id: root-sync.config-management-system.RootSync
 data:
   weekday: "monday"
 `)
@@ -263,6 +273,8 @@ metadata:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _configmap_bookstore_wrong-cm2
     configsync.gke.io/manager: :root
+  labels:
+    configsync.gke.io/parent-package-id: root-sync.config-management-system.RootSync
 data:
   weekday: "monday"
 `)
@@ -301,7 +313,9 @@ metadata:
   annotations:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _configmap_bookstore_test-cm3
-    configsync.gke.io/manager: :abcdef
+    configsync.gke.io/manager: config-management-system_abcdef
+  labels:
+    configsync.gke.io/parent-package-id: abcdef.config-management-system.RootSync
 data:
   weekday: "monday"
 `)
@@ -318,8 +332,10 @@ data:
 	// Wait 5 seconds so that the reconciler can process the event.
 	time.Sleep(5 * time.Second)
 
-	// The `configsync.gke.io/manager` annotation of `test-ns3` suggests that its manager is ':abcdef'.
-	// The root reconciler does not manage `test-ns3`, therefore should not remove `test-ns3`.
+	// The `configsync.gke.io/manager` annotation of `test-ns3` suggests that
+	// its manager is 'config-management-system_abcdef' (RootSync named 'abcdef').
+	// The root reconciler (RootSync named 'root-sync') does not manage `test-ns3`,
+	// therefore should not remove `test-ns3`.
 	err = nt.Validate("test-cm3", "bookstore", &corev1.ConfigMap{},
 		testpredicates.HasExactlyAnnotationKeys(
 			metadata.ResourceManagementKey,
@@ -340,6 +356,8 @@ metadata:
   annotations:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _configmap_bookstore_test-cm4
+  labels:
+    configsync.gke.io/parent-package-id: root-sync.config-management-system.RootSync
 data:
   weekday: "monday"
 `)
@@ -378,6 +396,8 @@ metadata:
     configmanagement.gke.io/managed: enabled
     configsync.gke.io/resource-id: _configmap_bookstore_test-secret
     configsync.gke.io/manager: :root
+  labels:
+    configsync.gke.io/parent-package-id: root-sync.config-management-system.RootSync
 `)
 
 	if err := os.WriteFile(filepath.Join(nt.TmpDir, "test-secret.yaml"), cm, 0644); err != nil {

e2e/testcases/multi_sync_test.go

@@ -453,20 +453,20 @@ func TestConflictingDefinitions_NamespaceToRoot(t *testing.T) {
 	}
 
 	// Validate reconciler error metric is emitted from namespace reconciler.
-	rootSyncLabels, err := nomostest.MetricLabelsForRepoSync(nt, repoSyncNN)
+	repoSyncLabels, err := nomostest.MetricLabelsForRepoSync(nt, repoSyncNN)
 	if err != nil {
 		nt.T.Fatal(err)
 	}
 	commitHash := nt.NonRootRepos[repoSyncNN].MustHash(nt.T)
 
-	err = nomostest.ValidateMetrics(nt,
-		nomostest.ReconcilerSyncError(nt, rootSyncLabels, commitHash),
-		nomostest.ReconcilerErrorMetrics(nt, rootSyncLabels, commitHash, metrics.ErrorSummary{
-			Sync: 1,
-		}))
-	if err != nil {
-		nt.T.Fatal(err)
-	}
+	// Reverting a manual change IS considered a "conflict" when the new
+	// owner is a different reconciler.
+	nt.Must(nomostest.ValidateMetrics(nt,
+		nomostest.ReconcilerSyncError(nt, repoSyncLabels, commitHash),
+		nomostest.ReconcilerErrorMetrics(nt, repoSyncLabels, commitHash, metrics.ErrorSummary{
+			Conflicts: 1, // from the remediator
+			Sync:      1, // remediator conflict error replaces applier conflict error
+		})))
 
 	nt.T.Logf("Ensure the Role matches the one in the Root repo %s", configsync.RootSyncName)
 	err = nt.Validate("pods", testNs, &rbacv1.Role{},

e2e/testcases/preserve_fields_test.go

@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	kubectlapply "k8s.io/kubectl/pkg/cmd/apply"
 	"kpt.dev/configsync/e2e/nomostest"
 	"kpt.dev/configsync/e2e/nomostest/metrics"
 	"kpt.dev/configsync/e2e/nomostest/retry"
@@ -333,7 +334,11 @@ func TestAddUpdateDeleteLabels(t *testing.T) {
 		nt.T.Fatal(err)
 	}
 
-	var defaultLabels = []string{metadata.ManagedByKey, metadata.DeclaredVersionLabel}
+	var defaultLabels = []string{
+		metadata.ManagedByKey,
+		metadata.DeclaredVersionLabel,
+		kubectlapply.ApplysetPartOfLabel,
+	}
 
 	// Checking that the configmap with no labels appears on cluster, and
 	// that no user labels are specified
@@ -350,12 +355,13 @@ func TestAddUpdateDeleteLabels(t *testing.T) {
 		nt.T.Fatal(err)
 	}
 
-	// Checking that label is updated after syncing an update.
-	err = nt.Validate(cmName, ns, &corev1.ConfigMap{},
-		testpredicates.HasExactlyLabelKeys(append(defaultLabels, "baz")...))
-	if err != nil {
-		nt.T.Fatal(err)
-	}
+	var updatedLabels []string
+	updatedLabels = append(updatedLabels, defaultLabels...)
+	updatedLabels = append(updatedLabels, "baz")
+
+	// Checking that label was added after syncing.
+	nt.Must(nt.Validate(cmName, ns, &corev1.ConfigMap{},
+		testpredicates.HasExactlyLabelKeys(updatedLabels...)))
 
 	delete(cm.Labels, "baz")
 	nt.Must(nt.RootRepos[configsync.RootSyncName].Add(cmPath, cm))
@@ -365,11 +371,8 @@ func TestAddUpdateDeleteLabels(t *testing.T) {
 	}
 
 	// Check that the label is deleted after syncing.
-	err = nt.Validate(cmName, ns, &corev1.ConfigMap{},
-		testpredicates.HasExactlyLabelKeys(metadata.ManagedByKey, metadata.DeclaredVersionLabel))
-	if err != nil {
-		nt.T.Fatal(err)
-	}
+	nt.Must(nt.Validate(cmName, ns, &corev1.ConfigMap{},
+		testpredicates.HasExactlyLabelKeys(defaultLabels...)))
 
 	nt.MetricsExpectations.AddObjectApply(configsync.RootSyncKind, rootSyncNN, nsObj)
 	nt.MetricsExpectations.AddObjectApply(configsync.RootSyncKind, rootSyncNN, cm)

pkg/api/configsync/register.go

@@ -14,7 +14,9 @@
 
 package configsync
 
-import "time"
+import (
+	"time"
+)
 
 const (
 	// GroupName is the name of the group of configsync resources.
@@ -29,6 +31,14 @@ const (
 
 	// ControllerNamespace is the Namespace used for Nomos controllers
 	ControllerNamespace = "config-management-system"
+
+	// ApplySetToolingName is the name used to represent Config Sync in the
+	// ApplySet tooling annotation.
+	ApplySetToolingName = GroupName
+
+	// ApplySetToolingVersion is the version used to represent Config Sync in
+	// the ApplySet tooling annotation.
+	ApplySetToolingVersion = "v1beta1"
 )
 
 // API type constants

pkg/applier/applier_test.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	kubectlapply "k8s.io/kubectl/pkg/cmd/apply"
 	"kpt.dev/configsync/pkg/api/configsync/v1beta1"
 	"kpt.dev/configsync/pkg/applier/stats"
 	"kpt.dev/configsync/pkg/core"
@@ -89,19 +90,20 @@ func TestApply(t *testing.T) {
 	abandonObj := deploymentObj.DeepCopy()
 	abandonObj.SetName("abandon-me")
 	abandonObj.SetAnnotations(map[string]string{
-		common.LifecycleDeleteAnnotation: common.PreventDeletion,
+		common.LifecycleDeleteAnnotation: common.PreventDeletion, // not removed
 		metadata.ResourceManagementKey:   metadata.ResourceManagementEnabled,
 		metadata.ResourceIDKey:           core.GKNN(abandonObj),
 		metadata.ResourceManagerKey:      resourceManager,
 		metadata.OwningInventoryKey:      "anything",
 		metadata.SyncTokenAnnotationKey:  "anything",
-		"example-to-not-delete":          "anything",
+		"example-to-not-delete":          "anything", // not removed
 	})
 	abandonObj.SetLabels(map[string]string{
-		metadata.ManagedByKey:   metadata.ManagedByValue,
-		metadata.SystemLabel:    "anything",
-		metadata.ArchLabel:      "anything",
-		"example-to-not-delete": "anything",
+		metadata.ManagedByKey:            metadata.ManagedByValue,
+		metadata.SystemLabel:             "anything",
+		metadata.ArchLabel:               "anything",
+		kubectlapply.ApplysetPartOfLabel: "anything", // not removed
+		"example-to-not-delete":          "anything", // not removed
 	})
 	abandonObjID := core.IDOf(abandonObj)
 
@@ -322,15 +324,18 @@ func TestApply(t *testing.T) {
 			expectedServerObjs: []client.Object{
 				func() client.Object {
 					obj := abandonObj.DeepCopy()
-					obj.SetAnnotations(map[string]string{
-						common.LifecycleDeleteAnnotation: common.PreventDeletion,
-						// all configsync annotations removed
-						"example-to-not-delete": "anything",
-					})
-					obj.SetLabels(map[string]string{
-						// all configsync labels removed
-						"example-to-not-delete": "anything",
-					})
+					// all configsync annotations removed
+					core.RemoveAnnotations(obj,
+						metadata.ResourceManagementKey,
+						metadata.ResourceIDKey,
+						metadata.ResourceManagerKey,
+						metadata.OwningInventoryKey,
+						metadata.SyncTokenAnnotationKey)
+					// all configsync labels removed
+					core.RemoveLabels(obj,
+						metadata.ManagedByKey,
+						metadata.SystemLabel,
+						metadata.ArchLabel)
 					obj.SetUID("1")
 					obj.SetResourceVersion("2")
 					obj.SetGeneration(1)

pkg/applier/clientset.go

@@ -19,12 +19,15 @@ import (
 
 	"github.com/GoogleContainerTools/kpt/pkg/live"
 	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/klog/v2"
+	kubectlapply "k8s.io/kubectl/pkg/cmd/apply"
 	"k8s.io/kubectl/pkg/cmd/util"
 	"sigs.k8s.io/cli-utils/pkg/apply"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/watcher"
 	"sigs.k8s.io/cli-utils/pkg/object"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -53,7 +56,7 @@ type ClientSet struct {
 }
 
 // NewClientSet constructs a new ClientSet.
-func NewClientSet(c client.Client, configFlags *genericclioptions.ConfigFlags, statusMode string) (*ClientSet, error) {
+func NewClientSet(c client.Client, configFlags *genericclioptions.ConfigFlags, statusMode, applySetID string) (*ClientSet, error) {
 	matchVersionKubeConfigFlags := util.NewMatchVersionFlags(configFlags)
 	f := util.NewFactory(matchVersionKubeConfigFlags)
 
@@ -71,9 +74,19 @@ func NewClientSet(c client.Client, configFlags *genericclioptions.ConfigFlags, s
 		return nil, err
 	}
 
+	// Only watch objects applied by this reconciler for status updates.
+	// This reduces both the number of events processed and the memory used by
+	// the informer cache.
+	watchFilters := &watcher.Filters{
+		Labels: labels.Set{
+			kubectlapply.ApplysetPartOfLabel: applySetID,
+		}.AsSelector(),
+	}
+
 	applier, err := apply.NewApplierBuilder().
 		WithInventoryClient(invClient).
 		WithFactory(f).
+		WithStatusWatcherFilters(watchFilters).
 		Build()
 	if err != nil {
 		return nil, err
@@ -82,6 +95,7 @@ func NewClientSet(c client.Client, configFlags *genericclioptions.ConfigFlags, s
 	destroyer, err := apply.NewDestroyerBuilder().
 		WithInventoryClient(invClient).
 		WithFactory(f).
+		WithStatusWatcherFilters(watchFilters).
 		Build()
 	if err != nil {
 		return nil, err

pkg/core/decorate.go

@@ -68,6 +68,15 @@ func RemoveAnnotations(obj client.Object, annotations ...string) {
 	obj.SetAnnotations(as)
 }
 
+// RemoveLabels removes the passed set of labels from obj.
+func RemoveLabels(obj client.Object, labels ...string) {
+	actual := obj.GetLabels()
+	for _, label := range labels {
+		delete(actual, label)
+	}
+	obj.SetLabels(actual)
+}
+
 // AddAnnotations adds the specified annotations to the object.
 func AddAnnotations(obj client.Object, annotations map[string]string) {
 	existing := obj.GetAnnotations()