schedule volume resource

.gitignore

@@ -25,3 +25,4 @@ cscope.*
 *~
 .DS_Store
 tools/updatev4.go
+*.swp

Makefile

@@ -10,7 +10,6 @@ GO_LDFLAGS ?= -s -X $(REPO_PATH)/versioninfo.REVISION=$(REVISION) \
 
 grpc:
 	cd ./rpc/gen/; protoc --go_out=plugins=grpc:. core.proto
-	cd ./rpc/gen/; python3 -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. core.proto;
 
 deps:
 	env GO111MODULE=on go mod download

cluster/calcium/create.go

@@ -68,7 +68,7 @@ func (c *Calcium) doCreateContainer(ctx context.Context, opts *types.DeployOptio
 					ch <- m
 					if m.Error != nil && m.ContainerID == "" {
 						if err := c.withNodeLocked(ctx, nodeInfo.Name, func(node *types.Node) error {
-							return c.store.UpdateNodeResource(ctx, node, m.CPU, opts.CPUQuota, opts.Memory, opts.Storage, store.ActionIncr)
+							return c.store.UpdateNodeResource(ctx, node, m.CPU, opts.CPUQuota, opts.Memory, opts.Storage, m.VolumePlan.Consumed(), store.ActionIncr)
 						}); err != nil {
 							log.Errorf("[doCreateContainer] Reset node %s failed %v", nodeInfo.Name, err)
 						}
@@ -100,7 +100,15 @@ func (c *Calcium) doCreateContainerOnNode(ctx context.Context, nodeInfo types.No
 			if len(nodeInfo.CPUPlan) > 0 {
 				cpu = nodeInfo.CPUPlan[i]
 			}
-			ms[i] = &types.CreateContainerMessage{Error: err, CPU: cpu}
+			volumePlan := map[string]types.VolumeMap{}
+			if len(nodeInfo.VolumePlans) > 0 {
+				volumePlan = nodeInfo.VolumePlans[i]
+			}
+			ms[i] = &types.CreateContainerMessage{
+				Error:      err,
+				CPU:        cpu,
+				VolumePlan: volumePlan,
+			}
 		}
 		return ms
 	}
@@ -111,7 +119,11 @@ func (c *Calcium) doCreateContainerOnNode(ctx context.Context, nodeInfo types.No
 		if len(nodeInfo.CPUPlan) > 0 {
 			cpu = nodeInfo.CPUPlan[i]
 		}
-		ms[i] = c.doCreateAndStartContainer(ctx, i+index, node, opts, cpu)
+		volumePlan := types.VolumePlan{}
+		if len(nodeInfo.VolumePlans) > 0 {
+			volumePlan = nodeInfo.VolumePlans[i]
+		}
+		ms[i] = c.doCreateAndStartContainer(ctx, i+index, node, opts, cpu, volumePlan)
 		if !ms[i].Success {
 			log.Errorf("[doCreateContainerOnNode] Error when create and start a container, %v", ms[i].Error)
 			continue
@@ -136,6 +148,7 @@ func (c *Calcium) doCreateAndStartContainer(
 	no int, node *types.Node,
 	opts *types.DeployOptions,
 	cpu types.CPUMap,
+	volumePlan types.VolumePlan,
 ) *types.CreateContainerMessage {
 	container := &types.Container{
 		Podname:    opts.Podname,
@@ -152,16 +165,18 @@ func (c *Calcium) doCreateAndStartContainer(
 		Env:        opts.Env,
 		User:       opts.User,
 		Volumes:    opts.Volumes,
+		VolumePlan: volumePlan,
 	}
 	createContainerMessage := &types.CreateContainerMessage{
-		Podname:  container.Podname,
-		Nodename: container.Nodename,
-		Success:  false,
-		CPU:      cpu,
-		Quota:    opts.CPUQuota,
-		Memory:   opts.Memory,
-		Storage:  opts.Storage,
-		Publish:  map[string][]string{},
+		Podname:    container.Podname,
+		Nodename:   container.Nodename,
+		Success:    false,
+		CPU:        cpu,
+		Quota:      opts.CPUQuota,
+		Memory:     opts.Memory,
+		Storage:    opts.Storage,
+		VolumePlan: volumePlan,
+		Publish:    map[string][]string{},
 	}
 	var err error
 
@@ -177,7 +192,7 @@ func (c *Calcium) doCreateAndStartContainer(
 	}()
 
 	// get config
-	config := c.doMakeContainerOptions(no, cpu, opts, node)
+	config := c.doMakeContainerOptions(no, cpu, volumePlan, opts, node)
 	container.Name = config.Name
 	container.Labels = config.Labels
 	createContainerMessage.ContainerName = container.Name
@@ -242,11 +257,11 @@ func (c *Calcium) doCreateAndStartContainer(
 	return createContainerMessage
 }
 
-func (c *Calcium) doMakeContainerOptions(index int, cpumap types.CPUMap, opts *types.DeployOptions, node *types.Node) *enginetypes.VirtualizationCreateOptions {
+func (c *Calcium) doMakeContainerOptions(index int, cpumap types.CPUMap, volumePlan types.VolumePlan, opts *types.DeployOptions, node *types.Node) *enginetypes.VirtualizationCreateOptions {
 	config := &enginetypes.VirtualizationCreateOptions{}
 	// general
 	config.Seq = index
-	config.CPU = cpumap.Map()
+	config.CPU = cpumap
 	config.Quota = opts.CPUQuota
 	config.Memory = opts.Memory
 	config.Storage = opts.Storage
@@ -263,6 +278,12 @@ func (c *Calcium) doMakeContainerOptions(index int, cpumap types.CPUMap, opts *t
 	config.Debug = opts.Debug
 	config.Network = opts.NetworkMode
 	config.Networks = opts.Networks
+
+	// volumes
+	for idx, volume := range config.Volumes {
+		config.Volumes[idx] = volumePlan.GetVolumeString(volume)
+	}
+
 	// entry
 	entry := opts.Entrypoint
 	config.WorkingDir = entry.Dir

cluster/calcium/dissociate.go

@@ -20,7 +20,7 @@ func (c *Calcium) DissociateContainer(ctx context.Context, IDs []string) (chan *
 						return err
 					}
 					log.Infof("[DissociateContainer] Container %s dissociated", container.ID)
-					return c.store.UpdateNodeResource(ctx, node, container.CPU, container.Quota, container.Memory, container.Storage, store.ActionIncr)
+					return c.store.UpdateNodeResource(ctx, node, container.CPU, container.Quota, container.Memory, container.Storage, container.VolumePlan.Consumed(), store.ActionIncr)
 				})
 			})
 			if err != nil {

cluster/calcium/dissociate_test.go

@@ -57,7 +57,7 @@ func TestDissociateContainer(t *testing.T) {
 	}
 	store.On("RemoveContainer", mock.Anything, mock.Anything).Return(nil)
 	// success
-	store.On("UpdateNodeResource", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	store.On("UpdateNodeResource", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	ch, err = c.DissociateContainer(ctx, []string{"c1"})
 	assert.NoError(t, err)
 	for r := range ch {

cluster/calcium/helper.go

@@ -140,12 +140,14 @@ func getNodesInfo(nodes map[string]*types.Node, cpu float64, memory, storage int
 		nodeInfo := types.NodeInfo{
 			Name:         node.Name,
 			CPUMap:       node.CPU,
+			VolumeMap:    node.Volume,
 			MemCap:       node.MemCap,
 			StorageCap:   node.AvailableStorage(),
 			CPURate:      cpu / float64(len(node.InitCPU)),
 			MemRate:      float64(memory) / float64(node.InitMemCap),
 			StorageRate:  float64(storage) / float64(node.InitStorageCap),
 			CPUUsed:      node.CPUUsed / float64(len(node.InitCPU)),
+			VolumeUsed:   float64(node.VolumeUsed) / float64(len(node.InitVolume)),
 			MemUsage:     1.0 - float64(node.MemCap)/float64(node.InitMemCap),
 			StorageUsage: node.StorageUsage(),
 			Capacity:     0,

cluster/calcium/node.go

@@ -10,10 +10,13 @@ import (
 )
 
 // AddNode add a node in pod
+/*
 func (c *Calcium) AddNode(ctx context.Context, nodename, endpoint, podname, ca, cert, key string,
 	cpu, share int, memory, storage int64, labels map[string]string,
 	numa types.NUMA, numaMemory types.NUMAMemory) (*types.Node, error) {
-	return c.store.AddNode(ctx, nodename, endpoint, podname, ca, cert, key, cpu, share, memory, storage, labels, numa, numaMemory)
+*/
+func (c *Calcium) AddNode(ctx context.Context, opts *types.AddNodeOptions) (*types.Node, error) {
+	return c.store.AddNode(ctx, opts)
 }
 
 // RemoveNode remove a node
@@ -126,6 +129,22 @@ func (c *Calcium) SetNode(ctx context.Context, opts *types.SetNodeOptions) (*typ
 				}
 			}
 		}
+		// update volume
+		for volumeDir, changeCap := range opts.DeltaVolume {
+			if _, ok := n.Volume[volumeDir]; !ok && changeCap > 0 {
+				n.Volume[volumeDir] = changeCap
+				n.InitVolume[volumeDir] = changeCap
+			} else if ok && changeCap == 0 {
+				delete(n.Volume, volumeDir)
+				delete(n.InitVolume, volumeDir)
+			} else if ok {
+				n.Volume[volumeDir] += changeCap
+				n.InitVolume[volumeDir] += changeCap
+				if n.Volume[volumeDir] < 0 {
+					return types.ErrBadVolume
+				}
+			}
+		}
 		return c.store.UpdateNode(ctx, n)
 	})
 }

cluster/calcium/node_test.go

@@ -28,7 +28,7 @@ func TestAddNode(t *testing.T) {
 		mock.Anything, mock.Anything, mock.Anything).Return(node, nil)
 	c.store = store
 
-	n, err := c.AddNode(ctx, "", "", "", "", "", "", 0, 0, int64(0), int64(0), nil, nil, nil)
+	n, err := c.AddNode(ctx, &types.AddNodeOptions{})
 	assert.NoError(t, err)
 	assert.Equal(t, n.Name, name)
 }
@@ -210,10 +210,10 @@ func TestSetNode(t *testing.T) {
 	assert.NoError(t, err)
 	_, ok := n.CPU["1"]
 	assert.False(t, ok)
-	assert.Equal(t, n.CPU["2"], 1)
-	assert.Equal(t, n.InitCPU["2"], 9)
-	assert.Equal(t, n.CPU["3"], 10)
-	assert.Equal(t, n.InitCPU["3"], 10)
+	assert.Equal(t, n.CPU["2"], int64(1))
+	assert.Equal(t, n.InitCPU["2"], int64(9))
+	assert.Equal(t, n.CPU["3"], int64(10))
+	assert.Equal(t, n.InitCPU["3"], int64(10))
 	assert.Equal(t, len(n.CPU), 2)
 	assert.Equal(t, len(n.InitCPU), 2)
 }

cluster/calcium/realloc.go

@@ -19,7 +19,7 @@ type nodeContainers map[string][]*types.Container
 type cpuMemNodeContainers map[float64]map[int64]nodeContainers
 
 // ReallocResource allow realloc container resource
-func (c *Calcium) ReallocResource(ctx context.Context, IDs []string, cpu float64, memory int64) (chan *types.ReallocResourceMessage, error) {
+func (c *Calcium) ReallocResource(ctx context.Context, IDs []string, cpu float64, memory int64, volumes []string) (chan *types.ReallocResourceMessage, error) {
 	ch := make(chan *types.ReallocResourceMessage)
 	go func() {
 		defer close(ch)
@@ -52,7 +52,7 @@ func (c *Calcium) ReallocResource(ctx context.Context, IDs []string, cpu float64
 			for pod, nodeContainersInfo := range containersInfo {
 				go func(pod *types.Pod, nodeContainersInfo nodeContainers) {
 					defer wg.Done()
-					c.doReallocContainer(ctx, ch, pod, nodeContainersInfo, cpu, memory)
+					c.doReallocContainer(ctx, ch, pod, nodeContainersInfo, cpu, memory, volumes)
 				}(pod, nodeContainersInfo)
 			}
 			wg.Wait()
@@ -72,7 +72,7 @@ func (c *Calcium) doReallocContainer(
 	ch chan *types.ReallocResourceMessage,
 	pod *types.Pod,
 	nodeContainersInfo nodeContainers,
-	cpu float64, memory int64) {
+	cpu float64, memory int64, volumes []string) {
 
 	cpuMemNodeContainersInfo := cpuMemNodeContainers{}
 	for nodename, containers := range nodeContainersInfo {
@@ -138,7 +138,7 @@ func (c *Calcium) doReallocContainer(
 					}
 
 					for _, container := range containers {
-						newResource := &enginetypes.VirtualizationResource{Quota: newCPU, Memory: newMemory, SoftLimit: container.SoftLimit}
+						newResource := &enginetypes.VirtualizationResource{Quota: newCPU, Memory: newMemory, SoftLimit: container.SoftLimit, Volumes: volumes}
 						if len(container.CPU) > 0 {
 							newResource.CPU = cpusets[0]
 							newResource.NUMANode = node.GetNUMANode(cpusets[0])

cluster/calcium/realloc_test.go

@@ -67,36 +67,36 @@ func TestRealloc(t *testing.T) {
 	store.On("GetContainers", mock.Anything, []string{"c1", "c2"}).Return([]*types.Container{c1, c2}, nil)
 	// failed by lock
 	store.On("CreateLock", mock.Anything, mock.Anything).Return(nil, types.ErrNoETCD).Once()
-	ch, err := c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB))
+	ch, err := c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil)
 	// failed by GetPod
 	store.On("GetPod", mock.Anything, mock.Anything).Return(pod1, types.ErrNoETCD).Once()
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	store.On("GetPod", mock.Anything, mock.Anything).Return(pod1, nil)
 	// failed by newCPU < 0
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, -1, 2*int64(units.GiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	// failed by GetNode
 	store.On("GetNode", mock.Anything, "node1").Return(nil, types.ErrNoETCD).Once()
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.GiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.GiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	store.On("GetNode", mock.Anything, "node1").Return(node1, nil)
 	// failed by memory not enough
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.GiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.GiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
@@ -105,14 +105,14 @@ func TestRealloc(t *testing.T) {
 	simpleMockScheduler := &schedulermocks.Scheduler{}
 	c.scheduler = simpleMockScheduler
 	simpleMockScheduler.On("SelectCPUNodes", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, 0, types.ErrInsufficientMEM).Once()
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.MiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.MiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	// failed by wrong total
 	simpleMockScheduler.On("SelectCPUNodes", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, 0, nil).Once()
-	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.MiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1"}, 0.1, 2*int64(units.MiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
@@ -139,19 +139,19 @@ func TestRealloc(t *testing.T) {
 		Engine:   engine,
 		Endpoint: "http://1.1.1.1:1",
 	}
-	ch, err = c.ReallocResource(ctx, []string{"c1", "c2"}, 0.1, 2*int64(units.MiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1", "c2"}, 0.1, 2*int64(units.MiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
 	}
 	// check node resource as usual
-	assert.Equal(t, node1.CPU["2"], 10)
+	assert.Equal(t, node1.CPU["2"], int64(10))
 	assert.Equal(t, node1.MemCap, int64(units.GiB))
 	engine.On("VirtualizationUpdateResource", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	store.On("UpdateNode", mock.Anything, mock.Anything).Return(nil)
 	// failed by update container
 	store.On("UpdateContainer", mock.Anything, mock.Anything).Return(types.ErrBadContainerID).Twice()
-	ch, err = c.ReallocResource(ctx, []string{"c1", "c2"}, 0.1, 2*int64(units.MiB))
+	ch, err = c.ReallocResource(ctx, []string{"c1", "c2"}, 0.1, 2*int64(units.MiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.False(t, r.Success)
@@ -194,12 +194,12 @@ func TestRealloc(t *testing.T) {
 	simpleMockScheduler.On("SelectCPUNodes", mock.Anything, mock.Anything, mock.Anything).Return(nil, nodeCPUPlans, 2, nil)
 	store.On("GetNode", mock.Anything, "node2").Return(node2, nil)
 	store.On("GetContainers", mock.Anything, []string{"c3", "c4"}).Return([]*types.Container{c3, c4}, nil)
-	ch, err = c.ReallocResource(ctx, []string{"c3", "c4"}, 0.1, 2*int64(units.MiB))
+	ch, err = c.ReallocResource(ctx, []string{"c3", "c4"}, 0.1, 2*int64(units.MiB), nil)
 	assert.NoError(t, err)
 	for r := range ch {
 		assert.True(t, r.Success)
 	}
-	assert.Equal(t, node2.CPU["3"], 0)
-	assert.Equal(t, node2.CPU["2"], 100)
+	assert.Equal(t, node2.CPU["3"], int64(0))
+	assert.Equal(t, node2.CPU["2"], int64(100))
 	assert.Equal(t, node2.MemCap, int64(units.GiB)-4*int64(units.MiB))
 }

cluster/calcium/remove.go

@@ -35,7 +35,7 @@ func (c *Calcium) RemoveContainer(ctx context.Context, IDs []string, force bool,
 							return err
 						}
 						log.Infof("[RemoveContainer] Container %s removed", container.ID)
-						if err = c.store.UpdateNodeResource(ctx, node, container.CPU, container.Quota, container.Memory, container.Storage, store.ActionIncr); err == nil {
+						if err = c.store.UpdateNodeResource(ctx, node, container.CPU, container.Quota, container.Memory, container.Storage, container.VolumePlan.Consumed(), store.ActionIncr); err == nil {
 							success = true
 						}
 						return err

cluster/calcium/remove_test.go

@@ -57,7 +57,7 @@ func TestRemoveContainer(t *testing.T) {
 	engine.On("VirtualizationRemove", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	store.On("GetContainers", mock.Anything, mock.Anything).Return([]*types.Container{container}, nil)
 	store.On("RemoveContainer", mock.Anything, mock.Anything).Return(nil)
-	store.On("UpdateNodeResource", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	store.On("UpdateNodeResource", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	// success
 	ch, err = c.RemoveContainer(ctx, []string{"xx"}, false, 0)
 	assert.NoError(t, err)

cluster/calcium/replace.go

@@ -137,7 +137,7 @@ func (c *Calcium) doReplaceContainer(
 	}
 	// 不涉及资源消耗，创建容器失败会被回收容器而不回收资源
 	// 创建成功容器会干掉之前的老容器也不会动资源，实际上实现了动态捆绑
-	createMessage := c.doCreateAndStartContainer(ctx, index, node, &opts.DeployOptions, container.CPU)
+	createMessage := c.doCreateAndStartContainer(ctx, index, node, &opts.DeployOptions, container.CPU, container.VolumePlan)
 	if createMessage.Error != nil {
 		// 重启老容器
 		message, err := c.doStartContainer(ctx, container, opts.IgnoreHook)