Use JSON-RPC over HTTP for communication between plugins

Signed-off-by: Bill Farner <[email protected]>

Author: wfarner

cli/serverutil.go

@@ -5,16 +5,15 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/infrakit/discovery"
-	"github.com/docker/infrakit/rpc"
+	"github.com/docker/infrakit/rpc/server"
 )
 
 // RunPlugin runs a plugin server, advertising with the provided name for discovery.
-// THe plugin should conform to the rpc call convention as implemented in the rpc package.
+// The plugin should conform to the rpc call convention as implemented in the rpc package.
 func RunPlugin(name string, plugin interface{}) {
-	_, stopped, err := rpc.StartPluginAtPath(path.Join(discovery.Dir(), name), plugin)
+	stoppable, err := server.StartPluginAtPath(path.Join(discovery.Dir(), name), plugin)
 	if err != nil {
 		log.Error(err)
 	}
-
-	<-stopped // block until done
+	stoppable.AwaitStopped()
 }

cmd/cli/flavor.go

@@ -31,10 +31,7 @@ func flavorPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 				return err
 			}
 
-			flavorPlugin, err = flavor_plugin.NewClient(endpoint.Protocol, endpoint.Address)
-			if err != nil {
-				return err
-			}
+			flavorPlugin = flavor_plugin.NewClient(endpoint.Address)
 
 			return nil
 		},

cmd/cli/group.go

@@ -35,10 +35,7 @@ func groupPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 				return err
 			}
 
-			groupPlugin, err = group_plugin.NewClient(endpoint.Protocol, endpoint.Address)
-			if err != nil {
-				return err
-			}
+			groupPlugin = group_plugin.NewClient(endpoint.Address)
 
 			return nil
 		},

cmd/cli/instance.go

@@ -30,10 +30,7 @@ func instancePluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 				return err
 			}
 
-			instancePlugin, err = instance_plugin.NewClient(endpoint.Protocol, endpoint.Address)
-			if err != nil {
-				return err
-			}
+			instancePlugin = instance_plugin.NewClient(endpoint.Address)
 
 			return nil
 		},

cmd/group/main.go

@@ -40,15 +40,15 @@ func main() {
 				if err != nil {
 					return nil, err
 				}
-				return instance_client.NewClient(endpoint.Protocol, endpoint.Address)
+				return instance_client.NewClient(endpoint.Address), nil
 			}
 
 			flavorPluginLookup := func(n string) (flavor.Plugin, error) {
 				endpoint, err := plugins.Find(n)
 				if err != nil {
 					return nil, err
 				}
-				return flavor_client.NewClient(endpoint.Protocol, endpoint.Address)
+				return flavor_client.NewClient(endpoint.Address), nil
 			}
 
 			cli.RunPlugin(name, group_server.PluginServer(

cmd/manager/main.go

@@ -9,7 +9,6 @@ import (
 	"github.com/docker/infrakit/discovery"
 	"github.com/docker/infrakit/leader"
 	"github.com/docker/infrakit/manager"
-	"github.com/docker/infrakit/rpc"
 	group_rpc "github.com/docker/infrakit/rpc/group"
 	"github.com/docker/infrakit/store"
 	"github.com/spf13/cobra"
@@ -52,28 +51,20 @@ func runMain(backend *backend) error {
 
 	log.Infoln("Starting up manager:", backend)
 
-	manager, err := manager.NewManager(backend.plugins,
+	mgr, err := manager.NewManager(backend.plugins,
 		backend.leader, backend.snapshot, backend.pluginName)
 	if err != nil {
 		return err
 	}
 
-	_, err = manager.Start()
+	_, err = mgr.Start()
 	if err != nil {
 		return err
 	}
 
-	_, stopped, err := rpc.StartPluginAtPath(
-		filepath.Join(discovery.Dir(), backend.id),
-		group_rpc.PluginServer(manager),
-	)
-	if err != nil {
-		return err
-	}
-
-	<-stopped // block until done
+	cli.RunPlugin(backend.id, group_rpc.PluginServer(mgr))
 
-	manager.Stop()
+	mgr.Stop()
 	log.Infoln("Manager stopped")
 
 	return err

discovery/dir_test.go

@@ -7,8 +7,8 @@ import (
 	"testing"
 	"time"
 
-	server "github.com/docker/infrakit/rpc"
 	rpc "github.com/docker/infrakit/rpc/instance"
+	"github.com/docker/infrakit/rpc/server"
 	"github.com/stretchr/testify/require"
 )
 
@@ -31,17 +31,15 @@ func TestDirDiscovery(t *testing.T) {
 
 	name1 := "server1"
 	path1 := filepath.Join(dir, name1)
-	stop1, errors1, err1 := server.StartPluginAtPath(path1, rpc.PluginServer(nil))
-	require.NoError(t, err1)
-	require.NotNil(t, stop1)
-	require.NotNil(t, errors1)
+	server1, err := server.StartPluginAtPath(path1, rpc.PluginServer(nil))
+	require.NoError(t, err)
+	require.NotNil(t, server1)
 
 	name2 := "server2"
 	path2 := filepath.Join(dir, name2)
-	stop2, errors2, err2 := server.StartPluginAtPath(path2, rpc.PluginServer(nil))
-	require.NoError(t, err2)
-	require.NotNil(t, stop2)
-	require.NotNil(t, errors2)
+	server2, err := server.StartPluginAtPath(path2, rpc.PluginServer(nil))
+	require.NoError(t, err)
+	require.NotNil(t, server2)
 
 	discover, err := newDirPluginDiscovery(dir)
 	require.NoError(t, err)
@@ -55,7 +53,7 @@ func TestDirDiscovery(t *testing.T) {
 	require.NotNil(t, p)
 
 	// Now we stop the servers
-	close(stop1)
+	server1.Stop()
 	blockWhileFileExists(path1)
 
 	p, err = discover.Find(name1)
@@ -65,7 +63,7 @@ func TestDirDiscovery(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, p)
 
-	close(stop2)
+	server2.Stop()
 
 	blockWhileFileExists(path2)
 

example/flavor/combo/main.go

@@ -32,7 +32,7 @@ func main() {
 				if err != nil {
 					return nil, err
 				}
-				return flavor_rpc.NewClient(endpoint.Protocol, endpoint.Address)
+				return flavor_rpc.NewClient(endpoint.Address), nil
 			}
 
 			cli.SetLogLevel(logLevel)

leader/swarm/swarm_test.go

@@ -19,8 +19,10 @@ func TestSwarmDetector(t *testing.T) {
 
 	ctx := context.Background()
 	nodeInfo := types.Info{
-		Swarm: swarm.Info{
-			NodeID: "node",
+		InfoBase: &types.InfoBase{
+			Swarm: swarm.Info{
+				NodeID: "node",
+			},
 		},
 	}
 	node := swarm.Node{

manager/group_plugin_impl.go

@@ -18,13 +18,8 @@ func (m *manager) proxyForGroupPlugin(name string) (group.Plugin, error) {
 		return nil, err
 	}
 
-	client, err := rpc.NewClient(endpoint.Protocol, endpoint.Address)
-	if err != nil {
-		return nil, err
-	}
-
 	m.backendName = name
-	return client, nil
+	return rpc.NewClient(endpoint.Address), nil
 }
 
 // This implements the Group Plugin interface to support single group-only operations