Fix graceful shutdown "broken pipe" error (#527)

Author: busser

tfexec/cmd.go

@@ -252,20 +252,30 @@ func mergeWriters(writers ...io.Writer) io.Writer {
 	return io.MultiWriter(compact...)
 }
 
-func writeOutput(ctx context.Context, r io.ReadCloser, w io.Writer) error {
-	// ReadBytes will block until bytes are read, which can cause a delay in
-	// returning even if the command's context has been canceled. Use a separate
-	// goroutine to prompt ReadBytes to return on cancel
-	closeCtx, closeCancel := context.WithCancel(ctx)
-	defer closeCancel()
-	go func() {
-		select {
-		case <-ctx.Done():
-			r.Close()
-		case <-closeCtx.Done():
-			return
-		}
-	}()
+func (tf *Terraform) writeOutput(ctx context.Context, r io.ReadCloser, w io.Writer) error {
+	// ReadBytes will block until all bytes are read, which can cause a delay in
+	// returning even if the command's context has been canceled. When the
+	// context is canceled, Terraform receives an interrupt signal and will exit
+	// after a short while. Once the process has exited, the stdio pipes will
+	// close, allowing this function to return.
+
+	if tf.enableLegacyPipeClosing {
+		// Rather than wait for the stdio pipes to close naturally, we can close
+		// them ourselves when the command's context is canceled, causing the
+		// process to exit immediately. This works around a bug in Terraform
+		// < v1.1 that would otherwise leave the process (and this function)
+		// hanging after the context is canceled.
+		closeCtx, closeCancel := context.WithCancel(ctx)
+		defer closeCancel()
+		go func() {
+			select {
+			case <-ctx.Done():
+				r.Close()
+			case <-closeCtx.Done():
+				return
+			}
+		}()
+	}
 
 	buf := bufio.NewReader(r)
 	for {

tfexec/cmd_default.go

@@ -59,13 +59,13 @@ func (tf *Terraform) runTerraformCmd(ctx context.Context, cmd *exec.Cmd) error {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		errStdout = writeOutput(ctx, stdoutPipe, stdoutWriter)
+		errStdout = tf.writeOutput(ctx, stdoutPipe, stdoutWriter)
 	}()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		errStderr = writeOutput(ctx, stderrPipe, stderrWriter)
+		errStderr = tf.writeOutput(ctx, stderrPipe, stderrWriter)
 	}()
 
 	// Reads from pipes must be completed before calling cmd.Wait(). Otherwise

tfexec/cmd_linux.go

@@ -64,13 +64,13 @@ func (tf *Terraform) runTerraformCmd(ctx context.Context, cmd *exec.Cmd) error {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		errStdout = writeOutput(ctx, stdoutPipe, stdoutWriter)
+		errStdout = tf.writeOutput(ctx, stdoutPipe, stdoutWriter)
 	}()
 
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		errStderr = writeOutput(ctx, stderrPipe, stderrWriter)
+		errStderr = tf.writeOutput(ctx, stderrPipe, stderrWriter)
 	}()
 
 	// Reads from pipes must be completed before calling cmd.Wait(). Otherwise

tfexec/internal/e2etest/errors_test.go

@@ -23,6 +23,8 @@ import (
 
 var (
 	protocol5MinVersion = version.Must(version.NewVersion("0.12.0"))
+
+	gracefulShutdownMinVersion = version.Must(version.NewVersion("1.1.0"))
 )
 
 func TestUnparsedError(t *testing.T) {
@@ -168,9 +170,47 @@ func TestContext_sleepTimeoutExpired(t *testing.T) {
 			t.Skip("the ability to interrupt an apply was added in protocol 5.0 in Terraform 0.12, so test is not valid")
 		}
 
-		// sleep will not react to SIGINT
-		// This ensures that process is killed within the expected time limit.
-		tf.SetWaitDelay(500 * time.Millisecond)
+		if !tfv.GreaterThanOrEqual(gracefulShutdownMinVersion) {
+			// Versions < 1.1 will not react to SIGINT.
+			// This ensures the process is killed within the expected time limit.
+			tf.SetEnableLegacyPipeClosing(true)
+			tf.SetWaitDelay(500 * time.Millisecond)
+		}
+
+		err := tf.Init(context.Background())
+		if err != nil {
+			t.Fatalf("err during init: %s", err)
+		}
+
+		ctx := context.Background()
+		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+
+		errCh := make(chan error)
+		go func() {
+			err = tf.Apply(ctx)
+			if err != nil {
+				errCh <- err
+			}
+		}()
+
+		select {
+		case err := <-errCh:
+			if !errors.Is(err, context.DeadlineExceeded) {
+				t.Fatalf("expected context.DeadlineExceeded, got %T %s", err, err)
+			}
+		case <-time.After(time.Second * 10):
+			t.Fatal("terraform apply should have canceled and returned in ~5s")
+		}
+	})
+}
+
+func TestContext_sleepGracefulShutdown(t *testing.T) {
+	runTest(t, "sleep", func(t *testing.T, tfv *version.Version, tf *tfexec.Terraform) {
+		// only testing versions that can shut down gracefully
+		if !tfv.GreaterThanOrEqual(gracefulShutdownMinVersion) {
+			t.Skip("graceful shutdown was added in Terraform 1.1, so test is not valid")
+		}
 
 		err := tf.Init(context.Background())
 		if err != nil {
@@ -194,6 +234,16 @@ func TestContext_sleepTimeoutExpired(t *testing.T) {
 			if !errors.Is(err, context.DeadlineExceeded) {
 				t.Fatalf("expected context.DeadlineExceeded, got %T %s", err, err)
 			}
+			var ee *exec.ExitError
+			if !errors.As(err, &ee) {
+				t.Fatalf("expected exec.ExitError, got %T, %s", err, err)
+			}
+			if !ee.Exited() {
+				t.Fatalf("expected process to have exited, but it did not (%s)", ee.ProcessState.String())
+			}
+			if ee.ExitCode() != 1 {
+				t.Fatalf("expected exit code 1, got %d", ee.ExitCode())
+			}
 		case <-time.After(time.Second * 10):
 			t.Fatal("terraform apply should have canceled and returned in ~5s")
 		}

tfexec/terraform.go

@@ -73,6 +73,9 @@ type Terraform struct {
 	// waitDelay represents the WaitDelay field of the [exec.Cmd] of Terraform
 	waitDelay time.Duration
 
+	// enableLegacyPipeClosing closes the stdout/stderr pipes before calling [exec.Cmd.Wait]
+	enableLegacyPipeClosing bool
+
 	versionLock  sync.Mutex
 	execVersion  *version.Version
 	provVersions map[string]*version.Version
@@ -232,6 +235,16 @@ func (tf *Terraform) SetWaitDelay(delay time.Duration) error {
 	return nil
 }
 
+// SetEnableLegacyPipeClosing causes the library to "force-close" stdio pipes.
+// This works around a bug in Terraform < v1.1 that would otherwise leave
+// the process (and caller) hanging after graceful shutdown.
+//
+// This option can be safely ignored (set to false) with Terraform 1.1+.
+func (tf *Terraform) SetEnableLegacyPipeClosing(enabled bool) error {
+	tf.enableLegacyPipeClosing = enabled
+	return nil
+}
+
 // WorkingDir returns the working directory for Terraform.
 func (tf *Terraform) WorkingDir() string {
 	return tf.workingDir