Fix graceful shutdown "broken pipe" error (#527)
diff --git a/tfexec/cmd.go b/tfexec/cmd.go
index c8b18ff..cd5a7e2 100644
--- a/tfexec/cmd.go
+++ b/tfexec/cmd.go
@@ -252,20 +252,30 @@
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 {
diff --git a/tfexec/cmd_default.go b/tfexec/cmd_default.go
index 3af11c8..7a5aa81 100644
--- a/tfexec/cmd_default.go
+++ b/tfexec/cmd_default.go
@@ -59,13 +59,13 @@
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
diff --git a/tfexec/cmd_linux.go b/tfexec/cmd_linux.go
index 0565372..43e44a5 100644
--- a/tfexec/cmd_linux.go
+++ b/tfexec/cmd_linux.go
@@ -64,13 +64,13 @@
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
diff --git a/tfexec/internal/e2etest/errors_test.go b/tfexec/internal/e2etest/errors_test.go
index c99aad9..74b695f 100644
--- a/tfexec/internal/e2etest/errors_test.go
+++ b/tfexec/internal/e2etest/errors_test.go
@@ -23,6 +23,8 @@
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,12 @@
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 {
@@ -200,6 +205,51 @@
})
}
+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 {
+ 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)
+ }
+ 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")
+ }
+ })
+}
+
func TestContext_alreadyCancelled(t *testing.T) {
runTest(t, "", func(t *testing.T, tfv *version.Version, tf *tfexec.Terraform) {
ctx, cancel := context.WithCancel(context.Background())
diff --git a/tfexec/terraform.go b/tfexec/terraform.go
index 9088616..9ea21c5 100644
--- a/tfexec/terraform.go
+++ b/tfexec/terraform.go
@@ -73,6 +73,9 @@
// 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 @@
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
