Support plugins that listen on network port (#488)

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

Author: David Chung

pkg/cli/serverutil.go

@@ -19,20 +19,47 @@ func EnsureDirExists(dir string) {
 // 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 server.VersionedInterface, more ...server.VersionedInterface) {
-
 	dir := local.Dir()
 	EnsureDirExists(dir)
 
 	socketPath := path.Join(dir, name)
 	pidPath := path.Join(dir, name+".pid")
+	run("", socketPath, pidPath, plugin, more...)
+}
 
-	stoppable, err := server.StartPluginAtPath(socketPath, plugin, more...)
-	if err != nil {
-		logrus.Error(err)
+// RunListener runs a plugin server, listening at listen address, and
+// advertising with the provided name for discovery.
+// The plugin should conform to the rpc call convention as implemented in the rpc package.
+func RunListener(listen, name string, plugin server.VersionedInterface, more ...server.VersionedInterface) {
+	dir := local.Dir()
+	EnsureDirExists(dir)
+
+	discoverPath := path.Join(dir, name+".listen")
+	pidPath := path.Join(dir, name+".pid")
+	run(listen, discoverPath, pidPath, plugin, more...)
+}
+
+func run(listen, discoverPath, pidPath string,
+	plugin server.VersionedInterface, more ...server.VersionedInterface) {
+
+	var stoppable server.Stoppable
+
+	if listen != "" {
+		s, err := server.StartListenerAtPath(listen, discoverPath, plugin, more...)
+		if err != nil {
+			logrus.Error(err)
+		}
+		stoppable = s
+	} else {
+		s, err := server.StartPluginAtPath(discoverPath, plugin, more...)
+		if err != nil {
+			logrus.Error(err)
+		}
+		stoppable = s
 	}
 
 	// write PID file
-	err = ioutil.WriteFile(pidPath, []byte(fmt.Sprintf("%v", os.Getpid())), 0644)
+	err := ioutil.WriteFile(pidPath, []byte(fmt.Sprintf("%v", os.Getpid())), 0644)
 	if err != nil {
 		logrus.Error(err)
 	}
@@ -44,4 +71,7 @@ func RunPlugin(name string, plugin server.VersionedInterface, more ...server.Ver
 	// clean up
 	os.Remove(pidPath)
 	logrus.Infoln("Removed PID file at", pidPath)
+
+	os.Remove(discoverPath)
+	logrus.Infoln("Removed discover file at", discoverPath)
 }

pkg/discovery/discovery.go

@@ -18,16 +18,16 @@ const (
 	PluginDirEnvVar = "INFRAKIT_PLUGINS_DIR"
 )
 
-// ErrNotUnixSocket is the error raised when the file is not a unix socket
-type ErrNotUnixSocket string
+// ErrNotUnixSocketOrListener is the error raised when the file is not a unix socket
+type ErrNotUnixSocketOrListener string
 
-func (e ErrNotUnixSocket) Error() string {
-	return fmt.Sprintf("not a unix socket:%s", string(e))
+func (e ErrNotUnixSocketOrListener) Error() string {
+	return fmt.Sprintf("not a unix socket or listener:%s", string(e))
 }
 
-// IsErrNotUnixSocket returns true if the error is due to the file not being a valid unix socket.
-func IsErrNotUnixSocket(e error) bool {
-	_, is := e.(ErrNotUnixSocket)
+// IsErrNotUnixSocketOrListener returns true if the error is due to the file not being a valid unix socket.
+func IsErrNotUnixSocketOrListener(e error) bool {
+	_, is := e.(ErrNotUnixSocketOrListener)
 	return is
 }
 

pkg/discovery/discovery_test.go

@@ -6,8 +6,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestErrNotUnixSocket(t *testing.T) {
-	err := ErrNotUnixSocket("no socket!")
+func TestErrNotUnixSocketOrListener(t *testing.T) {
+	err := ErrNotUnixSocketOrListener("no socket!")
 	require.Error(t, err)
-	require.True(t, IsErrNotUnixSocket(err))
+	require.True(t, IsErrNotUnixSocketOrListener(err))
 }

pkg/discovery/local/dir.go

@@ -3,9 +3,11 @@ package local
 import (
 	"fmt"
 	"io/ioutil"
+	"net/url"
 	"os"
 	"os/user"
 	"path/filepath"
+	"strings"
 	"sync"
 
 	"github.com/docker/infrakit/pkg/discovery"
@@ -81,16 +83,32 @@ func newDirPluginDiscovery(dir string) (*dirPluginDiscovery, error) {
 }
 
 func (r *dirPluginDiscovery) dirLookup(entry os.FileInfo) (*plugin.Endpoint, error) {
-	socketPath := filepath.Join(r.dir, entry.Name())
-	if entry.Mode()&os.ModeSocket != 0 {
+	path := filepath.Join(r.dir, entry.Name())
+
+	switch {
+
+	case entry.Mode()&os.ModeSocket != 0:
 		return &plugin.Endpoint{
 			Protocol: "unix",
-			Address:  socketPath,
+			Address:  path,
 			Name:     entry.Name(),
 		}, nil
+
+	case entry.Mode()&os.ModeType == 0 && filepath.Ext(path) == ".listen":
+		if buff, err := ioutil.ReadFile(path); err == nil {
+			// content should be a url
+			i := strings.Index(entry.Name(), ".listen")
+			if u, err := url.Parse(string(buff)); err == nil {
+				return &plugin.Endpoint{
+					Protocol: u.Scheme,
+					Address:  u.String(),
+					Name:     entry.Name()[0:i],
+				}, nil
+			}
+		}
 	}
 
-	return nil, discovery.ErrNotUnixSocket(socketPath)
+	return nil, discovery.ErrNotUnixSocketOrListener(path)
 }
 
 // List returns a list of plugins known, keyed by the name
@@ -112,7 +130,7 @@ func (r *dirPluginDiscovery) List() (map[string]*plugin.Endpoint, error) {
 			instance, err := r.dirLookup(entry)
 
 			if err != nil {
-				if !discovery.IsErrNotUnixSocket(err) {
+				if !discovery.IsErrNotUnixSocketOrListener(err) {
 					log.Warn("Err loading plugin", "err", err)
 				}
 				continue

pkg/discovery/local/dir_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/docker/infrakit/pkg/plugin"
 	rpc "github.com/docker/infrakit/pkg/rpc/instance"
 	"github.com/docker/infrakit/pkg/rpc/server"
+	. "github.com/docker/infrakit/pkg/testing"
 	"github.com/stretchr/testify/require"
 )
 
@@ -30,15 +31,17 @@ func TestDirDiscovery(t *testing.T) {
 	dir, err := ioutil.TempDir("", "infrakit_dir_test")
 	require.NoError(t, err)
 
+	T(100).Infoln("Starting server1")
 	name1 := "server1"
 	path1 := filepath.Join(dir, name1)
 	server1, err := server.StartPluginAtPath(path1, rpc.PluginServer(nil))
 	require.NoError(t, err)
 	require.NotNil(t, server1)
 
+	T(100).Infoln("Starting server2")
 	name2 := "server2"
-	path2 := filepath.Join(dir, name2)
-	server2, err := server.StartPluginAtPath(path2, rpc.PluginServer(nil))
+	path2 := filepath.Join(dir, name2+".listen")
+	server2, err := server.StartListenerAtPath("localhost:7777", path2, rpc.PluginServer(nil))
 	require.NoError(t, err)
 	require.NotNil(t, server2)
 

pkg/rpc/client/client.go

@@ -3,10 +3,12 @@ package client
 import (
 	"bytes"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"path"
 	"sync"
 
 	log "github.com/Sirupsen/logrus"
@@ -22,10 +24,6 @@ type client struct {
 
 // New creates a new Client that communicates with a unix socket and validates the remote API.
 func New(address string, api spi.InterfaceSpec) (Client, error) {
-	u, err := url.Parse(address)
-	if err != nil {
-		return nil, err
-	}
 
 	u, httpC, err := parseAddress(address)
 	if err != nil {
@@ -45,6 +43,15 @@ func New(address string, api spi.InterfaceSpec) (Client, error) {
 }
 
 func parseAddress(address string) (*url.URL, *http.Client, error) {
+
+	if path.Ext(address) == ".listen" {
+		buff, err := ioutil.ReadFile(address)
+		if err != nil {
+			return nil, nil, err
+		}
+		address = string(buff)
+	}
+
 	u, err := url.Parse(address)
 	if err != nil {
 		return nil, nil, err
@@ -60,7 +67,10 @@ func parseAddress(address string) (*url.URL, *http.Client, error) {
 				return net.Dial("unix", address)
 			},
 		}}, nil
-	case "http", "https", "tcp":
+	case "tcp":
+		u.Scheme = "http"
+		return u, &http.Client{}, nil
+	case "http", "https":
 		return u, &http.Client{}, nil
 
 	default:

pkg/rpc/client/client_test.go

@@ -21,7 +21,7 @@ func TestParseAddress(t *testing.T) {
 	u, c, err = parseAddress("tcp://host:9090/foo/bar/baz")
 	require.NoError(t, err)
 	require.Nil(t, c.Transport)
-	require.Equal(t, "tcp://host:9090/foo/bar/baz", u.String())
+	require.Equal(t, "http://host:9090/foo/bar/baz", u.String())
 
 	u, c, err = parseAddress("https://host:9090")
 	require.NoError(t, err)

pkg/rpc/server/server.go

@@ -2,10 +2,12 @@ package server
 
 import (
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
+	"os"
 	"time"
 
 	log "github.com/Sirupsen/logrus"
@@ -76,9 +78,22 @@ type VersionedInterface interface {
 	ImplementedInterface() spi.InterfaceSpec
 }
 
+// StartListenerAtPath starts an HTTP server listening on tcp port with discovery entry at specified path.
+// Returns a Stoppable that can be used to stop or block on the server.
+func StartListenerAtPath(listen, discoverPath string,
+	receiver VersionedInterface, more ...VersionedInterface) (Stoppable, error) {
+	return startAtPath(listen, discoverPath, receiver, more...)
+}
+
 // StartPluginAtPath starts an HTTP server listening on a unix socket at the specified path.
 // Returns a Stoppable that can be used to stop or block on the server.
 func StartPluginAtPath(socketPath string, receiver VersionedInterface, more ...VersionedInterface) (Stoppable, error) {
+	return startAtPath("", socketPath, receiver, more...)
+}
+
+func startAtPath(listen, discoverPath string,
+	receiver VersionedInterface, more ...VersionedInterface) (Stoppable, error) {
+
 	server := rpc.NewServer()
 	server.RegisterCodec(json2.NewCodec(), "application/json")
 
@@ -160,22 +175,50 @@ func StartPluginAtPath(socketPath string, receiver VersionedInterface, more ...V
 
 	gracefulServer := graceful.Server{
 		Timeout: 10 * time.Second,
-		Server:  &http.Server{Addr: fmt.Sprintf("unix://%s", socketPath), Handler: router},
 	}
 
-	listener, err := net.Listen("unix", socketPath)
-	if err != nil {
-		return nil, err
-	}
+	var listener net.Listener
+
+	if listen != "" {
+		gracefulServer.Server = &http.Server{
+			Addr:    listen,
+			Handler: router,
+		}
+		l, err := net.Listen("tcp", listen)
+		if err != nil {
+			return nil, err
+		}
+		listener = l
+
+		if err := ioutil.WriteFile(discoverPath, []byte(fmt.Sprintf("tcp://%s", listen)), 0644); err != nil {
+			return nil, err
+		}
 
-	log.Infof("Listening at: %s", socketPath)
+		log.Infof("Listening at: %s, discoverable at %s", listen, discoverPath)
+
+	} else {
+		gracefulServer.Server = &http.Server{
+			Addr:    fmt.Sprintf("unix://%s", discoverPath),
+			Handler: router,
+		}
+		l, err := net.Listen("unix", discoverPath)
+		if err != nil {
+			return nil, err
+		}
+		listener = l
+		log.Infof("Listening at: %s", discoverPath)
+
+	}
 
 	go func() {
 		err := gracefulServer.Serve(listener)
 		if err != nil {
 			log.Warn(err)
 		}
 		events.Stop()
+		if listen != "" {
+			os.Remove(discoverPath)
+		}
 	}()
 
 	return &stoppableServer{server: &gracefulServer}, nil

pkg/rpc/server/server_test.go

@@ -59,3 +59,46 @@ func TestUnixSocketServer(t *testing.T) {
 
 	server.Stop()
 }
+
+func TestTCPServer(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mock := plugin_mock.NewMockPlugin(ctrl)
+
+	instanceID := instance.ID("id")
+	spec := instance.Spec{
+		Tags: map[string]string{
+			"tag1": "value1",
+		},
+		Init: "init",
+	}
+
+	properties := types.AnyString(`{"foo":"bar"}`)
+	validateErr := errors.New("validate-error")
+
+	gomock.InOrder(
+		mock.EXPECT().Validate(properties).Return(validateErr),
+		mock.EXPECT().Provision(spec).Return(&instanceID, nil),
+	)
+
+	service := plugin_rpc.PluginServer(mock)
+
+	discover := filepath.Join(os.TempDir(), fmt.Sprintf("%d.listen", time.Now().Unix()))
+	name := plugin.Name(filepath.Base(discover))
+	server, err := StartListenerAtPath("localhost:7777", discover, service)
+	require.NoError(t, err)
+
+	c, err := plugin_rpc.NewClient(name, discover)
+	require.NoError(t, err)
+
+	err = c.Validate(properties)
+	require.Error(t, err)
+	require.Equal(t, validateErr.Error(), err.Error())
+
+	id, err := c.Provision(spec)
+	require.NoError(t, err)
+	require.Equal(t, instanceID, *id)
+
+	server.Stop()
+}