fix transactions for realloc and replace

cluster/calcium/realloc.go

@@ -145,18 +145,18 @@ func (c *Calcium) doReallocContainersOnNode(ctx context.Context, ch chan *types.
 
 				// if: commit changes of realloc resources
 				func(ctx context.Context) (err error) {
-
 					for _, plan := range planMap {
 						plan.ApplyChangesOnNode(node, utils.Range(len(containers))...)
 					}
+					rollbacks = utils.Range(len(containers))
+					for _, container := range containers {
+						originalContainers = append(originalContainers, *container)
+					}
 					return c.store.UpdateNodes(ctx, node)
 				},
 
 				// then: update instances' resources
 				func(ctx context.Context) error {
-					for _, container := range containers {
-						originalContainers = append(originalContainers, *container)
-					}
 					rollbacks, err = c.doUpdateResourceOnInstances(ctx, ch, node, planMap, containers, hardVbsMap)
 					return err
 				},

cluster/calcium/replace.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/projecteru2/core/store"
 	"github.com/projecteru2/core/types"
 	"github.com/projecteru2/core/utils"
 	log "github.com/sirupsen/logrus"
@@ -151,23 +150,9 @@ func (c *Calcium) doReplaceContainer(
 				ctx,
 				// if
 				func(ctx context.Context) error {
-					return utils.Txn(
-						ctx,
-						func(ctx context.Context) error {
-							return c.doDeployOneWorkload(ctx, node, &opts.DeployOptions, createMessage, index, -1)
-						},
-						nil,
-						func(ctx context.Context) error {
-							log.Errorf("[doReplaceContainer] Error when create and start a container, %v", createMessage.Error)
-							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,
-					)
+					ctx, cancel := context.WithCancel(ctx)
+					cancel()
+					return c.doDeployOneWorkload(ctx, node, &opts.DeployOptions, createMessage, index, -1)
 				},
 				// then
 				func(ctx context.Context) (err error) {