extracted plugin and volume extension point specific code from original plugin_mode branch

Author: progrium

daemon/container.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net"
 	"os"
 	"path"
 	"path/filepath"
@@ -30,6 +31,7 @@ import (
 	"github.com/docker/docker/pkg/networkfs/resolvconf"
 	"github.com/docker/docker/pkg/promise"
 	"github.com/docker/docker/pkg/symlink"
+	"github.com/docker/docker/plugins"
 	"github.com/docker/docker/runconfig"
 	"github.com/docker/docker/utils"
 )
@@ -1330,9 +1332,53 @@ func (container *Container) waitForStart() error {
 		return err
 	}
 
+	if container.hostConfig.Plugin {
+		if err := container.waitForPluginSock(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
+func (container *Container) waitForPluginSock() error {
+	pluginSock, err := container.getPluginSocketPath()
+	if err != nil {
+		return err
+	}
+
+	chConn := make(chan net.Conn)
+	chStop := make(chan struct{})
+	go func() {
+		log.Debugf("waiting for plugin socket at: %s", pluginSock)
+		for {
+			conn, err := net.DialTimeout("unix", pluginSock, 100*time.Millisecond)
+			// If the file doesn't exist yet, that's ok, maybe plugin hasn't created it yet
+			if err != nil {
+				select {
+				case <-chStop:
+					return
+				default:
+					continue
+				}
+			}
+			log.Debugf("got plugin socket")
+			chConn <- conn
+			return
+		}
+	}()
+
+	select {
+	case conn := <-chConn:
+		// We can close this net.Conn since the plugin system will establish it's own connection
+		conn.Close()
+		return plugins.Repo.RegisterPlugin(pluginSock)
+	case <-time.After(30 * time.Second):
+		chStop <- struct{}{}
+		return fmt.Errorf("connection to plugin sock timed out")
+	}
+}
+
 func (container *Container) allocatePort(eng *engine.Engine, port nat.Port, bindings nat.PortMap) error {
 	binding := bindings[port]
 	if container.hostConfig.PublishAllPorts && len(binding) == 0 {
@@ -1415,3 +1461,7 @@ func (container *Container) getNetworkedContainer() (*Container, error) {
 func (container *Container) Stats() (*execdriver.ResourceStats, error) {
 	return container.daemon.Stats(container)
 }
+
+func (container *Container) getPluginSocketPath() (string, error) {
+	return container.getRootResourcePath(filepath.Join("p", "plugin.sock"))
+}

daemon/daemon.go

@@ -329,6 +329,7 @@ func (daemon *Daemon) restore() error {
 		return err
 	}
 
+	// TODO: sort out plugins from normal containers and load plugins first
 	for _, v := range dir {
 		id := v.Name()
 		container, err := daemon.load(id)

daemon/volumes.go

@@ -130,7 +130,7 @@ func (container *Container) registerVolumes() {
 		if rw, exists := container.VolumesRW[path]; exists {
 			writable = rw
 		}
-		v, err := container.daemon.volumes.FindOrCreateVolume(path, writable)
+		v, err := container.daemon.volumes.FindOrCreateVolume(path, container.ID, writable)
 		if err != nil {
 			log.Debugf("error registering volume %s: %v", path, err)
 			continue
@@ -159,7 +159,7 @@ func (container *Container) parseVolumeMountConfig() (map[string]*Mount, error)
 			return nil, err
 		}
 		// Check if a volume already exists for this and use it
-		vol, err := container.daemon.volumes.FindOrCreateVolume(path, writable)
+		vol, err := container.daemon.volumes.FindOrCreateVolume(path, container.ID, writable)
 		if err != nil {
 			return nil, err
 		}
@@ -184,7 +184,7 @@ func (container *Container) parseVolumeMountConfig() (map[string]*Mount, error)
 			continue
 		}
 
-		vol, err := container.daemon.volumes.FindOrCreateVolume("", true)
+		vol, err := container.daemon.volumes.FindOrCreateVolume("", container.ID, true)
 		if err != nil {
 			return nil, err
 		}
@@ -318,6 +318,20 @@ func (container *Container) setupMounts() error {
 		mounts = append(mounts, execdriver.Mount{Source: container.HostsPath, Destination: "/etc/hosts", Writable: true, Private: true})
 	}
 
+	if container.hostConfig.Plugin {
+		// We are going to create this socket then close/unlink it so it can be
+		// bind-mounted into the container and used by the plugin process.
+		socketPath, err := container.getPluginSocketPath()
+		if err != nil {
+			return err
+		}
+		pluginDir := filepath.Dir(socketPath)
+		if err := os.MkdirAll(pluginDir, 0700); err != nil {
+			return err
+		}
+		mounts = append(mounts, execdriver.Mount{Source: pluginDir, Destination: "/var/run/docker-plugin", Writable: true, Private: true})
+	}
+
 	for _, m := range mounts {
 		if err := label.SetFileLabel(m.Source, container.MountLabel); err != nil {
 			return err

plugins/client.go

@@ -0,0 +1,64 @@
+package plugins
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/docker/docker/pkg/ioutils"
+)
+
+const pluginApiVersion = "v1"
+
+func connect(addr string) (*httputil.ClientConn, error) {
+	c, err := net.DialTimeout("unix", addr, 30*time.Second)
+	if err != nil {
+		return nil, err
+	}
+	return httputil.NewClientConn(c, nil), nil
+}
+
+func call(addr, method, path string, data interface{}) (io.ReadCloser, error) {
+	client, err := connect(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	reqBody, err := json.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Debugf("sending request for extension:\n%s", string(reqBody))
+	path = "/" + pluginApiVersion + "/" + path
+	req, err := http.NewRequest(method, path, bytes.NewBuffer(reqBody))
+	if err != nil {
+		client.Close()
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := client.Do(req)
+	if err != nil {
+		client.Close()
+		return nil, err
+	}
+
+	// FIXME: this should be better defined
+	if resp.StatusCode >= 400 {
+		return nil, fmt.Errorf("got bad status: %s", resp.Status)
+	}
+
+	return ioutils.NewReadCloserWrapper(resp.Body, func() error {
+		if err := resp.Body.Close(); err != nil {
+			return err
+		}
+		return client.Close()
+	}), nil
+}

plugins/plugin.go

@@ -0,0 +1,36 @@
+package plugins
+
+import (
+	"encoding/json"
+	"io"
+)
+
+type Plugin struct {
+	addr string
+	kind string
+}
+
+type handshakeResp struct {
+	InterestedIn []string
+	Name         string
+	Author       string
+	Org          string
+	Website      string
+}
+
+func (p *Plugin) Call(method, path string, data interface{}) (io.ReadCloser, error) {
+	path = p.kind + "/" + path
+	return call(p.addr, method, path, data)
+}
+
+func (p *Plugin) handshake() (*handshakeResp, error) {
+	// Don't use the local `call` because this shouldn't be namespaced
+	respBody, err := call(p.addr, "POST", "handshake", nil)
+	if err != nil {
+		return nil, err
+	}
+	defer respBody.Close()
+
+	var data handshakeResp
+	return &data, json.NewDecoder(respBody).Decode(&data)
+}

plugins/repository.go

@@ -0,0 +1,61 @@
+package plugins
+
+import (
+	"errors"
+	"fmt"
+)
+
+// Temporary singleton
+var Repo = NewRepository()
+
+var ErrNotRegistered = errors.New("plugin type is not registered")
+
+type Repository struct {
+	plugins map[string]Plugins
+}
+
+type Plugins []*Plugin
+
+func (repository *Repository) GetPlugins(kind string) (Plugins, error) {
+	plugins, exists := repository.plugins[kind]
+	// TODO: check whether 'kind' is a supportedPluginType
+	if !exists {
+		// If no plugins have been registered for this kind yet, that's
+		// OK. Just set and return an empty list.
+		repository.plugins[kind] = make([]*Plugin, 0)
+		return repository.plugins[kind], nil
+	}
+	return plugins, nil
+}
+
+var supportedPluginTypes = map[string]struct{}{
+	"volume": {},
+}
+
+func NewRepository() *Repository {
+	return &Repository{
+		plugins: make(map[string]Plugins),
+	}
+}
+
+func (repository *Repository) RegisterPlugin(addr string) error {
+	plugin := &Plugin{addr: addr}
+	resp, err := plugin.handshake()
+	if err != nil {
+		return fmt.Errorf("error in plugin handshake: %v", err)
+	}
+
+	for _, interest := range resp.InterestedIn {
+		if _, exists := supportedPluginTypes[interest]; !exists {
+			return fmt.Errorf("plugin type %s is not supported", interest)
+		}
+
+		if _, exists := repository.plugins[interest]; !exists {
+			repository.plugins[interest] = []*Plugin{}
+		}
+		plugin.kind = interest
+		repository.plugins[interest] = append(repository.plugins[interest], plugin)
+	}
+
+	return nil
+}

runconfig/hostconfig.go

@@ -119,6 +119,7 @@ type HostConfig struct {
 	RestartPolicy   RestartPolicy
 	SecurityOpt     []string
 	ReadonlyRootfs  bool
+	Plugin          bool
 }
 
 // This is used by the create command when you want to set both the
@@ -150,6 +151,7 @@ func ContainerHostConfigFromJob(job *engine.Job) *HostConfig {
 		IpcMode:         IpcMode(job.Getenv("IpcMode")),
 		PidMode:         PidMode(job.Getenv("PidMode")),
 		ReadonlyRootfs:  job.GetenvBool("ReadonlyRootfs"),
+		Plugin:          job.GetenvBool("Plugin"),
 	}
 
 	job.GetenvJson("LxcConf", &hostConfig.LxcConf)

runconfig/parse.go

@@ -64,6 +64,7 @@ func Parse(cmd *flag.FlagSet, args []string) (*Config, *HostConfig, *flag.FlagSe
 		flIpcMode         = cmd.String([]string{"-ipc"}, "", "Default is to create a private IPC namespace (POSIX SysV IPC) for the container\n'container:<name|id>': reuses another container shared memory, semaphores and message queues\n'host': use the host shared memory,semaphores and message queues inside the container.  Note: the host mode gives the container full access to local shared memory and is therefore considered insecure.")
 		flRestartPolicy   = cmd.String([]string{"-restart"}, "", "Restart policy to apply when a container exits (no, on-failure[:max-retry], always)")
 		flReadonlyRootfs  = cmd.Bool([]string{"-read-only"}, false, "Mount the container's root filesystem as read only")
+		flPlugin          = cmd.Bool([]string{"-plugin"}, false, "Enable plugin mode!")
 	)
 
 	cmd.Var(&flAttach, []string{"a", "-attach"}, "Attach to STDIN, STDOUT or STDERR.")
@@ -314,6 +315,7 @@ func Parse(cmd *flag.FlagSet, args []string) (*Config, *HostConfig, *flag.FlagSe
 		RestartPolicy:   restartPolicy,
 		SecurityOpt:     flSecurityOpt.GetAll(),
 		ReadonlyRootfs:  *flReadonlyRootfs,
+		Plugin:          *flPlugin,
 	}
 
 	// When allocating stdin in attached mode, close stdin at client disconnect

volumes/repository.go

@@ -1,6 +1,7 @@
 package volumes
 
 import (
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -9,9 +10,19 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/docker/daemon/graphdriver"
+	"github.com/docker/docker/plugins"
 	"github.com/docker/docker/utils"
 )
 
+type VolumeExtensionReq struct {
+	HostPath    string
+	ContainerID string
+}
+
+type VolumeExtensionResp struct {
+	ModifiedHostPath string
+}
+
 type Repository struct {
 	configPath string
 	driver     graphdriver.Driver
@@ -195,10 +206,40 @@ func (r *Repository) createNewVolumePath(id string) (string, error) {
 	return path, nil
 }
 
-func (r *Repository) FindOrCreateVolume(path string, writable bool) (*Volume, error) {
+func (r *Repository) FindOrCreateVolume(path, containerId string, writable bool) (*Volume, error) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
+	plugins, err := plugins.Repo.GetPlugins("volume")
+	if err != nil {
+		return nil, err
+	}
+
+	for _, plugin := range plugins {
+		data := VolumeExtensionReq{
+			HostPath:    path,
+			ContainerID: containerId,
+		}
+
+		resp, err := plugin.Call("POST", "volumes", data)
+		if err != nil {
+			return nil, fmt.Errorf("got error calling volume extension: %v", err)
+		}
+		defer resp.Close()
+
+		var extResp VolumeExtensionResp
+		log.Debugf("decoding volume extension response")
+		if err := json.NewDecoder(resp).Decode(&extResp); err != nil {
+			return nil, err
+		}
+
+		// Use the path provided by the extension instead of creating one
+		if extResp.ModifiedHostPath != "" {
+			log.Debugf("using modified host path for volume extension")
+			path = extResp.ModifiedHostPath
+		}
+	}
+
 	if path == "" {
 		return r.newVolume(path, writable)
 	}