fix: kubelet --root-dir can be used

Signed-off-by: niuli33 <[email protected]>

Author: niuli33

cmd/k8s-rdma-shared-dp/main.go

@@ -75,6 +75,10 @@ func main() {
 		log.Fatalf("Exiting.. one or more invalid configuration(s) given: %v", err)
 	}
 
+	if err := rm.SetWatchMode(); err != nil {
+		log.Fatalln(err.Error())
+	}
+
 	if err := rm.ValidateRdmaSystemMode(); err != nil {
 		log.Fatalf("Exiting.. can not change : %v", err)
 	}

deployment/k8s/base/configmap.yaml

@@ -6,6 +6,7 @@ metadata:
 data:
   config.json: |
     {
+        "kubeletRootDir": "/var/lib/kubelet",
         "periodicUpdateInterval": 300,
         "configList": [{
              "resourceName": "hca_shared_devices_a",

deployment/k8s/base/daemonset.yaml

@@ -21,6 +21,8 @@ spec:
         securityContext:
           privileged: true
         volumeMounts:
+          # 如果configmap.yaml中指定了kubeletRootDir，则需要调整为相同的路径
+          # 如：kubeletRootDir=/data/kubelet，则mountPath=/data/kubelet/device-plugins
           - name: device-plugin
             mountPath: /var/lib/kubelet/device-plugins
             readOnly: false

pkg/resources/resources_manager.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"path"
 	"regexp"
 	"strconv"
 	"time"
@@ -52,6 +53,9 @@ const (
 
 	// RDMA subsystem network namespace mode
 	rdmaExclusive = "exclusive"
+
+	// Default kubelet root dir
+	defaultKubeletRootDir = "/var/lib/kubelet"
 )
 
 var (
@@ -72,20 +76,14 @@ type resourceManager struct {
 	rds                    types.RdmaDeviceSpec
 	PeriodicUpdateInterval time.Duration
 	useCdi                 bool
+	kubeletRootDir         string
 }
 
 func NewResourceManager(configFile string, useCdi bool) types.ResourceManager {
-	watcherMode := detectPluginWatchMode(activeSockDir)
-	if watcherMode {
-		fmt.Println("Using Kubelet Plugin Registry Mode")
-	} else {
-		fmt.Println("Using Deprecated Devie Plugin Registry Path")
-	}
 	return &resourceManager{
 		configFile:            configFile,
 		defaultResourcePrefix: rdmaHcaResourcePrefix,
 		socketSuffix:          socketSuffix,
-		watchMode:             watcherMode,
 		netlinkManager:        &netlinkManager{},
 		rds:                   NewRdmaDeviceSpec(requiredRdmaDevices),
 		useCdi:                useCdi,
@@ -107,6 +105,12 @@ func (rm *resourceManager) ReadConfig() error {
 
 	log.Printf("loaded config: %+v \n", config.ConfigList)
 
+	// check kubelet root dir config
+	rm.kubeletRootDir = defaultKubeletRootDir
+	if config.KubeletRootDir != "" {
+		rm.kubeletRootDir = config.KubeletRootDir
+	}
+
 	// if periodic update is not set then use the default value
 	if config.PeriodicUpdateInterval == nil {
 		log.Println("no periodic update interval is set, use default interval 60 seconds")
@@ -228,7 +232,7 @@ func (rm *resourceManager) InitServers() error {
 				return err
 			}
 		}
-		rs, err := newResourceServer(config, filteredDevices, rm.watchMode, rm.socketSuffix, rm.useCdi)
+		rs, err := newResourceServer(config, filteredDevices, rm.watchMode, rm.socketSuffix, rm.useCdi, rm.kubeletRootDir)
 		if err != nil {
 			return err
 		}
@@ -418,3 +422,16 @@ func (rm *resourceManager) PeriodicUpdate() func() {
 		}
 	}
 }
+
+func (rm *resourceManager) SetWatchMode() error {
+	sockPathDir := path.Join(rm.kubeletRootDir, "plugins_registry")
+	watcherMode := detectPluginWatchMode(sockPathDir)
+	if watcherMode {
+		fmt.Println("Using Kubelet Plugin Registry Mode")
+	} else {
+		fmt.Println("Using Deprecated Devie Plugin Registry Path")
+	}
+	rm.watchMode = watcherMode
+
+	return nil
+}

pkg/resources/server.go

@@ -23,6 +23,7 @@ import (
 	"log"
 	"net"
 	"os"
+	"path"
 	"path/filepath"
 	"strconv"
 	"sync"
@@ -68,6 +69,7 @@ type resourceServer struct {
 	useCdi          bool
 	cdi             cdi.CDI
 	cdiResourceName string
+	kubeletRootDir  string
 }
 
 func (rsc *resourcesServerPort) GetServer() *grpc.Server {
@@ -123,10 +125,10 @@ func (rsc *resourcesServerPort) Dial(unixSocketPath string, timeout time.Duratio
 
 // newResourceServer returns an initialized server
 func newResourceServer(config *types.UserConfig, devices []types.PciNetDevice, watcherMode bool,
-	socketSuffix string, useCdi bool) (types.ResourceServer, error) {
+	socketSuffix string, useCdi bool, kubeletRootDir string) (types.ResourceServer, error) {
 	var devs []*pluginapi.Device
 
-	sockDir := activeSockDir
+	sockDir := path.Join(kubeletRootDir, "plugins_registry")
 
 	if config.RdmaHcaMax < 0 {
 		return nil, fmt.Errorf("error: Invalid value for rdmaHcaMax < 0: %d", config.RdmaHcaMax)
@@ -151,7 +153,7 @@ func newResourceServer(config *types.UserConfig, devices []types.PciNetDevice, w
 	}
 
 	if !watcherMode {
-		sockDir = deprecatedSockDir
+		sockDir = path.Join(kubeletRootDir, "device-plugins")
 	}
 
 	socketName := fmt.Sprintf("%s.%s", config.ResourceName, socketSuffix)
@@ -172,6 +174,7 @@ func newResourceServer(config *types.UserConfig, devices []types.PciNetDevice, w
 		useCdi:          useCdi,
 		cdi:             cdi.New(),
 		cdiResourceName: config.ResourceName,
+		kubeletRootDir:  kubeletRootDir,
 	}, nil
 }
 
@@ -277,7 +280,7 @@ func (rs *resourceServer) Watch() {
 
 // Register registers the device plugin for the given resourceName with Kubelet.
 func (rs *resourceServer) register() error {
-	kubeletEndpoint := filepath.Join(deprecatedSockDir, kubeEndPoint)
+	kubeletEndpoint := filepath.Join(rs.kubeletRootDir, "device-plugins", kubeEndPoint)
 	conn, err := rs.rsConnector.Dial(kubeletEndpoint, cDialTimeout)
 	if err != nil {
 		return err
@@ -424,7 +427,7 @@ func (rs *resourceServer) GetInfo(ctx context.Context, rqt *registerapi.InfoRequ
 	pluginInfoResponse := &registerapi.PluginInfo{
 		Type:              registerapi.DevicePlugin,
 		Name:              rs.resourceName,
-		Endpoint:          filepath.Join(activeSockDir, rs.socketName),
+		Endpoint:          filepath.Join(rs.kubeletRootDir, "plugins_registry", rs.socketName),
 		SupportedVersions: []string{"v1alpha1", "v1beta1"},
 	}
 	return pluginInfoResponse, nil

pkg/types/types.go

@@ -48,6 +48,7 @@ type UserConfig struct {
 
 // UserConfigList config list for servers
 type UserConfigList struct {
+	KubeletRootDir         string       `json:"kubeletRootDir"`
 	PeriodicUpdateInterval *int         `json:"periodicUpdateInterval"`
 	ConfigList             []UserConfig `json:"configList"`
 }
@@ -75,6 +76,7 @@ type ResourceManager interface {
 	RestartAllServers() error
 	GetFilteredDevices(devices []PciNetDevice, selector *Selectors) []PciNetDevice
 	PeriodicUpdate() func()
+	SetWatchMode() error
 }
 
 // ResourceServerPort to connect the resources server to k8s