Merge pull request #228 from vieux/overcommit

Move overcommit outside of binpacking

cluster/cluster.go

@@ -17,15 +17,17 @@ var (
 
 type Cluster struct {
 	sync.RWMutex
-	tlsConfig     *tls.Config
-	eventHandlers []EventHandler
-	nodes         map[string]*Node
+	tlsConfig       *tls.Config
+	eventHandlers   []EventHandler
+	nodes           map[string]*Node
+	overcommitRatio float64
 }
 
-func NewCluster(tlsConfig *tls.Config) *Cluster {
+func NewCluster(tlsConfig *tls.Config, overcommitRatio float64) *Cluster {
 	return &Cluster{
-		tlsConfig: tlsConfig,
-		nodes:     make(map[string]*Node),
+		tlsConfig:       tlsConfig,
+		nodes:           make(map[string]*Node),
+		overcommitRatio: overcommitRatio,
 	}
 }
 
@@ -63,7 +65,7 @@ func (c *Cluster) UpdateNodes(nodes []*discovery.Node) {
 	for _, addr := range nodes {
 		go func(node *discovery.Node) {
 			if c.Node(node.String()) == nil {
-				n := NewNode(node.String())
+				n := NewNode(node.String(), c.overcommitRatio)
 				if err := n.Connect(c.tlsConfig); err != nil {
 					log.Error(err)
 					return

cluster/cluster_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func createNode(t *testing.T, ID string, containers ...dockerclient.Container) *Node {
-	node := NewNode(ID)
+	node := NewNode(ID, 0)
 	node.Name = ID
 
 	assert.False(t, node.IsConnected())
@@ -32,7 +32,7 @@ func createNode(t *testing.T, ID string, containers ...dockerclient.Container) *
 }
 
 func TestAddNode(t *testing.T) {
-	c := NewCluster(nil)
+	c := NewCluster(nil, 0)
 
 	assert.Equal(t, len(c.Nodes()), 0)
 	assert.Nil(t, c.Node("test"))
@@ -52,7 +52,7 @@ func TestAddNode(t *testing.T) {
 }
 
 func TestLookupContainer(t *testing.T) {
-	c := NewCluster(nil)
+	c := NewCluster(nil, 0)
 	container := dockerclient.Container{
 		Id:    "container-id",
 		Names: []string{"/container-name1", "/container-name2"},

cluster/node.go

@@ -21,13 +21,14 @@ const (
 	requestTimeout = 10 * time.Second
 )
 
-func NewNode(addr string) *Node {
+func NewNode(addr string, overcommitRatio float64) *Node {
 	e := &Node{
-		Addr:       addr,
-		Labels:     make(map[string]string),
-		ch:         make(chan bool),
-		containers: make(map[string]*Container),
-		healthy:    true,
+		Addr:            addr,
+		Labels:          make(map[string]string),
+		ch:              make(chan bool),
+		containers:      make(map[string]*Container),
+		healthy:         true,
+		overcommitRatio: int64(overcommitRatio * 100),
 	}
 	return e
 }
@@ -43,11 +44,12 @@ type Node struct {
 	Memory int64
 	Labels map[string]string
 
-	ch           chan bool
-	containers   map[string]*Container
-	client       dockerclient.Client
-	eventHandler EventHandler
-	healthy      bool
+	ch              chan bool
+	containers      map[string]*Container
+	client          dockerclient.Client
+	eventHandler    EventHandler
+	healthy         bool
+	overcommitRatio int64
 }
 
 // Connect will initialize a connection to the Docker daemon running on the
@@ -268,6 +270,14 @@ func (n *Node) ReservedCpus() int64 {
 	return r
 }
 
+func (n *Node) UsableMemory() int64 {
+	return n.Memory + (n.Memory * n.overcommitRatio / 100)
+}
+
+func (n *Node) UsableCpus() int64 {
+	return n.Cpus + (n.Cpus * n.overcommitRatio / 100)
+}
+
 func (n *Node) Create(config *dockerclient.ContainerConfig, name string, pullImage bool) (*Container, error) {
 	var (
 		err    error

cluster/node_test.go

@@ -26,7 +26,7 @@ var (
 )
 
 func TestNodeConnectionFailure(t *testing.T) {
-	node := NewNode("test")
+	node := NewNode("test", 0)
 	assert.False(t, node.IsConnected())
 
 	// Always fail.
@@ -41,7 +41,7 @@ func TestNodeConnectionFailure(t *testing.T) {
 }
 
 func TestOutdatedNode(t *testing.T) {
-	node := NewNode("test")
+	node := NewNode("test", 0)
 	client := mockclient.NewMockClient()
 	client.On("Info").Return(&dockerclient.Info{}, nil)
 
@@ -52,7 +52,7 @@ func TestOutdatedNode(t *testing.T) {
 }
 
 func TestNodeCpusMemory(t *testing.T) {
-	node := NewNode("test")
+	node := NewNode("test", 0)
 	assert.False(t, node.IsConnected())
 
 	client := mockclient.NewMockClient()
@@ -71,7 +71,7 @@ func TestNodeCpusMemory(t *testing.T) {
 }
 
 func TestNodeSpecs(t *testing.T) {
-	node := NewNode("test")
+	node := NewNode("test", 0)
 	assert.False(t, node.IsConnected())
 
 	client := mockclient.NewMockClient()
@@ -95,7 +95,7 @@ func TestNodeSpecs(t *testing.T) {
 }
 
 func TestNodeState(t *testing.T) {
-	node := NewNode("test")
+	node := NewNode("test", 0)
 	assert.False(t, node.IsConnected())
 
 	client := mockclient.NewMockClient()
@@ -140,7 +140,7 @@ func TestCreateContainer(t *testing.T) {
 			Cmd:       []string{"date"},
 			Tty:       false,
 		}
-		node   = NewNode("test")
+		node   = NewNode("test", 0)
 		client = mockclient.NewMockClient()
 	)
 
@@ -186,3 +186,23 @@ func TestCreateContainer(t *testing.T) {
 	assert.Equal(t, container.Id, id)
 	assert.Len(t, node.Containers(), 2)
 }
+
+func TestUsableMemory(t *testing.T) {
+	node := NewNode("test", 0.05)
+	node.Memory = 1024
+	assert.Equal(t, node.UsableMemory(), 1024+1024*5/100)
+
+	node = NewNode("test", 0)
+	node.Memory = 1024
+	assert.Equal(t, node.UsableMemory(), 1024)
+}
+
+func TestUsableCpus(t *testing.T) {
+	node := NewNode("test", 0.05)
+	node.Cpus = 2
+	assert.Equal(t, node.UsableCpus(), 2+2*5/100)
+
+	node = NewNode("test", 0)
+	node.Cpus = 2
+	assert.Equal(t, node.UsableCpus(), 2)
+}

flags.go

@@ -50,10 +50,15 @@ var (
 		Name:  "tlsverify",
 		Usage: "use TLS and verify the remote",
 	}
+	flOverCommit = cli.Float64Flag{
+		Name:  "overcommit, oc",
+		Usage: "overcommit to apply on resources",
+		Value: 0.05,
+	}
 	flStrategy = cli.StringFlag{
 		Name:  "strategy",
 		Usage: "placement strategy to use [binpacking, random]",
-		Value: "binpacking:0.05",
+		Value: "binpacking",
 	}
 	flFilter = cli.StringSliceFlag{
 		Name:  "filter, f",

main.go

@@ -87,7 +87,7 @@ func main() {
 			Flags: []cli.Flag{
 				flDiscovery,
 				flStrategy, flFilter,
-				flHosts, flHeartBeat,
+				flHosts, flHeartBeat, flOverCommit,
 				flTls, flTlsCaCert, flTlsCert, flTlsKey, flTlsVerify,
 				flEnableCors},
 			Action: manage,

manage.go

@@ -71,7 +71,7 @@ func manage(c *cli.Context) {
 		}
 	}
 
-	cluster := cluster.NewCluster(tlsConfig)
+	cluster := cluster.NewCluster(tlsConfig, c.Float64("overcommit"))
 	cluster.Events(&logHandler{})
 
 	if !c.IsSet("discovery") {

scheduler/filter/constraint_test.go

@@ -12,9 +12,9 @@ func TestConstrainteFilter(t *testing.T) {
 	var (
 		f     = ConstraintFilter{}
 		nodes = []*cluster.Node{
-			cluster.NewNode("node-0"),
-			cluster.NewNode("node-1"),
-			cluster.NewNode("node-2"),
+			cluster.NewNode("node-0", 0),
+			cluster.NewNode("node-1", 0),
+			cluster.NewNode("node-2", 0),
 		}
 		result []*cluster.Node
 		err    error

scheduler/filter/port_test.go

@@ -24,9 +24,9 @@ func TestPortFilterNoConflicts(t *testing.T) {
 	var (
 		p     = PortFilter{}
 		nodes = []*cluster.Node{
-			cluster.NewNode("node-1"),
-			cluster.NewNode("node-2"),
-			cluster.NewNode("node-3"),
+			cluster.NewNode("node-1", 0),
+			cluster.NewNode("node-2", 0),
+			cluster.NewNode("node-3", 0),
 		}
 		result []*cluster.Node
 		err    error
@@ -70,9 +70,9 @@ func TestPortFilterSimple(t *testing.T) {
 	var (
 		p     = PortFilter{}
 		nodes = []*cluster.Node{
-			cluster.NewNode("node-1"),
-			cluster.NewNode("node-2"),
-			cluster.NewNode("node-3"),
+			cluster.NewNode("node-1", 0),
+			cluster.NewNode("node-2", 0),
+			cluster.NewNode("node-3", 0),
 		}
 		result []*cluster.Node
 		err    error
@@ -99,9 +99,9 @@ func TestPortFilterDifferentInterfaces(t *testing.T) {
 	var (
 		p     = PortFilter{}
 		nodes = []*cluster.Node{
-			cluster.NewNode("node-1"),
-			cluster.NewNode("node-2"),
-			cluster.NewNode("node-3"),
+			cluster.NewNode("node-1", 0),
+			cluster.NewNode("node-2", 0),
+			cluster.NewNode("node-3", 0),
 		}
 		result []*cluster.Node
 		err    error

scheduler/strategy/binpacking.go

@@ -3,7 +3,6 @@ package strategy
 import (
 	"errors"
 	"sort"
-	"strconv"
 
 	"github.com/docker/swarm/cluster"
 	"github.com/samalba/dockerclient"
@@ -13,22 +12,18 @@ var (
 	ErrNoResourcesAvailable = errors.New("no resources available to schedule container")
 )
 
-type BinPackingPlacementStrategy struct {
-	ratio int64
-}
+type BinPackingPlacementStrategy struct{}
 
-func (p *BinPackingPlacementStrategy) Initialize(opts string) error {
-	overcommitRatio, err := strconv.ParseFloat(opts, 64)
-	p.ratio = int64(overcommitRatio * 100)
-	return err
+func (p *BinPackingPlacementStrategy) Initialize() error {
+	return nil
 }
 
 func (p *BinPackingPlacementStrategy) PlaceContainer(config *dockerclient.ContainerConfig, nodes []*cluster.Node) (*cluster.Node, error) {
 	scores := scores{}
 
 	for _, node := range nodes {
-		nodeMemory := node.Memory + (node.Memory * p.ratio / 100)
-		nodeCpus := node.Cpus + (node.Cpus * p.ratio / 100)
+		nodeMemory := node.UsableMemory()
+		nodeCpus := node.UsableCpus()
 
 		// Skip nodes that are smaller than the requested resources.
 		if nodeMemory < int64(config.Memory) || nodeCpus < config.CpuShares {

scheduler/strategy/binpacking_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func createNode(ID string, memory int64, cpus int64) *cluster.Node {
-	node := cluster.NewNode(ID)
+	node := cluster.NewNode(ID, 0.05)
 	node.ID = ID
 	node.Memory = memory * 1024 * 1024 * 1024
 	node.Cpus = cpus
@@ -114,7 +114,7 @@ func TestPlaceContainerHuge(t *testing.T) {
 }
 
 func TestPlaceContainerOvercommit(t *testing.T) {
-	s, err := New("binpacking:0.05")
+	s, err := New("binpacking")
 	assert.NoError(t, err)
 
 	nodes := []*cluster.Node{createNode("node-1", 0, 1)}

scheduler/strategy/random.go

@@ -10,10 +10,9 @@ import (
 )
 
 // Randomly place the container into the cluster.
-type RandomPlacementStrategy struct {
-}
+type RandomPlacementStrategy struct{}
 
-func (p *RandomPlacementStrategy) Initialize(_ string) error {
+func (p *RandomPlacementStrategy) Initialize() error {
 	rand.Seed(time.Now().UTC().UnixNano())
 	return nil
 }

scheduler/strategy/strategy.go

@@ -2,7 +2,6 @@ package strategy
 
 import (
 	"errors"
-	"strings"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/docker/swarm/cluster"
@@ -10,7 +9,7 @@ import (
 )
 
 type PlacementStrategy interface {
-	Initialize(string) error
+	Initialize() error
 	// Given a container configuration and a set of nodes, select the target
 	// node where the container should be scheduled.
 	PlaceContainer(config *dockerclient.ContainerConfig, nodes []*cluster.Node) (*cluster.Node, error)
@@ -28,19 +27,10 @@ func init() {
 	}
 }
 
-func New(nameAndOpts string) (PlacementStrategy, error) {
-	var (
-		parts = strings.SplitN(nameAndOpts, ":", 2)
-		name  = parts[0]
-		opts  string
-	)
-	if len(parts) == 2 {
-		opts = parts[1]
-	}
-
+func New(name string) (PlacementStrategy, error) {
 	if strategy, exists := strategies[name]; exists {
-		log.Debugf("Initializing %q strategy with %q", name, opts)
-		err := strategy.Initialize(opts)
+		log.Debugf("Initializing %q strategy", name)
+		err := strategy.Initialize()
 		return strategy, err
 	}