Allow kubeconfig info to be passed in

Author: julianKatz

cmd/gce-pd-csi-driver/main.go

@@ -29,6 +29,7 @@ import (
 
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/strings/slices"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
@@ -51,6 +52,10 @@ var (
 	grpcLogCharCap       = flag.Int("grpc-log-char-cap", 10000, "The maximum amount of characters logged for every grpc responses")
 	enableOtelTracing    = flag.Bool("enable-otel-tracing", false, "If set, enable opentelemetry tracing for the driver. The tracing is disabled by default. Configure the exporter endpoint with OTEL_EXPORTER_OTLP_ENDPOINT and other env variables, see https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#general-sdk-configuration.")
 
+	// Kubernetes client configuration flags
+	master     = flag.String("master", "", "The address of the Kubernetes API server (overrides any value in kubeconfig)")
+	kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig file, specifying how to connect to the API server. If not set, in-cluster config will be used")
+
 	errorBackoffInitialDurationMs = flag.Int("backoff-initial-duration-ms", 200, "The amount of ms for the initial duration of the backoff condition for controller publish/unpublish CSI operations. Default is 200.")
 	errorBackoffMaxDurationMs     = flag.Int("backoff-max-duration-ms", 300000, "The amount of ms for the max duration of the backoff condition for controller publish/unpublish CSI operations. Default is 300000 (5m).")
 	extraVolumeLabelsStr          = flag.String("extra-labels", "", "Extra labels to attach to each PD created. It is a comma separated list of key value pairs like '<key1>=<value1>,<key2>=<value2>'. See https://cloud.google.com/compute/docs/labeling-resources for details")
@@ -281,7 +286,6 @@ func handle() {
 			EnableDataCache:          *enableDataCacheFlag,
 			DataCacheEnabledNodePool: isDataCacheEnabledNodePool,
 		}
-
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
 
 		if *maxConcurrentFormatAndMount > 0 {
@@ -321,10 +325,20 @@ func handle() {
 }
 
 func instantiateKubeClient() (*kubernetes.Clientset, error) {
-	cfg, err := rest.InClusterConfig()
+	var cfg *rest.Config
+	var err error
+
+	if *master != "" || *kubeconfig != "" {
+		klog.Infof("Either master or kubeconfig specified. building kube config from that..")
+		cfg, err = clientcmd.BuildConfigFromFlags(*master, *kubeconfig)
+	} else {
+		klog.Infof("No master or kubeconfig specified. building in-cluster kube config")
+		cfg, err = rest.InClusterConfig()
+	}
 	if err != nil {
 		return nil, fmt.Errorf("failed to create REST Config for k8s client: %w", err)
 	}
+
 	kubeClient, err := kubernetes.NewForConfig(cfg)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create k8s client: %w", err)

pkg/gce-pd-csi-driver/controller.go

@@ -629,12 +629,8 @@ func (gceCS *GCEControllerServer) createSingleDeviceDisk(ctx context.Context, re
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
 	defer gceCS.volumeLocks.Release(volumeID)
-<<<<<<< HEAD
-	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones, accessMode)
-=======
->>>>>>> 474af8f4 (Only add disk support Topology if all nodes have a disk support label)
 
-	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones)
+	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones, accessMode)
 	if err != nil {
 		return nil, common.LoggedError("CreateVolume failed: %v", err)
 	}

pkg/gce-pd-csi-driver/node.go

@@ -31,7 +31,6 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 
@@ -589,28 +588,6 @@ func (ns *GCENodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRe
 	return resp, err
 }
 
-// fetchGKETopologyLabels retrieves the node labels with the prefix
-// `topology.gke.io/` for the specified node.
-func (ns *GCENodeServer) fetchGKETopologyLabels(ctx context.Context, nodeName string) (map[string]string, error) {
-	klog.V(2).Infof("Retrieving node topology labels for node %q", nodeName)
-
-	node, err := ns.KubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
-	if err != nil {
-		// Q: Should we retry if we fail to get the node?
-		return nil, err
-	}
-
-	topology := make(map[string]string)
-	for k, v := range node.GetLabels() {
-		if common.IsGKETopologyLabel(k) {
-			klog.V(2).Infof("Including node topology label %q=%q", k, v)
-			topology[k] = v
-		}
-	}
-
-	return topology, nil
-}
-
 func (ns *GCENodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
 	if len(req.VolumeId) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty")