Simulate a scale test on the OCM Policy Propagator

main.go

package main

import (
	"context"
	"fmt"
	"os/user"
	"path"
	"sync"

	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	configpoliciesv1 "open-cluster-management.io/config-policy-controller/api/v1"
	policiesv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
	"open-cluster-management.io/governance-policy-propagator/test/utils"
	appsv1 "open-cluster-management.io/multicloud-operators-subscription/pkg/apis/apps/placementrule/v1"
)

const numClusters = 1000
const numUpdatesCycle = 100

func main() {
	clusters := []string{}
	for i := 0; i < numClusters; i++ {
		clusters = append(clusters, fmt.Sprintf("managed%d", i+1))
	}

	usr, _ := user.Current()

	config, err := clientcmd.BuildConfigFromFlags("", path.Join(usr.HomeDir, ".kube/config"))
	if err != nil {
		panic(err)
	}

	dynamicClient := dynamic.NewForConfigOrDie(config)

	mcGVR := schema.GroupVersionResource{
		Group: "cluster.open-cluster-management.io", Version: "v1", Resource: "managedclusters",
	}

	fmt.Println("Checking if the managed clusters already exist")

	_, err = dynamicClient.Resource(mcGVR).Get(context.TODO(), clusters[len(clusters)-1], metav1.GetOptions{})
	if err != nil {
		if !errors.IsNotFound(err) {
			panic(err)
		}

		// Create the managed clusters only if the last cluster does not exist so as to not try to recreate them all.
		wg := sync.WaitGroup{}
		clusterChannel := make(chan string, len(clusters))

		for i := 0; i < 10; i++ {
			wg.Add(1)
			go createManagedCluster(clusterChannel, &wg, config)
		}

		for _, cluster := range clusters {
			clusterChannel <- cluster
		}

		close(clusterChannel)

		wg.Wait()
	}

	fmt.Println("Creating the PlacementRule")

	plr := appsv1.PlacementRule{
		TypeMeta: metav1.TypeMeta{
			Kind:       "PlacementRule",
			APIVersion: "apps.open-cluster-management.io/v1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      "test-placement-rule",
			Namespace: "default",
		},
		Spec:   appsv1.PlacementRuleSpec{},
		Status: *utils.GeneratePlrStatus(clusters...),
	}
	plrUnstructured := &unstructured.Unstructured{}

	plrUnstructured.Object, err = runtime.DefaultUnstructuredConverter.ToUnstructured(&plr)
	if err != nil {
		panic(err)
	}

	gvrPlacementRule := schema.GroupVersionResource{
		Group: "apps.open-cluster-management.io", Version: "v1", Resource: "placementrules",
	}

	plrUnstructuredCreated, err := dynamicClient.Resource(gvrPlacementRule).Namespace("default").Create(
		context.TODO(), plrUnstructured, metav1.CreateOptions{},
	)
	if err != nil {
		if !errors.IsAlreadyExists(err) {
			panic(err)
		}

		plrUnstructuredCreated, err = dynamicClient.Resource(gvrPlacementRule).Namespace("default").Get(
			context.TODO(), plrUnstructured.GetName(), metav1.GetOptions{},
		)
		if err != nil {
			panic(err)
		}
	}

	plrUnstructuredCreated.Object["status"] = plrUnstructured.Object["status"]

	fmt.Println("Updating the PlacementRule status")

	_, err = dynamicClient.Resource(gvrPlacementRule).Namespace("default").UpdateStatus(
		context.TODO(), plrUnstructuredCreated, metav1.UpdateOptions{},
	)
	if err != nil {
		panic(err)
	}

	fmt.Println("Creating the Policy")
	configPolicy := configpoliciesv1.ConfigurationPolicy{
		TypeMeta: metav1.TypeMeta{
			Kind:       "ConfigurationPolicy",
			APIVersion: "policy.open-cluster-management.io/v1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: "test-policy",
		},
		Spec: configpoliciesv1.ConfigurationPolicySpec{
			Severity:          "low",
			RemediationAction: "inform",
			ObjectTemplates: []*configpoliciesv1.ObjectTemplate{
				{
					ComplianceType: configpoliciesv1.MustHave,
					ObjectDefinition: runtime.RawExtension{
						Object: &unstructured.Unstructured{
							Object: map[string]interface{}{
								"apiVersion": "v1",
								"kind":       "Namespace",
								"metadata": map[string]interface{}{
									"name": "test",
								},
							},
						},
					},
				},
			},
		},
	}

	policy := policiesv1.Policy{
		TypeMeta: metav1.TypeMeta{
			Kind:       "Policy",
			APIVersion: "policy.open-cluster-management.io/v1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      "test-policy",
			Namespace: "default",
		},
		Spec: policiesv1.PolicySpec{
			RemediationAction: "inform",
			Disabled:          false,
			PolicyTemplates: []*policiesv1.PolicyTemplate{
				{
					ObjectDefinition: runtime.RawExtension{
						Object: &configPolicy,
					},
				},
			},
		},
	}

	policyUnstructured := &unstructured.Unstructured{}
	policyUnstructured.Object, err = runtime.DefaultUnstructuredConverter.ToUnstructured(&policy)
	if err != nil {
		panic(err)
	}

	gvrPolicy := schema.GroupVersionResource{
		Group: "policy.open-cluster-management.io", Version: "v1", Resource: "policies",
	}

	_, err = dynamicClient.Resource(gvrPolicy).Namespace("default").Create(
		context.TODO(), policyUnstructured, metav1.CreateOptions{},
	)
	if err != nil && !errors.IsAlreadyExists(err) {
		panic(err)
	}

	fmt.Println("Creating the PlacementBinding")
	gvrPB := schema.GroupVersionResource{
		Group: "policy.open-cluster-management.io", Version: "v1", Resource: "placementbindings",
	}
	pb := policiesv1.PlacementBinding{
		TypeMeta: metav1.TypeMeta{
			Kind:       "PlacementBinding",
			APIVersion: "policy.open-cluster-management.io/v1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      "test-placementbinding",
			Namespace: "default",
		},
		PlacementRef: policiesv1.PlacementSubject{
			APIGroup: gvrPlacementRule.Group,
			Kind:     plr.Kind,
			Name:     plr.Name,
		},
		Subjects: []policiesv1.Subject{
			{
				APIGroup: gvrPolicy.Group,
				Kind:     policy.Kind,
				Name:     policy.Name,
			},
		},
	}

	pbUnstructured := &unstructured.Unstructured{}
	pbUnstructured.Object, err = runtime.DefaultUnstructuredConverter.ToUnstructured(&pb)
	if err != nil {
		panic(err)
	}

	_, err = dynamicClient.Resource(gvrPB).Namespace("default").Create(
		context.TODO(), pbUnstructured, metav1.CreateOptions{},
	)
	if err != nil && !errors.IsAlreadyExists(err) {
		panic(err)
	}

	for i := 0; i < numUpdatesCycle; i++ {
		var complianceState string

		if i%2 == 0 {
			complianceState = "Compliant"
		} else {
			complianceState = "NonCompliant"
		}

		wg := sync.WaitGroup{}
		clusterChannel := make(chan string, len(clusters))

		for i := 0; i < 10; i++ {
			wg.Add(1)
			go patchPolicy(clusterChannel, &wg, config, complianceState)
		}

		for _, cluster := range clusters {
			clusterChannel <- cluster
		}

		close(clusterChannel)

		wg.Wait()
	}
}

func patchPolicy(clusters <-chan string, wg *sync.WaitGroup, config *rest.Config, complianceState string) {
	dynamicClient := dynamic.NewForConfigOrDie(config)
	gvrPolicy := schema.GroupVersionResource{
		Group: "policy.open-cluster-management.io", Version: "v1", Resource: "policies",
	}

	for cluster := range clusters {
		replicatedName := "default.test-policy"
		fmt.Printf("Patching the status of %s/%s\n", cluster, replicatedName)

		replicatedPolicy, err := dynamicClient.Resource(gvrPolicy).Namespace(cluster).Get(
			context.TODO(), replicatedName, metav1.GetOptions{},
		)

		if err != nil {
			fmt.Printf("Failed to get the replicated policy for %s: %s\n", cluster, err.Error())

			continue
		}

		replicatedPolicy.Object["status"] = map[string]interface{}{"compliant": complianceState}

		_, err = dynamicClient.Resource(gvrPolicy).Namespace(cluster).UpdateStatus(
			context.TODO(), replicatedPolicy, metav1.UpdateOptions{},
		)
		if err != nil {
			fmt.Printf("Failed to update the replicated policy status for %s: %s\n", cluster, err.Error())
		}
	}

	wg.Done()
}

func createManagedCluster(clusters <-chan string, wg *sync.WaitGroup, config *rest.Config) {
	dynamicClient := dynamic.NewForConfigOrDie(config)
	mcGVR := schema.GroupVersionResource{
		Group: "cluster.open-cluster-management.io", Version: "v1", Resource: "managedclusters",
	}
	nsGVR := schema.GroupVersionResource{Version: "v1", Resource: "namespaces"}

	for cluster := range clusters {
		fmt.Println("Creating ManagedCluster " + cluster)
		mc := unstructured.Unstructured{
			Object: map[string]interface{}{
				"apiVersion": "cluster.open-cluster-management.io/v1",
				"kind":       "ManagedCluster",
				"metadata": map[string]interface{}{
					"name": cluster,
				},
			},
		}

		_, err := dynamicClient.Resource(mcGVR).Create(context.TODO(), &mc, metav1.CreateOptions{})
		if err != nil && !errors.IsAlreadyExists(err) {
			panic(err)
		}

		fmt.Println("Creating the namespace" + cluster)
		ns := unstructured.Unstructured{
			Object: map[string]interface{}{
				"apiVersion": "v1",
				"kind":       "Namespace",
				"metadata": map[string]interface{}{
					"name": cluster,
				},
			},
		}

		_, err = dynamicClient.Resource(nsGVR).Create(context.TODO(), &ns, metav1.CreateOptions{})
		if err != nil && !errors.IsAlreadyExists(err) {
			panic(err)
		}
	}

	wg.Done()
}