Move plugin client creation to the extension point

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

api/swagger.yaml

@@ -1859,6 +1859,13 @@ definitions:
                 type: "string"
                 x-nullable: false
                 example: "plugins.sock"
+              ProtocolScheme:
+                type: "string"
+                example: "some.protocol/v1.0"
+                description: "Protocol to use for clients connecting to the plugin."
+                enum:
+                  - ""
+                  - "moby.plugins.http/v1"
           Entrypoint:
             type: "array"
             items:

api/types/plugin.go

@@ -121,6 +121,9 @@ type PluginConfigArgs struct {
 // swagger:model PluginConfigInterface
 type PluginConfigInterface struct {
 
+	// Protocol to use for clients connecting to the plugin.
+	ProtocolScheme string `json:"ProtocolScheme,omitempty"`
+
 	// socket
 	// Required: true
 	Socket string `json:"Socket"`

daemon/graphdriver/plugin.go

@@ -5,7 +5,9 @@ import (
 	"path/filepath"
 
 	"github.com/docker/docker/pkg/plugingetter"
+	"github.com/docker/docker/pkg/plugins"
 	"github.com/docker/docker/plugin/v2"
+	"github.com/pkg/errors"
 )
 
 func lookupPlugin(name string, pg plugingetter.PluginGetter, config Options) (Driver, error) {
@@ -28,6 +30,22 @@ func newPluginDriver(name string, pl plugingetter.CompatPlugin, config Options)
 			}
 		}
 	}
-	proxy := &graphDriverProxy{name, pl, Capabilities{}}
+
+	var proxy *graphDriverProxy
+
+	pa, ok := pl.(plugingetter.PluginAddr)
+	if !ok {
+		proxy = &graphDriverProxy{name, pl, Capabilities{}, pl.Client()}
+	} else {
+		if pa.Protocol() != plugins.ProtocolSchemeHTTPV1 {
+			return nil, errors.Errorf("plugin protocol not supported: %s", pa.Protocol())
+		}
+		addr := pa.Addr()
+		client, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, pa.Timeout())
+		if err != nil {
+			return nil, errors.Wrap(err, "error creating plugin client")
+		}
+		proxy = &graphDriverProxy{name, pl, Capabilities{}, client}
+	}
 	return proxy, proxy.Init(filepath.Join(home, name), config.DriverOptions, config.UIDMaps, config.GIDMaps)
 }

daemon/graphdriver/proxy.go

@@ -16,6 +16,7 @@ type graphDriverProxy struct {
 	name   string
 	p      plugingetter.CompatPlugin
 	caps   Capabilities
+	client *plugins.Client
 }
 
 type graphDriverRequest struct {
@@ -57,7 +58,7 @@ func (d *graphDriverProxy) Init(home string, opts []string, uidMaps, gidMaps []i
 		GIDMaps: gidMaps,
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Init", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Init", args, &ret); err != nil {
 		return err
 	}
 	if ret.Err != "" {
@@ -74,7 +75,7 @@ func (d *graphDriverProxy) Init(home string, opts []string, uidMaps, gidMaps []i
 func (d *graphDriverProxy) fetchCaps() (Capabilities, error) {
 	args := &graphDriverRequest{}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Capabilities", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Capabilities", args, &ret); err != nil {
 		if !plugins.IsNotFound(err) {
 			return Capabilities{}, err
 		}
@@ -108,7 +109,7 @@ func (d *graphDriverProxy) create(method, id, parent string, opts *CreateOpts) e
 		args.StorageOpt = opts.StorageOpt
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call(method, args, &ret); err != nil {
+	if err := d.client.Call(method, args, &ret); err != nil {
 		return err
 	}
 	if ret.Err != "" {
@@ -120,7 +121,7 @@ func (d *graphDriverProxy) create(method, id, parent string, opts *CreateOpts) e
 func (d *graphDriverProxy) Remove(id string) error {
 	args := &graphDriverRequest{ID: id}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Remove", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Remove", args, &ret); err != nil {
 		return err
 	}
 	if ret.Err != "" {
@@ -135,7 +136,7 @@ func (d *graphDriverProxy) Get(id, mountLabel string) (containerfs.ContainerFS,
 		MountLabel: mountLabel,
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Get", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Get", args, &ret); err != nil {
 		return nil, err
 	}
 	var err error
@@ -148,7 +149,7 @@ func (d *graphDriverProxy) Get(id, mountLabel string) (containerfs.ContainerFS,
 func (d *graphDriverProxy) Put(id string) error {
 	args := &graphDriverRequest{ID: id}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Put", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Put", args, &ret); err != nil {
 		return err
 	}
 	if ret.Err != "" {
@@ -160,7 +161,7 @@ func (d *graphDriverProxy) Put(id string) error {
 func (d *graphDriverProxy) Exists(id string) bool {
 	args := &graphDriverRequest{ID: id}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Exists", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Exists", args, &ret); err != nil {
 		return false
 	}
 	return ret.Exists
@@ -169,7 +170,7 @@ func (d *graphDriverProxy) Exists(id string) bool {
 func (d *graphDriverProxy) Status() [][2]string {
 	args := &graphDriverRequest{}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Status", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Status", args, &ret); err != nil {
 		return nil
 	}
 	return ret.Status
@@ -180,7 +181,7 @@ func (d *graphDriverProxy) GetMetadata(id string) (map[string]string, error) {
 		ID: id,
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.GetMetadata", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.GetMetadata", args, &ret); err != nil {
 		return nil, err
 	}
 	if ret.Err != "" {
@@ -199,7 +200,7 @@ func (d *graphDriverProxy) Cleanup() error {
 
 	args := &graphDriverRequest{}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Cleanup", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Cleanup", args, &ret); err != nil {
 		return nil
 	}
 	if ret.Err != "" {
@@ -213,7 +214,7 @@ func (d *graphDriverProxy) Diff(id, parent string) (io.ReadCloser, error) {
 		ID:     id,
 		Parent: parent,
 	}
-	body, err := d.p.Client().Stream("GraphDriver.Diff", args)
+	body, err := d.client.Stream("GraphDriver.Diff", args)
 	if err != nil {
 		return nil, err
 	}
@@ -226,7 +227,7 @@ func (d *graphDriverProxy) Changes(id, parent string) ([]archive.Change, error)
 		Parent: parent,
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.Changes", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.Changes", args, &ret); err != nil {
 		return nil, err
 	}
 	if ret.Err != "" {
@@ -238,7 +239,7 @@ func (d *graphDriverProxy) Changes(id, parent string) ([]archive.Change, error)
 
 func (d *graphDriverProxy) ApplyDiff(id, parent string, diff io.Reader) (int64, error) {
 	var ret graphDriverResponse
-	if err := d.p.Client().SendFile(fmt.Sprintf("GraphDriver.ApplyDiff?id=%s&parent=%s", id, parent), diff, &ret); err != nil {
+	if err := d.client.SendFile(fmt.Sprintf("GraphDriver.ApplyDiff?id=%s&parent=%s", id, parent), diff, &ret); err != nil {
 		return -1, err
 	}
 	if ret.Err != "" {
@@ -253,7 +254,7 @@ func (d *graphDriverProxy) DiffSize(id, parent string) (int64, error) {
 		Parent: parent,
 	}
 	var ret graphDriverResponse
-	if err := d.p.Client().Call("GraphDriver.DiffSize", args, &ret); err != nil {
+	if err := d.client.Call("GraphDriver.DiffSize", args, &ret); err != nil {
 		return -1, err
 	}
 	if ret.Err != "" {

daemon/logger/plugin.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/docker/docker/api/types/plugins/logdriver"
 	getter "github.com/docker/docker/pkg/plugingetter"
+	"github.com/docker/docker/pkg/plugins"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/pkg/errors"
 )
@@ -37,11 +38,32 @@ func getPlugin(name string, mode int) (Creator, error) {
 		return nil, fmt.Errorf("error looking up logging plugin %s: %v", name, err)
 	}
 
-	d := &logPluginProxy{p.Client()}
+	client, err := makePluginClient(p)
-	return makePluginCreator(name, d, p.ScopedPath), nil
+	if err != nil {
+		return nil, err
+	}
+	return makePluginCreator(name, client, p.ScopedPath), nil
 }
 
-func makePluginCreator(name string, l *logPluginProxy, scopePath func(s string) string) Creator {
+func makePluginClient(p getter.CompatPlugin) (logPlugin, error) {
+	pa, ok := p.(getter.PluginAddr)
+	if !ok {
+		return &logPluginProxy{p.Client()}, nil
+	}
+
+	if pa.Protocol() != plugins.ProtocolSchemeHTTPV1 {
+		return nil, errors.Errorf("plugin protocol not supported: %s", p)
+	}
+
+	addr := pa.Addr()
+	c, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, pa.Timeout())
+	if err != nil {
+		return nil, errors.Wrap(err, "error making plugin client")
+	}
+	return &logPluginProxy{c}, nil
+}
+
+func makePluginCreator(name string, l logPlugin, scopePath func(s string) string) Creator {
 	return func(logCtx Info) (logger Logger, err error) {
 		defer func() {
 			if err != nil {

daemon/metrics.go

@@ -4,6 +4,7 @@ import (
 	"sync"
 
 	"github.com/docker/docker/pkg/plugingetter"
+	"github.com/docker/docker/pkg/plugins"
 	"github.com/docker/go-metrics"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -118,7 +119,15 @@ func (d *Daemon) cleanupMetricsPlugins() {
 		p := plugin
 		go func() {
 			defer wg.Done()
-			pluginStopMetricsCollection(p)
+
+			adapter, err := makePluginAdapter(p)
+			if err != nil {
+				logrus.WithError(err).WithField("plugin", p.Name()).Error("Error creating metrics plugin adapater")
+				return
+			}
+			if err := adapter.StopMetrics(); err != nil {
+				logrus.WithError(err).WithField("plugin", p.Name()).Error("Error stopping plugin metrics collection")
+			}
 		}()
 	}
 	wg.Wait()
@@ -128,12 +137,39 @@ func (d *Daemon) cleanupMetricsPlugins() {
 	}
 }
 
-func pluginStartMetricsCollection(p plugingetter.CompatPlugin) error {
+type metricsPlugin interface {
+	StartMetrics() error
+	StopMetrics() error
+}
+
+func makePluginAdapter(p plugingetter.CompatPlugin) (metricsPlugin, error) {
+	pa, ok := p.(plugingetter.PluginAddr)
+	if !ok {
+		return &metricsPluginAdapter{p.Client(), p.Name()}, nil
+	}
+	if pa.Protocol() != plugins.ProtocolSchemeHTTPV1 {
+		return nil, errors.Errorf("plugin protocol not supported: %s", pa.Protocol())
+	}
+
+	addr := pa.Addr()
+	client, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, pa.Timeout())
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating metrics plugin client")
+	}
+	return &metricsPluginAdapter{client, p.Name()}, nil
+}
+
+type metricsPluginAdapter struct {
+	c    *plugins.Client
+	name string
+}
+
+func (a *metricsPluginAdapter) StartMetrics() error {
 	type metricsPluginResponse struct {
 		Err string
 	}
 	var res metricsPluginResponse
-	if err := p.Client().Call(metricsPluginType+".StartMetrics", nil, &res); err != nil {
+	if err := a.c.Call(metricsPluginType+".StartMetrics", nil, &res); err != nil {
 		return errors.Wrap(err, "could not start metrics plugin")
 	}
 	if res.Err != "" {
@@ -142,8 +178,9 @@ func pluginStartMetricsCollection(p plugingetter.CompatPlugin) error {
 	return nil
 }
 
-func pluginStopMetricsCollection(p plugingetter.CompatPlugin) {
+func (a *metricsPluginAdapter) StopMetrics() error {
-	if err := p.Client().Call(metricsPluginType+".StopMetrics", nil, nil); err != nil {
+	if err := a.c.Call(metricsPluginType+".StopMetrics", nil, nil); err != nil {
-		logrus.WithError(err).WithField("name", p.Name()).Error("error stopping metrics collector")
+		return errors.Wrap(err, "error stopping metrics collector")
 	}
+	return nil
 }

daemon/metrics_unix.go

@@ -49,8 +49,12 @@ func registerMetricsPluginCallback(store *plugin.Store, sockPath string) {
 			return
 		}
 
-		if err := pluginStartMetricsCollection(p); err != nil {
+		adapter, err := makePluginAdapter(p)
-			logrus.WithError(err).WithField("name", name).Error("error while initializing metrics plugin")
+		if err != nil {
+			logrus.WithError(err).WithField("plugin", p.Name()).Error("Error creating plugin adapater")
+		}
+		if err := adapter.StartMetrics(); err != nil {
+			logrus.WithError(err).WithField("plugin", p.Name()).Error("Error starting metrics collector plugin")
 		}
 	})
 }

pkg/plugingetter/getter.go

@@ -1,6 +1,9 @@
 package plugingetter // import "github.com/docker/docker/pkg/plugingetter"
 
 import (
+	"net"
+	"time"
+
 	"github.com/docker/docker/pkg/plugins"
 )
 
@@ -21,6 +24,14 @@ type CompatPlugin interface {
 	IsV1() bool
 }
 
+// PluginAddr is a plugin that exposes the socket address for creating custom clients rather than the built-in `*plugins.Client`
+type PluginAddr interface {
+	CompatPlugin
+	Addr() net.Addr
+	Timeout() time.Duration
+	Protocol() string
+}
+
 // CountedPlugin is a plugin which is reference counted.
 type CountedPlugin interface {
 	Acquire()

pkg/plugins/plugins.go

@@ -31,6 +31,9 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+// ProtocolSchemeHTTPV1 is the name of the protocol used for interacting with plugins using this package.
+const ProtocolSchemeHTTPV1 = "moby.plugins.http/v1"
+
 var (
 	// ErrNotImplements is returned if the plugin does not implement the requested driver.
 	ErrNotImplements = errors.New("Plugin does not implement the requested driver")
@@ -88,6 +91,11 @@ func (p *Plugin) Client() *Client {
 	return p.client
 }
 
+// Protocol returns the protocol name/version used for plugins in this package.
+func (p *Plugin) Protocol() string {
+	return ProtocolSchemeHTTPV1
+}
+
 // IsV1 returns true for V1 plugins and false otherwise.
 func (p *Plugin) IsV1() bool {
 	return true

plugin/manager_linux.go

@@ -71,7 +71,12 @@ func (pm *Manager) enable(p *v2.Plugin, c *controller, force bool) error {
 
 func (pm *Manager) pluginPostStart(p *v2.Plugin, c *controller) error {
 	sockAddr := filepath.Join(pm.config.ExecRoot, p.GetID(), p.GetSocket())
-	client, err := plugins.NewClientWithTimeout("unix://"+sockAddr, nil, time.Duration(c.timeoutInSecs)*time.Second)
+	p.SetTimeout(time.Duration(c.timeoutInSecs) * time.Second)
+	addr := &net.UnixAddr{Net: "unix", Name: sockAddr}
+	p.SetAddr(addr)
+
+	if p.Protocol() == plugins.ProtocolSchemeHTTPV1 {
+		client, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, p.Timeout())
 		if err != nil {
 			c.restart = false
 			shutdownPlugin(p, c, pm.executor)
@@ -79,6 +84,7 @@ func (pm *Manager) pluginPostStart(p *v2.Plugin, c *controller) error {
 		}
 
 		p.SetPClient(client)
+	}
 
 	// Initial sleep before net Dial to allow plugin to listen on socket.
 	time.Sleep(500 * time.Millisecond)

plugin/v2/plugin.go

@@ -2,9 +2,11 @@ package v2 // import "github.com/docker/docker/plugin/v2"
 
 import (
 	"fmt"
+	"net"
 	"path/filepath"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/pkg/plugingetter"
@@ -27,6 +29,8 @@ type Plugin struct {
 	modifyRuntimeSpec func(*specs.Spec)
 
 	SwarmServiceID string
+	timeout        time.Duration
+	addr           net.Addr
 }
 
 const defaultPluginRuntimeDestination = "/run/docker/plugins"
@@ -50,6 +54,7 @@ func (p *Plugin) ScopedPath(s string) string {
 }
 
 // Client returns the plugin client.
+// Deprecated: use p.Addr() and manually create the client
 func (p *Plugin) Client() *plugins.Client {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
@@ -58,6 +63,7 @@ func (p *Plugin) Client() *plugins.Client {
 }
 
 // SetPClient set the plugin client.
+// Deprecated: Hardcoded plugin client is deprecated
 func (p *Plugin) SetPClient(client *plugins.Client) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -264,3 +270,42 @@ func (p *Plugin) SetSpecOptModifier(f func(*specs.Spec)) {
 	p.modifyRuntimeSpec = f
 	p.mu.Unlock()
 }
+
+// Timeout gets the currently configured connection timeout.
+// This should be used when dialing the plugin.
+func (p *Plugin) Timeout() time.Duration {
+	p.mu.RLock()
+	t := p.timeout
+	p.mu.RUnlock()
+	return t
+}
+
+// SetTimeout sets the timeout to use for dialing.
+func (p *Plugin) SetTimeout(t time.Duration) {
+	p.mu.Lock()
+	p.timeout = t
+	p.mu.Unlock()
+}
+
+// Addr returns the net.Addr to use to connect to the plugin socket
+func (p *Plugin) Addr() net.Addr {
+	p.mu.RLock()
+	addr := p.addr
+	p.mu.RUnlock()
+	return addr
+}
+
+// SetAddr sets the plugin address which can be used for dialing the plugin.
+func (p *Plugin) SetAddr(addr net.Addr) {
+	p.mu.Lock()
+	p.addr = addr
+	p.mu.Unlock()
+}
+
+// Protocol is the protocol that should be used for interacting with the plugin.
+func (p *Plugin) Protocol() string {
+	if p.PluginObj.Config.Interface.ProtocolScheme != "" {
+		return p.PluginObj.Config.Interface.ProtocolScheme
+	}
+	return plugins.ProtocolSchemeHTTPV1
+}

plugin/v2/plugin_unsupported.go

@@ -5,7 +5,7 @@ package v2 // import "github.com/docker/docker/plugin/v2"
 import (
 	"errors"
 
-	specs "github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/opencontainers/runtime-spec/specs-go"
 )
 
 // InitSpec creates an OCI spec from the plugin's config.

volume/drivers/adapter.go

@@ -17,7 +17,7 @@ type volumeDriverAdapter struct {
 	name         string
 	scopePath    func(s string) string
 	capabilities *volume.Capability
-	proxy        *volumeDriverProxy
+	proxy        volumeDriver
 }
 
 func (a *volumeDriverAdapter) Name() string {
@@ -114,7 +114,7 @@ func (a *volumeDriverAdapter) getCapabilities() volume.Capability {
 }
 
 type volumeAdapter struct {
-	proxy      *volumeDriverProxy
+	proxy      volumeDriver
 	name       string
 	scopePath  func(string) string
 	driverName string

volume/drivers/extpoint.go

@@ -10,6 +10,7 @@ import (
 	"github.com/docker/docker/errdefs"
 	"github.com/docker/docker/pkg/locker"
 	getter "github.com/docker/docker/pkg/plugingetter"
+	"github.com/docker/docker/pkg/plugins"
 	"github.com/docker/docker/volume"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -17,12 +18,6 @@ import (
 
 const extName = "VolumeDriver"
 
-// NewVolumeDriver returns a driver has the given name mapped on the given client.
-func NewVolumeDriver(name string, scopePath func(string) string, c client) volume.Driver {
-	proxy := &volumeDriverProxy{c}
-	return &volumeDriverAdapter{name: name, scopePath: scopePath, proxy: proxy}
-}
-
 // volumeDriver defines the available functions that volume plugins must implement.
 // This interface is only defined to generate the proxy objects.
 // It's not intended to be public or reused.
@@ -93,7 +88,10 @@ func (s *Store) lookup(name string, mode int) (volume.Driver, error) {
 			return nil, errors.Wrap(err, "error looking up volume plugin "+name)
 		}
 
-		d := NewVolumeDriver(p.Name(), p.ScopedPath, p.Client())
+		d, err := makePluginAdapter(p)
+		if err != nil {
+			return nil, errors.Wrap(err, "error making plugin client")
+		}
 		if err := validateDriver(d); err != nil {
 			if mode > 0 {
 				// Undo any reference count changes from the initial `Get`
@@ -201,7 +199,10 @@ func (s *Store) GetAllDrivers() ([]volume.Driver, error) {
 			continue
 		}
 
-		ext := NewVolumeDriver(name, p.ScopedPath, p.Client())
+		ext, err := makePluginAdapter(p)
+		if err != nil {
+			return nil, errors.Wrap(err, "error making plugin client")
+		}
 		if p.IsV1() {
 			s.extensions[name] = ext
 		}
@@ -209,3 +210,22 @@ func (s *Store) GetAllDrivers() ([]volume.Driver, error) {
 	}
 	return ds, nil
 }
+
+func makePluginAdapter(p getter.CompatPlugin) (*volumeDriverAdapter, error) {
+	pa, ok := p.(getter.PluginAddr)
+	if !ok {
+		return &volumeDriverAdapter{name: p.Name(), scopePath: p.ScopedPath, proxy: &volumeDriverProxy{p.Client()}}, nil
+	}
+
+	if pa.Protocol() != plugins.ProtocolSchemeHTTPV1 {
+		return nil, errors.Errorf("plugin protocol not supported: %s", p)
+	}
+
+	addr := pa.Addr()
+	client, err := plugins.NewClientWithTimeout(addr.Network()+"://"+addr.String(), nil, pa.Timeout())
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating plugin client")
+	}
+
+	return &volumeDriverAdapter{name: p.Name(), scopePath: p.ScopedPath, proxy: &volumeDriverProxy{client}}, nil
+}