Script run stage (#4720)

* Add option script run stage

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Implement Executor for script run

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Use StageStatus_STAGE_FAILURE

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Add error log

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Add Copyright

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Re-implement script run stage like custom sync

Signed-off-by: Yoshiki Fujikane <[email protected]>

* Delete comment

Signed-off-by: Yoshiki Fujikane <[email protected]>

---------

Signed-off-by: Yoshiki Fujikane <[email protected]>

pkg/app/piped/executor/registry/registry.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/ecs"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/kubernetes"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/lambda"
+	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/scriptrun"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/terraform"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/wait"
 	"github.com/pipe-cd/pipecd/pkg/app/piped/executor/waitapproval"
@@ -112,4 +113,5 @@ func init() {
 	wait.Register(defaultRegistry)
 	waitapproval.Register(defaultRegistry)
 	customsync.Register(defaultRegistry)
+	scriptrun.Register(defaultRegistry)
 }

pkg/app/piped/executor/scriptrun/scriptrun.go

@@ -0,0 +1,135 @@
+// Copyright 2023 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package scriptrun
+
+import (
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/pipe-cd/pipecd/pkg/app/piped/executor"
+	"github.com/pipe-cd/pipecd/pkg/model"
+)
+
+type registerer interface {
+	Register(stage model.Stage, f executor.Factory) error
+	RegisterRollback(kind model.RollbackKind, f executor.Factory) error
+}
+
+type Executor struct {
+	executor.Input
+
+	appDir string
+}
+
+func (e *Executor) Execute(sig executor.StopSignal) model.StageStatus {
+	e.LogPersister.Infof("Start executing the script run stage")
+
+	opts := e.Input.StageConfig.ScriptRunStageOptions
+	if opts == nil {
+		e.LogPersister.Error("option for script run stage not found")
+		return model.StageStatus_STAGE_FAILURE
+	}
+
+	if opts.Run == "" {
+		return model.StageStatus_STAGE_SUCCESS
+	}
+
+	var originalStatus = e.Stage.Status
+	ds, err := e.TargetDSP.Get(sig.Context(), e.LogPersister)
+	if err != nil {
+		e.LogPersister.Errorf("Failed to prepare target deploy source data (%v)", err)
+		return model.StageStatus_STAGE_FAILURE
+	}
+	e.appDir = ds.AppDir
+
+	timeout := e.StageConfig.ScriptRunStageOptions.Timeout.Duration()
+
+	c := make(chan model.StageStatus, 1)
+	go func() {
+		c <- e.executeCommand()
+	}()
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	for {
+		select {
+		case result := <-c:
+			return result
+		case <-timer.C:
+			e.LogPersister.Errorf("Canceled because of timeout")
+			return model.StageStatus_STAGE_FAILURE
+
+		case s := <-sig.Ch():
+			switch s {
+			case executor.StopSignalCancel:
+				e.LogPersister.Info("Canceled by user")
+				return model.StageStatus_STAGE_CANCELLED
+			case executor.StopSignalTerminate:
+				e.LogPersister.Info("Terminated by system")
+				return originalStatus
+			default:
+				e.LogPersister.Error("Unexpected")
+				return model.StageStatus_STAGE_FAILURE
+			}
+		}
+	}
+}
+
+func (e *Executor) executeCommand() model.StageStatus {
+	opts := e.StageConfig.ScriptRunStageOptions
+
+	e.LogPersister.Infof("Runnnig commands...")
+	for _, v := range strings.Split(opts.Run, "\n") {
+		if v != "" {
+			e.LogPersister.Infof("   %s", v)
+		}
+	}
+
+	envs := make([]string, 0, len(opts.Env))
+	for key, value := range opts.Env {
+		envs = append(envs, key+"="+value)
+	}
+
+	cmd := exec.Command("/bin/sh", "-l", "-c", opts.Run)
+	cmd.Dir = e.appDir
+	cmd.Env = append(os.Environ(), envs...)
+	cmd.Stdout = e.LogPersister
+	cmd.Stderr = e.LogPersister
+	if err := cmd.Run(); err != nil {
+		e.LogPersister.Errorf("failed to exec command: %w", err)
+		return model.StageStatus_STAGE_FAILURE
+	}
+	return model.StageStatus_STAGE_SUCCESS
+}
+
+type RollbackExecutor struct {
+	executor.Input
+}
+
+func (e *RollbackExecutor) Execute(sig executor.StopSignal) model.StageStatus {
+	e.LogPersister.Infof("Unimplement: rollbacking the script run stage")
+	return model.StageStatus_STAGE_FAILURE
+}
+
+// Register registers this executor factory into a given registerer.
+func Register(r registerer) {
+	r.Register(model.StageScriptRun, func(in executor.Input) executor.Executor {
+		return &Executor{
+			Input: in,
+		}
+	})
+}

pkg/config/application.go

@@ -225,6 +225,7 @@ type PipelineStage struct {
 	WaitStageOptions         *WaitStageOptions
 	WaitApprovalStageOptions *WaitApprovalStageOptions
 	AnalysisStageOptions     *AnalysisStageOptions
+	ScriptRunStageOptions    *ScriptRunStageOptions
 
 	K8sPrimaryRolloutStageOptions  *K8sPrimaryRolloutStageOptions
 	K8sCanaryRolloutStageOptions   *K8sCanaryRolloutStageOptions
@@ -291,6 +292,12 @@ func (s *PipelineStage) UnmarshalJSON(data []byte) error {
 		if len(gs.With) > 0 {
 			err = json.Unmarshal(gs.With, s.AnalysisStageOptions)
 		}
+	case model.StageScriptRun:
+		s.ScriptRunStageOptions = &ScriptRunStageOptions{}
+		if len(gs.With) > 0 {
+			err = json.Unmarshal(gs.With, s.ScriptRunStageOptions)
+		}
+
 	case model.StageK8sPrimaryRollout:
 		s.K8sPrimaryRolloutStageOptions = &K8sPrimaryRolloutStageOptions{}
 		if len(gs.With) > 0 {
@@ -485,6 +492,22 @@ func (a *AnalysisStageOptions) Validate() error {
 	return nil
 }
 
+// ScriptRunStageOptions contains all configurable values for a SCRIPT_RUN stage.
+type ScriptRunStageOptions struct {
+	Env        map[string]string `json:"env"`
+	Run        string            `json:"run"`
+	Timeout    Duration          `json:"timeout" default:"6h"`
+	OnRollback string            `json:"onRollback"`
+}
+
+// Validate checks the required fields of ScriptRunStageOptions.
+func (s *ScriptRunStageOptions) Validate() error {
+	if s.Run == "" {
+		return fmt.Errorf("SCRIPT_RUN stage requires run field")
+	}
+	return nil
+}
+
 type AnalysisTemplateRef struct {
 	Name    string            `json:"name"`
 	AppArgs map[string]string `json:"appArgs"`

pkg/config/application_test.go

@@ -702,3 +702,32 @@ func TestCustomSyncConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestScriptSycConfiguration(t *testing.T) {
+	testcases := []struct {
+		name    string
+		opts    ScriptRunStageOptions
+		wantErr bool
+	}{
+		{
+			name: "valid",
+			opts: ScriptRunStageOptions{
+				Run: "echo 'hello world'",
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid",
+			opts: ScriptRunStageOptions{
+				Run: "",
+			},
+			wantErr: true,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.opts.Validate()
+			assert.Equal(t, tc.wantErr, err != nil)
+		})
+	}
+}

pkg/model/stage.go

@@ -27,6 +27,9 @@ const (
 	// StageAnalysis represents the waiting state for analysing
 	// the application status based on metrics, log, http request...
 	StageAnalysis Stage = "ANALYSIS"
+	// StageScriptRun represents a state where
+	// the specified script will be executed.
+	StageScriptRun Stage = "SCRIPT_RUN"
 
 	// StageK8sSync represents the state where
 	// all resources should be synced with the Git state.