Handshake early to detect plugin version mismatch (#369)

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

Author: David Chung

cmd/cli/flavor.go

@@ -34,8 +34,11 @@ func flavorPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 			return err
 		}
 
-		flavorPlugin = flavor_plugin.NewClient(plugin.Name(*name), endpoint.Address)
-
+		p, err := flavor_plugin.NewClient(plugin.Name(*name), endpoint.Address)
+		if err != nil {
+			return err
+		}
+		flavorPlugin = p
 		return nil
 	}
 

cmd/cli/group.go

@@ -37,8 +37,11 @@ func groupPluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 			return err
 		}
 
-		groupPlugin = group_plugin.NewClient(endpoint.Address)
-
+		p, err := group_plugin.NewClient(endpoint.Address)
+		if err != nil {
+			return err
+		}
+		groupPlugin = p
 		return nil
 	}
 

cmd/cli/instance.go

@@ -32,8 +32,11 @@ func instancePluginCommand(plugins func() discovery.Plugins) *cobra.Command {
 			return err
 		}
 
-		instancePlugin = instance_plugin.NewClient(plugin.Name(*name), endpoint.Address)
-
+		p, err := instance_plugin.NewClient(plugin.Name(*name), endpoint.Address)
+		if err != nil {
+			return err
+		}
+		instancePlugin = p
 		return nil
 	}
 

cmd/group/main.go

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

examples/flavor/combo/main.go

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

pkg/manager/group_plugin_impl.go

@@ -22,7 +22,7 @@ func (m *manager) proxyForGroupPlugin(name string) (group.Plugin, error) {
 		if err != nil {
 			return nil, err
 		}
-		return rpc.NewClient(endpoint.Address), nil
+		return rpc.NewClient(endpoint.Address)
 	}), nil
 }
 

pkg/manager/manager.go

@@ -332,7 +332,7 @@ func (m *manager) execPlugins(config GlobalSpec, work func(group.Plugin, group.S
 			return err
 		}
 
-		gp := rpc.NewClient(ep.Address)
+		gp, err := rpc.NewClient(ep.Address)
 		if err != nil {
 			log.Warningln("Cannot contact group", id, " at plugin", name, "endpoint=", ep.Address)
 			return err

pkg/rpc/client/client.go

@@ -16,13 +16,22 @@ type client struct {
 }
 
 // New creates a new Client that communicates with a unix socket and validates the remote API.
-func New(socketPath string, api spi.InterfaceSpec) Client {
+func New(socketPath string, api spi.InterfaceSpec) (Client, error) {
 	dialUnix := func(proto, addr string) (conn net.Conn, err error) {
 		return net.Dial("unix", socketPath)
 	}
 
 	unvalidatedClient := &client{http: http.Client{Transport: &http.Transport{Dial: dialUnix}}}
-	return &handshakingClient{client: unvalidatedClient, iface: api, lock: &sync.Mutex{}}
+	cl := &handshakingClient{client: unvalidatedClient, iface: api, lock: &sync.Mutex{}}
+
+	// check handshake
+	if err := cl.handshake(); err != nil {
+		// Note - we still return the client with the possibility of doing a handshake later on
+		// if we provide an api for the plugin to recheck later.  This way, individual components
+		// can stay running and recalibrate themselves after the user has corrected the problems.
+		return cl, err
+	}
+	return cl, nil
 }
 
 func (c client) Call(method string, arg interface{}, result interface{}) error {

pkg/rpc/client/handshake.go

@@ -25,6 +25,19 @@ type handshakeResult struct {
 	err error
 }
 
+type errVersionMismatch string
+
+// Error implements error interface
+func (e errVersionMismatch) Error() string {
+	return string(e)
+}
+
+// IsErrVersionMismatch return true if the error is from mismatched api versions.
+func IsErrVersionMismatch(e error) bool {
+	_, is := e.(errVersionMismatch)
+	return is
+}
+
 func (c *handshakingClient) handshake() error {
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -45,11 +58,11 @@ func (c *handshakingClient) handshake() error {
 						err = nil
 						break
 					} else {
-						err = fmt.Errorf(
+						err = errVersionMismatch(fmt.Sprintf(
 							"Plugin supports %s interface version %s, client requires %s",
 							iface.Name,
 							iface.Version,
-							c.iface.Version)
+							c.iface.Version))
 					}
 				}
 			}

pkg/rpc/client/handshake_test.go

@@ -1,13 +1,15 @@
 package client
 
 import (
-	"github.com/docker/infrakit/pkg/rpc/server"
-	"github.com/docker/infrakit/pkg/spi"
-	"github.com/stretchr/testify/require"
+	"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{
@@ -27,20 +29,35 @@ func startPluginServer(t *testing.T) (server.Stoppable, string) {
 	return testServer, socket
 }
 
+func TestErrVersionMismatch(t *testing.T) {
+	var e error
+
+	e = errVersionMismatch("test")
+	require.True(t, IsErrVersionMismatch(e))
+
+	e = fmt.Errorf("untyped")
+	require.False(t, IsErrVersionMismatch(e))
+}
+
 func TestHandshakeSuccess(t *testing.T) {
 	testServer, socket := startPluginServer(t)
 	defer testServer.Stop()
 
-	client := rpcClient{client: New(socket, apiSpec)}
+	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()
 
-	client := rpcClient{client: New(socket, spi.InterfaceSpec{Name: "TestPlugin", Version: "0.2.0"})}
-	err := client.DoSomething()
+	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())
 }
@@ -49,8 +66,11 @@ func TestHandshakeFailWrongAPI(t *testing.T) {
 	testServer, socket := startPluginServer(t)
 	defer testServer.Stop()
 
-	client := rpcClient{client: New(socket, spi.InterfaceSpec{Name: "OtherPlugin", Version: "0.1.0"})}
-	err := client.DoSomething()
+	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())
 }