Merge pull request #4922 from hashicorp/f-drivermananger

add generic plugin manager interface and orchestration

client/pluginmanager/group.go

@@ -0,0 +1,60 @@
+package pluginmanager
+
+import (
+	"fmt"
+	"sync"
+
+	log "github.com/hashicorp/go-hclog"
+)
+
+// PluginGroup is a utility struct to manage a collectively orchestrate a
+// set of PluginManagers
+type PluginGroup struct {
+	// managers is the set of managers managed, access is synced by mLock
+	managers []PluginManager
+
+	// shutdown indicates if shutdown has been called, access is synced by mLock
+	shutdown bool
+
+	// mLock gaurds manangers and shutdown
+	mLock sync.Mutex
+
+	logger log.Logger
+}
+
+// New returns an initialized PluginGroup
+func New(logger log.Logger) *PluginGroup {
+	return &PluginGroup{
+		managers: []PluginManager{},
+		logger:   logger.Named("plugin"),
+	}
+}
+
+// RegisterAndRun registers the manager and starts it in a separate goroutine
+func (m *PluginGroup) RegisterAndRun(manager PluginManager) error {
+	m.mLock.Lock()
+	if m.shutdown {
+		return fmt.Errorf("plugin group already shutdown")
+	}
+	m.managers = append(m.managers, manager)
+	m.mLock.Unlock()
+
+	go func() {
+		m.logger.Info("starting plugin manager", "plugin-type", manager.PluginType())
+		manager.Run()
+		m.logger.Info("plugin manager finished", "plugin-type", manager.PluginType())
+	}()
+	return nil
+}
+
+// Shutdown shutsdown all registered PluginManagers in reverse order of how
+// they were started.
+func (m *PluginGroup) Shutdown() {
+	m.mLock.Lock()
+	defer m.mLock.Unlock()
+	for i := len(m.managers) - 1; i >= 0; i-- {
+		m.logger.Info("shutting down plugin manager", "plugin-type", m.managers[i].PluginType())
+		m.managers[i].Shutdown()
+	}
+	m.shutdown = true
+}

client/pluginmanager/group_test.go

@@ -0,0 +1,64 @@
+package pluginmanager
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/hashicorp/nomad/helper/testlog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPluginGroup_RegisterAndRun(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	var hasRun bool
+	var wg sync.WaitGroup
+	wg.Add(1)
+	mananger := &MockPluginManager{RunF: func() {
+		hasRun = true
+		wg.Done()
+	}}
+
+	group := New(testlog.HCLogger(t))
+	require.NoError(group.RegisterAndRun(mananger))
+	wg.Wait()
+	require.True(hasRun)
+}
+
+func TestPluginGroup_Shutdown(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	var stack []int
+	var stackMu sync.Mutex
+	var runWg sync.WaitGroup
+	var shutdownWg sync.WaitGroup
+	group := New(testlog.HCLogger(t))
+	for i := 1; i < 4; i++ {
+		i := i
+		runWg.Add(1)
+		shutdownWg.Add(1)
+		mananger := &MockPluginManager{RunF: func() {
+			stackMu.Lock()
+			defer stackMu.Unlock()
+			defer runWg.Done()
+			stack = append(stack, i)
+		}, ShutdownF: func() {
+			stackMu.Lock()
+			defer stackMu.Unlock()
+			defer shutdownWg.Done()
+			idx := len(stack) - 1
+			val := stack[idx]
+			require.Equal(i, val)
+			stack = stack[:idx]
+		}}
+		require.NoError(group.RegisterAndRun(mananger))
+		runWg.Wait()
+	}
+	group.Shutdown()
+	shutdownWg.Wait()
+	require.Empty(stack)
+
+	require.Error(group.RegisterAndRun(&MockPluginManager{}))
+}

client/pluginmanager/manager.go

@@ -0,0 +1,14 @@
+package pluginmanager
+
+// PluginManager orchestrates the lifecycle of a set of plugins
+type PluginManager interface {
+	// Run starts a plugin manager and must block until shutdown
+	Run()
+
+	// Shutdown should gracefully shutdown all plugins managed by the manager.
+	// It must block until shutdown is complete
+	Shutdown()
+
+	// PluginType is the type of plugin which the manager manages
+	PluginType() string
+}

client/pluginmanager/testing.go

@@ -0,0 +1,10 @@
+package pluginmanager
+
+type MockPluginManager struct {
+	RunF      func()
+	ShutdownF func()
+}
+
+func (m *MockPluginManager) Run()               { m.RunF() }
+func (m *MockPluginManager) Shutdown()          { m.ShutdownF() }
+func (m *MockPluginManager) PluginType() string { return "mock" }