meta

Author: DuodenumL

cluster/calcium/create.go

@@ -323,7 +323,18 @@ func (c *Calcium) doDeployOneWorkload(
 				}
 			}
 
-			// start first
+			// avoid be interrupted by MakeDeployStatus
+			processing := opts.GetProcessing(node.Name)
+			if !decrProcessing {
+				processing = nil
+			}
+			// add workload metadata first
+			if err := c.store.AddWorkload(ctx, workload, processing); err != nil {
+				return errors.WithStack(err)
+			}
+			log.Infof(ctx, "[doDeployOneWorkload] workload metadata created: %s", workload.ID)
+
+			// start workload
 			msg.Hook, err = c.doStartWorkload(ctx, workload, opts.IgnoreHook)
 			if err != nil {
 				return err
@@ -340,22 +351,20 @@ func (c *Calcium) doDeployOneWorkload(
 			if workloadInfo.Networks != nil {
 				msg.Publish = utils.MakePublishInfo(workloadInfo.Networks, opts.Entrypoint.Publish)
 			}
-			// reset users
-			if workloadInfo.User != workload.User {
+
+			// if workload metadata changed, then update
+			if workloadInfo.User != workload.User || workload.Hook != opts.Entrypoint.Hook {
+				// reset users
 				workload.User = workloadInfo.User
-			}
-			// reset workload.hook
-			workload.Hook = opts.Entrypoint.Hook
+				// reset workload.hook
+				workload.Hook = opts.Entrypoint.Hook
 
-			// avoid be interrupted by MakeDeployStatus
-			processing := opts.GetProcessing(node.Name)
-			if !decrProcessing {
-				processing = nil
-			}
-			if err := c.store.AddWorkload(ctx, workload, processing); err != nil {
-				return errors.WithStack(err)
+				if err := c.store.UpdateWorkload(ctx, workload); err != nil {
+					return errors.WithStack(err)
+				}
 			}
-			log.Infof(ctx, "[doDeployOneWorkload] workload created and saved: %s", workload.ID)
+
+			log.Infof(ctx, "[doDeployOneWorkload] workload updated: %s", workload.ID)
 			msg.WorkloadID = workload.ID
 			msg.WorkloadName = workload.Name
 			msg.Podname = workload.Podname
@@ -368,6 +377,11 @@ func (c *Calcium) doDeployOneWorkload(
 			if workload.ID == "" {
 				return nil
 			}
+
+			if err := c.store.RemoveWorkload(ctx, workload); err != nil {
+				log.Errorf(ctx, "[doDeployOneWorkload] failed to remove workload %s")
+			}
+
 			return workload.Remove(ctx, true)
 		},
 		c.config.GlobalTimeout,

rpc/rpc.go

@@ -615,6 +615,7 @@ func (v *Vibranium) CreateWorkload(opts *pb.DeployOptions, stream pb.CoreRPC_Cre
 		return grpcstatus.Error(CreateWorkload, err.Error())
 	}
 	for m := range ch {
+		log.Debugf(ctx, "create workload message: %v", m)
 		if err = stream.Send(toRPCCreateWorkloadMessage(m)); err != nil {
 			v.logUnsentMessages(ctx, "CreateWorkload", err, m)
 		}

store/etcdv3/meta/etcd.go

@@ -277,7 +277,7 @@ func (e *ETCD) bindStatusWithTTL(ctx context.Context, entityKey, statusKey, stat
 	// There isn't the entity kv pair.
 	if !entityTxn.Succeeded {
 		e.revokeLease(ctx, leaseID)
-		return nil
+		return types.ErrEntityNotExists
 	}
 
 	// There isn't a status bound to the entity.
@@ -311,10 +311,10 @@ func (e *ETCD) bindStatusWithoutTTL(ctx context.Context, statusKey, statusValue
 	updateStatus := []clientv3.Op{clientv3.OpPut(statusKey, statusValue)}
 	_, err := e.cliv3.Txn(ctx).
 		If(clientv3.Compare(clientv3.Version(statusKey), "!=", 0)). // if there's an existing status key
-		Then(clientv3.OpTxn(                                        // deal with existing status key
+		Then(clientv3.OpTxn( // deal with existing status key
 			[]clientv3.Cmp{clientv3.Compare(clientv3.Value(statusKey), "!=", statusValue)}, // if the new value != the old value
-			updateStatus,    // then the status has been changed.
-			[]clientv3.Op{}, // otherwise do nothing.
+			updateStatus,                                                                   // then the status has been changed.
+			[]clientv3.Op{},                                                                // otherwise do nothing.
 		)).
 		Else(updateStatus...). // otherwise deal with non-existing status key
 		Commit()

types/errors.go

@@ -85,6 +85,7 @@ var (
 
 	ErrNodeNotExists     = errors.New("node not exists")
 	ErrWorkloadNotExists = errors.New("workload not exists")
+	ErrEntityNotExists   = errors.New("entity not exists")
 
 	ErrUnregisteredWALEventType = errors.New("unregistered WAL event type")
 	ErrInvalidWALBucket         = errors.New("invalid WAL bucket")