2019-10-22 13:20:26 +00:00
|
|
|
package csimanager
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2020-04-02 20:04:56 +00:00
|
|
|
"fmt"
|
2019-10-22 13:20:26 +00:00
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/hashicorp/go-hclog"
|
|
|
|
"github.com/hashicorp/nomad/client/dynamicplugins"
|
|
|
|
"github.com/hashicorp/nomad/client/pluginmanager"
|
|
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
|
|
)
|
|
|
|
|
|
|
|
// defaultPluginResyncPeriod is the time interval used to do a full resync
|
|
|
|
// against the dynamicplugins, to account for missed updates.
|
|
|
|
const defaultPluginResyncPeriod = 30 * time.Second
|
|
|
|
|
|
|
|
// UpdateNodeCSIInfoFunc is the callback used to update the node from
|
|
|
|
// fingerprinting
|
|
|
|
type UpdateNodeCSIInfoFunc func(string, *structs.CSIInfo)
|
2020-03-31 21:13:52 +00:00
|
|
|
type TriggerNodeEvent func(*structs.NodeEvent)
|
2019-10-22 13:20:26 +00:00
|
|
|
|
|
|
|
type Config struct {
|
|
|
|
Logger hclog.Logger
|
|
|
|
DynamicRegistry dynamicplugins.Registry
|
|
|
|
UpdateNodeCSIInfoFunc UpdateNodeCSIInfoFunc
|
|
|
|
PluginResyncPeriod time.Duration
|
2020-03-31 21:13:52 +00:00
|
|
|
TriggerNodeEvent TriggerNodeEvent
|
2019-10-22 13:20:26 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// New returns a new PluginManager that will handle managing CSI plugins from
|
|
|
|
// the dynamicRegistry from the provided Config.
|
2020-01-08 12:47:07 +00:00
|
|
|
func New(config *Config) Manager {
|
2019-10-22 13:20:26 +00:00
|
|
|
// Use a dedicated internal context for managing plugin shutdown.
|
|
|
|
ctx, cancelFn := context.WithCancel(context.Background())
|
|
|
|
if config.PluginResyncPeriod == 0 {
|
|
|
|
config.PluginResyncPeriod = defaultPluginResyncPeriod
|
|
|
|
}
|
|
|
|
|
|
|
|
return &csiManager{
|
|
|
|
logger: config.Logger,
|
2020-03-31 21:13:52 +00:00
|
|
|
eventer: config.TriggerNodeEvent,
|
2019-10-22 13:20:26 +00:00
|
|
|
registry: config.DynamicRegistry,
|
|
|
|
instances: make(map[string]map[string]*instanceManager),
|
|
|
|
|
|
|
|
updateNodeCSIInfoFunc: config.UpdateNodeCSIInfoFunc,
|
|
|
|
pluginResyncPeriod: config.PluginResyncPeriod,
|
|
|
|
|
|
|
|
shutdownCtx: ctx,
|
|
|
|
shutdownCtxCancelFn: cancelFn,
|
|
|
|
shutdownCh: make(chan struct{}),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
type csiManager struct {
|
|
|
|
// instances should only be accessed from the run() goroutine and the shutdown
|
2022-02-23 20:23:07 +00:00
|
|
|
// fn. It is a map of PluginType : [PluginName : *instanceManager]
|
2019-10-22 13:20:26 +00:00
|
|
|
instances map[string]map[string]*instanceManager
|
|
|
|
|
|
|
|
registry dynamicplugins.Registry
|
|
|
|
logger hclog.Logger
|
2020-03-31 21:13:52 +00:00
|
|
|
eventer TriggerNodeEvent
|
2019-10-22 13:20:26 +00:00
|
|
|
pluginResyncPeriod time.Duration
|
|
|
|
|
|
|
|
updateNodeCSIInfoFunc UpdateNodeCSIInfoFunc
|
|
|
|
|
|
|
|
shutdownCtx context.Context
|
|
|
|
shutdownCtxCancelFn context.CancelFunc
|
|
|
|
shutdownCh chan struct{}
|
|
|
|
}
|
|
|
|
|
2020-01-08 12:47:07 +00:00
|
|
|
func (c *csiManager) PluginManager() pluginmanager.PluginManager {
|
|
|
|
return c
|
|
|
|
}
|
|
|
|
|
2020-04-02 20:04:56 +00:00
|
|
|
func (c *csiManager) MounterForPlugin(ctx context.Context, pluginID string) (VolumeMounter, error) {
|
2020-01-08 12:47:07 +00:00
|
|
|
nodePlugins, hasAnyNodePlugins := c.instances["csi-node"]
|
|
|
|
if !hasAnyNodePlugins {
|
2020-04-02 20:04:56 +00:00
|
|
|
return nil, fmt.Errorf("no storage node plugins found")
|
2020-01-08 12:47:07 +00:00
|
|
|
}
|
|
|
|
|
2020-04-02 20:04:56 +00:00
|
|
|
mgr, hasPlugin := nodePlugins[pluginID]
|
2020-01-28 16:27:32 +00:00
|
|
|
if !hasPlugin {
|
2020-04-02 20:04:56 +00:00
|
|
|
return nil, fmt.Errorf("plugin %s for type csi-node not found", pluginID)
|
2020-01-08 12:47:07 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return mgr.VolumeMounter(ctx)
|
|
|
|
}
|
|
|
|
|
2019-10-22 13:20:26 +00:00
|
|
|
// Run starts a plugin manager and should return early
|
|
|
|
func (c *csiManager) Run() {
|
|
|
|
go c.runLoop()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *csiManager) runLoop() {
|
2020-03-19 15:27:48 +00:00
|
|
|
timer := time.NewTimer(0) // ensure we sync immediately in first pass
|
|
|
|
controllerUpdates := c.registry.PluginsUpdatedCh(c.shutdownCtx, "csi-controller")
|
|
|
|
nodeUpdates := c.registry.PluginsUpdatedCh(c.shutdownCtx, "csi-node")
|
2019-10-22 13:20:26 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-timer.C:
|
|
|
|
c.resyncPluginsFromRegistry("csi-controller")
|
|
|
|
c.resyncPluginsFromRegistry("csi-node")
|
|
|
|
timer.Reset(c.pluginResyncPeriod)
|
2020-03-19 15:27:48 +00:00
|
|
|
case event := <-controllerUpdates:
|
|
|
|
c.handlePluginEvent(event)
|
|
|
|
case event := <-nodeUpdates:
|
|
|
|
c.handlePluginEvent(event)
|
|
|
|
case <-c.shutdownCtx.Done():
|
|
|
|
close(c.shutdownCh)
|
|
|
|
return
|
2019-10-22 13:20:26 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-03-19 15:27:48 +00:00
|
|
|
// resyncPluginsFromRegistry does a full sync of the running instance
|
|
|
|
// managers against those in the registry. we primarily will use update
|
|
|
|
// events from the registry.
|
2019-10-22 13:20:26 +00:00
|
|
|
func (c *csiManager) resyncPluginsFromRegistry(ptype string) {
|
|
|
|
plugins := c.registry.ListPlugins(ptype)
|
|
|
|
seen := make(map[string]struct{}, len(plugins))
|
|
|
|
|
|
|
|
// For every plugin in the registry, ensure that we have an existing plugin
|
|
|
|
// running. Also build the map of valid plugin names.
|
2020-03-19 15:27:48 +00:00
|
|
|
// Note: monolith plugins that run as both controllers and nodes get a
|
|
|
|
// separate instance manager for both modes.
|
2019-10-22 13:20:26 +00:00
|
|
|
for _, plugin := range plugins {
|
|
|
|
seen[plugin.Name] = struct{}{}
|
2020-03-19 15:27:48 +00:00
|
|
|
c.ensureInstance(plugin)
|
2019-10-22 13:20:26 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// For every instance manager, if we did not find it during the plugin
|
|
|
|
// iterator, shut it down and remove it from the table.
|
2020-03-19 15:27:48 +00:00
|
|
|
instances := c.instancesForType(ptype)
|
|
|
|
for name, mgr := range instances {
|
2019-10-22 13:20:26 +00:00
|
|
|
if _, ok := seen[name]; !ok {
|
2020-03-19 15:27:48 +00:00
|
|
|
c.ensureNoInstance(mgr.info)
|
2019-10-22 13:20:26 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-03-19 15:27:48 +00:00
|
|
|
// handlePluginEvent syncs a single event against the plugin registry
|
|
|
|
func (c *csiManager) handlePluginEvent(event *dynamicplugins.PluginUpdateEvent) {
|
|
|
|
if event == nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
c.logger.Trace("dynamic plugin event",
|
|
|
|
"event", event.EventType,
|
|
|
|
"plugin_id", event.Info.Name,
|
|
|
|
"plugin_alloc_id", event.Info.AllocID)
|
|
|
|
|
|
|
|
switch event.EventType {
|
|
|
|
case dynamicplugins.EventTypeRegistered:
|
|
|
|
c.ensureInstance(event.Info)
|
|
|
|
case dynamicplugins.EventTypeDeregistered:
|
|
|
|
c.ensureNoInstance(event.Info)
|
|
|
|
default:
|
|
|
|
c.logger.Error("received unknown dynamic plugin event type",
|
|
|
|
"type", event.EventType)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Ensure we have an instance manager for the plugin and add it to
|
|
|
|
// the CSI manager's tracking table for that plugin type.
|
|
|
|
func (c *csiManager) ensureInstance(plugin *dynamicplugins.PluginInfo) {
|
|
|
|
name := plugin.Name
|
|
|
|
ptype := plugin.Type
|
|
|
|
instances := c.instancesForType(ptype)
|
2022-02-23 20:23:07 +00:00
|
|
|
mgr, ok := instances[name]
|
|
|
|
if !ok {
|
|
|
|
c.logger.Debug("detected new CSI plugin", "name", name, "type", ptype, "alloc", plugin.AllocID)
|
2020-03-31 21:13:52 +00:00
|
|
|
mgr := newInstanceManager(c.logger, c.eventer, c.updateNodeCSIInfoFunc, plugin)
|
2020-03-19 15:27:48 +00:00
|
|
|
instances[name] = mgr
|
|
|
|
mgr.run()
|
2022-02-23 20:23:07 +00:00
|
|
|
} else if mgr.allocID != plugin.AllocID {
|
|
|
|
mgr.shutdown()
|
|
|
|
c.logger.Debug("detected update for CSI plugin", "name", name, "type", ptype, "alloc", plugin.AllocID)
|
|
|
|
mgr := newInstanceManager(c.logger, c.eventer, c.updateNodeCSIInfoFunc, plugin)
|
|
|
|
instances[name] = mgr
|
|
|
|
mgr.run()
|
|
|
|
|
2020-03-19 15:27:48 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Shut down the instance manager for a plugin and remove it from
|
|
|
|
// the CSI manager's tracking table for that plugin type.
|
|
|
|
func (c *csiManager) ensureNoInstance(plugin *dynamicplugins.PluginInfo) {
|
|
|
|
name := plugin.Name
|
|
|
|
ptype := plugin.Type
|
|
|
|
instances := c.instancesForType(ptype)
|
|
|
|
if mgr, ok := instances[name]; ok {
|
2022-02-23 20:23:07 +00:00
|
|
|
if mgr.allocID == plugin.AllocID {
|
|
|
|
c.logger.Debug("shutting down CSI plugin", "name", name, "type", ptype, "alloc", plugin.AllocID)
|
|
|
|
mgr.shutdown()
|
|
|
|
delete(instances, name)
|
|
|
|
}
|
2020-03-19 15:27:48 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Get the instance managers table for a specific plugin type,
|
|
|
|
// ensuring it's been initialized if it doesn't exist.
|
|
|
|
func (c *csiManager) instancesForType(ptype string) map[string]*instanceManager {
|
|
|
|
pluginMap, ok := c.instances[ptype]
|
|
|
|
if !ok {
|
|
|
|
pluginMap = make(map[string]*instanceManager)
|
|
|
|
c.instances[ptype] = pluginMap
|
|
|
|
}
|
|
|
|
return pluginMap
|
|
|
|
}
|
|
|
|
|
2019-10-22 13:20:26 +00:00
|
|
|
// Shutdown should gracefully shutdown all plugins managed by the manager.
|
|
|
|
// It must block until shutdown is complete
|
|
|
|
func (c *csiManager) Shutdown() {
|
|
|
|
// Shut down the run loop
|
|
|
|
c.shutdownCtxCancelFn()
|
|
|
|
|
2020-03-19 15:27:48 +00:00
|
|
|
// Wait for plugin manager shutdown to complete so that we
|
|
|
|
// don't try to shutdown instance managers while runLoop is
|
|
|
|
// doing a resync
|
2019-10-22 13:20:26 +00:00
|
|
|
<-c.shutdownCh
|
|
|
|
|
|
|
|
// Shutdown all the instance managers in parallel
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
for _, pluginMap := range c.instances {
|
|
|
|
for _, mgr := range pluginMap {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(mgr *instanceManager) {
|
|
|
|
mgr.shutdown()
|
|
|
|
wg.Done()
|
|
|
|
}(mgr)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
}
|
|
|
|
|
|
|
|
// PluginType is the type of plugin which the manager manages
|
|
|
|
func (c *csiManager) PluginType() string {
|
|
|
|
return "csi"
|
|
|
|
}
|