Merge pull request #686 from BrianBland/storagedriver-versioning

Adds versioning for out-of-process storage driver
This commit is contained in:
Olivier Gambier 2014-11-07 14:55:13 -08:00
commit da205085f3
5 changed files with 106 additions and 17 deletions

View file

@ -40,6 +40,8 @@ Storage drivers should call `factory.Register` with their driver name in an `ini
### Out-of-process drivers
As many users will run the registry as a pre-constructed docker container, storage drivers should also be distributable as IPC server executables. Drivers written in go should model the main method provided in `storagedriver/filesystem/registry-storage-filesystem/filesystem.go`. Parameters to IPC drivers will be provided as a JSON-serialized map in the first argument to the process. These parameters should be validated and then a blocking call to `ipc.StorageDriverServer` should be made with a new storage driver.
Out-of-process drivers must also implement the `ipc.IPCStorageDriver` interface, which exposes a `Version` check for the storage driver. This is used to validate storage driver api compatibility at driver load-time.
## Testing
Storage driver test suites are provided in `storagedriver/testsuites/testsuites.go` and may be used for any storage driver written in go. Two methods are provided for registering test suites, `RegisterInProcessSuite` and `RegisterIPCSuite`, which run the same set of tests for the driver imported or managed over IPC respectively.

View file

@ -10,6 +10,7 @@ import (
"os/exec"
"syscall"
"github.com/docker/docker-registry/storagedriver"
"github.com/docker/libchan"
"github.com/docker/libchan/spdy"
)
@ -25,6 +26,7 @@ type StorageDriverClient 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
@ -63,42 +65,62 @@ func (driver *StorageDriverClient) Start() error {
}
childSocket := os.NewFile(uintptr(fileDescriptors[0]), "childSocket")
parentSocket := os.NewFile(uintptr(fileDescriptors[1]), "parentSocket")
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 {
parentSocket.Close()
driver.Stop()
return err
}
if err = childSocket.Close(); err != nil {
parentSocket.Close()
driver.Stop()
return err
}
connection, err := net.FileConn(parentSocket)
connection, err := net.FileConn(driver.socket)
if err != nil {
parentSocket.Close()
driver.Stop()
return err
}
transport, err := spdy.NewClientTransport(connection)
driver.transport, err = spdy.NewClientTransport(connection)
if err != nil {
parentSocket.Close()
driver.Stop()
return err
}
sender, err := transport.NewSendChannel()
driver.sender, err = driver.transport.NewSendChannel()
if err != nil {
transport.Close()
parentSocket.Close()
driver.Stop()
return err
}
driver.socket = parentSocket
driver.transport = transport
driver.sender = sender
// 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
}
driver.version = response.Version
if driver.version.Major() != storagedriver.CurrentVersion.Major() || driver.version.Minor() > storagedriver.CurrentVersion.Minor() {
return IncompatibleVersionError{driver.version}
}
return nil
}
@ -106,10 +128,20 @@ func (driver *StorageDriverClient) Start() error {
// Stop stops the child process storage driver
// storagedriver.StorageDriver methods called after Stop will fail
func (driver *StorageDriverClient) Stop() error {
closeSenderErr := driver.sender.Close()
closeTransportErr := driver.transport.Close()
closeSocketErr := driver.socket.Close()
killErr := driver.subprocess.Process.Kill()
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 closeSenderErr != nil {
return closeSenderErr

View file

@ -5,9 +5,29 @@ import (
"io"
"reflect"
"github.com/docker/docker-registry/storagedriver"
"github.com/docker/libchan"
)
// IPCStorageDriver 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 IPCStorageDriver 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 {
@ -38,6 +58,12 @@ func (err *responseError) Error() string {
// IPC method call response object definitions
// VersionResponse is a response for a Version request
type VersionResponse struct {
Version storagedriver.Version
Error *responseError
}
// ReadStreamResponse is a response for a ReadStream request
type ReadStreamResponse struct {
Reader io.ReadCloser

View file

@ -61,6 +61,11 @@ func receive(driver storagedriver.StorageDriver, receiver libchan.Receiver) {
// 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)

View file

@ -3,8 +3,32 @@ package storagedriver
import (
"fmt"
"io"
"strconv"
"strings"
)
// Version is a string representing the storage driver version, of the form Major.Minor.
// The registry must accept storage drivers with equal major version and greater minor version,
// but may not be compatible with older storage driver versions.
type Version string
// Major returns the major (primary) component of a version
func (version Version) Major() uint {
majorPart := strings.Split(string(version), ".")[0]
major, _ := strconv.ParseUint(majorPart, 10, 0)
return uint(major)
}
// Minor returns the minor (secondary) component of a version
func (version Version) Minor() uint {
minorPart := strings.Split(string(version), ".")[1]
minor, _ := strconv.ParseUint(minorPart, 10, 0)
return uint(minor)
}
// CurrentVersion is the current storage driver Version
const CurrentVersion Version = "0.1"
// StorageDriver defines methods that a Storage Driver must implement for a filesystem-like
// key/value object storage
type StorageDriver interface {