[QT-644] Concurrently decode scenarios if possible (#116)

Fix incorrect logic that led to us always decoding scenarios using the
serial method. Now we'll actually utilize all of our cores when decoding
a scenario that has a matrix and our machine has more than 2 CPUs.

* Fix concurrency path bug
* Fix a data race on the eval context during decode
* Fix our concurrent wait logic
* Improve collection workers in concurrent decode
* Bump version
* Please new linters

Signed-off-by: Ryan Cragun <[email protected]>

Author: ryancragun

acceptance/acceptance_test.go

@@ -245,7 +245,7 @@ func requireEqualOperationResponses(t *testing.T, expected *pb.OperationResponse
 	sortResponses(expectedResponses)
 	sortResponses(gotResponses)
 
-	for i := range expected.Responses {
+	for i := range expected.GetResponses() {
 		got := gotResponses[i]
 		expected := expectedResponses[i]
 

acceptance/fmt_test.go

@@ -47,10 +47,10 @@ func TestAcc_Cmd_Fmt(t *testing.T) {
 	got := &pb.FormatResponse{}
 	require.NoErrorf(t, protojson.Unmarshal(out, got), string(out))
 	require.Len(t, got.GetResponses(), len(expected.GetResponses()))
-	for i := range expected.Responses {
-		got := got.Responses[i]
-		expected := expected.Responses[i]
-		require.Equal(t, expected.Path, got.Path)
-		require.Equal(t, expected.Changed, got.Changed)
+	for i := range expected.GetResponses() {
+		got := got.GetResponses()[i]
+		expected := expected.GetResponses()[i]
+		require.Equal(t, expected.GetPath(), got.GetPath())
+		require.Equal(t, expected.GetChanged(), got.GetChanged())
 	}
 }

acceptance/scenario_check_test.go

@@ -82,7 +82,7 @@ func TestAcc_Cmd_Scenario_Check(t *testing.T) {
 					},
 				}
 
-				expected.Responses = append(expected.Responses, &pb.Operation_Response{
+				expected.Responses = append(expected.GetResponses(), &pb.Operation_Response{
 					Op: &pb.Ref_Operation{
 						Scenario: scenarioRef,
 					},
@@ -189,7 +189,7 @@ func TestAcc_Cmd_Scenario_Check_WithWarnings(t *testing.T) {
 					}
 				}
 
-				expected.Responses = append(expected.Responses, &pb.Operation_Response{
+				expected.Responses = append(expected.GetResponses(), &pb.Operation_Response{
 					Op: &pb.Ref_Operation{
 						Scenario: scenarioRef,
 					},

acceptance/scenario_e2e_aws_test.go

@@ -8,7 +8,6 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/hashicorp/enos/proto/hashicorp/enos/v1/pb"
@@ -98,7 +97,7 @@ func TestAcc_Cmd_Scenario_E2E_AWS(t *testing.T) {
 				// Lets try one more time to destroy resources that might have been
 				// created
 				out, err := enos.run(context.Background(), fmt.Sprintf("scenario destroy --chdir %s --out %s", path, outDir))
-				assert.NoErrorf(t, err, string(out))
+				require.NoErrorf(t, err, string(out))
 			})
 
 			expected := &pb.OperationResponses{
@@ -149,7 +148,7 @@ func TestAcc_Cmd_Scenario_E2E_AWS(t *testing.T) {
 					},
 				}
 
-				expected.Responses = append(expected.Responses, res)
+				expected.Responses = append(expected.GetResponses(), res)
 			}
 
 			cmd := fmt.Sprintf("scenario run --chdir %s --out %s --format json", path, outDir)

acceptance/scenario_list_test.go

@@ -121,8 +121,8 @@ func TestAcc_Cmd_Scenario_List(t *testing.T) {
 			got := &pb.ListScenariosResponse{}
 			require.NoError(t, protojson.Unmarshal(out, got))
 			require.Len(t, got.GetScenarios(), len(test.out.GetScenarios()))
-			for i := range test.out.Scenarios {
-				require.Equal(t, test.out.Scenarios[i].String(), got.Scenarios[i].String())
+			for i := range test.out.GetScenarios() {
+				require.Equal(t, test.out.GetScenarios()[i].String(), got.GetScenarios()[i].String())
 			}
 		})
 	}

acceptance/scenario_sample_list_test.go

@@ -61,8 +61,8 @@ func TestAcc_Cmd_Scenario_Sample_List(t *testing.T) {
 			got := &pb.ListSamplesResponse{}
 			require.NoError(t, protojson.Unmarshal(out, got))
 			require.Len(t, got.GetSamples(), len(test.out.GetSamples()))
-			for i := range test.out.Samples {
-				require.Equal(t, test.out.Samples[i].String(), got.Samples[i].String())
+			for i := range test.out.GetSamples() {
+				require.Equal(t, test.out.GetSamples()[i].String(), got.GetSamples()[i].String())
 			}
 		})
 	}

internal/command/enos/cmd/fmt.go

@@ -114,7 +114,7 @@ func runFmtCmd(cmd *cobra.Command, args []string) error {
 			return nil
 		}
 
-		if fmtCfg.Recursive {
+		if fmtCfg.GetRecursive() {
 			err = filepath.Walk(path, readRawFiles)
 		} else {
 			err = readRawFiles(path, nil, nil)
@@ -148,7 +148,7 @@ func runFmtCmd(cmd *cobra.Command, args []string) error {
 
 	/// Scan STDIN for content if we've been told to use STDIN either implicitly
 	// of explicitly.
-	if (argP == "-" || argP == "") && len(req.Files) == 0 {
+	if (argP == "-" || argP == "") && len(req.GetFiles()) == 0 {
 		bytes, err := io.ReadAll(cmd.InOrStdin())
 		if err != nil {
 			res.Diagnostics = diagnostics.FromErr(err)

internal/command/enos/cmd/scenario_sample_observe.go

@@ -29,14 +29,14 @@ func (t *sampleObserveFilter) Proto() *pb.Sample_Filter {
 		if i == 0 {
 			f.Subsets = []*pb.Sample_Subset_ID{}
 		}
-		f.Subsets = append(f.Subsets, &pb.Sample_Subset_ID{Name: t.OnlySubsets[i]})
+		f.Subsets = append(f.GetSubsets(), &pb.Sample_Subset_ID{Name: t.OnlySubsets[i]})
 	}
 
 	for i := range t.ExcludeSubsets {
 		if i == 0 {
 			f.ExcludeSubsets = []*pb.Sample_Subset_ID{}
 		}
-		f.ExcludeSubsets = append(f.ExcludeSubsets, &pb.Sample_Subset_ID{Name: t.ExcludeSubsets[i]})
+		f.ExcludeSubsets = append(f.GetExcludeSubsets(), &pb.Sample_Subset_ID{Name: t.ExcludeSubsets[i]})
 	}
 
 	return f

internal/diagnostics/diagnostics.go

@@ -59,7 +59,7 @@ func hasSeverity(sev pb.Diagnostic_Severity, diags ...[]*pb.Diagnostic) bool {
 	}
 
 	for _, diag := range combined {
-		if diag.Severity == sev {
+		if diag.GetSeverity() == sev {
 			return true
 		}
 	}
@@ -118,14 +118,14 @@ func FromTFJSON(in []tfjson.Diagnostic) []*pb.Diagnostic {
 				Filename: din.Range.Filename,
 			}
 
-			if d.Range.Start != nil {
+			if d.GetRange().GetStart() != nil {
 				d.Range.Start = &pb.Range_Pos{
 					Line:   int64(din.Range.Start.Line),
 					Column: int64(din.Range.Start.Column),
 					Byte:   int64(din.Range.Start.Byte),
 				}
 			}
-			if d.Range.End != nil {
+			if d.GetRange().GetEnd() != nil {
 				d.Range.End = &pb.Range_Pos{
 					Line:   int64(din.Range.End.Line),
 					Column: int64(din.Range.End.Column),
@@ -150,7 +150,7 @@ func FromTFJSON(in []tfjson.Diagnostic) []*pb.Diagnostic {
 				if i == 0 {
 					d.Snippet.Values = []*pb.Diagnostic_ExpressionValue{}
 				}
-				d.Snippet.Values = append(d.Snippet.Values, &pb.Diagnostic_ExpressionValue{
+				d.Snippet.Values = append(d.GetSnippet().GetValues(), &pb.Diagnostic_ExpressionValue{
 					Traversal: expr.Traversal,
 					Statement: expr.Statement,
 				})
@@ -321,7 +321,7 @@ func FromHCL(files map[string]*hcl.File, diags hcl.Diagnostics) []*pb.Diagnostic
 						}
 					}
 					sort.Slice(values, func(i, j int) bool {
-						return values[i].Traversal < values[j].Traversal
+						return values[i].GetTraversal() < values[j].GetTraversal()
 					})
 					pbDiag.Snippet.Values = values
 				}
@@ -393,7 +393,7 @@ func String(diag *pb.Diagnostic, opts ...StringOpt) string {
 		width = int(cfg.uiSettings.GetWidth())
 	}
 
-	switch diag.Severity {
+	switch diag.GetSeverity() {
 	case pb.Diagnostic_SEVERITY_ERROR:
 		buf.WriteString(cfg.color.Color("[bold][red]Error: [reset]"))
 		leftRuleLine = cfg.color.Color("[red]│[reset] ")
@@ -415,22 +415,22 @@ func String(diag *pb.Diagnostic, opts ...StringOpt) string {
 	// We don't wrap the summary, since we expect it to be terse, and since
 	// this is where we put the text of a native Go error it may not always
 	// be pure text that lends itself well to word-wrapping.
-	fmt.Fprintf(&buf, cfg.color.Color("[bold]%s[reset]\n\n"), diag.Summary)
+	fmt.Fprintf(&buf, cfg.color.Color("[bold]%s[reset]\n\n"), diag.GetSummary())
 
 	appendSourceSnippets(&buf, diag, cfg.color)
 
-	if diag.Detail != "" {
+	if diag.GetDetail() != "" {
 		paraWidth := width - leftRuleWidth - 1 // leave room for the left rule
 		if paraWidth > 0 {
-			lines := strings.Split(diag.Detail, "\n")
+			lines := strings.Split(diag.GetDetail(), "\n")
 			for _, line := range lines {
 				if !strings.HasPrefix(line, " ") {
 					line = wordwrap.WrapString(line, uint(paraWidth))
 				}
 				fmt.Fprintf(&buf, "%s\n", line)
 			}
 		} else {
-			fmt.Fprintf(&buf, "%s\n", diag.Detail)
+			fmt.Fprintf(&buf, "%s\n", diag.GetDetail())
 		}
 	}
 

internal/flightplan/decoder.go

@@ -432,7 +432,7 @@ func DecodeProto(
 
 	hclDiags := dec.Parse()
 	if len(hclDiags) > 0 {
-		res.Diagnostics = append(res.Diagnostics, diagnostics.FromHCL(dec.ParserFiles(), hclDiags)...)
+		res.Diagnostics = append(res.GetDiagnostics(), diagnostics.FromHCL(dec.ParserFiles(), hclDiags)...)
 	}
 
 	if diagnostics.HasErrors(res.GetDiagnostics()) {
@@ -441,7 +441,7 @@ func DecodeProto(
 
 	fp, hclDiags := dec.Decode(ctx)
 	if len(hclDiags) > 0 {
-		res.Diagnostics = append(res.Diagnostics, diagnostics.FromHCL(dec.ParserFiles(), hclDiags)...)
+		res.Diagnostics = append(res.GetDiagnostics(), diagnostics.FromHCL(dec.ParserFiles(), hclDiags)...)
 	}
 
 	return fp, res