refactor: create storage classes directly instead of using CRS (#539)

**What problem does this PR solve?**:

**Which issue(s) this PR fixes**:
Removes our dependency for Cluster Resource Sets entirely for Nutanix
Clusters

**How Has This Been Tested?**:
<!--
Please describe the tests that you ran to verify your changes.
Provide output from the tests and any manual steps needed to replicate
the tests.
-->

Create a cluster and check if storage class is created.


```sh
02:55 PM  faiqus @ archlinux  ~/go/src/github.com/d2iq-labs/cluster-api-runtime-extensions-nutanix  faiq/storageclass-direct? 
$ kubectl --kubeconfig=faiqcsi.conf get sc
NAME                       PROVISIONER       RECLAIMPOLICY   VOLUMEBINDINGMODE      ALLOWVOLUMEEXPANSION   AGE
nutanix-volume (default)   csi.nutanix.com   Delete          WaitForFirstConsumer   false                  3m

02:56 PM  faiqus @ archlinux  ~/go/src/github.com/d2iq-labs/cluster-api-runtime-extensions-nutanix  faiq/storageclass-direct? 
$ kubectl get clusterresourcesets
No resources found in default namespace.

02:56 PM  faiqus @ archlinux  ~/go/src/github.com/d2iq-labs/cluster-api-runtime-extensions-nutanix  faiq/storageclass-direct? 
$ kubectl get clusterresourcesets -A
No resources found

02:56 PM  faiqus @ archlinux  ~/go/src/github.com/d2iq-labs/cluster-api-runtime-extensions-nutanix  faiq/storageclass-direct? 
$ kubectl get helmreleaseproxy -A
NAMESPACE   NAME                                   CLUSTER   READY   REASON   STATUS     REVISION
default     cilium-faiqcsi-ndb7v                   faiqcsi   True             deployed   1
default     cluster-autoscaler-faiqcsi-v58lv       faiqcsi   True             deployed   1
default     node-feature-discovery-faiqcsi-q9n2w   faiqcsi   True             deployed   1
default     nutanix-cloud-provider-faiqcsi-2d4zq   faiqcsi   True             deployed   1
default     nutanix-csi-snapshot-faiqcsi-wjmjr     faiqcsi   True             deployed   1
default     nutanix-csi-storage-faiqcsi-6567j      faiqcsi   True             deployed   1
```

**Special notes for your reviewer**:
<!--
Use this to provide any additional information to the reviewers.
This may include:
- Best way to review the PR.
- Where the author wants the most review attention on.
- etc.
-->

Author: faiq

pkg/handlers/generic/lifecycle/csi/aws-ebs/handler.go

@@ -11,7 +11,6 @@ import (
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	runtimehooksv1 "sigs.k8s.io/cluster-api/exp/runtime/hooks/api/v1alpha1"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -22,7 +21,7 @@ import (
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/pkg/handlers/options"
 )
 
-var defaultStorageClassParams = map[string]string{
+var defaultStorageClassParameters = map[string]string{
 	"csi.storage.k8s.io/fstype": "ext4",
 	"type":                      "gp3",
 }
@@ -74,49 +73,20 @@ func (a *AWSEBS) Apply(
 	default:
 		return fmt.Errorf("stategy %s not implemented", strategy)
 	}
-	return a.createStorageClasses(
+	err := lifecycleutils.CreateStorageClassOnRemote(
 		ctx,
+		a.client,
 		provider.StorageClassConfig,
 		&req.Cluster,
 		defaultStorageConfig,
+		v1alpha1.CSIProviderAWSEBS,
+		v1alpha1.AWSEBSProvisioner,
+		defaultStorageClassParameters,
 	)
-}
-
-func (a *AWSEBS) createStorageClasses(ctx context.Context,
-	configs []v1alpha1.StorageClassConfig,
-	cluster *clusterv1.Cluster,
-	defaultStorageConfig *v1alpha1.DefaultStorage,
-) error {
-	allStorageClasses := make([]runtime.Object, 0, len(configs))
-	for _, config := range configs {
-		setAsDefault := config.Name == defaultStorageConfig.StorageClassConfigName &&
-			v1alpha1.CSIProviderAWSEBS == defaultStorageConfig.ProviderName
-		allStorageClasses = append(allStorageClasses, lifecycleutils.CreateStorageClass(
-			config,
-			v1alpha1.AWSEBSProvisioner,
-			setAsDefault,
-			defaultStorageClassParams,
-		))
-	}
-	cm, err := lifecycleutils.CreateConfigMapForCRS(
-		fmt.Sprintf("aws-storageclass-cm-%s", cluster.Name),
-		a.config.DefaultsNamespace(),
-		allStorageClasses...,
-	)
-	if err != nil {
-		return err
-	}
-	err = client.ServerSideApply(ctx, a.client, cm)
 	if err != nil {
-		return err
+		return fmt.Errorf("error creating StorageClasses for the AWS EBS CSI driver: %w", err)
 	}
-	return lifecycleutils.EnsureCRSForClusterFromObjects(
-		ctx,
-		"aws-storageclass-crs",
-		a.client,
-		cluster,
-		cm,
-	)
+	return nil
 }
 
 func (a *AWSEBS) handleCRSApply(ctx context.Context,

pkg/handlers/generic/lifecycle/csi/nutanix-csi/handler.go

@@ -10,7 +10,6 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/spf13/pflag"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	runtimehooksv1 "sigs.k8s.io/cluster-api/exp/runtime/hooks/api/v1alpha1"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -117,16 +116,19 @@ func (n *NutanixCSI) Apply(
 		}
 	}
 
-	err := n.createStorageClasses(
+	err := lifecycleutils.CreateStorageClassOnRemote(
 		ctx,
+		n.client,
 		provider.StorageClassConfig,
 		&req.Cluster,
 		defaultStorageConfig,
+		v1alpha1.CSIProviderNutanix,
+		v1alpha1.NutanixProvisioner,
+		defaultStorageClassParameters,
 	)
 	if err != nil {
 		return fmt.Errorf("error creating StorageClasses for the Nutanix CSI driver: %w", err)
 	}
-
 	return nil
 }
 
@@ -222,41 +224,3 @@ func (n *NutanixCSI) handleHelmAddonApply(
 
 	return nil
 }
-
-func (n *NutanixCSI) createStorageClasses(
-	ctx context.Context,
-	configs []v1alpha1.StorageClassConfig,
-	cluster *clusterv1.Cluster,
-	defaultStorageConfig *v1alpha1.DefaultStorage,
-) error {
-	allStorageClasses := make([]runtime.Object, 0, len(configs))
-	for _, config := range configs {
-		setAsDefault := config.Name == defaultStorageConfig.StorageClassConfigName &&
-			v1alpha1.CSIProviderNutanix == defaultStorageConfig.ProviderName
-		allStorageClasses = append(allStorageClasses, lifecycleutils.CreateStorageClass(
-			config,
-			v1alpha1.NutanixProvisioner,
-			setAsDefault,
-			defaultStorageClassParameters,
-		))
-	}
-	cm, err := lifecycleutils.CreateConfigMapForCRS(
-		fmt.Sprintf("nutanix-storageclass-cm-%s", cluster.Name),
-		n.config.DefaultsNamespace(),
-		allStorageClasses...,
-	)
-	if err != nil {
-		return err
-	}
-	err = client.ServerSideApply(ctx, n.client, cm)
-	if err != nil {
-		return err
-	}
-	return lifecycleutils.EnsureCRSForClusterFromObjects(
-		ctx,
-		"nutanix-storageclass-crs",
-		n.client,
-		cluster,
-		cm,
-	)
-}

pkg/handlers/generic/lifecycle/utils/scs.go

@@ -4,11 +4,18 @@
 package utils
 
 import (
+	"context"
+	"fmt"
+
 	storagev1 "k8s.io/api/storage/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/api/v1alpha1"
+	"github.com/nutanix-cloud-native/cluster-api-runtime-extensions-nutanix/common/pkg/k8s/client"
 )
 
 const (
@@ -50,3 +57,38 @@ func CreateStorageClass(
 	}
 	return &sc
 }
+
+func CreateStorageClassOnRemote(
+	ctx context.Context,
+	cl ctrlclient.Client,
+	configs []v1alpha1.StorageClassConfig,
+	cluster *clusterv1.Cluster,
+	defaultStorageConfig *v1alpha1.DefaultStorage,
+	csiProvider string,
+	provisioner v1alpha1.StorageProvisioner,
+	defaultParameters map[string]string,
+) error {
+	allStorageClasses := make([]*storagev1.StorageClass, 0, len(configs))
+	for _, config := range configs {
+		setAsDefault := config.Name == defaultStorageConfig.StorageClassConfigName &&
+			csiProvider == defaultStorageConfig.ProviderName
+		allStorageClasses = append(allStorageClasses, CreateStorageClass(
+			config,
+			provisioner,
+			setAsDefault,
+			defaultParameters,
+		))
+	}
+	clusterKey := ctrlclient.ObjectKeyFromObject(cluster)
+	remoteClient, err := remote.NewClusterClient(ctx, "", cl, clusterKey)
+	if err != nil {
+		return fmt.Errorf("error creating client for remote cluster: %w", err)
+	}
+	for _, sc := range allStorageClasses {
+		err = client.ServerSideApply(ctx, remoteClient, sc)
+		if err != nil {
+			return fmt.Errorf("error creating storage class %v on remote cluster %w", sc, err)
+		}
+	}
+	return nil
+}