Add watch filtering by package-id label

- Update k8s & cli-utils & controller-runtime
- Fix Fprintf calls to handle errors

.golangci.yaml

@@ -26,7 +26,7 @@ linters:
 
 linters-settings:
   staticcheck:
-    go: "1.21"
+    go: "1.22"
 
 issues:
   exclude-rules:

Makefile

@@ -56,7 +56,7 @@ BIN_DIR := $(GO_DIR)/bin
 ADDLICENSE_VERSION := v1.1.1
 ADDLICENSE := $(BIN_DIR)/addlicense
 
-GOLANGCI_LINT_VERSION := v1.56.2
+GOLANGCI_LINT_VERSION := v1.59.1
 GOLANGCI_LINT := $(BIN_DIR)/golangci-lint
 
 KUSTOMIZE_VERSION := v5.4.2-gke.0

cmd/nomos/status/cluster_state.go

@@ -48,15 +48,15 @@ type ClusterState struct {
 }
 
 func (c *ClusterState) printRows(writer io.Writer) {
-	fmt.Fprintln(writer, "")
-	fmt.Fprintf(writer, "%s\n", c.Ref)
+	util.MustFprintf(writer, "\n")
+	util.MustFprintf(writer, "%s\n", c.Ref)
 	if c.status != "" || c.Error != "" {
-		fmt.Fprintf(writer, "%s%s\n", util.Indent, util.Separator)
-		fmt.Fprintf(writer, "%s%s\t%s\n", util.Indent, c.status, c.Error)
+		util.MustFprintf(writer, "%s%s\n", util.Indent, util.Separator)
+		util.MustFprintf(writer, "%s%s\t%s\n", util.Indent, c.status, c.Error)
 	}
 	for _, repo := range c.repos {
 		if name == "" || name == repo.syncName {
-			fmt.Fprintf(writer, "%s%s\n", util.Indent, util.Separator)
+			util.MustFprintf(writer, "%s%s\n", util.Indent, util.Separator)
 			repo.printRows(writer)
 		}
 	}
@@ -90,44 +90,44 @@ type RepoState struct {
 }
 
 func (r *RepoState) printRows(writer io.Writer) {
-	fmt.Fprintf(writer, "%s%s:%s\t%s\t\n", util.Indent, r.scope, r.syncName, sourceString(r.sourceType, r.git, r.oci, r.helm))
+	util.MustFprintf(writer, "%s%s:%s\t%s\t\n", util.Indent, r.scope, r.syncName, sourceString(r.sourceType, r.git, r.oci, r.helm))
 	if r.status == syncedMsg {
-		fmt.Fprintf(writer, "%s%s @ %v\t%s\t\n", util.Indent, r.status, r.lastSyncTimestamp, r.commit)
+		util.MustFprintf(writer, "%s%s @ %v\t%s\t\n", util.Indent, r.status, r.lastSyncTimestamp, r.commit)
 	} else {
-		fmt.Fprintf(writer, "%s%s\t%s\t\n", util.Indent, r.status, r.commit)
+		util.MustFprintf(writer, "%s%s\t%s\t\n", util.Indent, r.status, r.commit)
 	}
 
 	if r.errorSummary != nil && r.errorSummary.TotalCount > 0 {
 		if r.errorSummary.Truncated {
-			fmt.Fprintf(writer, "%sTotalErrorCount: %d, ErrorTruncated: %v, ErrorCountAfterTruncation: %d\n", util.Indent,
+			util.MustFprintf(writer, "%sTotalErrorCount: %d, ErrorTruncated: %v, ErrorCountAfterTruncation: %d\n", util.Indent,
 				r.errorSummary.TotalCount, r.errorSummary.Truncated, r.errorSummary.ErrorCountAfterTruncation)
 		} else {
-			fmt.Fprintf(writer, "%sTotalErrorCount: %d\n", util.Indent, r.errorSummary.TotalCount)
+			util.MustFprintf(writer, "%sTotalErrorCount: %d\n", util.Indent, r.errorSummary.TotalCount)
 		}
 	}
 
 	for _, err := range r.errors {
-		fmt.Fprintf(writer, "%sError:\t%s\t\n", util.Indent, err)
+		util.MustFprintf(writer, "%sError:\t%s\t\n", util.Indent, err)
 	}
 
 	if resourceStatus && len(r.resources) > 0 {
 		sort.Sort(byNamespaceAndType(r.resources))
-		fmt.Fprintf(writer, "%sManaged resources:\n", util.Indent)
+		util.MustFprintf(writer, "%sManaged resources:\n", util.Indent)
 		hasSourceHash := r.resources[0].SourceHash != ""
 		if !hasSourceHash {
-			fmt.Fprintf(writer, "%s\tNAMESPACE\tNAME\tSTATUS\n", util.Indent)
+			util.MustFprintf(writer, "%s\tNAMESPACE\tNAME\tSTATUS\n", util.Indent)
 		} else {
-			fmt.Fprintf(writer, "%s\tNAMESPACE\tNAME\tSTATUS\tSOURCEHASH\n", util.Indent)
+			util.MustFprintf(writer, "%s\tNAMESPACE\tNAME\tSTATUS\tSOURCEHASH\n", util.Indent)
 		}
 		for _, r := range r.resources {
 			if !hasSourceHash {
-				fmt.Fprintf(writer, "%s\t%s\t%s\t%s\n", util.Indent, r.Namespace, r.String(), r.Status)
+				util.MustFprintf(writer, "%s\t%s\t%s\t%s\n", util.Indent, r.Namespace, r.String(), r.Status)
 			} else {
-				fmt.Fprintf(writer, "%s\t%s\t%s\t%s\t%s\n", util.Indent, r.Namespace, r.String(), r.Status, r.SourceHash)
+				util.MustFprintf(writer, "%s\t%s\t%s\t%s\t%s\n", util.Indent, r.Namespace, r.String(), r.Status, r.SourceHash)
 			}
 			if len(r.Conditions) > 0 {
 				for _, condition := range r.Conditions {
-					fmt.Fprintf(writer, "%s%s%s%s%s\n", util.Indent, util.Indent, util.Indent, util.Indent, condition.Message)
+					util.MustFprintf(writer, "%s%s%s%s%s\n", util.Indent, util.Indent, util.Indent, util.Indent, condition.Message)
 				}
 			}
 		}

cmd/nomos/util/util.go

@@ -45,12 +45,20 @@ func MonoRepoNotice(writer io.Writer, monoRepoClusters ...string) {
 	clusterCount := len(monoRepoClusters)
 	if clusterCount != 0 {
 		if clusterCount == 1 {
-			fmt.Fprintf(writer, "%sNotice: The cluster %q is still running in the legacy mode.\n",
+			MustFprintf(writer, "%sNotice: The cluster %q is still running in the legacy mode.\n",
 				ColorYellow, monoRepoClusters[0])
 		} else {
-			fmt.Fprintf(writer, "%sNotice: The following clusters are still running in the legacy mode:\n%s%s\n",
+			MustFprintf(writer, "%sNotice: The following clusters are still running in the legacy mode:\n%s%s\n",
 				ColorYellow, Bullet, strings.Join(monoRepoClusters, "\n"+Bullet))
 		}
-		fmt.Fprintf(writer, "Run `nomos migrate` to enable multi-repo mode. It provides you with additional features and gives you the flexibility to sync to a single repository, or multiple repositories.%s\n", ColorDefault)
+		MustFprintf(writer, "Run `nomos migrate` to enable multi-repo mode. It provides you with additional features and gives you the flexibility to sync to a single repository, or multiple repositories.%s\n", ColorDefault)
+	}
+}
+
+// MustFprintf prints to the writer and panics if it errors.
+func MustFprintf(writer io.Writer, format string, a ...any) {
+	_, err := fmt.Fprintf(writer, format, a...)
+	if err != nil {
+		panic(fmt.Sprintf("Error writing output: %v", err))
 	}
 }

cmd/nomos/version/version.go

@@ -284,15 +284,15 @@ func tabulate(es []entry, out io.Writer) {
 	w := util.NewWriter(out)
 	defer func() {
 		if err := w.Flush(); err != nil {
-			fmt.Fprintf(os.Stderr, "error on Flush(): %v", err)
+			util.MustFprintf(os.Stderr, "error on Flush(): %v", err)
 		}
 	}()
-	fmt.Fprintf(w, format, "CURRENT", "CLUSTER_CONTEXT_NAME", "COMPONENT", "VERSION")
+	util.MustFprintf(w, format, "CURRENT", "CLUSTER_CONTEXT_NAME", "COMPONENT", "VERSION")
 	for _, e := range es {
 		if e.err != nil {
-			fmt.Fprintf(w, format, e.current, e.name, e.component, fmt.Sprintf("<error: %v>", e.err))
+			util.MustFprintf(w, format, e.current, e.name, e.component, fmt.Sprintf("<error: %v>", e.err))
 			continue
 		}
-		fmt.Fprintf(w, format, e.current, e.name, e.component, e.version)
+		util.MustFprintf(w, format, e.current, e.name, e.component, e.version)
 	}
 }

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)