csimanager: Cleanup volumemanager setup

client/pluginmanager/csimanager/fingerprint.go

@@ -27,6 +27,15 @@ type pluginFingerprinter struct {
 	fingerprintController bool
 
 	hadFirstSuccessfulFingerprint bool
+	// hadFirstSuccessfulFingerprintCh is closed the first time a fingerprint
+	// is completed successfully.
+	hadFirstSuccessfulFingerprintCh chan struct{}
+
+	// requiresStaging is set on a first successful fingerprint. It allows the
+	// csimanager to efficiently query this as it shouldn't change after a plugin
+	// is started. Removing this bool will require storing a cache of recent successful
+	// results that can be used by subscribers of the `hadFirstSuccessfulFingerprintCh`.
+	requiresStaging bool
 }
 
 func (p *pluginFingerprinter) fingerprint(ctx context.Context) *structs.CSIInfo {
@@ -61,7 +70,13 @@ func (p *pluginFingerprinter) fingerprint(ctx context.Context) *structs.CSIInfo
 		info.HealthDescription = fmt.Sprintf("failed fingerprinting with error: %v", err)
 	} else {
 		info = fp
-		p.hadFirstSuccessfulFingerprint = true
+		if !p.hadFirstSuccessfulFingerprint {
+			p.hadFirstSuccessfulFingerprint = true
+			if p.fingerprintNode {
+				p.requiresStaging = info.NodeInfo.RequiresNodeStageVolume
+			}
+			close(p.hadFirstSuccessfulFingerprintCh)
+		}
 	}
 
 	return info

client/pluginmanager/csimanager/instance.go

@@ -2,7 +2,6 @@ package csimanager
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/hashicorp/go-hclog"
@@ -31,9 +30,7 @@ type instanceManager struct {
 	fp *pluginFingerprinter
 
 	volumeManager        *volumeManager
-	volumeManagerMu      sync.RWMutex
 	volumeManagerSetupCh chan struct{}
-	volumeManagerSetup   bool
 
 	client csi.CSIPlugin
 }
@@ -47,10 +44,11 @@ func newInstanceManager(logger hclog.Logger, updater UpdateNodeCSIInfoFunc, p *d
 		updater: updater,
 
 		fp: &pluginFingerprinter{
-			logger:                logger.Named("fingerprinter"),
-			info:                  p,
-			fingerprintNode:       p.Type == dynamicplugins.PluginTypeCSINode,
-			fingerprintController: p.Type == dynamicplugins.PluginTypeCSIController,
+			logger:                          logger.Named("fingerprinter"),
+			info:                            p,
+			fingerprintNode:                 p.Type == dynamicplugins.PluginTypeCSINode,
+			fingerprintController:           p.Type == dynamicplugins.PluginTypeCSIController,
+			hadFirstSuccessfulFingerprintCh: make(chan struct{}),
 		},
 
 		mountPoint: p.Options["MountPoint"],
@@ -73,30 +71,34 @@ func (i *instanceManager) run() {
 	i.client = c
 	i.fp.client = c
 
+	go i.setupVolumeManager()
 	go i.runLoop()
 }
 
-// VolumeMounter returns the volume manager that is configured for the given plugin
-// instance. If called before the volume manager has been setup, it will block until
-// the volume manager is ready or the context is closed.
-func (i *instanceManager) VolumeMounter(ctx context.Context) (VolumeMounter, error) {
-	var vm VolumeMounter
-	i.volumeManagerMu.RLock()
-	if i.volumeManagerSetup {
-		vm = i.volumeManager
+func (i *instanceManager) setupVolumeManager() {
+	if i.info.Type != dynamicplugins.PluginTypeCSINode {
+		i.logger.Debug("Skipping volume manager setup - not managing a Node plugin", "type", i.info.Type)
+		return
 	}
-	i.volumeManagerMu.RUnlock()
 
-	if vm != nil {
-		return vm, nil
+	select {
+	case <-i.shutdownCtx.Done():
+		return
+	case <-i.fp.hadFirstSuccessfulFingerprintCh:
+		i.volumeManager = newVolumeManager(i.logger, i.client, i.mountPoint, i.fp.requiresStaging)
+		i.logger.Debug("Setup volume manager")
+		close(i.volumeManagerSetupCh)
+		return
 	}
+}
 
+// VolumeMounter returns the volume manager that is configured for the given plugin
+// instance. If called before the volume manager has been setup, it will block until
+// the volume manager is ready or the context is closed.
+func (i *instanceManager) VolumeMounter(ctx context.Context) (VolumeMounter, error) {
 	select {
 	case <-i.volumeManagerSetupCh:
-		i.volumeManagerMu.RLock()
-		vm = i.volumeManager
-		i.volumeManagerMu.RUnlock()
-		return vm, nil
+		return i.volumeManager, nil
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	}
@@ -124,21 +126,6 @@ func (i *instanceManager) runLoop() {
 			cancelFn()
 			i.updater(i.info.Name, info)
 
-			// TODO: refactor this lock into a faster, goroutine-local check
-			i.volumeManagerMu.RLock()
-			// When we've had a successful fingerprint, and the volume manager is not yet setup,
-			// and one is required (we're running a node plugin), then set one up now.
-			if i.fp.hadFirstSuccessfulFingerprint && !i.volumeManagerSetup && i.fp.fingerprintNode {
-				i.volumeManagerMu.RUnlock()
-				i.volumeManagerMu.Lock()
-				i.volumeManager = newVolumeManager(i.logger, i.client, i.mountPoint, info.NodeInfo.RequiresNodeStageVolume)
-				i.volumeManagerSetup = true
-				close(i.volumeManagerSetupCh)
-				i.volumeManagerMu.Unlock()
-			} else {
-				i.volumeManagerMu.RUnlock()
-			}
-
 			timer.Reset(managerFingerprintInterval)
 		}
 	}