improve resource managements on replace and realloc (#213)

* replace rollback doCreateAndStart properly

* realloc calculates node resource properly

* pass tests

Author: zc

cluster/calcium/build.go

@@ -83,12 +83,6 @@ func (c *Calcium) buildFromContent(ctx context.Context, node *types.Node, refs [
 }
 
 func (c *Calcium) buildFromExist(ctx context.Context, ref, existID string) (chan *types.BuildImageMessage, error) {
-	buildErrMsg := func(err error) *types.BuildImageMessage {
-		msg := &types.BuildImageMessage{Error: err.Error()}
-		msg.ErrorDetail.Message = err.Error()
-		return msg
-	}
-
 	return withImageBuiltChannel(func(ch chan *types.BuildImageMessage) {
 		node, err := c.getContainerNode(ctx, existID)
 		if err != nil {
@@ -184,3 +178,9 @@ func cleanupNodeImages(node *types.Node, IDs []string, ttl time.Duration) {
 		log.Infof("[BuildImage] Clean cached image and release space %d", spaceReclaimed)
 	}
 }
+
+func buildErrMsg(err error) *types.BuildImageMessage {
+	msg := &types.BuildImageMessage{Error: err.Error()}
+	msg.ErrorDetail.Message = err.Error()
+	return msg
+}

cluster/calcium/create.go

@@ -289,7 +289,6 @@ func (c *Calcium) doCreateAndStartContainer(
 		},
 		c.config.GlobalTimeout,
 	)
-
 	return createContainerMessage
 }
 

cluster/calcium/realloc.go

@@ -229,17 +229,17 @@ func (c *Calcium) updateContainersResources(ctx context.Context, ch chan *types.
 }
 
 func (c *Calcium) updateResource(ctx context.Context, node *types.Node, container *types.Container, newResource *enginetypes.VirtualizationResource) error {
-	if err := node.Engine.VirtualizationUpdateResource(ctx, container.ID, newResource); err == nil {
-		oldVolumes := container.Volumes
+	updateResourceErr := node.Engine.VirtualizationUpdateResource(ctx, container.ID, newResource)
+	if updateResourceErr == nil {
+		oldVolumeSize := container.Volumes.TotalSize()
 		container.CPU = newResource.CPU
 		container.Quota = newResource.Quota
 		container.Memory = newResource.Memory
 		container.Volumes, _ = types.MakeVolumeBindings(newResource.Volumes)
 		container.VolumePlan = types.MustToVolumePlan(newResource.VolumePlan)
-		container.Storage += container.Volumes.TotalSize() - oldVolumes.TotalSize()
+		container.Storage += container.Volumes.TotalSize() - oldVolumeSize
 	} else {
-		log.Errorf("[updateResource] When Realloc container, VirtualizationUpdateResource %s failed %v", container.ID, err)
-		return err
+		log.Errorf("[updateResource] When Realloc container, VirtualizationUpdateResource %s failed %v", container.ID, updateResourceErr)
 	}
 	// 成功失败都需要修改 node 的占用
 	// 成功的话，node 占用为新资源
@@ -257,10 +257,9 @@ func (c *Calcium) updateResource(ctx context.Context, node *types.Node, containe
 	// since we don't rollback VirutalUpdateResource, client can't interrupt
 	if err := c.store.UpdateContainer(context.Background(), container); err != nil {
 		log.Errorf("[updateResource] Realloc finish but update container %s failed %v", container.ID, err)
-
 		return err
 	}
-	return nil
+	return updateResourceErr
 }
 
 func (c *Calcium) reallocVolume(node *types.Node, containers []*types.Container, vbs types.VolumeBindings) (plans map[*types.Container]types.VolumePlan, err error) {

cluster/calcium/realloc_test.go

@@ -2,9 +2,10 @@ package calcium
 
 import (
 	"context"
-	complexscheduler "github.com/projecteru2/core/scheduler/complex"
 	"testing"
 
+	complexscheduler "github.com/projecteru2/core/scheduler/complex"
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/docker/go-units"
@@ -135,6 +136,7 @@ func TestRealloc(t *testing.T) {
 	simpleMockScheduler.On("SelectCPUNodes", mock.Anything, mock.Anything, mock.Anything).Return(nil, nodeCPUPlans, 2, nil).Times(5)
 	// failed by apply resource
 	engine.On("VirtualizationUpdateResource", mock.Anything, mock.Anything, mock.Anything).Return(types.ErrBadContainerID).Twice()
+	store.On("UpdateContainer", mock.Anything, mock.Anything).Return(nil).Twice()
 	// update node failed
 	store.On("UpdateNode", mock.Anything, mock.Anything).Return(types.ErrNoETCD).Once()
 	// reset node

cluster/calcium/replace.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"sync"
 
+	"github.com/projecteru2/core/store"
 	"github.com/projecteru2/core/types"
 	"github.com/projecteru2/core/utils"
 	log "github.com/sirupsen/logrus"
@@ -141,24 +142,43 @@ func (c *Calcium) doReplaceContainer(
 				ctx,
 				// if
 				func(ctx context.Context) error {
-					createMessage = c.doCreateAndStartContainer(ctx, index, node, &opts.DeployOptions, container.CPU, container.VolumePlan)
-					return createMessage.Error
+					return utils.Txn(
+						ctx,
+						func(ctx context.Context) error {
+							createMessage = c.doCreateAndStartContainer(ctx, index, node, &opts.DeployOptions, container.CPU, container.VolumePlan)
+							return createMessage.Error
+						},
+						nil,
+						func(ctx context.Context) error {
+							log.Errorf("[doReplaceContainer] Error when create and start a container, %v", createMessage.Error)
+							if createMessage.ContainerID != "" {
+								log.Warnf("[doReplaceContainer] Create container failed %v, and container %s not removed", createMessage.Error, createMessage.ContainerID)
+								return nil
+							}
+							if err = c.withNodeLocked(ctx, node.Name, func(node *types.Node) error {
+								return c.store.UpdateNodeResource(ctx, node, createMessage.CPU, createMessage.Quota, createMessage.Memory, createMessage.Storage, createMessage.VolumePlan.IntoVolumeMap(), store.ActionIncr)
+							}); err != nil {
+								log.Errorf("[doReplaceContainer] Reset node resource %s failed %v", node.Name, err)
+							}
+							return nil
+						},
+						c.config.GlobalTimeout,
+					)
 				},
 				// then
 				func(ctx context.Context) (err error) {
 					if err = c.doRemoveContainer(ctx, container, true); err != nil {
-						log.Errorf("[replaceAndRemove] the new started but the old failed to stop")
+						log.Errorf("[doReplaceContainer] the new started but the old failed to stop")
 						return
 					}
 					removeMessage.Success = true
 					return
 				},
-				// else
 				nil,
 				c.config.GlobalTimeout,
 			)
 		},
-		// else
+		// rollback
 		func(ctx context.Context) (err error) {
 			messages, err := c.doStartContainer(ctx, container, opts.IgnoreHook)
 			if err != nil {

cluster/calcium/replace_test.go

@@ -134,6 +134,7 @@ func TestReplaceContainer(t *testing.T) {
 	// failed by VirtualizationCreate
 	engine.On("VirtualizationCreate", mock.Anything, mock.Anything).Return(nil, types.ErrCannotGetEngine).Once()
 	engine.On("VirtualizationStart", mock.Anything, mock.Anything).Return(types.ErrCannotGetEngine).Once()
+	store.On("UpdateNodeResource", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
 	ch, err = c.ReplaceContainer(ctx, opts)
 	assert.NoError(t, err)
 	for r := range ch {

cluster/calcium/resource.go

@@ -211,7 +211,7 @@ func (c *Calcium) doAllocResource(ctx context.Context, opts *types.DeployOptions
 						nodesInfo[i].VolumePlans = nodeVolumePlans[nodeInfo.Name][:nodeInfo.Deploy]
 					}
 					if err = c.store.UpdateNodeResource(ctx, nodes[nodeInfo.Name], cpuCost, quotaCost, memoryCost, storageCost, volumeCost, store.ActionDecr); err != nil {
-						return err
+						return err // due to ctx lifecircle, this will be interrupted by client
 					}
 					track = i
 				}