Merge pull request #236 from aluzzardi/state-store

Store requested state persistently.

cluster/cluster.go

@@ -8,6 +8,8 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/swarm/discovery"
+	"github.com/docker/swarm/state"
+	"github.com/samalba/dockerclient"
 )
 
 var (
@@ -17,20 +19,55 @@ var (
 
 type Cluster struct {
 	sync.RWMutex
+	store           *state.Store
 	tlsConfig       *tls.Config
 	eventHandlers   []EventHandler
 	nodes           map[string]*Node
 	overcommitRatio float64
 }
 
-func NewCluster(tlsConfig *tls.Config, overcommitRatio float64) *Cluster {
+func NewCluster(store *state.Store, tlsConfig *tls.Config, overcommitRatio float64) *Cluster {
 	return &Cluster{
 		tlsConfig:       tlsConfig,
 		nodes:           make(map[string]*Node),
+		store:           store,
 		overcommitRatio: overcommitRatio,
 	}
 }
 
+// Deploy a container into a `specific` node on the cluster.
+func (c *Cluster) DeployContainer(node *Node, config *dockerclient.ContainerConfig, name string) (*Container, error) {
+	container, err := node.Create(config, name, true)
+	if err != nil {
+		return nil, err
+	}
+
+	// Commit the requested state.
+	st := &state.RequestedState{
+		ID:     container.Id,
+		Name:   name,
+		Config: config,
+	}
+	if err := c.store.Add(container.Id, st); err != nil {
+		return nil, err
+	}
+	return container, nil
+}
+
+// Destroys a given `container` from the cluster.
+func (c *Cluster) DestroyContainer(container *Container, force bool) error {
+	if err := container.Node.Destroy(container, force); err != nil {
+		return err
+	}
+	if err := c.store.Remove(container.Id); err != nil {
+		if err == state.ErrNotFound {
+			log.Debugf("Container %s not found in the store", container.Id)
+		}
+		return err
+	}
+	return nil
+}
+
 func (c *Cluster) Handle(e *Event) error {
 	for _, eventHandler := range c.eventHandlers {
 		if err := eventHandler.Handle(e); err != nil {

cluster/cluster_test.go

@@ -1,8 +1,10 @@
 package cluster
 
 import (
+	"io/ioutil"
 	"testing"
 
+	"github.com/docker/swarm/state"
 	"github.com/samalba/dockerclient"
 	"github.com/samalba/dockerclient/mockclient"
 	"github.com/stretchr/testify/assert"
@@ -31,9 +33,14 @@ func createNode(t *testing.T, ID string, containers ...dockerclient.Container) *
 	return node
 }
 
-func TestAddNode(t *testing.T) {
+func newCluster(t *testing.T) *Cluster {
-	c := NewCluster(nil, 0)
+	dir, err := ioutil.TempDir("", "store-test")
+	assert.NoError(t, err)
+	return NewCluster(state.NewStore(dir), nil, 0)
+}
 
+func TestAddNode(t *testing.T) {
+	c := newCluster(t)
 	assert.Equal(t, len(c.Nodes()), 0)
 	assert.Nil(t, c.Node("test"))
 	assert.Nil(t, c.Node("test2"))
@@ -51,8 +58,8 @@ func TestAddNode(t *testing.T) {
 	assert.NotNil(t, c.Node("test2"))
 }
 
-func TestLookupContainer(t *testing.T) {
+func TestContainerLookup(t *testing.T) {
-	c := NewCluster(nil, 0)
+	c := newCluster(t)
 	container := dockerclient.Container{
 		Id:    "container-id",
 		Names: []string{"/container-name1", "/container-name2"},
@@ -74,3 +81,23 @@ func TestLookupContainer(t *testing.T) {
 	assert.NotNil(t, c.Container("test-node/container-name1"))
 	assert.NotNil(t, c.Container("test-node/container-name2"))
 }
+
+func TestDeployContainer(t *testing.T) {
+	// Create a test node.
+	node := createNode(t, "test")
+
+	// Create a test cluster.
+	c := newCluster(t)
+	assert.NoError(t, c.AddNode(node))
+
+	// Fake dockerclient calls to deploy a container.
+	client := node.client.(*mockclient.MockClient)
+	client.On("CreateContainer", mock.Anything, mock.Anything).Return("id", nil).Once()
+	client.On("ListContainers", true, false, mock.Anything).Return([]dockerclient.Container{{Id: "id"}}, nil).Once()
+	client.On("InspectContainer", "id").Return(&dockerclient.ContainerInfo{Config: &dockerclient.ContainerConfig{CpuShares: 100}}, nil).Once()
+
+	// Ensure the container gets deployed.
+	container, err := c.DeployContainer(node, &dockerclient.ContainerConfig{}, "name")
+	assert.NoError(t, err)
+	assert.Equal(t, container.Id, "id")
+}

flags.go

@@ -1,8 +1,27 @@
 package main
 
-import "github.com/codegangsta/cli"
+import (
+	"github.com/codegangsta/cli"
+	"os/user"
+	"path"
+
+	log "github.com/Sirupsen/logrus"
+)
+
+func homepath(p string) string {
+	usr, err := user.Current()
+	if err != nil {
+		log.Fatal(err)
+	}
+	return path.Join(usr.HomeDir, p)
+}
 
 var (
+	flStore = cli.StringFlag{
+		Name:  "rootdir",
+		Value: homepath(".swarm"),
+		Usage: "",
+	}
 	flDiscovery = cli.StringFlag{
 		Name:   "discovery",
 		Value:  "",

main.go

@@ -86,6 +86,7 @@ func main() {
 			Usage:     "manage a docker cluster",
 			Flags: []cli.Flag{
 				flDiscovery,
+				flStore,
 				flStrategy, flFilter,
 				flHosts, flHeartBeat, flOverCommit,
 				flTls, flTlsCaCert, flTlsCert, flTlsKey, flTlsVerify,

manage.go

@@ -5,6 +5,7 @@ import (
 	"crypto/x509"
 	"fmt"
 	"io/ioutil"
+	"path"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/codegangsta/cli"
@@ -14,6 +15,7 @@ import (
 	"github.com/docker/swarm/scheduler"
 	"github.com/docker/swarm/scheduler/filter"
 	"github.com/docker/swarm/scheduler/strategy"
+	"github.com/docker/swarm/state"
 )
 
 type logHandler struct {
@@ -71,7 +73,12 @@ func manage(c *cli.Context) {
 		}
 	}
 
-	cluster := cluster.NewCluster(tlsConfig, c.Float64("overcommit"))
+	store := state.NewStore(path.Join(c.String("rootdir"), "state"))
+	if err := store.Initialize(); err != nil {
+		log.Fatal(err)
+	}
+
+	cluster := cluster.NewCluster(store, tlsConfig, c.Float64("overcommit"))
 	cluster.Events(&logHandler{})
 
 	if !c.IsSet("discovery") {

scheduler/scheduler.go

@@ -52,7 +52,7 @@ func (s *Scheduler) CreateContainer(config *dockerclient.ContainerConfig, name s
 	if err != nil {
 		return nil, err
 	}
-	return node.Create(config, name, true)
+	return s.cluster.DeployContainer(node, config, name)
 }
 
 // Remove a container from the cluster. Containers should always be destroyed

state/state.go

@@ -5,6 +5,7 @@ import (
 )
 
 type RequestedState struct {
+	ID     string
 	Name   string
 	Config *dockerclient.ContainerConfig
 }