Plugin activation (#356)

Signed-off-by: David Chung <[email protected]>

Author: David Chung

cmd/cli/plugin.go

@@ -2,7 +2,12 @@ package main
 
 import (
 	"fmt"
+	"sync"
+
 	"github.com/docker/infrakit/pkg/discovery"
+	"github.com/docker/infrakit/pkg/launch"
+	"github.com/docker/infrakit/pkg/launch/os"
+	"github.com/docker/infrakit/pkg/template"
 	"github.com/spf13/cobra"
 )
 
@@ -13,7 +18,7 @@ func pluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 		Short: "Manage plugins",
 	}
 
-	ls := cobra.Command{
+	ls := &cobra.Command{
 		Use:   "ls",
 		Short: "List available plugins",
 	}
@@ -34,7 +39,94 @@ func pluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 		return nil
 	}
 
-	cmd.AddCommand(&ls)
+	start := &cobra.Command{
+		Use:   "start",
+		Short: "Start named plugins. Args are a list of plugin names",
+	}
+
+	configURL := start.Flags().String("config-url", "", "URL for the startup configs")
+	osExec := start.Flags().Bool("os", false, "True to use os plugin binaries")
+	doWait := start.Flags().BoolP("wait", "w", false, "True to wait in the foreground; Ctrl-C to exit")
+
+	start.RunE = func(c *cobra.Command, args []string) error {
+
+		configTemplate, err := template.NewTemplate(*configURL, template.Options{
+			SocketDir: discovery.Dir(),
+		})
+		if err != nil {
+			return err
+		}
+
+		view, err := configTemplate.Render(nil)
+		if err != nil {
+			return err
+		}
+
+		configs := launch.Config([]byte(view))
+
+		parsedRules := []launch.Rule{}
+		err = configs.Unmarshal(&parsedRules)
+		if err != nil {
+			return err
+		}
+
+		monitors := []*launch.Monitor{}
+
+		if *osExec {
+			exec, err := os.NewLauncher()
+			if err != nil {
+				return err
+			}
+			monitors = append(monitors, launch.NewMonitor(exec, parsedRules))
+		}
+
+		input := []chan<- launch.StartPlugin{}
+		for _, m := range monitors {
+			ch, err := m.Start()
+			if err != nil {
+				return err
+			}
+			input = append(input, ch)
+		}
+
+		var wait sync.WaitGroup
+
+		if *doWait {
+			wait.Add(1)
+		}
+
+		// now start all the plugins
+		for _, plugin := range args {
+			fmt.Println("Starting up", plugin)
+
+			wait.Add(1)
+
+			for _, ch := range input {
+
+				name := plugin
+				ch <- launch.StartPlugin{
+					Plugin: name,
+					Started: func(config *launch.Config) {
+						fmt.Println(name, "started.")
+						wait.Done()
+					},
+					Error: func(config *launch.Config, err error) {
+						fmt.Println("Error starting", name, "err=", err)
+						wait.Done()
+					},
+				}
+			}
+		}
+
+		wait.Wait() // wait for everyone to complete
+
+		for _, monitor := range monitors {
+			monitor.Stop()
+		}
+		return nil
+	}
+
+	cmd.AddCommand(ls, start)
 
 	return cmd
 }

pkg/launch/config.go

@@ -0,0 +1,45 @@
+package launch
+
+import (
+	"encoding/json"
+)
+
+// Config is the raw configuration on how to launch the plugin.
+type Config json.RawMessage
+
+// Unmarshal decodes the raw config container in this object to the typed struct.
+func (c *Config) Unmarshal(typed interface{}) error {
+	if c == nil || len([]byte(*c)) == 0 {
+		return nil // no effect on typed
+	}
+	return json.Unmarshal([]byte(*c), typed)
+}
+
+// Marshal populates this raw message with a decoded form of the input struct.
+func (c *Config) Marshal(typed interface{}) error {
+	buff, err := json.MarshalIndent(typed, "", "  ")
+	if err != nil {
+		return err
+	}
+	*c = Config(json.RawMessage(buff))
+	return nil
+}
+
+// String returns the string representation.
+func (c *Config) String() string {
+	return string([]byte(*c))
+}
+
+// MarshalJSON implements the json Marshaler interface
+func (c *Config) MarshalJSON() ([]byte, error) {
+	if c == nil {
+		return nil, nil
+	}
+	return []byte(*c), nil
+}
+
+// UnmarshalJSON implements the json Unmarshaler interface
+func (c *Config) UnmarshalJSON(data []byte) error {
+	*c = Config(json.RawMessage(data))
+	return nil
+}

pkg/launch/exec.go

@@ -0,0 +1,18 @@
+package launch
+
+// Exec is a service that is able to start plugins based on different
+// mechanisms from running local binary to pulling and running docker containers or engine plugins
+type Exec interface {
+
+	// Name returns the name of the launcher.  This is used to identify
+	// which launcher to use in configurations or command line flags
+	Name() string
+
+	// Exec starts the plugin given the name of the plugin and
+	// the command and args to start it.
+	// This can be an async process but the launcher will poll for the running
+	// status of the plugin.
+	// The client can receive and block on the returned channel
+	// and add optional timeout in its own select statement.
+	Exec(name string, config *Config) (<-chan error, error)
+}

pkg/launch/monitor.go

@@ -0,0 +1,134 @@
+package launch
+
+import (
+	"errors"
+	"sync"
+
+	log "github.com/Sirupsen/logrus"
+)
+
+var errNoConfig = errors.New("no-counfig")
+
+// Rule provides the instructions on starting the plugin
+type Rule struct {
+
+	// Plugin is the name of the plugin
+	Plugin string
+
+	// Exec is the name of the exec to use to start the plugin
+	Exec string
+
+	// Launch is encoded form of the rule on how to start/exec the process
+	Launch *Config
+}
+
+// Monitor runs continuously receiving requests to start a plugin.
+// Monitor uses a launcher to actually start the process of the plugin.
+type Monitor struct {
+	exec      Exec
+	rules     map[string]Rule
+	startChan <-chan StartPlugin
+	inputChan chan<- StartPlugin
+	stop      chan interface{}
+	lock      sync.Mutex
+}
+
+// NewMonitor returns a monitor that continuously watches for input
+// requests and launches the process for the plugin, if not already running.
+func NewMonitor(l Exec, rules []Rule) *Monitor {
+	m := map[string]Rule{}
+	// index by name of plugin
+	for _, r := range rules {
+		if r.Exec == l.Name() {
+			m[r.Plugin] = r
+		}
+	}
+	return &Monitor{
+		exec:  l,
+		rules: m,
+	}
+}
+
+// StartPlugin is the command to start a plugin
+type StartPlugin struct {
+	Plugin  string
+	Started func(*Config)
+	Error   func(*Config, error)
+}
+
+func (s StartPlugin) reportError(config *Config, e error) {
+	if s.Error != nil {
+		go s.Error(config, e)
+	}
+}
+
+func (s StartPlugin) reportSuccess(config *Config) {
+	if s.Started != nil {
+		go s.Started(config)
+	}
+}
+
+// Start starts the monitor and returns a channel for sending
+// requests to launch plugins.  Closing the channel effectively stops
+// the monitor loop.
+func (m *Monitor) Start() (chan<- StartPlugin, error) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if m.startChan != nil {
+		return m.inputChan, nil
+	}
+
+	ch := make(chan StartPlugin)
+	m.startChan = ch
+	m.inputChan = ch
+
+	go func() {
+	loop:
+		for {
+			req, open := <-m.startChan
+			if !open {
+				log.Infoln("Plugin activation input closed. Stopping.")
+				m.inputChan = nil
+				return
+			}
+
+			r, has := m.rules[req.Plugin]
+			if !has {
+				log.Warningln("no plugin:", req)
+				req.reportError(r.Launch, errNoConfig)
+				continue loop
+			}
+
+			configCopy := &Config{}
+			if r.Launch != nil {
+				*configCopy = *r.Launch
+			}
+
+			block, err := m.exec.Exec(r.Plugin, configCopy)
+			if err != nil {
+				req.reportError(configCopy, err)
+				continue loop
+			}
+
+			log.Infoln("Waiting for", r.Plugin, "to start:", configCopy.String())
+			err = <-block
+			if err != nil {
+				req.reportError(configCopy, err)
+				continue loop
+			}
+
+			req.reportSuccess(configCopy)
+		}
+	}()
+
+	return m.inputChan, nil
+}
+
+// Stop stops the monitor
+func (m *Monitor) Stop() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if m.inputChan != nil {
+		close(m.inputChan)
+	}
+}