add vpa tests

e2e/flags.go

@@ -65,6 +65,10 @@ var Stress = flag.Bool("stress", false,
 var KCC = flag.Bool("kcc", false,
 	"If true, run kcc tests.")
 
+// VPA enables running the e2e tests for vertical pod autoscaling.
+var VPA = flag.Bool("vpa", false,
+	"If true, run VPA tests.")
+
 // GceNode enables running the e2e tests for 'gcenode' auth type
 var GceNode = flag.Bool("gcenode", false,
 	"If true, run test with 'gcenode' auth type.")

e2e/nomostest/client.go

@@ -30,6 +30,7 @@ import (
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
+	autoscalingv1vpa "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/autoscaling.k8s.io/v1"
 	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 	"kpt.dev/configsync/e2e"
 	"kpt.dev/configsync/e2e/nomostest/clusters"
@@ -91,6 +92,7 @@ func newScheme(t testing.NTB) *runtime.Scheme {
 		rbacv1beta1.SchemeBuilder,
 		resourcegroupv1alpha1.SchemeBuilder.SchemeBuilder,
 		apiregistrationv1.SchemeBuilder,
+		autoscalingv1vpa.SchemeBuilder,
 	}
 	for _, b := range builders {
 		err := b.AddToScheme(s)

e2e/nomostest/clusters/gke.go

@@ -238,6 +238,9 @@ func createGKECluster(t testing.NTB, name string) error {
 		if len(addons) > 0 {
 			args = append(args, "--addons", strings.Join(addons, ","))
 		}
+		if *e2e.VPA {
+			args = append(args, "--enable-vertical-pod-autoscaling")
+		}
 	}
 	t.Logf("gcloud %s", strings.Join(args, " "))
 	cmd := exec.Command("gcloud", args...)

e2e/nomostest/new.go

@@ -85,6 +85,10 @@ func newOptStruct(testName, tmpDir string, t nomostesting.NTB, ntOptions ...ntop
 		t.Skip("Test skipped since it is a KCC test")
 	}
 
+	if !*e2e.VPA && optsStruct.VPATest {
+		t.Skip("Test skipped since it is a VPA test")
+	}
+
 	if !*e2e.GceNode && optsStruct.GCENodeTest {
 		t.Skip("Test skipped since it is a test for GCENode auth type, which requires a GKE cluster without workload identity")
 	}

e2e/nomostest/ntopts/test_type.go

@@ -25,6 +25,9 @@ type TestType struct {
 	// KCCTest specifies the test is for KCC resources.
 	KCCTest bool
 
+	// VPATest specifies the test requires VPA to be installed.
+	VPATest bool
+
 	// GCENodeTest specifies the test is for verifying the gcenode auth type.
 	// It requires a GKE cluster with workload identity disabled.
 	GCENodeTest bool
@@ -45,6 +48,11 @@ func KCCTest(opt *New) {
 	opt.KCCTest = true
 }
 
+// VPATest specifies the test requires VPA to be installed.
+func VPATest(opt *New) {
+	opt.VPATest = true
+}
+
 // GCENodeTest specifies the test is for verifying the gcenode auth type.
 func GCENodeTest(opt *New) {
 	opt.GCENodeTest = true

e2e/nomostest/testpredicates/predicates.go

@@ -256,6 +256,16 @@ func HasCorrectResourceRequestsLimits(containerName string, cpuRequest, cpuLimit
 		if o == nil {
 			return ErrObjectNotFound
 		}
+		if uObj, ok := o.(*unstructured.Unstructured); ok {
+			rObj, err := kinds.ToTypedObject(uObj, core.Scheme)
+			if err != nil {
+				return err
+			}
+			o, err = kinds.ObjectAsClientObject(rObj)
+			if err != nil {
+				return err
+			}
+		}
 		dep, ok := o.(*appsv1.Deployment)
 		if !ok {
 			return WrongTypeErr(dep, &appsv1.Deployment{})

e2e/testcases/stress_test.go

@@ -23,10 +23,12 @@ import (
 
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	autoscalingv1 "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/autoscaling.k8s.io/v1"
 	"kpt.dev/configsync/e2e/nomostest"
 	"kpt.dev/configsync/e2e/nomostest/ntopts"
 	nomostesting "kpt.dev/configsync/e2e/nomostest/testing"
@@ -38,6 +40,8 @@ import (
 	"kpt.dev/configsync/pkg/kinds"
 	"kpt.dev/configsync/pkg/metadata"
 	"kpt.dev/configsync/pkg/testing/fake"
+	"kpt.dev/configsync/pkg/util/log"
+	kstatus "sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -121,7 +125,7 @@ func TestStressCRD(t *testing.T) {
 	}
 
 	nt.T.Logf("Verify that there are exactly 1000 CronTab CRs managed by Config Sync on the cluster")
-	crList := &unstructured.UnstructuredList{}
+	crList := &metav1.PartialObjectMetadataList{}
 	crList.SetGroupVersionKind(crontabGVK)
 	if err := nt.KubeClient.List(crList, client.MatchingLabels{metadata.ManagedByKey: metadata.ManagedByValue}); err != nil {
 		nt.T.Error(err)
@@ -161,19 +165,118 @@ func TestStressLargeNamespace(t *testing.T) {
 	}
 
 	nt.T.Log("Verify there are 5000 ConfigMaps in the namespace")
-	cmList := &corev1.ConfigMapList{}
+	cmList := &metav1.PartialObjectMetadataList{}
+	cmList.SetGroupVersionKind(kinds.ConfigMap())
+	if err := nt.KubeClient.List(cmList, &client.ListOptions{Namespace: ns}, client.MatchingLabels{labelKey: labelValue}); err != nil {
+		nt.T.Error(err)
+	}
+	if len(cmList.Items) != 5000 {
+		nt.T.Errorf("The %s namespace should include 5000 ConfigMaps having the `%s: %s` label exactly, found %v instead", ns, labelKey, labelValue, len(cmList.Items))
+	}
+}
+
+// TestStressLargeNamespaceAutoscaling tests that Config Sync can sync a
+// namespace including 5000 resources successfully, when autoscaling it set to
+// Auto, with the smaller initial resource requests.
+// Ideally, the reconciler should be OOMKilled and/or evicted at least once and
+// be replaced with more CPU/Mem.
+func TestStressLargeNamespaceAutoscaling(t *testing.T) {
+	nt := nomostest.New(t, nomostesting.Reconciliation1, ntopts.Unstructured, ntopts.StressTest,
+		ntopts.WithReconcileTimeout(configsync.DefaultReconcileTimeout))
+	nt.T.Log("Stop the CS webhook by removing the webhook configuration")
+	nomostest.StopWebhook(nt)
+
+	nt.T.Log("Enable autoscaling")
+	rootSync := fake.RootSyncObjectV1Beta1(configsync.RootSyncName)
+	if err := nt.KubeClient.Get(rootSync.Name, rootSync.Namespace, rootSync); err != nil {
+		nt.T.Fatal(err)
+	}
+	core.SetAnnotation(rootSync, metadata.ReconcilerAutoscalingStrategyAnnotationKey, string(metadata.ReconcilerAutoscalingStrategyAuto))
+	reconcilerResourceSpec := v1beta1.ContainerResourcesSpec{
+		ContainerName: "reconciler",
+		CPURequest:    resource.MustParse("10m"),
+		CPULimit:      resource.MustParse("1"),
+		MemoryRequest: resource.MustParse("5Mi"),
+		MemoryLimit:   resource.MustParse("10Mi"),
+	}
+	rootSync.Spec.Override.Resources = []v1beta1.ContainerResourcesSpec{
+		reconcilerResourceSpec,
+	}
+	if err := nt.KubeClient.Update(rootSync); err != nil {
+		nt.T.Fatal(err)
+	}
+
+	// Wait for the reconciler Deployment to reflect the RootSync changes
+	reconcilerKey := core.RootReconcilerObjectKey(configsync.RootSyncName)
+	err := nt.Watcher.WatchObject(kinds.Deployment(), reconcilerKey.Name, reconcilerKey.Namespace, []testpredicates.Predicate{
+		testpredicates.HasCorrectResourceRequestsLimits(reconcilerResourceSpec.ContainerName,
+			reconcilerResourceSpec.CPURequest, reconcilerResourceSpec.CPULimit,
+			reconcilerResourceSpec.MemoryRequest, reconcilerResourceSpec.MemoryLimit),
+		testpredicates.StatusEquals(nt.Scheme, kstatus.CurrentStatus),
+	})
+	if err != nil {
+		nt.T.Fatal(err)
+	}
+
+	if err := nt.WatchForAllSyncs(); err != nil {
+		nt.T.Fatal(err)
+	}
 
+	reconcilerPod, err := nt.KubeClient.GetDeploymentPod(reconcilerKey.Name, reconcilerKey.Namespace, nt.DefaultWaitTimeout)
+	if err != nil {
+		nt.T.Fatal(err)
+	}
+	nt.T.Log("Reconciler container specs (before):")
+	for _, container := range reconcilerPod.Spec.Containers {
+		nt.T.Logf("%s: %s", container.Name, log.AsJSON(container.Resources))
+	}
+
+	ns := "my-ns-1"
+	nt.Must(nt.RootRepos[configsync.RootSyncName].Add("acme/ns.yaml", fake.NamespaceObject(ns)))
+
+	labelKey := "StressTestName"
+	labelValue := "TestStressLargeNamespace"
+	for i := 1; i <= 5000; i++ {
+		nt.Must(nt.RootRepos[configsync.RootSyncName].Add(fmt.Sprintf("acme/cm-%d.yaml", i), fake.ConfigMapObject(
+			core.Name(fmt.Sprintf("cm-%d", i)), core.Namespace(ns), core.Label(labelKey, labelValue))))
+	}
+	nt.Must(nt.RootRepos[configsync.RootSyncName].CommitAndPush("Add 5000 ConfigMaps and 1 Namespace"))
+	err = nt.WatchForAllSyncs(nomostest.WithTimeout(10 * time.Minute))
+	if err != nil {
+		nt.T.Fatal(err)
+	}
+
+	nt.T.Log("Verify there are 5000 ConfigMaps in the namespace")
+	cmList := &metav1.PartialObjectMetadataList{}
+	cmList.SetGroupVersionKind(kinds.ConfigMap())
 	if err := nt.KubeClient.List(cmList, &client.ListOptions{Namespace: ns}, client.MatchingLabels{labelKey: labelValue}); err != nil {
 		nt.T.Error(err)
 	}
 	if len(cmList.Items) != 5000 {
 		nt.T.Errorf("The %s namespace should include 5000 ConfigMaps having the `%s: %s` label exactly, found %v instead", ns, labelKey, labelValue, len(cmList.Items))
 	}
+
+	reconcilerPod, err = nt.KubeClient.GetDeploymentPod(reconcilerKey.Name, reconcilerKey.Namespace, nt.DefaultWaitTimeout)
+	if err != nil {
+		nt.T.Fatal(err)
+	}
+	nt.T.Log("Reconciler container specs (after):")
+	for _, container := range reconcilerPod.Spec.Containers {
+		nt.T.Logf("%s: %s", container.Name, log.AsJSON(container.Resources))
+	}
+
+	vpa := &autoscalingv1.VerticalPodAutoscaler{}
+	if err := nt.KubeClient.Get(reconcilerKey.Name, reconcilerKey.Namespace, vpa); err != nil {
+		nt.T.Fatal(err)
+	}
+	nt.T.Log("Reconciler VPA recommendations:")
+	nt.T.Log(log.AsYAML(vpa.Status.Recommendation.ContainerRecommendations))
 }
 
 // TestStressFrequentGitCommits adds 100 Git commits, and verifies that Config Sync can sync the changes in these commits successfully.
 func TestStressFrequentGitCommits(t *testing.T) {
-	nt := nomostest.New(t, nomostesting.Reconciliation1, ntopts.Unstructured, ntopts.StressTest,
+	nt := nomostest.New(t, nomostesting.Reconciliation1, ntopts.Unstructured,
+		ntopts.StressTest, ntopts.VPATest,
 		ntopts.WithReconcileTimeout(configsync.DefaultReconcileTimeout))
 	nt.T.Log("Stop the CS webhook by removing the webhook configuration")
 	nomostest.StopWebhook(nt)
@@ -200,7 +303,8 @@ func TestStressFrequentGitCommits(t *testing.T) {
 	}
 
 	nt.T.Logf("Verify that there are exactly 100 ConfigMaps under the %s namespace", ns)
-	cmList := &corev1.ConfigMapList{}
+	cmList := &metav1.PartialObjectMetadataList{}
+	cmList.SetGroupVersionKind(kinds.ConfigMap())
 	if err := nt.KubeClient.List(cmList, &client.ListOptions{Namespace: ns}, client.MatchingLabels{labelKey: labelValue}); err != nil {
 		nt.T.Error(err)
 	}

pkg/core/scheme.go

@@ -17,6 +17,7 @@ package core
 import (
 	admissionv1 "k8s.io/api/admissionregistration/v1"
 	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1hpa "k8s.io/api/autoscaling/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
@@ -26,8 +27,7 @@ import (
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-
-	// autoscalingv1 "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/autoscaling.k8s.io/v1"
+	autoscalingv1vpa "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/autoscaling.k8s.io/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	clusterregistry "k8s.io/cluster-registry/pkg/apis/clusterregistry/v1alpha1"
 	k8sadmissionv1 "k8s.io/kubernetes/pkg/apis/admission/v1"
@@ -62,7 +62,7 @@ var Scheme = scheme.Scheme
 func init() {
 	mustRegisterKubernetesResources()
 	mustRegisterAPIExtensionsResources()
-	// mustRegisterAutosclaingResources()
+	mustRegisterAutoscalingResources()
 
 	// Config Sync types
 	utilruntime.Must(clusterregistry.AddToScheme(scheme.Scheme))
@@ -126,7 +126,8 @@ func mustRegisterAPIExtensionsResources() {
 	utilruntime.Must(scheme.Scheme.SetVersionPriority(apiextensionsv1.SchemeGroupVersion, apiextensionsv1beta1.SchemeGroupVersion))
 }
 
-// func mustRegisterAutosclaingResources() {
-// 	utilruntime.Must(autoscalingv1.AddToScheme(scheme.Scheme))
-// 	// autoscaling API has no generated defaults or conversions
-// }
+func mustRegisterAutoscalingResources() {
+	utilruntime.Must(autoscalingv1hpa.AddToScheme(scheme.Scheme))
+	utilruntime.Must(autoscalingv1vpa.AddToScheme(scheme.Scheme))
+	// autoscaling API has no generated defaults or conversions
+}