pipe-cd · omlahore · Sep 4, 2026
@@ -22,6 +22,8 @@ import (
 	"os/exec"
 	"strconv"
 	"strings"
+	"syscall"
+	"time"
 
 	sdk "github.com/pipe-cd/piped-plugin-sdk-go"
 )
@@ -30,7 +32,13 @@ const (
 	stageScriptRun         = "SCRIPT_RUN"
 	stageScriptRunRollback = "SCRIPT_RUN_ROLLBACK"
 	metadataKeyPrefix      = "started-"
-	nonEmptyValue          = "_"
+
+	// commandTerminationGracePeriod is how long the script's process group gets
+	// after SIGTERM before it is SIGKILLed. Long-running commands such as
+	// terraform apply may want this configurable; it is a constant for now
+	// because nothing in the stage config exposes it yet.
+	commandTerminationGracePeriod = 2 * time.Second
+	nonEmptyValue                 = "_"
 )
 
 type ContextInfo struct {
@@ -110,7 +118,7 @@ func executeScriptRun(ctx context.Context, request sdk.ExecuteStageRequest[struc
 	}
 	c := make(chan sdk.StageStatus, 1)
 	go func() {
-		c <- executeCommand(opts.Run, opts.Env, request, lp)
+		c <- executeCommand(ctx, opts.Run, opts.Env, request, lp)
 	}()
 	select {
 	case result := <-c:
@@ -143,7 +151,7 @@ func executeRollback(ctx context.Context, request sdk.ExecuteStageRequest[struct
 	}
 	c := make(chan sdk.StageStatus, 1)
 	go func() {
-		c <- executeCommand(opts.OnRollback, opts.Env, request, lp)
+		c <- executeCommand(ctx, opts.OnRollback, opts.Env, request, lp)
 	}()
 	select {
 	case result := <-c:
@@ -158,7 +166,7 @@ func executeRollback(ctx context.Context, request sdk.ExecuteStageRequest[struct
 func (p *plugin) FetchDefinedStages() []string {
 	return []string{stageScriptRun, stageScriptRunRollback}
 }
-func executeCommand(commands string, customEnv map[string]string, request sdk.ExecuteStageRequest[struct{}], lp sdk.StageLogPersister) sdk.StageStatus {
+func executeCommand(ctx context.Context, commands string, customEnv map[string]string, request sdk.ExecuteStageRequest[struct{}], lp sdk.StageLogPersister) sdk.StageStatus {
 	lp.Infof("Running commands...")
 	for _, v := range strings.Split(commands, "\n") {
 		if v != "" {
@@ -191,13 +199,43 @@ func executeCommand(commands string, customEnv map[string]string, request sdk.Ex
 		envs = append(envs, key+"="+value)
 	}
 
-	cmd := exec.Command("/bin/sh", "-l", "-c", commands)
+	cmd := exec.CommandContext(ctx, "/bin/sh", "-l", "-c", commands)
 	cmd.Env = append(os.Environ(), envs...)
 	cmd.Dir = request.TargetDeploymentSource.ApplicationDirectory
 	cmd.Stdout = lp
 	cmd.Stderr = lp
-	if err := cmd.Run(); err != nil {
-		lp.Errorf("failed to exec command: %w", err)
+
+	// Run the shell as its own process group leader so the whole tree can be
+	// signalled. Without this, cancelling the stage kills /bin/sh and leaves
+	// whatever it spawned still running against the cluster.
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+
+	// exec.CommandContext's default cancel only signals the direct child, and
+	// its WaitDelay fallback calls Process.Kill(), which is also just the
+	// child. Signal the group instead, then escalate to SIGKILL on the group so
+	// a grandchild that ignores SIGTERM cannot outlive the stage.
+	stopEscalation := make(chan struct{})
+	cmd.Cancel = func() error {
+		pgid := cmd.Process.Pid
+		lp.Infof("Cancelling script, sending SIGTERM to process group %d", pgid)
+		if err := syscall.Kill(-pgid, syscall.SIGTERM); err != nil {
+			return err
+		}
+		go func() {
+			select {
+			case <-stopEscalation:
+			case <-time.After(commandTerminationGracePeriod):
+				lp.Infof("Script did not exit within %s, sending SIGKILL to process group %d", commandTerminationGracePeriod, pgid)
+				_ = syscall.Kill(-pgid, syscall.SIGKILL)
+			}
+		}()
+		return nil
+	}
+
+	err = cmd.Run()
+	close(stopEscalation)
+	if err != nil {
+		lp.Errorf("failed to exec command: %v", err)
 		return sdk.StageStatusFailure
 	} else {
 		return sdk.StageStatusSuccess

@@ -0,0 +1,113 @@
+// Copyright 2026 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 main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	sdk "github.com/pipe-cd/piped-plugin-sdk-go"
+	"github.com/pipe-cd/piped-plugin-sdk-go/logpersister/logpersistertest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// alive reports whether pid is still running. Signal 0 performs the permission
+// and existence checks without delivering anything.
+func alive(pid int) bool {
+	return syscall.Kill(pid, 0) == nil
+}
+
+func waitGone(pid int, d time.Duration) bool {
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		if !alive(pid) {
+			return true
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return !alive(pid)
+}
+
+// A cancelled stage must take the whole process tree with it. The script here
+// backgrounds a child that traps SIGTERM and keeps running, which is what a
+// naive kill of /bin/sh alone would leave behind holding cluster state.
+func TestExecuteCommandKillsProcessGroupOnCancel(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	pidFile := filepath.Join(dir, "child.pid")
+
+	script := fmt.Sprintf(`
+trap '' TERM
+( trap '' TERM; echo $$ > %q; while true; do sleep 0.05; done ) &
+while true; do sleep 0.05; done
+`, pidFile)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	done := make(chan sdk.StageStatus, 1)
+	go func() {
+		done <- executeCommand(ctx, script, nil, sdk.ExecuteStageRequest[struct{}]{
+			StageName:  stageScriptRun,
+			Deployment: sdk.Deployment{ID: "deployment-1", ApplicationID: "app-1"},
+		}, logpersistertest.NewTestLogPersister(t))
+	}()
+
+	// Wait for the grandchild to record its pid.
+	var childPID int
+	require.Eventually(t, func() bool {
+		b, err := os.ReadFile(pidFile)
+		if err != nil {
+			return false
+		}
+		_, err = fmt.Sscanf(string(b), "%d", &childPID)
+		return err == nil && childPID > 0
+	}, 10*time.Second, 20*time.Millisecond, "grandchild never started")
+
+	require.True(t, alive(childPID), "grandchild should be running before cancel")
+
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("executeCommand did not return after cancel")
+	}
+
+	assert.True(t, waitGone(childPID, 15*time.Second),
+		"grandchild %d survived cancellation; the process group was not terminated", childPID)
+}
+
+// The normal path must be unaffected: a command that finishes on its own still
+// reports success and does not wait out the grace period.
+func TestExecuteCommandSucceedsWithoutCancel(t *testing.T) {
+	t.Parallel()
+
+	start := time.Now()
+	status := executeCommand(context.Background(), "echo hello", nil, sdk.ExecuteStageRequest[struct{}]{
+		StageName:  stageScriptRun,
+		Deployment: sdk.Deployment{ID: "deployment-1", ApplicationID: "app-1"},
+	}, logpersistertest.NewTestLogPersister(t))
+
+	assert.Equal(t, sdk.StageStatusSuccess, status)
+	assert.Less(t, time.Since(start), commandTerminationGracePeriod,
+		"a command that exits on its own must not wait for the termination grace period")
+}