test: add ResourceGroup assertions to WatchForSync

This adds validation for the ResourceGroup status in the WatchForSync
function, which is used extensively in the test framework to validate
RSyncs and wait for them to finish syncing.

This will cause WatchForSync to now also wait for all synced objects to
reconcile and for the ResourceGroup to be current. This is intended to
provide test coverage for the resource-group-controller when integrated
with Config Sync.

Author: sdowell

e2e/nomostest/sync.go

@@ -21,8 +21,10 @@ import (
 	"kpt.dev/configsync/e2e/nomostest/testpredicates"
 	"kpt.dev/configsync/pkg/api/configsync"
 	"kpt.dev/configsync/pkg/api/configsync/v1beta1"
+	"kpt.dev/configsync/pkg/api/kpt.dev/v1alpha1"
 	"kpt.dev/configsync/pkg/parse"
 	"kpt.dev/configsync/pkg/reposync"
+	"kpt.dev/configsync/pkg/resourcegroup"
 	"kpt.dev/configsync/pkg/rootsync"
 	"kpt.dev/configsync/pkg/util/log"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -263,3 +265,58 @@ func statusHasSyncDirAndNoErrors(status v1beta1.Status, sourceType configsync.So
 	}
 	return nil
 }
+
+func resourceGroupHasReconciled(sourceHash string) testpredicates.Predicate {
+	return func(o client.Object) error {
+		if o == nil {
+			return testpredicates.ErrObjectNotFound
+		}
+		rg, ok := o.(*v1alpha1.ResourceGroup)
+		if !ok {
+			return testpredicates.WrongTypeErr(o, &v1alpha1.ResourceGroup{})
+		}
+
+		// Status is disabled for this ResourceGroup, skip all status checks
+		if resourcegroup.IsStatusDisabled(rg) {
+			return testpredicates.ResourceGroupHasNoStatus()(o)
+		}
+
+		if len(rg.Spec.Resources) != len(rg.Status.ResourceStatuses) {
+			return fmt.Errorf("length of spec.resources (%d) does not equal length of status.resourceStatuses (%d)",
+				len(rg.Spec.Resources), len(rg.Status.ResourceStatuses))
+		}
+
+		resourceCount := make(map[v1alpha1.ObjMetadata]int)
+		for _, s := range rg.Status.ResourceStatuses {
+			if err := resourceStatusIsCurrent(s, sourceHash); err != nil {
+				return fmt.Errorf("object %s is not current: %w", s.ObjMetadata, err)
+			}
+			resourceCount[s.ObjMetadata]++
+		}
+
+		for _, o := range rg.Spec.Resources {
+			if count, ok := resourceCount[o]; ok && count > 0 {
+				resourceCount[o]--
+			} else {
+				return fmt.Errorf("spec.resources does not equal status.resourceStatuses")
+			}
+		}
+		return nil
+	}
+}
+
+func resourceStatusIsCurrent(rs v1alpha1.ResourceStatus, sourceHash string) error {
+	if rs.Status != v1alpha1.Current {
+		return fmt.Errorf("resourceStatus.status is not %s. Got %s", v1alpha1.Current, rs.Status)
+	}
+	if rs.Actuation != v1alpha1.ActuationSucceeded {
+		return fmt.Errorf("resourceStatus.actuation is not %s. Got %s", v1alpha1.ActuationSucceeded, rs.Actuation)
+	}
+	if rs.Reconcile != v1alpha1.ReconcileSucceeded {
+		return fmt.Errorf("resourceStatus.reconcile is not %s. Got %s", v1alpha1.ReconcileSucceeded, rs.Reconcile)
+	}
+	if rs.SourceHash != resourcegroup.TruncateSourceHash(sourceHash) {
+		return fmt.Errorf("resourceStatus.sourceHash is not %s. Got %s", sourceHash, rs.SourceHash)
+	}
+	return nil
+}

e2e/nomostest/testpredicates/predicates.go

@@ -1319,6 +1319,25 @@ func containerArgsContains(container *corev1.Container, expectedArg string) erro
 	return fmt.Errorf("expected arg not found: %s", expectedArg)
 }
 
+// ResourceGroupHasNoStatus verifies the status of the ResourceGroup is empty.
+func ResourceGroupHasNoStatus() Predicate {
+	return func(obj client.Object) error {
+		if obj == nil {
+			return ErrObjectNotFound
+		}
+		rg, ok := obj.(*v1alpha1.ResourceGroup)
+		if !ok {
+			return WrongTypeErr(obj, &v1alpha1.ResourceGroup{})
+		}
+		emptyStatus := v1alpha1.ResourceGroupStatus{}
+		if !equality.Semantic.DeepEqual(emptyStatus, rg.Status) {
+			return fmt.Errorf("found non-empty status in %s:\nDiff (- Expected, + Found)\n%s",
+				kinds.ObjectSummary(rg), log.AsYAMLDiff(emptyStatus, rg.Status))
+		}
+		return nil
+	}
+}
+
 // ResourceGroupStatusEquals checks that the RootSync's spec.override matches
 // the specified RootSyncOverrideSpec.
 func ResourceGroupStatusEquals(expected v1alpha1.ResourceGroupStatus) Predicate {

e2e/nomostest/testresourcegroup/resourcegroup.go

@@ -26,8 +26,8 @@ import (
 	"kpt.dev/configsync/pkg/api/configmanagement"
 	"kpt.dev/configsync/pkg/api/kpt.dev/v1alpha1"
 	"kpt.dev/configsync/pkg/metadata"
+	resourcegroup2 "kpt.dev/configsync/pkg/resourcegroup"
 	"kpt.dev/configsync/pkg/resourcegroup/controllers/resourcegroup"
-	"kpt.dev/configsync/pkg/resourcegroup/controllers/status"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -112,8 +112,8 @@ func CreateOrUpdateResources(kubeClient *testkubeclient.KubeClient, resources []
 			Kind:    r.Kind,
 		})
 		u.SetAnnotations(map[string]string{
-			"config.k8s.io/owning-inventory": id,
-			status.SourceHashAnnotationKey:   "1234567890",
+			"config.k8s.io/owning-inventory":       id,
+			resourcegroup2.SourceHashAnnotationKey: "1234567890",
 		})
 
 		err := kubeClient.Get(r.Name, r.Namespace, u.DeepCopy())

e2e/nomostest/wait_for_sync.go

@@ -19,6 +19,7 @@ import (
 	"strings"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -29,6 +30,7 @@ import (
 	"kpt.dev/configsync/e2e/nomostest/testwatcher"
 	"kpt.dev/configsync/pkg/api/configsync"
 	"kpt.dev/configsync/pkg/api/configsync/v1beta1"
+	"kpt.dev/configsync/pkg/api/kpt.dev/v1alpha1"
 	"kpt.dev/configsync/pkg/core/k8sobjects"
 	"kpt.dev/configsync/pkg/kinds"
 	"kpt.dev/configsync/pkg/reposync"
@@ -100,6 +102,14 @@ func SkipReadyCheck() WatchForAllSyncsOptions {
 	}
 }
 
+// SkipAllResourceGroupChecks is an optional parameter which specifies to skip
+// ResourceGroup checks for all RSyncs after they finish syncing.
+func SkipAllResourceGroupChecks() WatchForAllSyncsOptions {
+	return func(options *watchForAllSyncsOptions) {
+		options.watchForSyncOpts = append(options.watchForSyncOpts, SkipResourceGroupCheck())
+	}
+}
+
 // WatchForAllSyncs calls WatchForSync on all Syncs in nt.SyncSources.
 //
 // If you want to validate specific fields of a Sync object, use
@@ -159,7 +169,8 @@ func (nt *NT) WatchForAllSyncs(options ...WatchForAllSyncsOptions) error {
 }
 
 type watchForSyncOptions struct {
-	watchOptions []testwatcher.WatchOption
+	watchOptions           []testwatcher.WatchOption
+	skipResourceGroupCheck bool
 }
 
 // WatchForSyncOption is an optional parameter for WatchForSync.
@@ -173,6 +184,14 @@ func WithWatchOptions(watchOpts ...testwatcher.WatchOption) WatchForSyncOption {
 	}
 }
 
+// SkipResourceGroupCheck is an optional parameter to skip the ResourceGroup
+// watch after the RSync finishes syncing.
+func SkipResourceGroupCheck() WatchForSyncOption {
+	return func(options *watchForSyncOptions) {
+		options.skipResourceGroupCheck = true
+	}
+}
+
 // WatchForSync watches the specified sync object until it's synced.
 //
 //   - gvk (required) is the sync object GroupVersionKind
@@ -188,7 +207,8 @@ func (nt *NT) WatchForSync(
 ) error {
 	nt.T.Helper()
 	opts := watchForSyncOptions{
-		watchOptions: []testwatcher.WatchOption{},
+		skipResourceGroupCheck: false,
+		watchOptions:           []testwatcher.WatchOption{},
 	}
 	// Override defaults with specified options
 	for _, option := range options {
@@ -239,6 +259,26 @@ func (nt *NT) WatchForSync(
 		return fmt.Errorf("waiting for sync: %w", err)
 	}
 	nt.T.Logf("%s %s/%s is synced", gvk.Kind, namespace, name)
+	if opts.skipResourceGroupCheck {
+		return nil
+	}
+	rgWatchOptions := []testwatcher.WatchOption{
+		testwatcher.WatchPredicates([]testpredicates.Predicate{
+			// Wait until status.observedGeneration matches metadata.generation
+			testpredicates.HasObservedLatestGeneration(nt.Scheme),
+			// Wait until metadata.deletionTimestamp is missing, and conditions do not include Reconciling=True or Stalled=True
+			testpredicates.StatusEquals(nt.Scheme, kstatus.CurrentStatus),
+			// Make sure the ResourceGroup has the Stalled condition and it is "False"
+			// This ensures the condition exists and wasn't removed by the applier
+			testpredicates.HasConditionStatus(nt.Scheme, string(v1alpha1.Stalled), corev1.ConditionFalse),
+			// Wait until all resourceStatuses are consistent with spec and current/reconciled
+			resourceGroupHasReconciled(expectedCommit),
+		}...),
+	}
+	rgWatchOptions = append(rgWatchOptions, opts.watchOptions...)
+	if err := nt.Watcher.WatchObject(kinds.ResourceGroup(), name, namespace, rgWatchOptions...); err != nil {
+		return fmt.Errorf("waiting for ResourceGroup: %w", err)
+	}
 	return nil
 }
 

e2e/testcases/basic_test.go

@@ -111,7 +111,7 @@ func TestSyncDeploymentAndReplicaSet(t *testing.T) {
 	nt.T.Log("Add a corresponding deployment")
 	nt.Must(rootSyncGitRepo.Copy(fmt.Sprintf("%s/deployment-helloworld.yaml", yamlDir), "acme/namespaces/dir/deployment.yaml"))
 	nt.Must(rootSyncGitRepo.CommitAndPush("Add corresponding deployment"))
-	nt.Must(nt.WatchForAllSyncs())
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 
 	nt.T.Log("check that the deployment was created")
 	if err := nt.Validate("hello-world", "dir", &appsv1.Deployment{}); err != nil {

e2e/testcases/multi_sync_test.go

@@ -486,7 +486,9 @@ func TestConflictingDefinitions_NamespaceToRoot(t *testing.T) {
 	nt.T.Logf("Remove the Role from the Namespace repo %s", repoSyncKey)
 	nt.Must(repoSyncGitRepo.Remove(podRoleFilePath))
 	nt.Must(repoSyncGitRepo.CommitAndPush("remove conflicting pod role from Namespace repo"))
-	nt.Must(nt.WatchForAllSyncs())
+	// rs-test tries to delete the Role, but since it is also being managed by root-sync
+	// it may Timeout waiting for the deletion to reconcile.
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 
 	nt.T.Logf("Ensure the Role still matches the one in the Root repo %s", rootSyncKey.Name)
 	err = nt.Validate("pods", testNs, &rbacv1.Role{},
@@ -620,7 +622,9 @@ func TestConflictingDefinitions_RootToRoot(t *testing.T) {
 	nt.T.Logf("Remove the declaration from RootSync %s", rootSyncID.Name)
 	nt.Must(rootSyncGitRepo.Remove(podRoleFilePath))
 	nt.Must(rootSyncGitRepo.CommitAndPush("remove conflicting pod role"))
-	nt.Must(nt.WatchForAllSyncs())
+	// root-sync tries to delete the Role, but since it is also being managed by root-test
+	// it may Timeout waiting for the deletion to reconcile.
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 
 	nt.T.Logf("Ensure the Role is managed by RootSync %s", rootSync2ID.Name)
 	// The pod role may be deleted from the cluster after it was removed from the `root-sync` Root repo.

e2e/testcases/override_reconcile_timeout_test.go

@@ -96,7 +96,7 @@ func TestOverrideReconcileTimeout(t *testing.T) {
 	nt.Must(rootSyncGitRepo.Add("acme/pod-1.yaml", pod1))
 	nt.Must(rootSyncGitRepo.Add("acme/ns-1.yaml", k8sobjects.NamespaceObject(namespaceName)))
 	nt.Must(rootSyncGitRepo.CommitAndPush(fmt.Sprintf("Add namespace/%s & pod/%s (never ready)", namespaceName, pod1Name)))
-	nt.Must(nt.WatchForAllSyncs())
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 	expectActuationStatus := "Succeeded"
 	expectReconcileStatus := "Timeout"
 	nt.Must(nt.Watcher.WatchObject(kinds.ResourceGroup(), "root-sync", "config-management-system",

e2e/testcases/status_enablement_test.go

@@ -16,7 +16,6 @@ package e2e
 
 import (
 	"fmt"
-	"reflect"
 	"testing"
 
 	"kpt.dev/configsync/e2e/nomostest"
@@ -46,18 +45,18 @@ func TestStatusEnabledAndDisabled(t *testing.T) {
 	rootSync := k8sobjects.RootSyncObjectV1Alpha1(configsync.RootSyncName)
 	// Override the statusMode for root-reconciler
 	nt.MustMergePatch(rootSync, `{"spec": {"override": {"statusMode": "disabled"}}}`)
-	nt.Must(nt.WatchForAllSyncs())
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 
 	namespaceName := "status-test"
 	nt.Must(rootSyncGitRepo.Add("acme/ns.yaml", namespaceObject(namespaceName, nil)))
 	nt.Must(rootSyncGitRepo.Add("acme/cm1.yaml", k8sobjects.ConfigMapObject(core.Name("cm1"), core.Namespace(namespaceName))))
 	nt.Must(rootSyncGitRepo.CommitAndPush("Add a namespace and a configmap"))
-	nt.Must(nt.WatchForAllSyncs())
+	nt.Must(nt.WatchForAllSyncs(nomostest.SkipAllResourceGroupChecks()))
 
 	nt.Must(nt.Watcher.WatchObject(kinds.ResourceGroup(),
 		configsync.RootSyncName, configsync.ControllerNamespace,
 		testwatcher.WatchPredicates(
-			resourceGroupHasNoStatus,
+			testpredicates.ResourceGroupHasNoStatus(),
 			testpredicates.HasLabel(common.InventoryLabel, id),
 		),
 		testwatcher.WatchTimeout(nt.DefaultWaitTimeout)))
@@ -75,24 +74,6 @@ func TestStatusEnabledAndDisabled(t *testing.T) {
 		testwatcher.WatchTimeout(nt.DefaultWaitTimeout)))
 }
 
-func resourceGroupHasNoStatus(obj client.Object) error {
-	if obj == nil {
-		return testpredicates.ErrObjectNotFound
-	}
-	rg, ok := obj.(*resourcegroupv1alpha1.ResourceGroup)
-	if !ok {
-		return testpredicates.WrongTypeErr(obj, &resourcegroupv1alpha1.ResourceGroup{})
-	}
-	// We can't check that the status field is missing, because the
-	// ResourceGroup object doesn't use a pointer for status.
-	// But we can check that the status is empty, which is what we really care
-	// about, to reduce the size of the object in etcd.
-	if !reflect.ValueOf(rg.Status).IsZero() {
-		return fmt.Errorf("found non-empty status in %s", core.IDOf(obj))
-	}
-	return nil
-}
-
 func resourceGroupHasStatus(obj client.Object) error {
 	if obj == nil {
 		return testpredicates.ErrObjectNotFound

pkg/resourcegroup/controllers/root/root_controller.go

@@ -28,8 +28,9 @@ import (
 	"kpt.dev/configsync/pkg/api/kpt.dev/v1alpha1"
 	"kpt.dev/configsync/pkg/metadata"
 	"kpt.dev/configsync/pkg/reconcilermanager/controllers"
+	"kpt.dev/configsync/pkg/resourcegroup"
 	"kpt.dev/configsync/pkg/resourcegroup/controllers/handler"
-	"kpt.dev/configsync/pkg/resourcegroup/controllers/resourcegroup"
+	resourcegroupcontroller "kpt.dev/configsync/pkg/resourcegroup/controllers/resourcegroup"
 	"kpt.dev/configsync/pkg/resourcegroup/controllers/resourcemap"
 	"kpt.dev/configsync/pkg/resourcegroup/controllers/typeresolver"
 	"kpt.dev/configsync/pkg/resourcegroup/controllers/watch"
@@ -42,9 +43,7 @@ import (
 
 //nolint:revive // TODO: add comments for public constants and enable linting
 const (
-	KptGroup           = "kpt"
-	DisableStatusKey   = "configsync.gke.io/status"
-	DisableStatusValue = "disabled"
+	KptGroup = "kpt"
 )
 
 // Reconciler reconciles a ResourceGroup object
@@ -100,7 +99,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 
 	// Skip ResourceGroup status updates if the status is disabled and has
 	// already been removed.
-	if isStatusDisabled(resgroup) {
+	if resourcegroup.IsStatusDisabled(resgroup) {
 		r.Logger(ctx).V(3).Info("Skipping update event: ResourceGroup status disabled")
 		return r.reconcileDisabledResourceGroup(ctx, req, resgroup)
 	}
@@ -155,23 +154,14 @@ func (r *Reconciler) reconcileDisabledResourceGroup(ctx context.Context, req ctr
 	}
 	resgroup.Status = emptyStatus
 	// Use `r.Status().Update()` here instead of `r.Update()` to update only resgroup.Status.
-	err := r.client.Status().Update(ctx, resgroup, client.FieldOwner(resourcegroup.FieldManager))
+	err := r.client.Status().Update(ctx, resgroup, client.FieldOwner(resourcegroupcontroller.FieldManager))
 	if err != nil {
 		return ctrl.Result{}, err
 	}
 	// update the resMap
 	return r.reconcile(ctx, req.NamespacedName, []v1alpha1.ObjMetadata{}, true)
 }
 
-func isStatusDisabled(resgroup *v1alpha1.ResourceGroup) bool {
-	annotations := resgroup.GetAnnotations()
-	if annotations == nil {
-		return false
-	}
-	val, found := annotations[DisableStatusKey]
-	return found && val == DisableStatusValue
-}
-
 // NewController creates a new Reconciler and registers it with the provided manager
 func NewController(mgr manager.Manager, channel chan event.GenericEvent,
 	logger logr.Logger, resolver *typeresolver.TypeResolver, group string, resMap *resourcemap.ResourceMap) error {