bulky operate remove and diss by node
@@ -13,14 +13,23 @@ import (
// DissociateWorkload dissociate workload from eru, return it resource but not modity it
func (c *Calcium) DissociateWorkload(ctx context.Context, ids []string) (chan *types.DissociateWorkloadMessage, error) {
logger := log.WithField("Caliucm", "Dissociate").WithField("ids", ids)
+
+ nodeWorkloadGroup, err := c.groupWorkloadsByNode(ctx, ids)
+ if err != nil {
+ logger.Errorf("failed to group workloads by node: %+v", err)
+ return nil, errors.WithStack(err)
+ }
+
ch := make(chan *types.DissociateWorkloadMessage)
go func() {
defer close(ch)
- // TODO@zc: group ids by node
- for _, id := range ids {
- err := c.withWorkloadLocked(ctx, id, func(ctx context.Context, workload *types.Workload) error {
- return c.withNodeLocked(ctx, workload.Nodename, func(ctx context.Context, node *types.Node) (err error) {
- err = utils.Txn(
+
+ for nodename, workloadIDs := range nodeWorkloadGroup {
+ if err := c.withNodeLocked(ctx, nodename, func(ctx context.Context, node *types.Node) error {
+ for _, workloadID := range workloadIDs {
+ if err := c.withWorkloadLocked(ctx, workloadID, func(ctx context.Context, workload *types.Workload) error {
+ msg := &types.DissociateWorkloadMessage{WorkloadID: workloadID}
+ if err := utils.Txn(
ctx,
// if
func(ctx context.Context) error {
@@ -32,11 +41,33 @@ func (c *Calcium) DissociateWorkload(ctx context.Context, ids []string) (chan *t
return errors.WithStack(c.store.RemoveWorkload(ctx, workload))
},
// rollback
- func(ctx context.Context, _ bool) error {
+ func(ctx context.Context, failedByCond bool) error {
+ if failedByCond {
+ return nil
+ }
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, store.ActionDecr))
},
c.config.GlobalTimeout,
- )
+ ); err != nil {
+ msg.Error = err
+ logger.WithField("id", workloadID).Errorf("failed to diss workload: %+v", err)
+ }
+ ch <- msg
+ return nil
+ }); err != nil {
+ logger.WithField("id", workloadID).Errorf("failed to lock workload: %+v", err)
+ }
+ }
+ c.doRemapResourceAndLog(ctx, logger, node)
+ return nil
+ }); err != nil {
+ logger.WithField("nodename", nodename).Errorf("failed to lock node: %+v", err)
+ }
+ }
+
+ for _, id := range ids {
+ err := c.withWorkloadLocked(ctx, id, func(ctx context.Context, workload *types.Workload) error {
+ return c.withNodeLocked(ctx, workload.Nodename, func(ctx context.Context, node *types.Node) (err error) {
c.doRemapResourceAndLog(ctx, logger, node)
return err
