csimanager: Cleanup volumemanager setup
This commit is contained in:
parent
ee85c468c0
commit
94e87fbe9c
|
@ -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
|
||||
|
|
|
@ -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()
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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) {
|
||||
var vm VolumeMounter
|
||||
i.volumeManagerMu.RLock()
|
||||
if i.volumeManagerSetup {
|
||||
vm = i.volumeManager
|
||||
}
|
||||
i.volumeManagerMu.RUnlock()
|
||||
|
||||
if vm != nil {
|
||||
return vm, nil
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue