Flakeguard: improve report aggregation performance and omit output fo…

…r passed tests (#1476)

* flakeguard: Improve performance of report aggregation

* Omit test outputs and package outputs for tests that pass

* fix tests

* do not run docker env test on changes in tools/

.github/workflows/docker-test.yaml

@@ -4,6 +4,8 @@ on:
     types: [submitted]
   pull_request:
     types: [labeled]
+    paths-ignore:
+      - 'tools/**'    
 
 jobs:
   eth_env:

tools/flakeguard/cmd/aggregate_results.go

@@ -36,25 +36,17 @@ var AggregateResultsCmd = &cobra.Command{
 
 		// Start spinner for loading test reports
 		s := spinner.New(spinner.CharSets[11], 100*time.Millisecond)
-		s.Suffix = " Loading test reports..."
+		s.Suffix = " Aggregating test reports..."
 		s.Start()
 
-		// Load test reports from JSON files
-		testReports, err := reports.LoadReports(resultsPath)
+		// Load test reports from JSON files and aggregate them
+		aggregatedReport, err := reports.LoadAndAggregate(resultsPath)
 		if err != nil {
 			s.Stop()
 			return fmt.Errorf("error loading test reports: %w", err)
 		}
 		s.Stop()
-		fmt.Println("Test reports loaded successfully.")
-
-		// Start spinner for aggregating reports
-		s = spinner.New(spinner.CharSets[11], 100*time.Millisecond)
-		s.Suffix = " Aggregating test reports..."
-		s.Start()
-
-		// Aggregate the reports
-		aggregatedReport, err := reports.Aggregate(testReports...)
+		fmt.Println("Test reports aggregated successfully.")
 
 		// Add metadata to the aggregated report
 		aggregatedReport.HeadSHA = headSHA

tools/flakeguard/cmd/run.go

@@ -31,6 +31,7 @@ var RunTestsCmd = &cobra.Command{
 		selectTests, _ := cmd.Flags().GetStringSlice("select-tests")
 		useShuffle, _ := cmd.Flags().GetBool("shuffle")
 		shuffleSeed, _ := cmd.Flags().GetString("shuffle-seed")
+		omitOutputsOnSuccess, _ := cmd.Flags().GetBool("omit-test-outputs-on-success")
 
 		// Check if project dependencies are correctly set up
 		if err := checkDependencies(projectPath); err != nil {
@@ -62,6 +63,7 @@ var RunTestsCmd = &cobra.Command{
 			SelectedTestPackages: testPackages,
 			UseShuffle:           useShuffle,
 			ShuffleSeed:          shuffleSeed,
+			OmitOutputsOnSuccess: omitOutputsOnSuccess,
 		}
 
 		// Run the tests
@@ -119,6 +121,7 @@ func init() {
 	RunTestsCmd.Flags().StringSlice("skip-tests", nil, "Comma-separated list of test names to skip from running")
 	RunTestsCmd.Flags().StringSlice("select-tests", nil, "Comma-separated list of test names to specifically run")
 	RunTestsCmd.Flags().Float64("max-pass-ratio", 1.0, "The maximum pass ratio threshold for a test to be considered flaky. Any tests below this pass rate will be considered flaky.")
+	RunTestsCmd.Flags().Bool("omit-test-outputs-on-success", true, "Omit test outputs and package outputs for tests that pass")
 }
 
 func checkDependencies(projectPath string) error {

tools/flakeguard/reports/data.go

@@ -128,13 +128,13 @@ func FilterTests(results []TestResult, predicate func(TestResult) bool) []TestRe
 	return filtered
 }
 
-func Aggregate(reports ...*TestReport) (*TestReport, error) {
+func aggregate(reportChan <-chan *TestReport) (*TestReport, error) {
 	testMap := make(map[string]TestResult)
 	fullReport := &TestReport{}
 	excludedTests := map[string]struct{}{}
 	selectedTests := map[string]struct{}{}
 
-	for _, report := range reports {
+	for report := range reportChan {
 		if fullReport.GoProject == "" {
 			fullReport.GoProject = report.GoProject
 		} else if fullReport.GoProject != report.GoProject {
@@ -159,13 +159,15 @@ func Aggregate(reports ...*TestReport) (*TestReport, error) {
 		}
 	}
 
+	// Finalize excluded and selected tests
 	for test := range excludedTests {
 		fullReport.ExcludedTests = append(fullReport.ExcludedTests, test)
 	}
 	for test := range selectedTests {
 		fullReport.SelectedTests = append(fullReport.SelectedTests, test)
 	}
 
+	// Prepare final results
 	var aggregatedResults []TestResult
 	for _, result := range testMap {
 		aggregatedResults = append(aggregatedResults, result)
@@ -177,6 +179,15 @@ func Aggregate(reports ...*TestReport) (*TestReport, error) {
 	return fullReport, nil
 }
 
+func aggregateFromReports(reports ...*TestReport) (*TestReport, error) {
+	reportChan := make(chan *TestReport, len(reports))
+	for _, report := range reports {
+		reportChan <- report
+	}
+	close(reportChan)
+	return aggregate(reportChan)
+}
+
 func mergeTestResults(a, b TestResult) TestResult {
 	a.Runs += b.Runs
 	a.Durations = append(a.Durations, b.Durations...)

tools/flakeguard/reports/data_test.go

@@ -287,7 +287,7 @@ func TestAggregate(t *testing.T) {
 		},
 	}
 
-	aggregatedReport, err := Aggregate(report1, report2)
+	aggregatedReport, err := aggregateFromReports(report1, report2)
 	if err != nil {
 		t.Fatalf("Error aggregating reports: %v", err)
 	}
@@ -371,7 +371,7 @@ func TestAggregateOutputs(t *testing.T) {
 		},
 	}
 
-	aggregatedReport, err := Aggregate(report1, report2)
+	aggregatedReport, err := aggregateFromReports(report1, report2)
 	if err != nil {
 		t.Fatalf("Error aggregating reports: %v", err)
 	}
@@ -432,7 +432,7 @@ func TestAggregateIdenticalOutputs(t *testing.T) {
 		},
 	}
 
-	aggregatedReport, err := Aggregate(report1, report2)
+	aggregatedReport, err := aggregateFromReports(report1, report2)
 	if err != nil {
 		t.Fatalf("Error aggregating reports: %v", err)
 	}
@@ -569,7 +569,7 @@ func TestAggregate_AllSkippedTests(t *testing.T) {
 		},
 	}
 
-	aggregatedReport, err := Aggregate(report1, report2)
+	aggregatedReport, err := aggregateFromReports(report1, report2)
 	if err != nil {
 		t.Fatalf("Error aggregating reports: %v", err)
 	}

tools/flakeguard/reports/io.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"os"
 	"path/filepath"
 )
@@ -31,30 +32,157 @@ func (OSFileSystem) WriteFile(filename string, data []byte, perm os.FileMode) er
 	return os.WriteFile(filename, data, perm)
 }
 
-// LoadReports reads JSON files from a directory and returns a slice of TestReport pointers
-func LoadReports(resultsPath string) ([]*TestReport, error) {
-	var testReports []*TestReport
-	err := filepath.Walk(resultsPath, func(path string, info os.FileInfo, err error) error {
+func LoadAndAggregate(resultsPath string) (*TestReport, error) {
+	reportChan := make(chan *TestReport)
+	errChan := make(chan error, 1)
+
+	// Start file processing in a goroutine
+	go func() {
+		defer close(reportChan)
+		defer close(errChan)
+
+		err := filepath.Walk(resultsPath, func(path string, info os.FileInfo, err error) error {
+			if err != nil {
+				return fmt.Errorf("error accessing path %s: %w", path, err)
+			}
+			if !info.IsDir() && filepath.Ext(path) == ".json" {
+				log.Printf("Processing file: %s", path)
+				processLargeFile(path, reportChan, errChan)
+			}
+			return nil
+		})
+		if err != nil {
+			errChan <- err
+		}
+	}()
+
+	// Aggregate results as they are being loaded
+	aggregatedReport, err := aggregate(reportChan)
+	if err != nil {
+		return nil, fmt.Errorf("error aggregating reports: %w", err)
+	}
+	return aggregatedReport, nil
+}
+
+func processLargeFile(filePath string, reportChan chan<- *TestReport, errChan chan<- error) {
+	file, err := os.Open(filePath)
+	if err != nil {
+		errChan <- fmt.Errorf("error opening file %s: %w", filePath, err)
+		log.Printf("Error opening file: %s, Error: %v", filePath, err)
+		return
+	}
+	defer file.Close()
+
+	decoder := json.NewDecoder(file)
+
+	var report TestReport
+	token, err := decoder.Token() // Read opening brace '{'
+	if err != nil || token != json.Delim('{') {
+		errChan <- fmt.Errorf("error reading JSON object start from file %s: %w", filePath, err)
+		log.Printf("Error reading JSON object start from file: %s, Error: %v", filePath, err)
+		return
+	}
+
+	// Parse fields until we reach the end of the object
+	for decoder.More() {
+		token, err := decoder.Token()
 		if err != nil {
-			return fmt.Errorf("error accessing path %s: %w", path, err)
+			errChan <- fmt.Errorf("error reading JSON token from file %s: %w", filePath, err)
+			log.Printf("Error reading JSON token from file: %s, Error: %v", filePath, err)
+			return
 		}
-		if !info.IsDir() && filepath.Ext(path) == ".json" {
-			data, readErr := os.ReadFile(path)
-			if readErr != nil {
-				return fmt.Errorf("error reading file %s: %w", path, readErr)
+
+		fieldName, ok := token.(string)
+		if !ok {
+			errChan <- fmt.Errorf("unexpected JSON token in file %s", filePath)
+			log.Printf("Unexpected JSON token in file: %s, Token: %v", filePath, token)
+			return
+		}
+
+		switch fieldName {
+		case "GoProject":
+			if err := decoder.Decode(&report.GoProject); err != nil {
+				log.Printf("Error decoding GoProject in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "HeadSHA":
+			if err := decoder.Decode(&report.HeadSHA); err != nil {
+				log.Printf("Error decoding HeadSHA in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "BaseSHA":
+			if err := decoder.Decode(&report.BaseSHA); err != nil {
+				log.Printf("Error decoding BaseSHA in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "RepoURL":
+			if err := decoder.Decode(&report.RepoURL); err != nil {
+				log.Printf("Error decoding RepoURL in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "GitHubWorkflowName":
+			if err := decoder.Decode(&report.GitHubWorkflowName); err != nil {
+				log.Printf("Error decoding GitHubWorkflowName in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "TestRunCount":
+			if err := decoder.Decode(&report.TestRunCount); err != nil {
+				log.Printf("Error decoding TestRunCount in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "RaceDetection":
+			if err := decoder.Decode(&report.RaceDetection); err != nil {
+				log.Printf("Error decoding RaceDetection in file: %s, Error: %v", filePath, err)
+				return
+			}
+		case "ExcludedTests":
+			if err := decoder.Decode(&report.ExcludedTests); err != nil {
+				log.Printf("Error decoding ExcludedTests in file: %s, Error: %v", filePath, err)
+				return
 			}
-			var report TestReport
-			if jsonErr := json.Unmarshal(data, &report); jsonErr != nil {
-				return fmt.Errorf("error unmarshaling JSON from file %s: %w", path, jsonErr)
+		case "SelectedTests":
+			if err := decoder.Decode(&report.SelectedTests); err != nil {
+				log.Printf("Error decoding SelectedTests in file: %s, Error: %v", filePath, err)
+				return
 			}
-			testReports = append(testReports, &report)
+		case "Results":
+			token, err := decoder.Token() // Read opening bracket '['
+			if err != nil || token != json.Delim('[') {
+				log.Printf("Error reading Results array start in file: %s, Error: %v", filePath, err)
+				return
+			}
+
+			for decoder.More() {
+				var result TestResult
+				if err := decoder.Decode(&result); err != nil {
+					log.Printf("Error decoding TestResult in file: %s, Error: %v", filePath, err)
+					return
+				}
+				report.Results = append(report.Results, result)
+			}
+
+			if _, err := decoder.Token(); err != nil {
+				log.Printf("Error reading Results array end in file: %s, Error: %v", filePath, err)
+				return
+			}
+		default:
+			// Skip unknown fields
+			var skip interface{}
+			if err := decoder.Decode(&skip); err != nil {
+				log.Printf("Error skipping unknown field: %s in file: %s, Error: %v", fieldName, filePath, err)
+				return
+			}
+			log.Printf("Skipped unknown field: %s in file: %s", fieldName, filePath)
 		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
 	}
-	return testReports, nil
+
+	// Read closing brace '}'
+	if _, err := decoder.Token(); err != nil {
+		log.Printf("Error reading JSON object end in file: %s, Error: %v", filePath, err)
+		return
+	}
+
+	reportChan <- &report
 }
 
 // LoadReport reads a JSON file and returns a TestReport pointer

tools/flakeguard/runner/runner.go

@@ -39,6 +39,7 @@ type Runner struct {
 	SelectTests          []string      // Test names to include.
 	SelectedTestPackages []string      // Explicitly selected packages to run.
 	CollectRawOutput     bool          // Set to true to collect test output for later inspection.
+	OmitOutputsOnSuccess bool          // Set to true to omit test outputs on success.
 	rawOutputs           map[string]*bytes.Buffer
 }
 
@@ -69,7 +70,7 @@ func (r *Runner) RunTests() (*reports.TestReport, error) {
 		}
 	}
 
-	results, err := parseTestResults(r.RunCount, jsonFilePaths)
+	results, err := r.parseTestResults(jsonFilePaths)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse test results: %w", err)
 	}
@@ -182,7 +183,7 @@ func (e entry) String() string {
 // panics and failures at that point.
 // Subtests add more complexity, as panics in subtests are only reported in their parent's output,
 // and cannot be accurately attributed to the subtest that caused them.
-func parseTestResults(expectedRuns int, filePaths []string) ([]reports.TestResult, error) {
+func (r *Runner) parseTestResults(filePaths []string) ([]reports.TestResult, error) {
 	var (
 		testDetails         = make(map[string]*reports.TestResult) // Holds run, pass counts, and other details for each test
 		panickedPackages    = map[string]struct{}{}                // Packages with tests that panicked
@@ -192,6 +193,7 @@ func parseTestResults(expectedRuns int, filePaths []string) ([]reports.TestResul
 		panicDetectionMode  = false
 		raceDetectionMode   = false
 		detectedEntries     = []entry{} // race or panic entries
+		expectedRuns        = r.RunCount
 	)
 
 	// Process each file
@@ -357,6 +359,10 @@ func parseTestResults(expectedRuns int, filePaths []string) ([]reports.TestResul
 					}
 					result.Durations = append(result.Durations, duration)
 					result.Successes++
+					if r.OmitOutputsOnSuccess {
+						// Clear outputs for passing tests
+						result.Outputs = nil
+					}
 				}
 			case "fail":
 				if entryLine.Test != "" {