refactor: make kube config optional on nas/oss node server

Author: iltyty

main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	_ "net/http/pprof"
@@ -48,6 +49,7 @@ import (
 	"github.com/sirupsen/logrus"
 	flag "github.com/spf13/pflag"
 	"golang.org/x/sys/unix"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/component-base/logs"
 	logsapi "k8s.io/component-base/logs/api/v1"
@@ -190,6 +192,8 @@ func main() {
 		}
 	}
 
+	csiCfg := getCSIPluginConfig()
+
 	for _, driverName := range driverNames {
 		wg.Add(1)
 		endPointName := replaceCsiEndpoint(driverName, *endpoint)
@@ -199,19 +203,19 @@ func main() {
 		case TypePluginNAS:
 			go func(endPoint string) {
 				defer wg.Done()
-				driver := nas.NewDriver(meta, endPoint, serviceType)
+				driver := nas.NewDriver(meta, endPoint, serviceType, csiCfg)
 				driver.Run()
 			}(endPointName)
 		case TypePluginOSS:
 			go func(endPoint string) {
 				defer wg.Done()
-				driver := oss.NewDriver(endPoint, meta, serviceType)
+				driver := oss.NewDriver(endPoint, meta, serviceType, csiCfg)
 				driver.Run()
 			}(endPointName)
 		case TypePluginDISK:
 			go func(endPoint string) {
 				defer wg.Done()
-				driver := disk.NewDriver(meta, endPoint, serviceType)
+				driver := disk.NewDriver(meta, endPoint, serviceType, csiCfg)
 				driver.Run()
 			}(endPointName)
 
@@ -296,3 +300,30 @@ func healthHandler(w http.ResponseWriter, r *http.Request) {
 	message := "Liveness probe is OK, time:" + time.String()
 	_, _ = w.Write([]byte(message))
 }
+
+func getCSIPluginConfig() (config utils.Config) {
+	cfg, err := options.GetRestConfig()
+	if err != nil {
+		klog.ErrorS(err, "error building kube config")
+		return
+	}
+
+	client, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		if cfg != nil {
+			klog.Fatal(err, "error building kube client")
+		} else {
+			klog.ErrorS(err, "error building kube client")
+			return
+		}
+	}
+
+	cm, err := client.CoreV1().ConfigMaps("kube-system").Get(context.Background(), "csi-plugin", metav1.GetOptions{})
+	if err != nil {
+		klog.ErrorS(err, "failed to get config map csi-plugin")
+		return
+	}
+
+	config.ConfigMap = cm.Data
+	return
+}

pkg/disk/disk.go

@@ -37,7 +37,6 @@ import (
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/version"
 	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
@@ -106,13 +105,12 @@ func initDriver() {
 }
 
 // NewDriver create the identity/node/controller server and disk driver
-func NewDriver(m metadata.MetadataProvider, endpoint string, serviceType utils.ServiceType) *DISK {
+func NewDriver(m metadata.MetadataProvider, endpoint string, serviceType utils.ServiceType, csiCfg utils.Config) *DISK {
 	initDriver()
 	tmpdisk := &DISK{}
 	tmpdisk.endpoint = endpoint
 
-	// Config Global vars
-	csiCfg := GlobalConfigSet(m)
+	GlobalConfigSet(m, csiCfg)
 
 	if serviceType&utils.Node != 0 {
 		GlobalConfigVar.NodeID = metadata.MustGet(m, metadata.InstanceID)
@@ -158,9 +156,7 @@ func (disk *DISK) Run() {
 }
 
 // GlobalConfigSet set Global Config
-func GlobalConfigSet(m metadata.MetadataProvider) utils.Config {
-	configMapName := "csi-plugin"
-
+func GlobalConfigSet(m metadata.MetadataProvider, csiCfg utils.Config) {
 	// Global Configs Set
 	cfg, err := options.GetRestConfig()
 	if err != nil {
@@ -179,14 +175,6 @@ func GlobalConfigSet(m metadata.MetadataProvider) utils.Config {
 		klog.Fatalf("Error building kubernetes snapclientset: %s", err.Error())
 	}
 
-	csiCfg := utils.Config{}
-	configMap, err := kubeClient.CoreV1().ConfigMaps("kube-system").Get(context.Background(), configMapName, metav1.GetOptions{})
-	if err != nil {
-		klog.Infof("Not found configmap named as csi-plugin under kube-system, with: %v", err)
-	} else {
-		csiCfg.ConfigMap = configMap.Data
-	}
-
 	// Env variables
 	avmfe := os.Getenv("ADDON_VM_FATAL_EVENTS")
 	fatalEvents := []string{}
@@ -247,7 +235,6 @@ func GlobalConfigSet(m metadata.MetadataProvider) utils.Config {
 		GlobalConfigVar.DetachBeforeDelete,
 		GlobalConfigVar.ClusterID,
 	)
-	return csiCfg
 }
 
 func newBatcher(fromNode bool) (waitstatus.StatusWaiter[ecs.Disk], batcher.Batcher[ecs.Disk]) {

pkg/mounter/fuse_pod_manager/fuse_pod_manager.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	mounterutils "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	corev1 "k8s.io/api/core/v1"
 	apiserrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -101,14 +102,11 @@ type FuseContainerConfig struct {
 	Extra       map[string]string
 }
 
-func ExtractFuseContainerConfig(configmap *corev1.ConfigMap, name string) (config FuseContainerConfig) {
+func ExtractFuseContainerConfig(csiCfg utils.Config, name string) (config FuseContainerConfig) {
 	config.Resources.Requests = make(corev1.ResourceList)
 	config.Resources.Limits = make(corev1.ResourceList)
 
-	if configmap == nil {
-		return
-	}
-	content := configmap.Data["fuse-"+name]
+	content := csiCfg.Get("fuse-"+name, "OSS_FUSE_"+strings.ToUpper(name), "")
 	for _, line := range strings.Split(content, "\n") {
 		line = strings.TrimSpace(line)
 		if line == "" {

pkg/mounter/fuse_pod_manager/fuse_pod_manager_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -17,8 +18,8 @@ import (
 )
 
 func Test_ExtractFuseContainerConfig(t *testing.T) {
-	configmap := &corev1.ConfigMap{
-		Data: map[string]string{
+	csiCfg := utils.Config{
+		ConfigMap: map[string]string{
 			"fuse-ossfs": `
 				image=ossfs:latest
 				cpu-request=100m
@@ -32,7 +33,7 @@ func Test_ExtractFuseContainerConfig(t *testing.T) {
 			`,
 		},
 	}
-	config := ExtractFuseContainerConfig(configmap, "ossfs")
+	config := ExtractFuseContainerConfig(csiCfg, "ossfs")
 	expected := FuseContainerConfig{
 		Resources: corev1.ResourceRequirements{
 			Limits: corev1.ResourceList{

pkg/mounter/fuse_pod_manager/oss/ossfs/manager.go

@@ -53,8 +53,8 @@ var ossfsDbglevels = map[string]string{
 	fpm.DebugLevelFatal: "crit",
 }
 
-func NewFuseOssfs(configmap *corev1.ConfigMap, m metadata.MetadataProvider) ossfpm.OSSFuseMounterType {
-	config := fpm.ExtractFuseContainerConfig(configmap, mounterutils.OssFsType)
+func NewFuseOssfs(csiCfg utils.Config, m metadata.MetadataProvider) ossfpm.OSSFuseMounterType {
+	config := fpm.ExtractFuseContainerConfig(csiCfg, mounterutils.OssFsType)
 
 	// set default image
 	ossfpm.SetDefaultImage(mounterutils.OssFsType, m, &config)

pkg/mounter/fuse_pod_manager/oss/ossfs/manager_test.go

@@ -10,6 +10,7 @@ import (
 	fpm "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/fuse_pod_manager"
 	ossfpm "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/fuse_pod_manager/oss"
 	mounterutils "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -102,7 +103,7 @@ func Test_buildAuthSpec_ossfs(t *testing.T) {
 
 func TestPrecheckAuthConfig_ossfs(t *testing.T) {
 	fakeMeta := metadata.NewMetadata()
-	fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+	fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 	tests := []struct {
 		name    string
 		opts    *ossfpm.Options
@@ -316,7 +317,7 @@ func TestMakeAuthConfig_ossfs(t *testing.T) {
 	t.Setenv("CLUSTER_ID", "cluster-id")
 	t.Setenv("ALIBABA_CLOUD_ACCOUNT_ID", "account-id")
 	fakeMeta := metadata.NewMetadata()
-	fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+	fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 	tests := []struct {
 		name           string
 		options        *ossfpm.Options
@@ -597,7 +598,7 @@ func TestMakeMountOptions_ossfs(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Setenv("REGION_ID", tt.region)
 			fakeMeta := metadata.NewMetadata()
-			fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+			fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 			mountOptions, err := fakeOssfs.MakeMountOptions(tt.opts, fakeMeta)
 			assert.Equal(t, tt.expectedError, err != nil)
 			assert.ElementsMatch(t, tt.expected, mountOptions)
@@ -710,7 +711,7 @@ func TestGetAuthOpttions_ossfs(t *testing.T) {
 
 func TestAddDefaultMountOptions_ossfs(t *testing.T) {
 	fakeMeta := metadata.NewMetadata()
-	fakeInter := NewFuseOssfs(nil, fakeMeta)
+	fakeInter := NewFuseOssfs(utils.Config{}, fakeMeta)
 	fakeOssfs, ok := fakeInter.(*fuseOssfs)
 	if !ok {
 		t.Fatalf("failed to cast to fuseOssfs")

pkg/mounter/fuse_pod_manager/oss/ossfs2/manager.go

@@ -39,8 +39,8 @@ var ossfs2Dbglevels = map[string]string{
 	fpm.DebugLevelInfo:  "info",
 }
 
-func NewFuseOssfs(configmap *corev1.ConfigMap, m metadata.MetadataProvider) ossfpm.OSSFuseMounterType {
-	config := fpm.ExtractFuseContainerConfig(configmap, mounterutils.OssFs2Type)
+func NewFuseOssfs(csiCfg utils.Config, m metadata.MetadataProvider) ossfpm.OSSFuseMounterType {
+	config := fpm.ExtractFuseContainerConfig(csiCfg, mounterutils.OssFs2Type)
 	// set default image
 	ossfpm.SetDefaultImage(mounterutils.OssFs2Type, m, &config)
 	// set default memory request

pkg/mounter/fuse_pod_manager/oss/ossfs2/manager_test.go

@@ -11,6 +11,7 @@ import (
 	fpm "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/fuse_pod_manager"
 	ossfpm "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/fuse_pod_manager/oss"
 	mounterutils "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -19,7 +20,7 @@ import (
 
 func TestPrecheckAuthConfig_ossfs2(t *testing.T) {
 	fakeMeta := metadata.NewMetadata()
-	fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+	fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 	tests := []struct {
 		name    string
 		opts    *ossfpm.Options
@@ -156,7 +157,7 @@ func TestMakeAuthConfig_ossfs2(t *testing.T) {
 	t.Setenv("CLUSTER_ID", "cluster-id")
 	t.Setenv("ALIBABA_CLOUD_ACCOUNT_ID", "account-id")
 	fakeMeta := metadata.NewMetadata()
-	fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+	fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 	tests := []struct {
 		name    string
 		options *ossfpm.Options
@@ -371,7 +372,7 @@ func TestMakeMountOptions_ossfs2(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Setenv("REGION_ID", tt.region)
 			fakeMeta := metadata.NewMetadata()
-			fakeOssfs := NewFuseOssfs(nil, fakeMeta)
+			fakeOssfs := NewFuseOssfs(utils.Config{}, fakeMeta)
 			opts, err := fakeOssfs.MakeMountOptions(tt.opts, fakeMeta)
 			assert.Equal(t, tt.expectedError, err != nil)
 			assert.Equal(t, tt.expected, opts)
@@ -381,7 +382,7 @@ func TestMakeMountOptions_ossfs2(t *testing.T) {
 
 func TestAddDefaultMountOptions_ossfs2(t *testing.T) {
 	fakeMeta := metadata.NewMetadata()
-	fakeInter := NewFuseOssfs(nil, fakeMeta)
+	fakeInter := NewFuseOssfs(utils.Config{}, fakeMeta)
 	fakeOssfs, ok := fakeInter.(*fuseOssfs)
 	if !ok {
 		t.Fatalf("failed to cast to fuseOssfs")

pkg/mounter/fuse_pod_manager/oss/registry.go

@@ -3,18 +3,18 @@ package oss
 import (
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/cloud/metadata"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
-	corev1 "k8s.io/api/core/v1"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"k8s.io/client-go/kubernetes"
 )
 
 var (
-	fstypeToFactory      = map[string]func(*corev1.ConfigMap, metadata.MetadataProvider) OSSFuseMounterType{}
+	fstypeToFactory      = map[string]func(utils.Config, metadata.MetadataProvider) OSSFuseMounterType{}
 	fstypeToPath         = map[string]string{}
 	fstypeToInterceptors = map[string][]mounter.MountInterceptor{}
 )
 
 // RegisterFuseMounter registers a fuse mounter factory for a given fstype
-func RegisterFuseMounter(fstype string, factory func(*corev1.ConfigMap, metadata.MetadataProvider) OSSFuseMounterType) {
+func RegisterFuseMounter(fstype string, factory func(utils.Config, metadata.MetadataProvider) OSSFuseMounterType) {
 	fstypeToFactory[fstype] = factory
 }
 
@@ -50,12 +50,12 @@ func GetFuseMountInterceptors(fstype string) ([]mounter.MountInterceptor, bool)
 }
 
 // GetFuseMounter returns a fuse mounter instance for the given fstype
-func GetFuseMounter(fstype string, configmap *corev1.ConfigMap, m metadata.MetadataProvider) (OSSFuseMounterType, error) {
+func GetFuseMounter(fstype string, csiCfg utils.Config, m metadata.MetadataProvider) (OSSFuseMounterType, error) {
 	factory, ok := fstypeToFactory[fstype]
 	if !ok {
 		return nil, &UnsupportedFstypeError{Fstype: fstype}
 	}
-	return factory(configmap, m), nil
+	return factory(csiCfg, m), nil
 }
 
 // GetAllRegisteredFuseTypes returns all registered fuse types
@@ -70,10 +70,10 @@ func GetAllRegisteredFuseTypes() []string {
 // GetAllOSSFusePodManagers creates a map of all registered OSS fuse pod managers
 // configmap can be nil if not available (e.g., in CSI agent mode)
 // client can be nil if not needed (e.g., in CSI agent mode)
-func GetAllOSSFusePodManagers(configmap *corev1.ConfigMap, m metadata.MetadataProvider, client kubernetes.Interface) map[string]*OSSFusePodManager {
+func GetAllOSSFusePodManagers(csiCfg utils.Config, m metadata.MetadataProvider, client kubernetes.Interface) map[string]*OSSFusePodManager {
 	fusePodManagers := make(map[string]*OSSFusePodManager, len(fstypeToFactory))
 	for fstype, factory := range fstypeToFactory {
-		fusePodManagers[fstype] = NewOSSFusePodManager(factory(configmap, m), client)
+		fusePodManagers[fstype] = NewOSSFusePodManager(factory(csiCfg, m), client)
 	}
 	return fusePodManagers
 }

pkg/mounter/fuse_pod_manager/oss/registry_test.go

@@ -7,6 +7,7 @@ import (
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
 	fpm "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/fuse_pod_manager"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/interceptors"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/utils"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -54,7 +55,7 @@ func TestGetFuseMounterInterceptors(t *testing.T) {
 func TestGetAllRegisteredFuseTypes(t *testing.T) {
 	// Register a test factory
 	testType := "test-fuse-type-3"
-	testFactory := func(*corev1.ConfigMap, metadata.MetadataProvider) OSSFuseMounterType {
+	testFactory := func(utils.Config, metadata.MetadataProvider) OSSFuseMounterType {
 		return nil
 	}
 	RegisterFuseMounter(testType, testFactory)
@@ -75,13 +76,13 @@ func TestGetAllOSSFusePodManagers(t *testing.T) {
 
 	// Register a test factory that returns a valid mounter
 	testType := "test-fuse-type-4"
-	testFactory := func(*corev1.ConfigMap, metadata.MetadataProvider) OSSFuseMounterType {
+	testFactory := func(utils.Config, metadata.MetadataProvider) OSSFuseMounterType {
 		return &testFuseMounter{name: testType}
 	}
 	RegisterFuseMounter(testType, testFactory)
 
 	// Test with nil configmap and nil client (CSI agent mode)
-	managers := GetAllOSSFusePodManagers(nil, fakeMeta, nil)
+	managers := GetAllOSSFusePodManagers(utils.Config{}, fakeMeta, nil)
 
 	// Should have at least the test manager
 	assert.GreaterOrEqual(t, len(managers), 1, "Should have at least 1 fuse pod manager")