// operateSubTaskStage returns TaskOp.String() additionally to record metrics.
operateSubTaskStage returns TaskOp.String() additionally to record metrics.
// operateSubTaskStage returns TaskOp . String () additionally to record metrics .
func (w *SourceWorker) operateSubTaskStage(stage ha.Stage, subTaskCfg config.SubTaskConfig) (string, error) {
	var op pb.TaskOp
	log.L().Info("operateSubTaskStage",
		zap.String("sourceID", subTaskCfg.SourceID),
		zap.String("task", subTaskCfg.Name),
		zap.Stringer("stage", stage))
	if st := w.subTaskHolder.findSubTask(stage.Task); st == nil {
		switch stage.Expect {
		case pb.Stage_Running, pb.Stage_Paused, pb.Stage_Stopped:
			log.L().Info("start to create subtask in operateSubTaskStage", zap.String("sourceID", subTaskCfg.SourceID), zap.String("task", subTaskCfg.Name))
			expectValidatorStage, err := getExpectValidatorStage(subTaskCfg.ValidatorCfg, w.etcdClient, stage.Source, stage.Task, stage.Revision)
			if err != nil {
				return opErrTypeBeforeOp, terror.Annotate(err, "fail to get validator stage from etcd")
			}
			return opErrTypeBeforeOp, w.StartSubTask(&subTaskCfg, stage.Expect, expectValidatorStage, true)
		default:
			return op.String(), w.OperateSubTask(stage.Task, op)
		}
	}
	switch stage.Expect {
	case pb.Stage_Stopped, pb.Stage_Paused:
		op = pb.TaskOp_Pause
	case pb.Stage_Running:
		op = pb.TaskOp_Resume
	}
	if stage.IsDeleted {
		op = pb.TaskOp_Delete
	}
	return op.String(), w.OperateSubTask(stage.Task, op)
}
func ( w * SourceWorker ) operateSubTaskStage ( stage ha . Stage , subTaskCfg config . SubTaskConfig ) ( string , error ) { var op pb . TaskOp log . L ( ) . Info ( " " , zap . String ( " " , subTaskCfg . SourceID ) , zap . String ( " " , subTaskCfg . Name ) , zap . Stringer ( " " , stage ) ) if st := w . subTaskHolder . findSubTask ( stage . Task ) ; st == nil { switch stage . Expect { case pb . Stage_Running , pb . Stage_Paused , pb . Stage_Stopped : log . L ( ) . Info ( " " , zap . String ( " " , subTaskCfg . SourceID ) , zap . String ( " " , subTaskCfg . Name ) ) expectValidatorStage , err := getExpectValidatorStage ( subTaskCfg . ValidatorCfg , w . etcdClient , stage . Source , stage . Task , stage . Revision ) if err != nil { return opErrTypeBeforeOp , terror . Annotate ( err , " " ) } return opErrTypeBeforeOp , w . StartSubTask ( & subTaskCfg , stage . Expect , expectValidatorStage , true ) default : return op . String ( ) , w . OperateSubTask ( stage . Task , op ) } } switch stage . Expect { case pb . Stage_Stopped , pb . Stage_Paused : op = pb . TaskOp_Pause case pb . Stage_Running : op = pb . TaskOp_Resume } if stage . IsDeleted { op = pb . TaskOp_Delete } return op . String ( ) , w . OperateSubTask ( stage . Task , op ) }
operateSubTaskStage returns TaskOp.String() additionally to record metrics.
operateSubTaskStage returns TaskOp . String () additionally to record metrics .
// for new added subtask // todo refactor here deciding if the expected stage is valid should be put inside StartSubTask and OperateSubTask // not valid stage // todo(ehco) remove pause and resume after using openapi to impl dmctl