Handle skipped stage by piped

pkg/app/piped/apistore/commandstore/store.go

@@ -126,7 +126,7 @@ func (s *store) sync(ctx context.Context) error {
 			applicationCommands = append(applicationCommands, s.makeReportableCommand(cmd))
 		case model.Command_CANCEL_DEPLOYMENT:
 			deploymentCommands = append(deploymentCommands, s.makeReportableCommand(cmd))
-		case model.Command_APPROVE_STAGE:
+		case model.Command_APPROVE_STAGE, model.Command_SKIP_STAGE:
 			stageCommands = append(stageCommands, s.makeReportableCommand(cmd))
 		case model.Command_BUILD_PLAN_PREVIEW:
 			planPreviewCommands = append(planPreviewCommands, s.makeReportableCommand(cmd))

pkg/app/piped/controller/scheduler.go

@@ -322,9 +322,9 @@ func (s *scheduler) Run(ctx context.Context) error {
 			break
 		}
 
-		// If all operations of the stage were completed successfully
+		// If all operations of the stage were completed successfully or skipped by a web user
 		// handle the next stage.
-		if result == model.StageStatus_STAGE_SUCCESS {
+		if result == model.StageStatus_STAGE_SUCCESS || result == model.StageStatus_STAGE_SKIPPED {
 			continue
 		}
 

pkg/app/piped/executor/analysis/analysis.go

@@ -33,6 +33,10 @@ import (
 	"github.com/pipe-cd/pipecd/pkg/model"
 )
 
+const (
+	skippedByKey = "SkippedBy"
+)
+
 type Executor struct {
 	executor.Input
 
@@ -96,6 +100,31 @@ func (e *Executor) Execute(sig executor.StopSignal) model.StageStatus {
 
 	eg, ctxWithTimeout := errgroup.WithContext(ctxWithTimeout)
 
+	// Sync the skip command.
+	var (
+		status = model.StageStatus_STAGE_SUCCESS
+		doneCh = make(chan struct{})
+	)
+	defer close(doneCh)
+	go func() {
+		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				if !e.checkSkipped(ctx) {
+					continue
+				}
+				status = model.StageStatus_STAGE_SKIPPED
+				// Stop the context to cancel all running analysises.
+				ctxWithTimeout.Done()
+				return
+			case <-doneCh:
+				return
+			}
+		}
+	}()
+
 	// Run analyses with metrics providers.
 	for i := range options.Metrics {
 		cfg, err := e.getMetricsConfig(options.Metrics[i], templateCfg)
@@ -146,7 +175,7 @@ func (e *Executor) Execute(sig executor.StopSignal) model.StageStatus {
 		return model.StageStatus_STAGE_FAILURE
 	}
 
-	status := executor.DetermineStageStatus(sig.Signal(), e.Stage.Status, model.StageStatus_STAGE_SUCCESS)
+	status = executor.DetermineStageStatus(sig.Signal(), e.Stage.Status, status)
 	if status != model.StageStatus_STAGE_SUCCESS {
 		return status
 	}
@@ -307,3 +336,29 @@ func (e *Executor) buildAppArgs(customArgs map[string]string) argsTemplate {
 	args.K8s.Namespace = namespace
 	return args
 }
+
+func (e *Executor) checkSkipped(ctx context.Context) bool {
+	var skipCmd *model.ReportableCommand
+	commands := e.CommandLister.ListCommands()
+
+	for i, cmd := range commands {
+		if cmd.GetSkipStage() != nil {
+			skipCmd = &commands[i]
+			break
+		}
+	}
+	if skipCmd == nil {
+		return false
+	}
+
+	if err := e.MetadataStore.Stage(e.Stage.Id).Put(ctx, skippedByKey, skipCmd.Commander); err != nil {
+		e.LogPersister.Errorf("Unable to save the commander who skipped the stage information to deployment, %v", err)
+	}
+	e.LogPersister.Infof("Got the skip command from %q", skipCmd.Commander)
+	e.LogPersister.Infof("This stage has been skipped by user (%s)", skipCmd.Commander)
+
+	if err := skipCmd.Report(ctx, model.CommandStatus_COMMAND_SUCCEEDED, nil, nil); err != nil {
+		e.Logger.Error("failed to report handled command", zap.Error(err))
+	}
+	return true
+}