Add flocker volume plugin

Flocker [1] is an open-source container data volume manager for
Dockerized applications.

This PR adds a volume plugin for Flocker.
The plugin interfaces the Flocker Control Service REST API [2] to
attachment attach the volume to the pod.

Each kubelet host should run Flocker agents (Container Agent and Dataset
Agent).

The kubelet will also require environment variables that contain the
host and port of the Flocker Control Service. (see Flocker architecture
[3] for more).

- `FLOCKER_CONTROL_SERVICE_HOST`
- `FLOCKER_CONTROL_SERVICE_PORT`

The contribution introduces a new 'flocker' volume type to the API with
fields:

- `datasetName`: which indicates the name of the dataset in Flocker
  added to metadata;
- `size`: a human-readable number that indicates the maximum size of the
  requested dataset.

Full documentation can be found docs/user-guide/volumes.md and examples
can be found at the examples/ folder

[1] https://clusterhq.com/flocker/introduction/
[2] https://docs.clusterhq.com/en/1.3.1/reference/api.html
[3] https://docs.clusterhq.com/en/1.3.1/concepts/architecture.html

Godeps/_workspace/src/github.com/ClusterHQ/flocker-go/README.md

@@ -0,0 +1,18 @@
+flocker-go
+==========
+
+[![circleci](https://circleci.com/gh/ClusterHQ/flocker-go.svg)](https://circleci.com/gh/ClusterHQ/flocker-go)
+
+flocker-go implements the package `flocker` that will let you easily interact
+with a Flocker Control Service.
+
+What can it do?
+---------------
+
+You can check the package documentation here: https://godoc.org/github.com/ClusterHQ/flocker-go
+
+TODO
+----
+
+- Define a proper interface `flockerClientable` with all the needed methods for
+  wrapping the Flocker API.

Godeps/_workspace/src/github.com/ClusterHQ/flocker-go/client.go

@@ -0,0 +1,323 @@
+package flocker
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// From https://github.com/ClusterHQ/flocker-docker-plugin/blob/master/flockerdockerplugin/adapter.py#L18
+const defaultVolumeSize = json.Number("107374182400")
+
+var (
+	// A volume can take a long time to be available, if we don't want
+	// Kubernetes to wait forever we need to stop trying after some time, that
+	// time is defined here
+	timeoutWaitingForVolume = 2 * time.Minute
+	tickerWaitingForVolume  = 5 * time.Second
+
+	errStateNotFound         = errors.New("State not found by Dataset ID")
+	errConfigurationNotFound = errors.New("Configuration not found by Name")
+
+	errFlockerControlServiceHost = errors.New("The volume config must have a key CONTROL_SERVICE_HOST defined in the OtherAttributes field")
+	errFlockerControlServicePort = errors.New("The volume config must have a key CONTROL_SERVICE_PORT defined in the OtherAttributes field")
+
+	errVolumeAlreadyExists = errors.New("The volume already exists")
+	errVolumeDoesNotExist  = errors.New("The volume does not exist")
+
+	errUpdatingDataset = errors.New("It was impossible to update the dataset")
+)
+
+// Clientable exposes the needed methods to implement your own Flocker Client.
+type Clientable interface {
+	CreateDataset(metaName string) (*DatasetState, error)
+
+	GetDatasetState(datasetID string) (*DatasetState, error)
+	GetDatasetID(metaName string) (datasetID string, err error)
+	GetPrimaryUUID() (primaryUUID string, err error)
+
+	UpdatePrimaryForDataset(primaryUUID, datasetID string) (*DatasetState, error)
+}
+
+// Client is a default Flocker Client.
+type Client struct {
+	*http.Client
+
+	schema  string
+	host    string
+	port    int
+	version string
+
+	clientIP string
+
+	maximumSize json.Number
+}
+
+// NewClient creates a wrapper over http.Client to communicate with the flocker control service.
+func NewClient(host string, port int, clientIP string, caCertPath, keyPath, certPath string) (*Client, error) {
+	client, err := newTLSClient(caCertPath, keyPath, certPath)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Client{
+		Client:      client,
+		schema:      "https",
+		host:        host,
+		port:        port,
+		version:     "v1",
+		maximumSize: defaultVolumeSize,
+		clientIP:    clientIP,
+	}, nil
+}
+
+/*
+request do a request using the http.Client embedded to the control service
+and returns the response or an error in case it happens.
+
+Note: you will need to deal with the response body call to Close if you
+don't want to deal with problems later.
+*/
+func (c Client) request(method, url string, payload interface{}) (*http.Response, error) {
+	var (
+		b   []byte
+		err error
+	)
+
+	if method == "POST" { // Just allow payload on POST
+		b, err = json.Marshal(payload)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	req, err := http.NewRequest(method, url, bytes.NewBuffer(b))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	// REMEMBER TO CLOSE THE BODY IN THE OUTSIDE FUNCTION
+	return c.Do(req)
+}
+
+// post performs a post request with the indicated payload
+func (c Client) post(url string, payload interface{}) (*http.Response, error) {
+	return c.request("POST", url, payload)
+}
+
+// get performs a get request
+func (c Client) get(url string) (*http.Response, error) {
+	return c.request("GET", url, nil)
+}
+
+// getURL returns a full URI to the control service
+func (c Client) getURL(path string) string {
+	return fmt.Sprintf("%s://%s:%d/%s/%s", c.schema, c.host, c.port, c.version, path)
+}
+
+type configurationPayload struct {
+	Primary     string          `json:"primary"`
+	DatasetID   string          `json:"dataset_id,omitempty"`
+	MaximumSize json.Number     `json:"maximum_size,omitempty"`
+	Metadata    metadataPayload `json:"metadata,omitempty"`
+}
+
+type metadataPayload struct {
+	Name string `json:"name,omitempty"`
+}
+
+type DatasetState struct {
+	Path        string      `json:"path"`
+	DatasetID   string      `json:"dataset_id"`
+	Primary     string      `json:"primary,omitempty"`
+	MaximumSize json.Number `json:"maximum_size,omitempty"`
+}
+
+type datasetStatePayload struct {
+	*DatasetState
+}
+
+type nodeStatePayload struct {
+	UUID string `json:"uuid"`
+	Host string `json:"host"`
+}
+
+// findIDInConfigurationsPayload returns the datasetID if it was found in the
+// configurations payload, otherwise it will return an error.
+func (c Client) findIDInConfigurationsPayload(body io.ReadCloser, name string) (datasetID string, err error) {
+	var configurations []configurationPayload
+	if err = json.NewDecoder(body).Decode(&configurations); err == nil {
+		for _, r := range configurations {
+			if r.Metadata.Name == name {
+				return r.DatasetID, nil
+			}
+		}
+		return "", errConfigurationNotFound
+	}
+	return "", err
+}
+
+// GetPrimaryUUID returns the UUID of the primary Flocker Control Service for
+// the given host.
+func (c Client) GetPrimaryUUID() (uuid string, err error) {
+	resp, err := c.get(c.getURL("state/nodes"))
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	var states []nodeStatePayload
+	if err = json.NewDecoder(resp.Body).Decode(&states); err == nil {
+		for _, s := range states {
+			if s.Host == c.clientIP {
+				return s.UUID, nil
+			}
+		}
+		return "", errStateNotFound
+	}
+	return "", err
+}
+
+// GetDatasetState performs a get request to get the state of the given datasetID, if
+// something goes wrong or the datasetID was not found it returns an error.
+func (c Client) GetDatasetState(datasetID string) (*DatasetState, error) {
+	resp, err := c.get(c.getURL("state/datasets"))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	var states []datasetStatePayload
+	if err = json.NewDecoder(resp.Body).Decode(&states); err == nil {
+		for _, s := range states {
+			if s.DatasetID == datasetID {
+				return s.DatasetState, nil
+			}
+		}
+		return nil, errStateNotFound
+	}
+
+	return nil, err
+}
+
+/*
+CreateDataset creates a volume in Flocker, waits for it to be ready and
+returns the dataset id.
+
+This process is a little bit complex but follows this flow:
+
+1. Find the Flocker Control Service UUID
+2. Try to create the dataset
+3. If it already exists an error is returned
+4. If it didn't previously exist, wait for it to be ready
+*/
+func (c Client) CreateDataset(metaName string) (*DatasetState, error) {
+	// 1) Find the primary Flocker UUID
+	// Note: it could be cached, but doing this query we health check it
+	primary, err := c.GetPrimaryUUID()
+	if err != nil {
+		return nil, err
+	}
+
+	// 2) Try to create the dataset in the given Primary
+	payload := configurationPayload{
+		Primary:     primary,
+		MaximumSize: json.Number(c.maximumSize),
+		Metadata: metadataPayload{
+			Name: metaName,
+		},
+	}
+
+	resp, err := c.post(c.getURL("configuration/datasets"), payload)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	// 3) Return if the dataset was previously created
+	if resp.StatusCode == http.StatusConflict {
+		return nil, errVolumeAlreadyExists
+	}
+
+	if resp.StatusCode >= 300 {
+		return nil, fmt.Errorf("Expected: {1,2}xx creating the volume, got: %d", resp.StatusCode)
+	}
+
+	var p configurationPayload
+	if err := json.NewDecoder(resp.Body).Decode(&p); err != nil {
+		return nil, err
+	}
+
+	// 4) Wait until the dataset is ready for usage. In case it never gets
+	// ready there is a timeoutChan that will return an error
+	timeoutChan := time.NewTimer(timeoutWaitingForVolume).C
+	tickChan := time.NewTicker(tickerWaitingForVolume).C
+
+	for {
+		if s, err := c.GetDatasetState(p.DatasetID); err == nil {
+			return s, nil
+		} else if err != errStateNotFound {
+			return nil, err
+		}
+
+		select {
+		case <-timeoutChan:
+			return nil, err
+		case <-tickChan:
+			break
+		}
+	}
+}
+
+// UpdatePrimaryForDataset will update the Primary for the given dataset
+// returning the current DatasetState.
+func (c Client) UpdatePrimaryForDataset(newPrimaryUUID, datasetID string) (*DatasetState, error) {
+	payload := struct {
+		Primary string `json:"primary"`
+	}{
+		Primary: newPrimaryUUID,
+	}
+
+	url := c.getURL(fmt.Sprintf("configuration/datasets/%s", datasetID))
+	resp, err := c.post(url, payload)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		return nil, errUpdatingDataset
+	}
+
+	var s DatasetState
+	if err := json.NewDecoder(resp.Body).Decode(&s); err != nil {
+		return nil, err
+	}
+
+	return &s, nil
+}
+
+// GetDatasetID will return the DatasetID found for the given metadata name.
+func (c Client) GetDatasetID(metaName string) (datasetID string, err error) {
+	resp, err := c.get(c.getURL("configuration/datasets"))
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	var configurations []configurationPayload
+	if err = json.NewDecoder(resp.Body).Decode(&configurations); err == nil {
+		for _, c := range configurations {
+			if c.Metadata.Name == metaName {
+				return c.DatasetID, nil
+			}
+		}
+		return "", errConfigurationNotFound
+	}
+	return "", err
+}

Godeps/_workspace/src/github.com/ClusterHQ/flocker-go/client_test.go

@@ -0,0 +1,316 @@
+package flocker
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"k8s.io/kubernetes/pkg/volume"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMaximumSizeIs1024Multiple(t *testing.T) {
+	assert := assert.New(t)
+
+	n, err := strconv.Atoi(string(defaultVolumeSize))
+	assert.NoError(err)
+	assert.Equal(0, n%1024)
+}
+
+func TestPost(t *testing.T) {
+	const (
+		expectedPayload    = "foobar"
+		expectedStatusCode = 418
+	)
+
+	assert := assert.New(t)
+
+	type payload struct {
+		Test string `json:"test"`
+	}
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var result payload
+		err := json.NewDecoder(r.Body).Decode(&result)
+		assert.NoError(err)
+		assert.Equal(expectedPayload, result.Test)
+		w.WriteHeader(expectedStatusCode)
+	}))
+	defer ts.Close()
+
+	c := Client{Client: &http.Client{}}
+
+	resp, err := c.post(ts.URL, payload{expectedPayload})
+	assert.NoError(err)
+	assert.Equal(expectedStatusCode, resp.StatusCode)
+}
+
+func TestGet(t *testing.T) {
+	const (
+		expectedStatusCode = 418
+	)
+
+	assert := assert.New(t)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(expectedStatusCode)
+	}))
+	defer ts.Close()
+
+	c := Client{Client: &http.Client{}}
+
+	resp, err := c.get(ts.URL)
+	assert.NoError(err)
+	assert.Equal(expectedStatusCode, resp.StatusCode)
+}
+
+func TestFindIDInConfigurationsPayload(t *testing.T) {
+	const (
+		searchedName = "search-for-this-name"
+		expected     = "The-42-id"
+	)
+	assert := assert.New(t)
+
+	c := Client{}
+
+	payload := fmt.Sprintf(
+		`[{"dataset_id": "1-2-3", "metadata": {"name": "test"}}, {"dataset_id": "The-42-id", "metadata": {"name": "%s"}}]`,
+		searchedName,
+	)
+
+	id, err := c.findIDInConfigurationsPayload(
+		ioutil.NopCloser(bytes.NewBufferString(payload)), searchedName,
+	)
+	assert.NoError(err)
+	assert.Equal(expected, id)
+
+	id, err = c.findIDInConfigurationsPayload(
+		ioutil.NopCloser(bytes.NewBufferString(payload)), "it will not be found",
+	)
+	assert.Equal(errConfigurationNotFound, err)
+
+	id, err = c.findIDInConfigurationsPayload(
+		ioutil.NopCloser(bytes.NewBufferString("invalid { json")), "",
+	)
+	assert.Error(err)
+}
+
+func TestFindPrimaryUUID(t *testing.T) {
+	const expectedPrimary = "primary-uuid"
+	assert := assert.New(t)
+
+	var (
+		mockedHost    = "127.0.0.1"
+		mockedPrimary = expectedPrimary
+	)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal("GET", r.Method)
+		assert.Equal("/v1/state/nodes", r.URL.Path)
+		w.Write([]byte(fmt.Sprintf(`[{"host": "%s", "uuid": "%s"}]`, mockedHost, mockedPrimary)))
+	}))
+
+	host, port, err := getHostAndPortFromTestServer(ts)
+	assert.NoError(err)
+
+	c := newFlockerTestClient(host, port)
+	assert.NoError(err)
+
+	mockedPrimary = expectedPrimary
+	primary, err := c.GetPrimaryUUID()
+	assert.NoError(err)
+	assert.Equal(expectedPrimary, primary)
+
+	c.clientIP = "not.found"
+	_, err = c.GetPrimaryUUID()
+	assert.Equal(errStateNotFound, err)
+}
+
+func TestGetURL(t *testing.T) {
+	const (
+		expectedHost = "host"
+		expectedPort = 42
+	)
+
+	assert := assert.New(t)
+
+	c := newFlockerTestClient(expectedHost, expectedPort)
+	var expectedURL = fmt.Sprintf("%s://%s:%d/v1/test", c.schema, expectedHost, expectedPort)
+
+	url := c.getURL("test")
+	assert.Equal(expectedURL, url)
+}
+
+func getHostAndPortFromTestServer(ts *httptest.Server) (string, int, error) {
+	tsURL, err := url.Parse(ts.URL)
+	if err != nil {
+		return "", 0, err
+	}
+
+	hostSplits := strings.Split(tsURL.Host, ":")
+
+	port, err := strconv.Atoi(hostSplits[1])
+	if err != nil {
+		return "", 0, nil
+	}
+	return hostSplits[0], port, nil
+}
+
+func getVolumeConfig(host string, port int) volume.VolumeConfig {
+	return volume.VolumeConfig{
+		OtherAttributes: map[string]string{
+			"CONTROL_SERVICE_HOST": host,
+			"CONTROL_SERVICE_PORT": strconv.Itoa(port),
+		},
+	}
+}
+
+func TestHappyPathCreateDatasetFromNonExistent(t *testing.T) {
+	const (
+		expectedDatasetName = "dir"
+		expectedPrimary     = "A-B-C-D"
+		expectedDatasetID   = "datasetID"
+	)
+	expectedPath := fmt.Sprintf("/flocker/%s", expectedDatasetID)
+
+	assert := assert.New(t)
+	var (
+		numCalls int
+		err      error
+	)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		numCalls++
+		switch numCalls {
+		case 1:
+			assert.Equal("GET", r.Method)
+			assert.Equal("/v1/state/nodes", r.URL.Path)
+			w.Write([]byte(fmt.Sprintf(`[{"host": "127.0.0.1", "uuid": "%s"}]`, expectedPrimary)))
+		case 2:
+			assert.Equal("POST", r.Method)
+			assert.Equal("/v1/configuration/datasets", r.URL.Path)
+
+			var c configurationPayload
+			err := json.NewDecoder(r.Body).Decode(&c)
+			assert.NoError(err)
+			assert.Equal(expectedPrimary, c.Primary)
+			assert.Equal(defaultVolumeSize, c.MaximumSize)
+			assert.Equal(expectedDatasetName, c.Metadata.Name)
+
+			w.Write([]byte(fmt.Sprintf(`{"dataset_id": "%s"}`, expectedDatasetID)))
+		case 3:
+			assert.Equal("GET", r.Method)
+			assert.Equal("/v1/state/datasets", r.URL.Path)
+			w.Write([]byte(`[]`))
+		case 4:
+			assert.Equal("GET", r.Method)
+			assert.Equal("/v1/state/datasets", r.URL.Path)
+			w.Write([]byte(fmt.Sprintf(`[{"dataset_id": "%s", "path": "/flocker/%s"}]`, expectedDatasetID, expectedDatasetID)))
+		}
+	}))
+
+	host, port, err := getHostAndPortFromTestServer(ts)
+	assert.NoError(err)
+
+	c := newFlockerTestClient(host, port)
+	assert.NoError(err)
+
+	tickerWaitingForVolume = 1 * time.Millisecond // TODO: this is overriding globally
+
+	s, err := c.CreateDataset(expectedDatasetName)
+	assert.NoError(err)
+	assert.Equal(expectedPath, s.Path)
+}
+
+func TestCreateDatasetThatAlreadyExists(t *testing.T) {
+	const (
+		datasetName     = "dir"
+		expectedPrimary = "A-B-C-D"
+	)
+
+	assert := assert.New(t)
+	var numCalls int
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		numCalls++
+		switch numCalls {
+		case 1:
+			assert.Equal("GET", r.Method)
+			assert.Equal("/v1/state/nodes", r.URL.Path)
+			w.Write([]byte(fmt.Sprintf(`[{"host": "127.0.0.1", "uuid": "%s"}]`, expectedPrimary)))
+		case 2:
+			assert.Equal("POST", r.Method)
+			assert.Equal("/v1/configuration/datasets", r.URL.Path)
+			w.WriteHeader(http.StatusConflict)
+		}
+	}))
+
+	host, port, err := getHostAndPortFromTestServer(ts)
+	assert.NoError(err)
+
+	c := newFlockerTestClient(host, port)
+	assert.NoError(err)
+
+	_, err = c.CreateDataset(datasetName)
+	assert.Equal(errVolumeAlreadyExists, err)
+}
+
+func TestUpdatePrimaryForDataset(t *testing.T) {
+	const (
+		dir               = "dir"
+		expectedPrimary   = "the-new-primary"
+		expectedDatasetID = "datasetID"
+	)
+	expectedURL := fmt.Sprintf("/v1/configuration/datasets/%s", expectedDatasetID)
+
+	assert := assert.New(t)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal("POST", r.Method)
+		assert.Equal(expectedURL, r.URL.Path)
+
+		var c configurationPayload
+		err := json.NewDecoder(r.Body).Decode(&c)
+		assert.NoError(err)
+
+		assert.Equal(expectedPrimary, c.Primary)
+
+		w.Write([]byte(fmt.Sprintf(`{"dataset_id": "%s", "path": "just-to-double-check"}`, expectedDatasetID)))
+	}))
+
+	host, port, err := getHostAndPortFromTestServer(ts)
+	assert.NoError(err)
+
+	c := newFlockerTestClient(host, port)
+	assert.NoError(err)
+
+	s, err := c.UpdatePrimaryForDataset(expectedPrimary, expectedDatasetID)
+	assert.NoError(err)
+	assert.Equal(expectedDatasetID, s.DatasetID)
+	assert.NotEqual("", s.Path)
+}
+
+func TestInterfaceIsImplemented(t *testing.T) {
+	assert.Implements(t, (*Clientable)(nil), Client{})
+}
+
+func newFlockerTestClient(host string, port int) *Client {
+	return &Client{
+		Client:      &http.Client{},
+		host:        host,
+		port:        port,
+		version:     "v1",
+		schema:      "http",
+		maximumSize: defaultVolumeSize,
+		clientIP:    "127.0.0.1",
+	}
+}

Godeps/_workspace/src/github.com/ClusterHQ/flocker-go/doc.go

@@ -0,0 +1,2 @@
+// flocker package allows you to easily interact with a Flocker Control Service.
+package flocker

Godeps/_workspace/src/github.com/ClusterHQ/flocker-go/util.go

@@ -0,0 +1,34 @@
+package flocker
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"io/ioutil"
+	"net/http"
+)
+
+// newTLSClient returns a new TLS http client
+func newTLSClient(caCertPath, keyPath, certPath string) (*http.Client, error) {
+	// Client certificate
+	cert, err := tls.LoadX509KeyPair(certPath, keyPath)
+	if err != nil {
+		return nil, err
+	}
+
+	// CA certificate
+	caCert, err := ioutil.ReadFile(caCertPath)
+	if err != nil {
+		return nil, err
+	}
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(caCert)
+
+	tlsConfig := &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      caCertPool,
+	}
+	tlsConfig.BuildNameToCertificate()
+	transport := &http.Transport{TLSClientConfig: tlsConfig}
+
+	return &http.Client{Transport: transport}, nil
+}