Add watch filtering by package-id label

go/config-sync-watch-filter

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

@@ -333,7 +333,11 @@ func TestAddUpdateDeleteLabels(t *testing.T) {
 		nt.T.Fatal(err)
 	}
 
-	var defaultLabels = []string{metadata.ManagedByKey, metadata.DeclaredVersionLabel}
+	var defaultLabels = []string{
+		metadata.ManagedByKey,
+		metadata.ParentPackageIDLabel,
+		metadata.DeclaredVersionLabel,
+	}
 
 	// Checking that the configmap with no labels appears on cluster, and
 	// that no user labels are specified
@@ -350,12 +354,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 +370,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/applier/applier_test.go

@@ -98,10 +98,11 @@ func TestApply(t *testing.T) {
 		"example-to-not-delete":          "anything",
 	})
 	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.ParentPackageIDLabel: "anything",
+		metadata.SystemLabel:          "anything",
+		metadata.ArchLabel:            "anything",
+		"example-to-not-delete":       "anything",
 	})
 	abandonObjID := core.IDOf(abandonObj)
 

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"
 	"k8s.io/kubectl/pkg/cmd/util"
+	"kpt.dev/configsync/pkg/metadata"
 	"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, packageID 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{
+			metadata.ParentPackageIDLabel: packageID,
+		}.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/metadata/labels.go

@@ -61,6 +61,12 @@ const (
 	// the resource. Similar to the well known app.kubernetes.io/managed-by label,
 	// but scoped to Config Sync.
 	ConfigSyncManagedByLabel = configsync.ConfigSyncPrefix + "managed-by"
+
+	// ParentPackageIDLabel indicates the parent package a managed object was sourced from.
+	ParentPackageIDLabel = configsync.ConfigSyncPrefix + "parent-package-id"
+
+	// PackageIDLabel indicates the ID of the package represented by a RootSync or RepoSync.
+	PackageIDLabel = configsync.ConfigSyncPrefix + "package-id"
 )
 
 // DepthSuffix is a label suffix for hierarchical namespace depth.

pkg/metadata/metadata_test.go

@@ -105,6 +105,12 @@ func TestHasConfigSyncMetadata(t *testing.T) {
 				core.Label(metadata.ManagedByKey, "random-value")),
 			want: false,
 		},
+		{
+			name: "An object with the `parent-package-id` label",
+			obj: fake.UnstructuredObject(kinds.Deployment(), core.Name("deploy"),
+				core.Label(metadata.ParentPackageIDLabel, "random-value")),
+			want: true,
+		},
 	}
 
 	for _, tc := range testcases {

pkg/metadata/package_id.go

@@ -0,0 +1,44 @@
+// Copyright 2024 Google LLC
+//
+// 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 metadata
+
+import (
+	"fmt"
+	"hash/fnv"
+
+	"k8s.io/apimachinery/pkg/util/validation"
+)
+
+const maxLabelLength = validation.LabelValueMaxLength // 63
+
+// PackageID is a label value that uniquely identifies a RootSync or RepoSync.
+// PACKAGE_ID = <PACKAGE_ID_FULL[0:53]>-<hex(fnv(PACKAGE_ID_FULL))>
+// PACKAGE_ID_FULL = <NAME>.<NAMESPACE>.<RootSync|RepoSync>
+// Design: go/config-sync-watch-filter
+func PackageID(syncName, syncNamespace, syncKind string) string {
+	packageID := fmt.Sprintf("%s.%s.%s", syncName, syncNamespace, syncKind)
+	if len(packageID) <= maxLabelLength {
+		return packageID
+	}
+	// fnv32a has slightly better avalanche characteristics than fnv32
+	hasher := fnv.New32a()
+	hasher.Write([]byte(fmt.Sprintf("%s.%s.%s", syncName, syncNamespace, syncKind)))
+	// Converting 32-bit fnv to hex results in at most 8 characters.
+	// Rarely it's fewer, so pad the prefix with zeros to make it consistent.
+	suffix := fmt.Sprintf("%08x", hasher.Sum32())
+	packageIDLen := maxLabelLength - len(suffix) - 1
+	packageIDShort := packageID[0 : packageIDLen-1]
+	return fmt.Sprintf("%s-%s", packageIDShort, suffix)
+}