Fixing modules output corruption (#3459)

* fix: modules output corruption

* chore: fix strict lint

* chore: fix fmt

* chore: fix tests

* chore: fix tests

* chore: fix tests

* chore: fix tests

* chore: fix tests

* chore: fix tests

* chore: update testdata

* fix: Removing non non-interactive logic in err stream redirect (#3464)

* chore: revert code

* Redirect engine messages to stderr (#3468)

* Add option to redirect stdout to stderr

* Engine tests update

* Tests update

* Lint issues update

* chore: log message

* chore: revert testdatda

---------

Co-authored-by: Yousif Akbar <[email protected]>
Co-authored-by: Denis O <[email protected]>

configstack/module.go

@@ -27,6 +27,7 @@ const existingModulesCacheName = "existingModules"
 // TerraformModule represents a single module (i.e. folder with Terraform templates), including the Terragrunt configuration for that
 // module and the list of other modules that this module depends on
 type TerraformModule struct {
+	*Stack
 	Path                 string
 	Dependencies         TerraformModules
 	Config               config.TerragruntConfig
@@ -52,6 +53,18 @@ func (module *TerraformModule) MarshalJSON() ([]byte, error) {
 	return json.Marshal(module.Path)
 }
 
+// FlushOutput flushes buffer data to the output writer.
+func (module *TerraformModule) FlushOutput() error {
+	if writer, ok := module.TerragruntOptions.Writer.(*ModuleWriter); ok {
+		module.outputMu.Lock()
+		defer module.outputMu.Unlock()
+
+		return writer.Flush()
+	}
+
+	return nil
+}
+
 // Check for cycles using a depth-first-search as described here:
 // https://en.wikipedia.org/wiki/Topological_sorting#Depth-first_search
 //

configstack/module_writer.go

@@ -0,0 +1,46 @@
+package configstack
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+
+	"github.com/gruntwork-io/terragrunt/internal/errors"
+)
+
+// ModuleWriter represents a Writer with data buffering.
+// We should avoid outputting data directly to the output out,
+// since when modules run in parallel, the output data may be mixed with each other, thereby spoiling each other's results.
+type ModuleWriter struct {
+	buffer *bytes.Buffer
+	out    io.Writer
+}
+
+// NewModuleWriter returns a new ModuleWriter instance.
+func NewModuleWriter(out io.Writer) *ModuleWriter {
+	return &ModuleWriter{
+		buffer: &bytes.Buffer{},
+		out:    out,
+	}
+}
+
+// Write appends the contents of p to the buffer.
+func (writer *ModuleWriter) Write(p []byte) (int, error) {
+	n, err := writer.buffer.Write(p)
+	if err != nil {
+		return n, errors.New(err)
+	}
+
+	return n, nil
+}
+
+// Flush flushes buffer data to the `out` writer.
+func (writer *ModuleWriter) Flush() error {
+	if _, err := fmt.Fprint(writer.out, writer.buffer); err != nil {
+		return errors.New(err)
+	}
+
+	writer.buffer.Reset()
+
+	return nil
+}

configstack/options.go

@@ -5,18 +5,17 @@ import (
 	"github.com/gruntwork-io/terragrunt/config/hclparse"
 )
 
-type Option func(Stack) Stack
+// Option is type for passing options to the Stack.
+type Option func(*Stack)
 
 func WithChildTerragruntConfig(config *config.TerragruntConfig) Option {
-	return func(stack Stack) Stack {
+	return func(stack *Stack) {
 		stack.childTerragruntConfig = config
-		return stack
 	}
 }
 
 func WithParseOptions(parserOptions []hclparse.Option) Option {
-	return func(stack Stack) Stack {
+	return func(stack *Stack) {
 		stack.parserOptions = parserOptions
-		return stack
 	}
 }

configstack/running_module.go

@@ -110,6 +110,15 @@ func (module *RunningModule) waitForDependencies() error {
 	return nil
 }
 
+func (module *RunningModule) runTerragrunt(ctx context.Context, opts *options.TerragruntOptions) error {
+	opts.Logger.Debugf("Running %s", module.Module.Path)
+	opts.Writer = NewModuleWriter(opts.Writer)
+
+	defer module.Module.FlushOutput() //nolint:errcheck
+
+	return opts.RunTerragrunt(ctx, opts)
+}
+
 // Run a module right now by executing the RunTerragrunt command of its TerragruntOptions field.
 func (module *RunningModule) runNow(ctx context.Context, rootOptions *options.TerragruntOptions) error {
 	module.Status = Running
@@ -118,9 +127,7 @@ func (module *RunningModule) runNow(ctx context.Context, rootOptions *options.Te
 		module.Module.TerragruntOptions.Logger.Debugf("Assuming module %s has already been applied and skipping it", module.Module.Path)
 		return nil
 	} else {
-		module.Module.TerragruntOptions.Logger.Debugf("Running module %s now", module.Module.Path)
-
-		if err := module.Module.TerragruntOptions.RunTerragrunt(ctx, module.Module.TerragruntOptions); err != nil {
+		if err := module.runTerragrunt(ctx, module.Module.TerragruntOptions); err != nil {
 			return err
 		}
 
@@ -311,6 +318,7 @@ func (modules RunningModules) runModules(ctx context.Context, opts *options.Terr
 
 		go func(module *RunningModule) {
 			defer waitGroup.Done()
+
 			module.runModuleWhenReady(ctx, opts, semaphore)
 		}(module)
 	}

configstack/stack.go

@@ -13,6 +13,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/gruntwork-io/go-commons/collections"
 	"github.com/gruntwork-io/terragrunt/cli/commands/terraform/creds"
@@ -35,6 +36,7 @@ type Stack struct {
 	terragruntOptions     *options.TerragruntOptions
 	childTerragruntConfig *config.TerragruntConfig
 	Modules               TerraformModules
+	outputMu              sync.Mutex
 }
 
 // FindStackInSubfolders finds all the Terraform modules in the subfolders of the working directory of the given TerragruntOptions and
@@ -78,7 +80,7 @@ func NewStack(terragruntOptions *options.TerragruntOptions, opts ...Option) *Sta
 
 func (stack *Stack) WithOptions(opts ...Option) *Stack {
 	for _, opt := range opts {
-		*stack = opt(*stack)
+		opt(stack)
 	}
 
 	return stack
@@ -617,7 +619,7 @@ func (stack *Stack) resolveTerraformModule(ctx context.Context, terragruntConfig
 		return nil, nil
 	}
 
-	return &TerraformModule{Path: modulePath, Config: *terragruntConfig, TerragruntOptions: opts}, nil
+	return &TerraformModule{Stack: stack, Path: modulePath, Config: *terragruntConfig, TerragruntOptions: opts}, nil
 }
 
 // resolveDependenciesForModule looks through the dependencies of the given module and resolve the dependency paths listed in the module's config.

internal/errors/errors.go

@@ -13,7 +13,7 @@ const (
 )
 
 // New creates a new instance of Error.
-// If the given value does not contain an stack trace, it will be created.
+// If the given value does not contain a stack trace, it will be created.
 func New(val any) error {
 	if val == nil {
 		return nil
@@ -24,7 +24,7 @@ func New(val any) error {
 
 // Errorf creates a new error with the given format and values.
 // It can be used as a drop-in replacement for fmt.Errorf() to provide descriptive errors in return values.
-// If none of the given values contains an stack trace, it will be created.
+// If none of the given values contains a stack trace, it will be created.
 func Errorf(format string, vals ...any) error {
 	return errorfWithSkip(errorfSkip, format, vals...)
 }

internal/errors/export.go

@@ -13,11 +13,6 @@ func Is(err, target error) bool {
 	return errors.Is(err, target)
 }
 
-// // New returns an error that formats as the given text.
-// func New(text string) error {
-// 	return errors.New(text)
-// }
-
 // Join returns an error that wraps the given errors.
 func Join(errs ...error) error {
 	return errors.Join(errs...)

test/fixtures/buffer-module-output/app/main.tf

@@ -0,0 +1,41 @@
+terraform {
+  required_providers {
+    null = {
+      source  = "registry.terraform.io/hashicorp/null"
+      version = "2.1.2"
+    }
+  }
+}
+
+provider "null" {}
+
+# Create a large string by repeating a smaller string multiple times
+resource "null_resource" "large_json" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_"
+    ])
+  }
+}
+
+resource "null_resource" "large_json_2" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_1024"
+    ])
+  }
+}
+
+
+output "large_json_output" {
+  value = null_resource.large_json[0].triggers.large_data
+}
+
+
+output "large_json_output_2" {
+  value = null_resource.large_json_2[0].triggers.large_data
+}

test/fixtures/buffer-module-output/app2/main.tf

@@ -0,0 +1,42 @@
+provider "null" {}
+
+terraform {
+  required_providers {
+    null = {
+      source  = "registry.terraform.io/hashicorp/null"
+      version = "2.1.2"
+    }
+  }
+}
+
+
+# Create a large string by repeating a smaller string multiple times
+resource "null_resource" "large_json" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_"
+    ])
+  }
+}
+
+resource "null_resource" "large_json_2" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_1024"
+    ])
+  }
+}
+
+
+output "large_json_output" {
+  value = null_resource.large_json[0].triggers.large_data
+}
+
+
+output "large_json_output_2" {
+  value = null_resource.large_json_2[0].triggers.large_data
+}

test/fixtures/buffer-module-output/app3/main.tf

@@ -0,0 +1,41 @@
+terraform {
+  required_providers {
+    null = {
+      source  = "registry.terraform.io/hashicorp/null"
+      version = "2.1.2"
+    }
+  }
+}
+
+provider "null" {}
+
+# Create a large string by repeating a smaller string multiple times
+resource "null_resource" "large_json" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_"
+    ])
+  }
+}
+
+resource "null_resource" "large_json_2" {
+  count = 1
+
+  triggers = {
+    large_data = join("", [
+      for i in range(0, 1024) : "ThisIsAVeryLongStringRepeatedManyTimesToCreateLargeDataBlock_1024"
+    ])
+  }
+}
+
+
+output "large_json_output" {
+  value = null_resource.large_json[0].triggers.large_data
+}
+
+
+output "large_json_output_2" {
+  value = null_resource.large_json_2[0].triggers.large_data
+}

test/integration_test.go

@@ -102,6 +102,7 @@ const (
 	testFixtureStdout                         = "fixtures/download/stdout-test"
 	testFixtureTfTest                         = "fixtures/tftest/"
 	testFixtureErrorPrint                     = "fixtures/error-print"
+	testFixtureBufferModuleOutput             = "fixtures/buffer-module-output"
 
 	terraformFolder = ".terraform"
 
@@ -116,6 +117,30 @@ const (
 	tofuBinary      = "tofu"
 )
 
+func TestBufferModuleOutput(t *testing.T) {
+	t.Parallel()
+
+	cleanupTerraformFolder(t, testFixtureBufferModuleOutput)
+	tmpEnvPath := copyEnvironment(t, testFixtureBufferModuleOutput)
+	rootPath := util.JoinPath(tmpEnvPath, testFixtureBufferModuleOutput)
+
+	_, _, err := runTerragruntCommandWithOutput(t, "terragrunt run-all plan -out planfile --terragrunt-log-disable --terragrunt-working-dir "+rootPath)
+	require.NoError(t, err)
+
+	stdout, _, err := runTerragruntCommandWithOutput(t, "terragrunt run-all show -json planfile --terragrunt-non-interactive --terragrunt-log-disable --terragrunt-working-dir "+rootPath)
+	require.NoError(t, err)
+
+	for _, stdout := range strings.Split(stdout, "\n") {
+		if stdout == "" {
+			continue
+		}
+
+		var objmap map[string]json.RawMessage
+		err = json.Unmarshal([]byte(stdout), &objmap)
+		require.NoError(t, err)
+	}
+}
+
 func TestDisableLogging(t *testing.T) {
 	t.Parallel()