Move storagedriver package to registry/storage/driver

This change is slightly more complex than previous package maves in that the package name changed. To address this, we simply always reference the package driver as storagedriver to avoid compatbility issues with existing code. While unfortunate, this can be cleaned up over time. Signed-off-by: Stephen J Day <stephen.day@docker.com>
2015-02-10 18:14:23 -08:00 · 2015-02-10 18:14:23 -08:00 · 6e4f9a2e3e
commit 6e4f9a2e3e
parent 71e7ac33ca
47 changed files with 5674 additions and 24 deletions
--- a/docs/storage/driver/ipc/client.go
+++ b/docs/storage/driver/ipc/client.go
@ -0,0 +1,454 @@
+// +build ignore
+
+package ipc
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"os"
+	"os/exec"
+	"syscall"
+
+	storagedriver "github.com/docker/distribution/registry/storage/driver"
+	"github.com/docker/libchan"
+	"github.com/docker/libchan/spdy"
+)
+
+// StorageDriverExecutablePrefix is the prefix which the IPC storage driver
+// loader expects driver executables to begin with. For example, the s3 driver
+// should be named "registry-storagedriver-s3".
+const StorageDriverExecutablePrefix = "registry-storagedriver-"
+
+// StorageDriverClient is a storagedriver.StorageDriver implementation using a
+// managed child process communicating over IPC using libchan with a unix domain
+// socket
+type StorageDriverClient struct {
+	subprocess *exec.Cmd
+	exitChan   chan error
+	exitErr    error
+	stopChan   chan struct{}
+	socket     *os.File
+	transport  *spdy.Transport
+	sender     libchan.Sender
+	version    storagedriver.Version
+}
+
+// NewDriverClient constructs a new out-of-process storage driver using the
+// driver name and configuration parameters
+// A user must call Start on this driver client before remote method calls can
+// be made
+//
+// Looks for drivers in the following locations in order:
+// - Storage drivers directory (to be determined, yet not implemented)
+// - $GOPATH/bin
+// - $PATH
+func NewDriverClient(name string, parameters map[string]string) (*StorageDriverClient, error) {
+	paramsBytes, err := json.Marshal(parameters)
+	if err != nil {
+		return nil, err
+	}
+
+	driverExecName := StorageDriverExecutablePrefix + name
+	driverPath, err := exec.LookPath(driverExecName)
+	if err != nil {
+		return nil, err
+	}
+
+	command := exec.Command(driverPath, string(paramsBytes))
+
+	return &StorageDriverClient{
+		subprocess: command,
+	}, nil
+}
+
+// Start starts the designated child process storage driver and binds a socket
+// to this process for IPC method calls
+func (driver *StorageDriverClient) Start() error {
+	driver.exitErr = nil
+	driver.exitChan = make(chan error)
+	driver.stopChan = make(chan struct{})
+
+	fileDescriptors, err := syscall.Socketpair(syscall.AF_LOCAL, syscall.SOCK_STREAM, 0)
+	if err != nil {
+		return err
+	}
+
+	childSocket := os.NewFile(uintptr(fileDescriptors[0]), "childSocket")
+	driver.socket = os.NewFile(uintptr(fileDescriptors[1]), "parentSocket")
+
+	driver.subprocess.Stdout = os.Stdout
+	driver.subprocess.Stderr = os.Stderr
+	driver.subprocess.ExtraFiles = []*os.File{childSocket}
+
+	if err = driver.subprocess.Start(); err != nil {
+		driver.Stop()
+		return err
+	}
+
+	go driver.handleSubprocessExit()
+
+	if err = childSocket.Close(); err != nil {
+		driver.Stop()
+		return err
+	}
+
+	connection, err := net.FileConn(driver.socket)
+	if err != nil {
+		driver.Stop()
+		return err
+	}
+	driver.transport, err = spdy.NewClientTransport(connection)
+	if err != nil {
+		driver.Stop()
+		return err
+	}
+	driver.sender, err = driver.transport.NewSendChannel()
+	if err != nil {
+		driver.Stop()
+		return err
+	}
+
+	// Check the driver's version to determine compatibility
+	receiver, remoteSender := libchan.Pipe()
+	err = driver.sender.Send(&Request{Type: "Version", ResponseChannel: remoteSender})
+	if err != nil {
+		driver.Stop()
+		return err
+	}
+
+	var response VersionResponse
+	err = receiver.Receive(&response)
+	if err != nil {
+		driver.Stop()
+		return err
+	}
+
+	if response.Error != nil {
+		return response.Error.Unwrap()
+	}
+
+	driver.version = response.Version
+
+	if driver.version.Major() != storagedriver.CurrentVersion.Major() || driver.version.Minor() > storagedriver.CurrentVersion.Minor() {
+		return IncompatibleVersionError{driver.version}
+	}
+
+	return nil
+}
+
+// Stop stops the child process storage driver
+// storagedriver.StorageDriver methods called after Stop will fail
+func (driver *StorageDriverClient) Stop() error {
+	var closeSenderErr, closeTransportErr, closeSocketErr, killErr error
+
+	if driver.sender != nil {
+		closeSenderErr = driver.sender.Close()
+	}
+	if driver.transport != nil {
+		closeTransportErr = driver.transport.Close()
+	}
+	if driver.socket != nil {
+		closeSocketErr = driver.socket.Close()
+	}
+	if driver.subprocess != nil {
+		killErr = driver.subprocess.Process.Kill()
+	}
+	if driver.stopChan != nil {
+		close(driver.stopChan)
+	}
+
+	if closeSenderErr != nil {
+		return closeSenderErr
+	} else if closeTransportErr != nil {
+		return closeTransportErr
+	} else if closeSocketErr != nil {
+		return closeSocketErr
+	}
+
+	return killErr
+}
+
+// Implement the storagedriver.StorageDriver interface over IPC
+
+// GetContent retrieves the content stored at "path" as a []byte.
+func (driver *StorageDriverClient) GetContent(path string) ([]byte, error) {
+	if err := driver.exited(); err != nil {
+		return nil, err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+
+	params := map[string]interface{}{"Path": path}
+	err := driver.sender.Send(&Request{Type: "GetContent", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return nil, err
+	}
+
+	response := new(ReadStreamResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, response.Error.Unwrap()
+	}
+
+	defer response.Reader.Close()
+	contents, err := ioutil.ReadAll(response.Reader)
+	if err != nil {
+		return nil, err
+	}
+	return contents, nil
+}
+
+// PutContent stores the []byte content at a location designated by "path".
+func (driver *StorageDriverClient) PutContent(path string, contents []byte) error {
+	if err := driver.exited(); err != nil {
+		return err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+
+	params := map[string]interface{}{"Path": path, "Reader": ioutil.NopCloser(bytes.NewReader(contents))}
+	err := driver.sender.Send(&Request{Type: "PutContent", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return err
+	}
+
+	response := new(WriteStreamResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return err
+	}
+
+	if response.Error != nil {
+		return response.Error.Unwrap()
+	}
+
+	return nil
+}
+
+// ReadStream retrieves an io.ReadCloser for the content stored at "path" with a
+// given byte offset.
+func (driver *StorageDriverClient) ReadStream(path string, offset int64) (io.ReadCloser, error) {
+	if err := driver.exited(); err != nil {
+		return nil, err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"Path": path, "Offset": offset}
+	err := driver.sender.Send(&Request{Type: "ReadStream", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return nil, err
+	}
+
+	response := new(ReadStreamResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, response.Error.Unwrap()
+	}
+
+	return response.Reader, nil
+}
+
+// WriteStream stores the contents of the provided io.ReadCloser at a location
+// designated by the given path.
+func (driver *StorageDriverClient) WriteStream(path string, offset, size int64, reader io.ReadCloser) error {
+	if err := driver.exited(); err != nil {
+		return err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"Path": path, "Offset": offset, "Size": size, "Reader": reader}
+	err := driver.sender.Send(&Request{Type: "WriteStream", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return err
+	}
+
+	response := new(WriteStreamResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return err
+	}
+
+	if response.Error != nil {
+		return response.Error.Unwrap()
+	}
+
+	return nil
+}
+
+// CurrentSize retrieves the curernt size in bytes of the object at the given
+// path.
+func (driver *StorageDriverClient) CurrentSize(path string) (uint64, error) {
+	if err := driver.exited(); err != nil {
+		return 0, err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"Path": path}
+	err := driver.sender.Send(&Request{Type: "CurrentSize", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return 0, err
+	}
+
+	response := new(CurrentSizeResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return 0, err
+	}
+
+	if response.Error != nil {
+		return 0, response.Error.Unwrap()
+	}
+
+	return response.Position, nil
+}
+
+// List returns a list of the objects that are direct descendants of the given
+// path.
+func (driver *StorageDriverClient) List(path string) ([]string, error) {
+	if err := driver.exited(); err != nil {
+		return nil, err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"Path": path}
+	err := driver.sender.Send(&Request{Type: "List", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return nil, err
+	}
+
+	response := new(ListResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.Error != nil {
+		return nil, response.Error.Unwrap()
+	}
+
+	return response.Keys, nil
+}
+
+// Move moves an object stored at sourcePath to destPath, removing the original
+// object.
+func (driver *StorageDriverClient) Move(sourcePath string, destPath string) error {
+	if err := driver.exited(); err != nil {
+		return err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"SourcePath": sourcePath, "DestPath": destPath}
+	err := driver.sender.Send(&Request{Type: "Move", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return err
+	}
+
+	response := new(MoveResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return err
+	}
+
+	if response.Error != nil {
+		return response.Error.Unwrap()
+	}
+
+	return nil
+}
+
+// Delete recursively deletes all objects stored at "path" and its subpaths.
+func (driver *StorageDriverClient) Delete(path string) error {
+	if err := driver.exited(); err != nil {
+		return err
+	}
+
+	receiver, remoteSender := libchan.Pipe()
+	params := map[string]interface{}{"Path": path}
+	err := driver.sender.Send(&Request{Type: "Delete", Parameters: params, ResponseChannel: remoteSender})
+	if err != nil {
+		return err
+	}
+
+	response := new(DeleteResponse)
+	err = driver.receiveResponse(receiver, response)
+	if err != nil {
+		return err
+	}
+
+	if response.Error != nil {
+		return response.Error.Unwrap()
+	}
+
+	return nil
+}
+
+// handleSubprocessExit populates the exit channel until we have explicitly
+// stopped the storage driver subprocess
+// Requests can select on driver.exitChan and response receiving and not hang if
+// the process exits
+func (driver *StorageDriverClient) handleSubprocessExit() {
+	exitErr := driver.subprocess.Wait()
+	if exitErr == nil {
+		exitErr = fmt.Errorf("Storage driver subprocess already exited cleanly")
+	} else {
+		exitErr = fmt.Errorf("Storage driver subprocess exited with error: %s", exitErr)
+	}
+
+	driver.exitErr = exitErr
+
+	for {
+		select {
+		case driver.exitChan <- exitErr:
+		case <-driver.stopChan:
+			close(driver.exitChan)
+			return
+		}
+	}
+}
+
+// receiveResponse populates the response value with the next result from the
+// given receiver, or returns an error if receiving failed or the driver has
+// stopped
+func (driver *StorageDriverClient) receiveResponse(receiver libchan.Receiver, response interface{}) error {
+	receiveChan := make(chan error, 1)
+	go func(receiver libchan.Receiver, receiveChan chan<- error) {
+		receiveChan <- receiver.Receive(response)
+	}(receiver, receiveChan)
+
+	var err error
+	var ok bool
+	select {
+	case err = <-receiveChan:
+	case err, ok = <-driver.exitChan:
+		if !ok {
+			err = driver.exitErr
+		}
+	}
+
+	return err
+}
+
+// exited returns an exit error if the driver has exited or nil otherwise
+func (driver *StorageDriverClient) exited() error {
+	select {
+	case err, ok := <-driver.exitChan:
+		if !ok {
+			return driver.exitErr
+		}
+		return err
+	default:
+		return nil
+	}
+}
--- a/docs/storage/driver/ipc/ipc.go
+++ b/docs/storage/driver/ipc/ipc.go
@ -0,0 +1,148 @@
+// +build ignore
+
+package ipc
+
+import (
+	"fmt"
+	"io"
+	"reflect"
+
+	storagedriver "github.com/docker/distribution/registry/storage/driver"
+	"github.com/docker/libchan"
+)
+
+// StorageDriver is the interface which IPC storage drivers must implement. As external storage
+// drivers may be defined to use a different version of the storagedriver.StorageDriver interface,
+// we use an additional version check to determine compatiblity.
+type StorageDriver interface {
+	// Version returns the storagedriver.StorageDriver interface version which this storage driver
+	// implements, which is used to determine driver compatibility
+	Version() (storagedriver.Version, error)
+}
+
+// IncompatibleVersionError is returned when a storage driver is using an incompatible version of
+// the storagedriver.StorageDriver api
+type IncompatibleVersionError struct {
+	version storagedriver.Version
+}
+
+func (e IncompatibleVersionError) Error() string {
+	return fmt.Sprintf("Incompatible storage driver version: %s", e.version)
+}
+
+// Request defines a remote method call request
+// A return value struct is to be sent over the ResponseChannel
+type Request struct {
+	Type            string                 `codec:",omitempty"`
+	Parameters      map[string]interface{} `codec:",omitempty"`
+	ResponseChannel libchan.Sender         `codec:",omitempty"`
+}
+
+// ResponseError is a serializable error type.
+// The Type and Parameters may be used to reconstruct the same error on the
+// client side, falling back to using the Type and Message if this cannot be
+// done.
+type ResponseError struct {
+	Type       string                 `codec:",omitempty"`
+	Message    string                 `codec:",omitempty"`
+	Parameters map[string]interface{} `codec:",omitempty"`
+}
+
+// WrapError wraps an error in a serializable struct containing the error's type
+// and message.
+func WrapError(err error) *ResponseError {
+	if err == nil {
+		return nil
+	}
+	v := reflect.ValueOf(err)
+	re := ResponseError{
+		Type:    v.Type().String(),
+		Message: err.Error(),
+	}
+
+	if v.Kind() == reflect.Struct {
+		re.Parameters = make(map[string]interface{})
+		for i := 0; i < v.NumField(); i++ {
+			field := v.Type().Field(i)
+			re.Parameters[field.Name] = v.Field(i).Interface()
+		}
+	}
+	return &re
+}
+
+// Unwrap returns the underlying error if it can be reconstructed, or the
+// original ResponseError otherwise.
+func (err *ResponseError) Unwrap() error {
+	var errVal reflect.Value
+	var zeroVal reflect.Value
+
+	switch err.Type {
+	case "storagedriver.PathNotFoundError":
+		errVal = reflect.ValueOf(&storagedriver.PathNotFoundError{})
+	case "storagedriver.InvalidOffsetError":
+		errVal = reflect.ValueOf(&storagedriver.InvalidOffsetError{})
+	}
+	if errVal == zeroVal {
+		return err
+	}
+
+	for k, v := range err.Parameters {
+		fieldVal := errVal.Elem().FieldByName(k)
+		if fieldVal == zeroVal {
+			return err
+		}
+		fieldVal.Set(reflect.ValueOf(v))
+	}
+
+	if unwrapped, ok := errVal.Elem().Interface().(error); ok {
+		return unwrapped
+	}
+
+	return err
+
+}
+
+func (err *ResponseError) Error() string {
+	return fmt.Sprintf("%s: %s", err.Type, err.Message)
+}
+
+// IPC method call response object definitions
+
+// VersionResponse is a response for a Version request
+type VersionResponse struct {
+	Version storagedriver.Version `codec:",omitempty"`
+	Error   *ResponseError        `codec:",omitempty"`
+}
+
+// ReadStreamResponse is a response for a ReadStream request
+type ReadStreamResponse struct {
+	Reader io.ReadCloser  `codec:",omitempty"`
+	Error  *ResponseError `codec:",omitempty"`
+}
+
+// WriteStreamResponse is a response for a WriteStream request
+type WriteStreamResponse struct {
+	Error *ResponseError `codec:",omitempty"`
+}
+
+// CurrentSizeResponse is a response for a CurrentSize request
+type CurrentSizeResponse struct {
+	Position uint64         `codec:",omitempty"`
+	Error    *ResponseError `codec:",omitempty"`
+}
+
+// ListResponse is a response for a List request
+type ListResponse struct {
+	Keys  []string       `codec:",omitempty"`
+	Error *ResponseError `codec:",omitempty"`
+}
+
+// MoveResponse is a response for a Move request
+type MoveResponse struct {
+	Error *ResponseError `codec:",omitempty"`
+}
+
+// DeleteResponse is a response for a Delete request
+type DeleteResponse struct {
+	Error *ResponseError `codec:",omitempty"`
+}
--- a/docs/storage/driver/ipc/server.go
+++ b/docs/storage/driver/ipc/server.go
@ -0,0 +1,178 @@
+// +build ignore
+
+package ipc
+
+import (
+	"bytes"
+	"io"
+	"io/ioutil"
+	"net"
+	"os"
+	"reflect"
+
+	storagedriver "github.com/docker/distribution/registry/storage/driver"
+	"github.com/docker/libchan"
+	"github.com/docker/libchan/spdy"
+)
+
+// StorageDriverServer runs a new IPC server handling requests for the given
+// storagedriver.StorageDriver
+// This explicitly uses file descriptor 3 for IPC communication, as storage drivers are spawned in
+// client.go
+//
+// To create a new out-of-process driver, create a main package which calls StorageDriverServer with
+// a storagedriver.StorageDriver
+func StorageDriverServer(driver storagedriver.StorageDriver) error {
+	childSocket := os.NewFile(3, "childSocket")
+	defer childSocket.Close()
+	conn, err := net.FileConn(childSocket)
+	if err != nil {
+		panic(err)
+	}
+	defer conn.Close()
+	if transport, err := spdy.NewServerTransport(conn); err != nil {
+		panic(err)
+	} else {
+		for {
+			receiver, err := transport.WaitReceiveChannel()
+			if err == io.EOF {
+				return nil
+			} else if err != nil {
+				panic(err)
+			}
+			go receive(driver, receiver)
+		}
+	}
+}
+
+// receive receives new storagedriver.StorageDriver method requests and creates a new goroutine to
+// handle each request
+// Requests are expected to be of type ipc.Request as the parameters are unknown until the request
+// type is deserialized
+func receive(driver storagedriver.StorageDriver, receiver libchan.Receiver) {
+	for {
+		var request Request
+		err := receiver.Receive(&request)
+		if err == io.EOF {
+			return
+		} else if err != nil {
+			panic(err)
+		}
+		go handleRequest(driver, request)
+	}
+}
+
+// handleRequest handles storagedriver.StorageDriver method requests as defined in client.go
+// Responds to requests using the Request.ResponseChannel
+func handleRequest(driver storagedriver.StorageDriver, request Request) {
+	switch request.Type {
+	case "Version":
+		err := request.ResponseChannel.Send(&VersionResponse{Version: storagedriver.CurrentVersion})
+		if err != nil {
+			panic(err)
+		}
+	case "GetContent":
+		path, _ := request.Parameters["Path"].(string)
+		content, err := driver.GetContent(path)
+		var response ReadStreamResponse
+		if err != nil {
+			response = ReadStreamResponse{Error: WrapError(err)}
+		} else {
+			response = ReadStreamResponse{Reader: ioutil.NopCloser(bytes.NewReader(content))}
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "PutContent":
+		path, _ := request.Parameters["Path"].(string)
+		reader, _ := request.Parameters["Reader"].(io.ReadCloser)
+		contents, err := ioutil.ReadAll(reader)
+		defer reader.Close()
+		if err == nil {
+			err = driver.PutContent(path, contents)
+		}
+		response := WriteStreamResponse{
+			Error: WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "ReadStream":
+		path, _ := request.Parameters["Path"].(string)
+		// Depending on serialization method, Offset may be convereted to any int/uint type
+		offset := reflect.ValueOf(request.Parameters["Offset"]).Convert(reflect.TypeOf(int64(0))).Int()
+		reader, err := driver.ReadStream(path, offset)
+		var response ReadStreamResponse
+		if err != nil {
+			response = ReadStreamResponse{Error: WrapError(err)}
+		} else {
+			response = ReadStreamResponse{Reader: reader}
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "WriteStream":
+		path, _ := request.Parameters["Path"].(string)
+		// Depending on serialization method, Offset may be convereted to any int/uint type
+		offset := reflect.ValueOf(request.Parameters["Offset"]).Convert(reflect.TypeOf(int64(0))).Int()
+		// Depending on serialization method, Size may be convereted to any int/uint type
+		size := reflect.ValueOf(request.Parameters["Size"]).Convert(reflect.TypeOf(int64(0))).Int()
+		reader, _ := request.Parameters["Reader"].(io.ReadCloser)
+		err := driver.WriteStream(path, offset, size, reader)
+		response := WriteStreamResponse{
+			Error: WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "CurrentSize":
+		path, _ := request.Parameters["Path"].(string)
+		position, err := driver.CurrentSize(path)
+		response := CurrentSizeResponse{
+			Position: position,
+			Error:    WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "List":
+		path, _ := request.Parameters["Path"].(string)
+		keys, err := driver.List(path)
+		response := ListResponse{
+			Keys:  keys,
+			Error: WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "Move":
+		sourcePath, _ := request.Parameters["SourcePath"].(string)
+		destPath, _ := request.Parameters["DestPath"].(string)
+		err := driver.Move(sourcePath, destPath)
+		response := MoveResponse{
+			Error: WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	case "Delete":
+		path, _ := request.Parameters["Path"].(string)
+		err := driver.Delete(path)
+		response := DeleteResponse{
+			Error: WrapError(err),
+		}
+		err = request.ResponseChannel.Send(&response)
+		if err != nil {
+			panic(err)
+		}
+	default:
+		panic(request)
+	}
+}