Auto start plugins via up + various bug fixes (#732)

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

Author: David Chung

cmd/infrakit/up/up.go

@@ -39,6 +39,8 @@ func Command(plugins func() discovery.Plugins) *cobra.Command {
 
 	waitDuration := up.Flags().String("wait", "1s", "Wait for plugins to be ready")
 	configURL := up.Flags().String("config-url", "", "URL for the startup configs")
+	stack := up.Flags().String("stack", "mystack", "Name of the stack")
+
 	up.Flags().AddFlagSet(services.ProcessTemplateFlags)
 	metadatas := up.Flags().StringSlice("metadata", []string{}, "key=value to set metadata")
 
@@ -60,7 +62,8 @@ func Command(plugins func() discovery.Plugins) *cobra.Command {
 		wait := types.MustParseDuration(*waitDuration)
 
 		log.Info("Starting up base plugins")
-		basePlugins := []string{"vars", "manager"}
+		baseStack := fmt.Sprintf("manager:%v", *stack)
+		basePlugins := []string{"vars", "group:group-stateless", baseStack}
 		for _, base := range basePlugins {
 			execName, kind, name, _ := local.StartPlugin(base).Parse()
 			err := pluginManager.Launch(execName, kind, name, nil)

cmd/infrakit/util/init/init.go

@@ -174,11 +174,17 @@ func Command(plugins func() discovery.Plugins) *cobra.Command {
 			for _, start := range *starts {
 				targets = append(targets, local.StartPlugin(start))
 			}
-			more, err := local.Plugins(gid, groupSpec)
-			if err != nil {
-				return targets, err
+
+			// make this either-or to allow manual overriding.
+			if len(*starts) == 0 {
+
+				more, err := local.Plugins(gid, groupSpec)
+				if err != nil {
+					return targets, err
+				}
+				targets = append(targets, more...)
+
 			}
-			targets = append(targets, more...)
 			log.Info("plugins to start", "targets", targets)
 			return
 		}

pkg/manager/api.go

@@ -95,4 +95,10 @@ type Options struct {
 
 	// MetadataRefreshInterval is the interval to check for updates to metadata
 	MetadataRefreshInterval types.Duration
+
+	// LeaderCommitSpecsRetries is how many times to retry commit specs when becomes leader
+	LeaderCommitSpecsRetries int
+
+	// LeaderCommitSpecsRetryInterval is how long to wait before next retry
+	LeaderCommitSpecsRetryInterval types.Duration
 }

pkg/manager/manager.go

@@ -361,7 +361,7 @@ func (m *manager) getCurrentState() (globalSpec, error) {
 	return global, nil
 }
 
-func (m *manager) loadSpecs() error {
+func (m *manager) loadAndCommitSpecs() error {
 	if m.Options.SpecStore == nil {
 		return nil
 	}
@@ -385,7 +385,7 @@ func (m *manager) loadMetadata() (err error) {
 		return nil
 	}
 
-	log.Debug("loading metadata and committing")
+	log.Debug("loading metadata and committing", "V", debugV2)
 
 	var saved interface{}
 	err = m.Options.MetadataStore.Load(&saved)
@@ -400,11 +400,11 @@ func (m *manager) loadMetadata() (err error) {
 	}
 
 	if any == nil {
-		log.Debug("no metadata stored")
+		log.Debug("no metadata stored", "V", debugV2)
 		return
 	}
 
-	log.Debug("loaded metadata", "data", any.String())
+	log.Debug("loaded metadata", "data", any.String(), "V", debugV2)
 	_, proposed, cas, e := m.Updatable.Changes([]metadata.Change{
 		{Path: types.Dot, Value: any},
 	})
@@ -413,7 +413,7 @@ func (m *manager) loadMetadata() (err error) {
 		return
 	}
 
-	log.Debug("updating backend with stored metadata", "cas", cas, "proposed", proposed)
+	log.Debug("updating backend with stored metadata", "cas", cas, "proposed", proposed, "V", debugV2)
 	return m.Updatable.Commit(proposed, cas)
 }
 
@@ -427,11 +427,39 @@ func (m *manager) onAssumeLeadership() (err error) {
 		log.Error("error loading metadata", "err", err)
 	}
 
-	err = m.loadSpecs()
-	if err != nil {
-		log.Error("error loading specs", "err", err)
+	err = m.loadAndCommitSpecs()
+	log.Debug("Loading and committing specs", "err", err)
+
+	if err != nil && m.Options.LeaderCommitSpecsRetries > 0 {
+
+		log.Info("Retry loading and committing specs",
+			"retries", m.Options.LeaderCommitSpecsRetries,
+			"interval", m.Options.LeaderCommitSpecsRetryInterval)
+
+		delay := 1 * time.Second
+		if m.Options.LeaderCommitSpecsRetryInterval > 0 {
+			delay = m.Options.LeaderCommitSpecsRetryInterval.Duration()
+		}
+
+		for i := 1; i < m.Options.LeaderCommitSpecsRetries; i++ {
+
+			<-time.After(delay)
+
+			err = m.loadAndCommitSpecs()
+			if err == nil {
+				log.Info("Loaded and committed specs")
+				return nil
+			}
+
+			if err != nil {
+				log.Error("error loading specs", "err", err, "attempt", i)
+			}
+
+		}
+
 	}
-	return
+
+	return err
 }
 
 func (m *manager) onLostLeadership() error {

pkg/rpc/client/handshake_test.go

@@ -2,33 +2,11 @@ package client
 
 import (
 	"fmt"
-	"io/ioutil"
-	"net/http"
-	"path/filepath"
 	"testing"
 
-	"github.com/docker/infrakit/pkg/rpc/server"
-	"github.com/docker/infrakit/pkg/spi"
 	"github.com/stretchr/testify/require"
 )
 
-var apiSpec = spi.InterfaceSpec{
-	Name:    "TestPlugin",
-	Version: "0.1.0",
-}
-
-func startPluginServer(t *testing.T) (server.Stoppable, string) {
-	dir, err := ioutil.TempDir("", "infrakit_handshake_test")
-	require.NoError(t, err)
-
-	name := "instance"
-	socket := filepath.Join(dir, name)
-
-	testServer, err := server.StartPluginAtPath(socket, &TestPlugin{spec: apiSpec})
-	require.NoError(t, err)
-	return testServer, socket
-}
-
 func TestErrVersionMismatch(t *testing.T) {
 	var e error
 
@@ -38,73 +16,3 @@ func TestErrVersionMismatch(t *testing.T) {
 	e = fmt.Errorf("untyped")
 	require.False(t, IsErrVersionMismatch(e))
 }
-
-func TestHandshakeSuccess(t *testing.T) {
-	testServer, socket := startPluginServer(t)
-	defer testServer.Stop()
-
-	r, err := New(socket, apiSpec)
-	require.NoError(t, err)
-	client := rpcClient{client: r}
-	require.NoError(t, client.DoSomething())
-}
-
-func TestHandshakeFailVersion(t *testing.T) {
-	testServer, socket := startPluginServer(t)
-	defer testServer.Stop()
-
-	r, err := New(socket, spi.InterfaceSpec{Name: "TestPlugin", Version: "0.2.0"})
-	require.Error(t, err)
-
-	client := rpcClient{client: r}
-	err = client.DoSomething()
-	require.Error(t, err)
-	require.Equal(t, "Plugin supports TestPlugin interface version 0.1.0, client requires 0.2.0", err.Error())
-}
-
-func TestHandshakeFailWrongAPI(t *testing.T) {
-	testServer, socket := startPluginServer(t)
-	defer testServer.Stop()
-
-	r, err := New(socket, spi.InterfaceSpec{Name: "OtherPlugin", Version: "0.1.0"})
-	require.Error(t, err)
-
-	client := rpcClient{client: r}
-	err = client.DoSomething()
-	require.Error(t, err)
-	require.Equal(t, "Plugin does not support interface OtherPlugin/0.1.0", err.Error())
-}
-
-type rpcClient struct {
-	client Client
-}
-
-func (c rpcClient) DoSomething() error {
-	req := EmptyMessage{}
-	resp := EmptyMessage{}
-	return c.client.Call("TestPlugin.DoSomething", req, &resp)
-}
-
-// TestPlugin is an RPC service for this unit test.
-type TestPlugin struct {
-	spec spi.InterfaceSpec
-}
-
-// ImplementedInterface returns the interface implemented by this RPC service.
-func (p *TestPlugin) ImplementedInterface() spi.InterfaceSpec {
-	return p.spec
-}
-
-// Types returns the types
-func (p *TestPlugin) Types() []string {
-	return []string{"."}
-}
-
-// EmptyMessage is an empty test message.
-type EmptyMessage struct {
-}
-
-// DoSomething is an empty test RPC.
-func (p *TestPlugin) DoSomething(_ *http.Request, req *EmptyMessage, resp *EmptyMessage) error {
-	return nil
-}

pkg/rpc/group/publisher.go

@@ -0,0 +1,29 @@
+package group
+
+import (
+	"github.com/docker/infrakit/pkg/spi/event"
+	"github.com/docker/infrakit/pkg/types"
+)
+
+// List returns a list of *child nodes* given a path for a topic.
+// A topic of "." is the top level
+func (p *Group) List(topic types.Path) (child []string, err error) {
+	m := map[string]interface{}{}
+
+	subs := p.keyed.Types()
+	if len(subs) > 0 {
+		for _, t := range subs {
+			types.Put([]string{t, "commit"}, "", m)
+			types.Put([]string{t, "describe"}, "", m)
+		}
+	} else {
+		types.Put([]string{"commit"}, "", m)
+		types.Put([]string{"describe"}, "", m)
+	}
+
+	return types.List(topic, m), nil
+}
+
+// PublishOn sets the channel to publish
+func (p *Group) PublishOn(chan<- *event.Event) {
+}