[SPVR-80] add workload metadata before workload start

Author: DuodenumL

cluster/calcium/create.go

@@ -299,6 +299,17 @@ func (c *Calcium) doDeployOneWorkload(
 		},
 
 		func(ctx context.Context) (err error) {
+			// 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 %s metadata created", workload.ID)
+
 			// Copy data to workload
 			if len(opts.Files) > 0 {
 				for _, file := range opts.Files {
@@ -323,12 +334,15 @@ func (c *Calcium) doDeployOneWorkload(
 				}
 			}
 
-			// start first
+			// start workload
 			msg.Hook, err = c.doStartWorkload(ctx, workload, opts.IgnoreHook)
 			if err != nil {
 				return err
 			}
 
+			// reset workload.hook
+			workload.Hook = opts.Entrypoint.Hook
+
 			// inspect real meta
 			var workloadInfo *enginetypes.VirtualizationInfo
 			workloadInfo, err = workload.Inspect(ctx) // 补充静态元数据
@@ -340,22 +354,18 @@ func (c *Calcium) doDeployOneWorkload(
 			if workloadInfo.Networks != nil {
 				msg.Publish = utils.MakePublishInfo(workloadInfo.Networks, opts.Entrypoint.Publish)
 			}
-			// reset users
+
+			// if workload metadata changed, then update
 			if workloadInfo.User != workload.User {
+				// reset users
 				workload.User = workloadInfo.User
-			}
-			// 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 %s metadata updated", workload.ID)
 			}
-			log.Infof(ctx, "[doDeployOneWorkload] workload created and saved: %s", workload.ID)
+
 			msg.WorkloadID = workload.ID
 			msg.WorkloadName = workload.Name
 			msg.Podname = workload.Podname
@@ -365,9 +375,15 @@ func (c *Calcium) doDeployOneWorkload(
 
 		// remove workload
 		func(ctx context.Context, _ bool) error {
+			log.Errorf(ctx, "[doDeployOneWorkload] failed to deploy workload %s, rollback", workload.ID)
 			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,

cluster/calcium/create_test.go

@@ -149,6 +149,7 @@ func TestCreateWorkloadTxn(t *testing.T) {
 	store.On("MakeDeployStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(
 		errors.Wrap(context.DeadlineExceeded, "MakeDeployStatus"),
 	).Once()
+	store.On("RemoveWorkload", mock.Anything, mock.Anything).Return(nil)
 
 	ch, err := c.CreateWorkload(ctx, opts)
 	assert.Nil(t, err)

engine/docker/helper_test.go

@@ -29,7 +29,7 @@ func TestWithDumpFiles(t *testing.T) {
 	}
 	fp := []string{}
 	for target, content := range data {
-		withTarfileDump(context.TODO(), target, bytes.NewBuffer(content), func(target, tarfile string) error {
+		withTarfileDump(context.TODO(), target, content, 0, 0, 0, func(target, tarfile string) error {
 			assert.True(t, strings.HasPrefix(target, "/tmp/test"))
 			fp = append(fp, tarfile)
 			_, err := os.Stat(tarfile)

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, "[CreateWorkload] 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.

store/etcdv3/meta/etcd_test.go

@@ -102,7 +102,7 @@ func TestBindStatusButEntityTxnUnsuccessful(t *testing.T) {
 
 	etcd.On("Grant", mock.Anything, mock.Anything).Return(&clientv3.LeaseGrantResponse{}, nil).Once()
 	etcd.On("Txn", mock.Anything).Return(txn).Once()
-	require.Equal(t, nil, e.BindStatus(context.Background(), "/entity", "/status", "status", 1))
+	require.Equal(t, types.ErrEntityNotExists, e.BindStatus(context.Background(), "/entity", "/status", "status", 1))
 }
 
 func TestBindStatusButStatusTxnUnsuccessful(t *testing.T) {

store/etcdv3/workload_test.go

@@ -153,7 +153,7 @@ func TestSetWorkloadStatus(t *testing.T) {
 	workload.StatusMeta.Nodename = "n1"
 	// no workload, err nil
 	err = m.SetWorkloadStatus(ctx, workload.StatusMeta, 10)
-	assert.NoError(t, err)
+	assert.Equal(t, err, types.ErrEntityNotExists)
 	assert.NoError(t, m.AddWorkload(ctx, workload, nil))
 	// no status key, put succ, err nil
 	err = m.SetWorkloadStatus(ctx, workload.StatusMeta, 10)

store/redis/rediaron.go

@@ -272,10 +272,10 @@ func (r *Rediaron) BindStatus(ctx context.Context, entityKey, statusKey, statusV
 	if err != nil {
 		return err
 	}
-	// doesn't exist, returns nil, does nothing
+	// doesn't exist, returns error
 	// to behave just like etcd
 	if count != 1 {
-		return nil
+		return types.ErrEntityNotExists
 	}
 
 	_, err = r.cli.Set(ctx, statusKey, statusValue, time.Duration(ttl)*time.Second).Result()

store/redis/workload_test.go

@@ -157,7 +157,7 @@ func (s *RediaronTestSuite) TestSetWorkloadStatus() {
 	workload.StatusMeta.Nodename = "n1"
 	// no workload, err nil
 	err = m.SetWorkloadStatus(ctx, workload.StatusMeta, 10)
-	s.NoError(err)
+	s.ErrorIs(err, types.ErrEntityNotExists)
 	s.NoError(m.AddWorkload(ctx, workload, nil))
 	// no status key, put succ, err nil
 	err = m.SetWorkloadStatus(ctx, workload.StatusMeta, 10)

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")