Add Version information to plugin APIs (#318)

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

Author: wfarner

docs/plugins/README.md

@@ -113,15 +113,15 @@ _InfraKit_ plugins are exposed via HTTP, using [JSON-RPC 2.0](http://www.jsonrpc
 API requests can be made manually with `curl`.  For example, the following command will list all groups:
 ```console
 $ curl -X POST --unix-socket ~/.infrakit/plugins/group http:/rpc \
-  -H 'Content-Type: application/json'
+  -H 'Content-Type: application/json' \
   -d '{"jsonrpc":"2.0","method":"Group.InspectGroups","params":{},"id":1}'
 {"jsonrpc":"2.0","result":{"Groups":null},"id":1}
 ```
 
 API errors are surfaced with the `error` response field:
 ```console
 $ curl -X POST --unix-socket ~/.infrakit/plugins/group http:/rpc \
-  -H 'Content-Type: application/json'
+  -H 'Content-Type: application/json' \
   -d '{"jsonrpc":"2.0","method":"Group.CommitGroup","params":{},"id":1}'
 {"jsonrpc":"2.0","error":{"code":-32000,"message":"Group ID must not be blank","data":null},"id":1}
 ```
@@ -135,3 +135,14 @@ for each plugin type:
 See also: documentation on common API [types](types.md).
 
 Additionally, all plugins will log each API HTTP request and response when run with the `--log 5` command line argument.
+
+##### API identification
+Plugins are required to identify the name and version of plugin APIs they implement.  This is done with a request
+like the following:
+
+```console
+$ curl -X POST --unix-socket ~/.infrakit/plugins/group http:/rpc \
+  -H 'Content-Type: application/json' \
+  -d '{"jsonrpc":"2.0","method":"Plugin.Implements","params":{},"id":1}'
+{"jsonrpc":"2.0","result":{"Interfaces":[{"Name":"Group","Version":"0.1.0"}]},"id":1}
+```

docs/plugins/flavor.md

@@ -1,6 +1,6 @@
 # Flavor plugin API
 
-<!-- SOURCE-CHECKSUM pkg/spi/flavor/* 81a2c81f42a56ce0baa54511ee621f885fc7080e -->
+<!-- SOURCE-CHECKSUM pkg/spi/flavor/* 921b81c90c2abc7aec298333e1e1cf9c039afca5 -->
 
 ## API
 

docs/plugins/group.md

@@ -1,6 +1,6 @@
 # Group plugin API
 
-<!-- SOURCE-CHECKSUM pkg/spi/group/* 0eec99ab5b4dc627b4025e29fb97dba4ced8c16f -->
+<!-- SOURCE-CHECKSUM pkg/spi/group/* 4bc86b2ae0893db92f880ab4bb2479b5def55746 -->
 
 ## API
 

docs/plugins/instance.md

@@ -1,6 +1,6 @@
 # Instance plugin API
 
-<!-- SOURCE-CHECKSUM pkg/spi/instance/* 0c778e96cbeb32043532709412e15e6cc86778d7338393f886f528c3824986fc97cb27410aefd8e2 -->
+<!-- SOURCE-CHECKSUM pkg/spi/instance/* 8fc5d1832d0d96d01d8d76ea1137230790fe51fe338393f886f528c3824986fc97cb27410aefd8e2 -->
 
 ## API
 

pkg/cli/serverutil.go

@@ -10,7 +10,7 @@ import (
 
 // 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.
-func RunPlugin(name string, plugin interface{}) {
+func RunPlugin(name string, plugin server.VersionedInterface) {
 	stoppable, err := server.StartPluginAtPath(path.Join(discovery.Dir(), name), plugin)
 	if err != nil {
 		log.Error(err)

pkg/rpc/client/client.go

@@ -3,34 +3,35 @@ package client
 import (
 	"bytes"
 	log "github.com/Sirupsen/logrus"
+	"github.com/docker/infrakit/pkg/spi"
 	"github.com/gorilla/rpc/v2/json2"
 	"net"
 	"net/http"
 	"net/http/httputil"
+	"sync"
 )
 
-// Client is an HTTP client for sending JSON-RPC requests.
-type Client struct {
+type client struct {
 	http http.Client
 }
 
-// New creates a new Client that communicates with a unix socke.
-func New(socketPath string) Client {
+// New creates a new Client that communicates with a unix socket and validates the remote API.
+func New(socketPath string, api spi.InterfaceSpec) Client {
 	dialUnix := func(proto, addr string) (conn net.Conn, err error) {
 		return net.Dial("unix", socketPath)
 	}
 
-	return Client{http: http.Client{Transport: &http.Transport{Dial: dialUnix}}}
+	unvalidatedClient := &client{http: http.Client{Transport: &http.Transport{Dial: dialUnix}}}
+	return &handshakingClient{client: unvalidatedClient, iface: api, lock: &sync.Mutex{}}
 }
 
-// Call sends an RPC with a method and argument.  The result must be a pointer to the response object.
-func (c Client) Call(method string, arg interface{}, result interface{}) error {
+func (c client) Call(method string, arg interface{}, result interface{}) error {
 	message, err := json2.EncodeClientRequest(method, arg)
 	if err != nil {
 		return err
 	}
 
-	req, err := http.NewRequest("POST", "http:///", bytes.NewReader(message))
+	req, err := http.NewRequest("POST", "http://a/", bytes.NewReader(message))
 	if err != nil {
 		return err
 	}
@@ -43,7 +44,7 @@ func (c Client) Call(method string, arg interface{}, result interface{}) error {
 		log.Error(err)
 	}
 
-	resp, err := c.http.Post("http://d/rpc", "application/json", bytes.NewReader(message))
+	resp, err := c.http.Do(req)
 	if err != nil {
 		return err
 	}

pkg/rpc/client/handshake.go

@@ -0,0 +1,70 @@
+package client
+
+import (
+	"fmt"
+	"github.com/docker/infrakit/pkg/rpc/plugin"
+	"github.com/docker/infrakit/pkg/spi"
+	"sync"
+)
+
+type handshakingClient struct {
+	client Client
+	iface  spi.InterfaceSpec
+
+	// handshakeResult handles the tri-state outcome of handshake state:
+	//  - handshake has not yet completed (nil)
+	//  - handshake completed successfully (non-nil result, nil error)
+	//  - handshake failed (non-nil result, non-nil error)
+	handshakeResult *handshakeResult
+
+	// lock guards handshakeResult
+	lock *sync.Mutex
+}
+
+type handshakeResult struct {
+	err error
+}
+
+func (c *handshakingClient) handshake() error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.handshakeResult == nil {
+		req := plugin.ImplementsRequest{}
+		resp := plugin.ImplementsResponse{}
+
+		if err := c.client.Call("Plugin.Implements", req, &resp); err != nil {
+			return err
+		}
+
+		err := fmt.Errorf("Plugin does not support interface %v", c.iface)
+		if resp.APIs != nil {
+			for _, iface := range resp.APIs {
+				if iface.Name == c.iface.Name {
+					if iface.Version == c.iface.Version {
+						err = nil
+						break
+					} else {
+						err = fmt.Errorf(
+							"Plugin supports %s interface version %s, client requires %s",
+							iface.Name,
+							iface.Version,
+							c.iface.Version)
+					}
+				}
+			}
+		}
+
+		c.handshakeResult = &handshakeResult{err: err}
+	}
+
+	return c.handshakeResult.err
+}
+
+func (c *handshakingClient) Call(method string, arg interface{}, result interface{}) error {
+	if err := c.handshake(); err != nil {
+		return err
+	}
+
+	return c.client.Call(method, arg, result)
+}

pkg/rpc/client/handshake_test.go

@@ -0,0 +1,85 @@
+package client
+
+import (
+	"github.com/docker/infrakit/pkg/rpc/server"
+	"github.com/docker/infrakit/pkg/spi"
+	"github.com/stretchr/testify/require"
+	"io/ioutil"
+	"net/http"
+	"path/filepath"
+	"testing"
+)
+
+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 TestHandshakeSuccess(t *testing.T) {
+	testServer, socket := startPluginServer(t)
+	defer testServer.Stop()
+
+	client := rpcClient{client: New(socket, apiSpec)}
+	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()
+	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()
+
+	client := rpcClient{client: New(socket, spi.InterfaceSpec{Name: "OtherPlugin", Version: "0.1.0"})}
+	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
+}
+
+// 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/client/rpc.go

@@ -0,0 +1,8 @@
+package client
+
+// Client allows execution of RPCs.
+type Client interface {
+
+	// Call invokes an RPC method with an argument and a pointer to a result that will hold the return value.
+	Call(method string, arg interface{}, result interface{}) error
+}

pkg/rpc/flavor/client.go

@@ -10,7 +10,7 @@ import (
 
 // NewClient returns a plugin interface implementation connected to a remote plugin
 func NewClient(socketPath string) flavor.Plugin {
-	return &client{client: rpc_client.New(socketPath)}
+	return &client{client: rpc_client.New(socketPath, flavor.InterfaceSpec)}
 }
 
 type client struct {