Delay CSI client initialization

pkg/volume/csi/csi_block.go

@@ -37,8 +37,8 @@ import (
 )
 
 type csiBlockMapper struct {
+	csiClientGetter
 	k8s        kubernetes.Interface
-	csiClient  csiClient
 	plugin     *csiPlugin
 	driverName csiDriverName
 	specName   string
@@ -247,14 +247,20 @@ func (m *csiBlockMapper) SetUpDevice() (string, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 
+	csiClient, err := m.csiClientGetter.Get()
+	if err != nil {
+		klog.Error(log("blockMapper.SetUpDevice failed to get CSI client: %v", err))
+		return "", err
+	}
+
 	// Call NodeStageVolume
-	stagingPath, err := m.stageVolumeForBlock(ctx, m.csiClient, accessMode, csiSource, attachment)
+	stagingPath, err := m.stageVolumeForBlock(ctx, csiClient, accessMode, csiSource, attachment)
 	if err != nil {
 		return "", err
 	}
 
 	// Call NodePublishVolume
-	publishPath, err := m.publishVolumeForBlock(ctx, m.csiClient, accessMode, csiSource, attachment, stagingPath)
+	publishPath, err := m.publishVolumeForBlock(ctx, csiClient, accessMode, csiSource, attachment, stagingPath)
 	if err != nil {
 		return "", err
 	}
@@ -326,6 +332,12 @@ func (m *csiBlockMapper) TearDownDevice(globalMapPath, devicePath string) error
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 
+	csiClient, err := m.csiClientGetter.Get()
+	if err != nil {
+		klog.Error(log("blockMapper.TearDownDevice failed to get CSI client: %v", err))
+		return err
+	}
+
 	// Call NodeUnpublishVolume
 	publishPath := m.getPublishPath()
 	if _, err := os.Stat(publishPath); err != nil {
@@ -335,7 +347,7 @@ func (m *csiBlockMapper) TearDownDevice(globalMapPath, devicePath string) error
 			return err
 		}
 	} else {
-		err := m.unpublishVolumeForBlock(ctx, m.csiClient, publishPath)
+		err := m.unpublishVolumeForBlock(ctx, csiClient, publishPath)
 		if err != nil {
 			return err
 		}
@@ -350,7 +362,7 @@ func (m *csiBlockMapper) TearDownDevice(globalMapPath, devicePath string) error
 			return err
 		}
 	} else {
-		err := m.unstageVolumeForBlock(ctx, m.csiClient, stagingPath)
+		err := m.unstageVolumeForBlock(ctx, csiClient, stagingPath)
 		if err != nil {
 			return err
 		}

pkg/volume/csi/csi_client.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 	"time"
 
 	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
@@ -720,3 +721,36 @@ func versionRequiresV0Client(version *utilversion.Version) bool {
 
 	return false
 }
+
+// CSI client getter with cache.
+// This provides a method to initialize CSI client with driver name and caches
+// it for later use. When CSI clients have not been discovered yet (e.g.
+// on kubelet restart), client initialization will fail. Users of CSI client (e.g.
+// mounter manager and block mapper) can use this to delay CSI client
+// initialization until needed.
+type csiClientGetter struct {
+	sync.RWMutex
+	csiClient  csiClient
+	driverName csiDriverName
+}
+
+func (c *csiClientGetter) Get() (csiClient, error) {
+	c.RLock()
+	if c.csiClient != nil {
+		c.RUnlock()
+		return c.csiClient, nil
+	}
+	c.RUnlock()
+	c.Lock()
+	defer c.Unlock()
+	// Double-checking locking criterion.
+	if c.csiClient != nil {
+		return c.csiClient, nil
+	}
+	csi, err := newCsiDriverClient(c.driverName)
+	if err != nil {
+		return nil, err
+	}
+	c.csiClient = csi
+	return c.csiClient, nil
+}

pkg/volume/csi/csi_mounter.go

@@ -55,7 +55,7 @@ var (
 )
 
 type csiMountMgr struct {
-	csiClient      csiClient
+	csiClientGetter
 	k8s            kubernetes.Interface
 	plugin         *csiPlugin
 	driverName     csiDriverName
@@ -115,7 +115,11 @@ func (c *csiMountMgr) SetUpAt(dir string, fsGroup *int64) error {
 		return err
 	}
 
-	csi := c.csiClient
+	csi, err := c.csiClientGetter.Get()
+	if err != nil {
+		klog.Error(log("mounter.SetUpAt failed to get CSI client: %v", err))
+		return err
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 
@@ -304,7 +308,11 @@ func (c *csiMountMgr) TearDownAt(dir string) error {
 	}
 
 	volID := c.volumeID
-	csi := c.csiClient
+	csi, err := c.csiClientGetter.Get()
+	if err != nil {
+		klog.Error(log("mounter.SetUpAt failed to get CSI client: %v", err))
+		return err
+	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()

pkg/volume/csi/csi_plugin.go

@@ -302,11 +302,6 @@ func (p *csiPlugin) NewMounter(
 		return nil, errors.New("failed to get a Kubernetes client")
 	}
 
-	csi, err := newCsiDriverClient(csiDriverName(pvSource.Driver))
-	if err != nil {
-		return nil, err
-	}
-
 	mounter := &csiMountMgr{
 		plugin:       p,
 		k8s:          k8s,
@@ -316,9 +311,9 @@ func (p *csiPlugin) NewMounter(
 		driverName:   csiDriverName(pvSource.Driver),
 		volumeID:     pvSource.VolumeHandle,
 		specVolumeID: spec.Name(),
-		csiClient:    csi,
 		readOnly:     readOnly,
 	}
+	mounter.csiClientGetter.driverName = csiDriverName(pvSource.Driver)
 
 	// Save volume info in pod dir
 	dir := mounter.GetPath()
@@ -374,10 +369,7 @@ func (p *csiPlugin) NewUnmounter(specName string, podUID types.UID) (volume.Unmo
 	}
 	unmounter.driverName = csiDriverName(data[volDataKey.driverName])
 	unmounter.volumeID = data[volDataKey.volHandle]
-	unmounter.csiClient, err = newCsiDriverClient(unmounter.driverName)
-	if err != nil {
-		return nil, err
-	}
+	unmounter.csiClientGetter.driverName = unmounter.driverName
 
 	return unmounter, nil
 }
@@ -489,10 +481,6 @@ func (p *csiPlugin) NewBlockVolumeMapper(spec *volume.Spec, podRef *api.Pod, opt
 	}
 
 	klog.V(4).Info(log("setting up block mapper for [volume=%v,driver=%v]", pvSource.VolumeHandle, pvSource.Driver))
-	client, err := newCsiDriverClient(csiDriverName(pvSource.Driver))
-	if err != nil {
-		return nil, err
-	}
 
 	k8s := p.host.GetKubeClient()
 	if k8s == nil {
@@ -501,7 +489,6 @@ func (p *csiPlugin) NewBlockVolumeMapper(spec *volume.Spec, podRef *api.Pod, opt
 	}
 
 	mapper := &csiBlockMapper{
-		csiClient:  client,
 		k8s:        k8s,
 		plugin:     p,
 		volumeID:   pvSource.VolumeHandle,
@@ -511,6 +498,7 @@ func (p *csiPlugin) NewBlockVolumeMapper(spec *volume.Spec, podRef *api.Pod, opt
 		specName:   spec.Name(),
 		podUID:     podRef.UID,
 	}
+	mapper.csiClientGetter.driverName = csiDriverName(pvSource.Driver)
 
 	// Save volume info in pod dir
 	dataDir := getVolumeDeviceDataDir(spec.Name(), p.host)
@@ -565,7 +553,7 @@ func (p *csiPlugin) NewBlockVolumeUnmapper(volName string, podUID types.UID) (vo
 	}
 	unmapper.driverName = csiDriverName(data[volDataKey.driverName])
 	unmapper.volumeID = data[volDataKey.volHandle]
-	unmapper.csiClient, err = newCsiDriverClient(unmapper.driverName)
+	unmapper.csiClientGetter.driverName = unmapper.driverName
 	if err != nil {
 		return nil, err
 	}

pkg/volume/csi/csi_plugin_test.go

@@ -268,7 +268,8 @@ func TestPluginNewMounter(t *testing.T) {
 	if csiMounter.podUID == types.UID("") {
 		t.Error("mounter podUID not set")
 	}
-	if csiMounter.csiClient == nil {
+	csiClient, err := csiMounter.csiClientGetter.Get()
+	if csiClient == nil {
 		t.Error("mounter csiClient is nil")
 	}
 
@@ -327,8 +328,9 @@ func TestPluginNewUnmounter(t *testing.T) {
 		t.Error("podUID not set")
 	}
 
-	if csiUnmounter.csiClient == nil {
-		t.Error("unmounter csiClient is nil")
+	csiClient, err := csiUnmounter.csiClientGetter.Get()
+	if csiClient == nil {
+		t.Error("mounter csiClient is nil")
 	}
 }
 
@@ -405,7 +407,8 @@ func TestPluginNewBlockMapper(t *testing.T) {
 	if csiMapper.podUID == types.UID("") {
 		t.Error("CSI block mapper missing pod.UID")
 	}
-	if csiMapper.csiClient == nil {
+	csiClient, err := csiMapper.csiClientGetter.Get()
+	if csiClient == nil {
 		t.Error("mapper csiClient is nil")
 	}
 
@@ -467,7 +470,8 @@ func TestPluginNewUnmapper(t *testing.T) {
 		t.Error("specName not set")
 	}
 
-	if csiUnmapper.csiClient == nil {
+	csiClient, err := csiUnmapper.csiClientGetter.Get()
+	if csiClient == nil {
 		t.Error("unmapper csiClient is nil")
 	}