CPU prior with mem limit

Author: CMGS

cluster/calcium/common.go

@@ -26,11 +26,11 @@ type NodeContainers map[*types.Node][]*types.Container
 //NodeCPUMap store cpu and node info
 type NodeCPUMap map[*types.Node][]types.CPUMap
 
-//CPUNodeContainers store cpu and nodecontainers
-type CPUNodeContainers map[float64]NodeContainers
+//CPUMemNodeContainers store cpu, mem and nodecontainers
+type CPUMemNodeContainers map[float64]map[int64]NodeContainers
 
-//CPUNodeContainersMap store cpu and nodecpumap
-type CPUNodeContainersMap map[float64]NodeCPUMap
+//CPUMemNodeContainersMap store cpu, mem and nodecpumap
+type CPUMemNodeContainersMap map[float64]map[int64]NodeCPUMap
 
 type imageBucket struct {
 	sync.Mutex

cluster/calcium/create_container.go

@@ -27,6 +27,16 @@ func (c *Calcium) CreateContainer(ctx context.Context, opts *types.DeployOptions
 		return nil, err
 	}
 	log.Infof("[CreateContainer] Creating container with options: %v", opts)
+
+	// 4194304 Byte = 4 MB, docker 创建容器的内存最低标准
+	// -1 means without limit
+	if opts.Memory < minMemory && !opts.RawResource {
+		return nil, fmt.Errorf("Minimum memory limit allowed is 4MB, got %d", opts.Memory)
+	}
+	if opts.Count <= 0 { // Count 要大于0
+		return nil, fmt.Errorf("Count must be positive, got %d", opts.Count)
+	}
+
 	if opts.RawResource || pod.Favor == scheduler.MEMORY_PRIOR {
 		return c.createContainerWithMemoryPrior(ctx, opts)
 	} else if pod.Favor == scheduler.CPU_PRIOR {
@@ -37,14 +47,6 @@ func (c *Calcium) CreateContainer(ctx context.Context, opts *types.DeployOptions
 
 func (c *Calcium) createContainerWithMemoryPrior(ctx context.Context, opts *types.DeployOptions) (chan *types.CreateContainerMessage, error) {
 	ch := make(chan *types.CreateContainerMessage)
-	// 4194304 Byte = 4 MB, docker 创建容器的内存最低标准
-	// -1 means without limit
-	if opts.Memory < minMemory && !opts.RawResource {
-		return ch, fmt.Errorf("Minimum memory limit allowed is 4MB, got %d", opts.Memory)
-	}
-	if opts.Count <= 0 { // Count 要大于0
-		return ch, fmt.Errorf("Count must be positive, got %d", opts.Count)
-	}
 
 	// TODO RFC 计算当前 app 部署情况的时候需要保证同一时间只有这个 app 的这个 entrypoint 在跑
 	// 因此需要在这里加个全局锁，直到部署完毕才释放
@@ -87,7 +89,7 @@ func (c *Calcium) doCreateContainerWithMemoryPrior(ctx context.Context, nodeInfo
 		for i := 0; i < nodeInfo.Deploy; i++ {
 			ms[i] = &types.CreateContainerMessage{Error: err}
 			if !opts.RawResource {
-				if err := c.store.UpdateNodeMem(ctx, opts.Podname, nodeInfo.Name, opts.Memory, "+"); err != nil {
+				if err := c.store.UpdateNodeResource(ctx, opts.Podname, nodeInfo.Name, types.CPUMap{}, opts.Memory, "+"); err != nil {
 					log.Errorf("[doCreateContainerWithMemoryPrior] reset node memory failed %v", err)
 				}
 			}
@@ -153,7 +155,7 @@ func (c *Calcium) doCreateContainerWithCPUPrior(ctx context.Context, nodeName st
 		log.Errorf("[doCreateContainerWithCPUPrior] Get and prepare node error %v", err)
 		for i := 0; i < deployCount; i++ {
 			ms[i] = &types.CreateContainerMessage{Error: err}
-			if err := c.store.UpdateNodeCPU(ctx, opts.Podname, nodeName, cpuMap[i], "+"); err != nil {
+			if err := c.store.UpdateNodeResource(ctx, opts.Podname, nodeName, cpuMap[i], opts.Memory, "+"); err != nil {
 				log.Errorf("[doCreateContainerWithCPUPrior] update node CPU failed %v", err)
 			}
 		}
@@ -296,7 +298,7 @@ func (c *Calcium) makeContainerOptions(index int, quota types.CPUMap, opts *type
 
 	var resource enginecontainer.Resources
 	if favor == scheduler.CPU_PRIOR {
-		resource = makeCPUPriorSetting(c.config.Scheduler.ShareBase, quota)
+		resource = makeCPUPriorSetting(c.config.Scheduler.ShareBase, quota, opts.Memory)
 	} else if favor == scheduler.MEMORY_PRIOR {
 		resource = makeMemoryPriorSetting(opts.Memory, opts.CPUQuota)
 	} else {

cluster/calcium/helper.go

@@ -52,23 +52,25 @@ func makeMemoryPriorSetting(memory int64, cpu float64) enginecontainer.Resources
 	return resource
 }
 
-func makeCPUPriorSetting(shareBase int64, quota types.CPUMap) enginecontainer.Resources {
+func makeCPUPriorSetting(shareBase int64, quota types.CPUMap, memory int64) enginecontainer.Resources {
 	// calculate CPUShares and CPUSet
 	// scheduler won't return more than 1 share quota
 	// so the smallest share is the share numerator
 	shareQuota := shareBase
 	cpuIDs := []string{}
-	for cpuid, share := range quota {
-		cpuIDs = append(cpuIDs, cpuid)
+	for cpuID, share := range quota {
+		cpuIDs = append(cpuIDs, cpuID)
 		if share < shareQuota {
 			shareQuota = share
 		}
 	}
-	cpuShares := int64(float64(shareQuota) / float64(shareQuota) * float64(CpuShareBase))
+	cpuShares := int64(float64(shareQuota) / float64(shareBase) * float64(CpuShareBase))
 	cpuSetCpus := strings.Join(cpuIDs, ",")
+	log.Debugf("[makeCPUPriorSetting] CPU core %v CPU share %v Memory soft limit %v", cpuSetCpus, cpuShares, memory)
 	resource := enginecontainer.Resources{
-		CPUShares:  cpuShares,
-		CpusetCpus: cpuSetCpus,
+		CPUShares:         cpuShares,
+		CpusetCpus:        cpuSetCpus,
+		MemoryReservation: memory,
 	}
 	return resource
 }

cluster/calcium/mock_test.go

@@ -423,9 +423,7 @@ func initMockConfig() {
 	mockStore.On("GetNode", podname, "node3").Return(n3, nil)
 	mockStore.On("GetNode", podname, updatenodename).Return(n4, nil)
 	mockStore.On("GetNode", "", "").Return(n2, nil)
-
-	mockStore.On("UpdateNodeMem", podname, mockStringType, mock.AnythingOfType("int64"), mockStringType).Return(nil)
-	mockStore.On("UpdateNodeCPU", podname, mockStringType, mockCPUMapType, mockStringType).Return(nil)
+	mockStore.On("UpdateNodeResource", podname, mockStringType, mockCPUMapType, mock.AnythingOfType("int64"), mockStringType).Return(nil)
 	mockStore.On("UpdateNode", mockNodeType).Return(nil)
 
 	lk := mockstore.MockLock{}