feat: worker per file (#1017)

* feat: worker per file

* test: log output on timeout

* refactor: review

* refactor: simplify error handling

* fix: wait for exit on close

* fix: file list error handling

* test: disable version check in e2e tests

* fix: add mutex around main report writing

* test: remove parallel e2e tests due to conflict

* style: ignore unchecked error

e2e/flags/.snapshots/TestInitCommand

@@ -19,6 +19,7 @@ scan:
     external-rule-dir: []
     force: false
     internal-domains: []
+    parallel: 2
     quiet: false
     scanner:
         - sast

e2e/flags/.snapshots/TestMetadataFlags-help-scan

@@ -30,6 +30,7 @@ Scan Flags
       --external-rule-dir strings            Specify directories paths that contain .yaml files with external rules configuration
       --force                                Disable the cache and runs the detections again
       --internal-domains strings             Define regular expressions for better classification of private or unreachable domains e.g. --internal-domains=".*.my-company.com,private.sh"
+      --parallel int                         Specify the amount of parallelism to use during the scan (default 2)
       --quiet                                Suppress non-essential messages
       --scanner strings                      Specify which scanner to use e.g. --scanner=secrets, --scanner=secrets,sast (default [sast])
       --skip-path strings                    Specify the comma separated files and directories to skip. Supports * syntax, e.g. --skip-path users/*.go,users/admin.sql

e2e/flags/.snapshots/TestMetadataFlags-scan-help

@@ -30,6 +30,7 @@ Scan Flags
       --external-rule-dir strings            Specify directories paths that contain .yaml files with external rules configuration
       --force                                Disable the cache and runs the detections again
       --internal-domains strings             Define regular expressions for better classification of private or unreachable domains e.g. --internal-domains=".*.my-company.com,private.sh"
+      --parallel int                         Specify the amount of parallelism to use during the scan (default 2)
       --quiet                                Suppress non-essential messages
       --scanner strings                      Specify which scanner to use e.g. --scanner=secrets, --scanner=secrets,sast (default [sast])
       --skip-path strings                    Specify the comma separated files and directories to skip. Supports * syntax, e.g. --skip-path users/*.go,users/admin.sql

e2e/flags/.snapshots/TestReportFlagsShouldFail-invalid-context-flag

@@ -31,6 +31,7 @@ Scan Flags
       --external-rule-dir strings            Specify directories paths that contain .yaml files with external rules configuration
       --force                                Disable the cache and runs the detections again
       --internal-domains strings             Define regular expressions for better classification of private or unreachable domains e.g. --internal-domains=".*.my-company.com,private.sh"
+      --parallel int                         Specify the amount of parallelism to use during the scan (default 2)
       --quiet                                Suppress non-essential messages
       --scanner strings                      Specify which scanner to use e.g. --scanner=secrets, --scanner=secrets,sast (default [sast])
       --skip-path strings                    Specify the comma separated files and directories to skip. Supports * syntax, e.g. --skip-path users/*.go,users/admin.sql

e2e/flags/.snapshots/TestReportFlagsShouldFail-invalid-format-flag

@@ -31,6 +31,7 @@ Scan Flags
       --external-rule-dir strings            Specify directories paths that contain .yaml files with external rules configuration
       --force                                Disable the cache and runs the detections again
       --internal-domains strings             Define regular expressions for better classification of private or unreachable domains e.g. --internal-domains=".*.my-company.com,private.sh"
+      --parallel int                         Specify the amount of parallelism to use during the scan (default 2)
       --quiet                                Suppress non-essential messages
       --scanner strings                      Specify which scanner to use e.g. --scanner=secrets, --scanner=secrets,sast (default [sast])
       --skip-path strings                    Specify the comma separated files and directories to skip. Supports * syntax, e.g. --skip-path users/*.go,users/admin.sql

e2e/flags/.snapshots/TestReportFlagsShouldFail-invalid-report-flag

@@ -31,6 +31,7 @@ Scan Flags
       --external-rule-dir strings            Specify directories paths that contain .yaml files with external rules configuration
       --force                                Disable the cache and runs the detections again
       --internal-domains strings             Define regular expressions for better classification of private or unreachable domains e.g. --internal-domains=".*.my-company.com,private.sh"
+      --parallel int                         Specify the amount of parallelism to use during the scan (default 2)
       --quiet                                Suppress non-essential messages
       --scanner strings                      Specify which scanner to use e.g. --scanner=secrets, --scanner=secrets,sast (default [sast])
       --skip-path strings                    Specify the comma separated files and directories to skip. Supports * syntax, e.g. --skip-path users/*.go,users/admin.sql

e2e/flags/report_flags_test.go

@@ -11,12 +11,16 @@ import (
 )
 
 func newScanTest(name string, arguments []string) testhelper.TestCase {
-	arguments = append([]string{"scan", filepath.Join("e2e", "flags", "testdata", "simple")}, arguments...)
+	arguments = append([]string{
+		"scan",
+		"--disable-version-check",
+		filepath.Join("e2e", "flags", "testdata", "simple")},
+		arguments...,
+	)
 	return testhelper.NewTestCase(name, arguments, testhelper.TestCaseOptions{})
 }
 
 func TestReportFlags(t *testing.T) {
-	t.Parallel()
 	tests := []testhelper.TestCase{
 		newScanTest("report-dataflow", []string{"--report=dataflow"}),
 	}
@@ -41,8 +45,7 @@ func TestReportFlagsShouldFail(t *testing.T) {
 }
 
 func TestOuputFlag(t *testing.T) {
-	t.Parallel()
-	outputPath := tmpfile.Create("", "test_output.jsonl")
+	outputPath := tmpfile.Create("test_output.jsonl")
 	defer func() {
 		err := os.Remove(outputPath)
 		if err != nil {

e2e/internal/testhelper/testhelper.go

@@ -51,6 +51,7 @@ func executeApp(t *testing.T, arguments []string) (string, error) {
 
 	timer := time.NewTimer(TestTimeout)
 	commandFinished := make(chan struct{}, 1)
+	combinedOutput := func() string { return buffOut.String() + "\n--\n" + buffErr.String() }
 
 	go func() {
 		err = cmd.Start()
@@ -67,14 +68,12 @@ func executeApp(t *testing.T, arguments []string) (string, error) {
 	select {
 	case <-timer.C:
 		cancel()
-		t.Fatalf("command failed to complete on time 'bearer %s'", strings.Join(arguments, " "))
+		t.Fatalf("command failed to complete on time 'bearer %s':\n%s", strings.Join(arguments, " "), combinedOutput())
 	case <-commandFinished:
 		cancel()
 	}
 
-	combinedOutput := buffOut.String() + "\n--\n" + buffErr.String()
-
-	return combinedOutput, err
+	return combinedOutput(), err
 }
 
 func CreateCommand(arguments []string) (*exec.Cmd, context.CancelFunc) {

e2e/rules/rules_test.go

@@ -8,8 +8,6 @@ import (
 )
 
 func TestSecrets(t *testing.T) {
-	t.Parallel()
-
 	testCases := []testhelper.TestCase{
 		testhelper.NewTestCase(
 			"secrets",
@@ -29,26 +27,21 @@ func TestSecrets(t *testing.T) {
 }
 
 func TestAuxilary(t *testing.T) {
-	t.Parallel()
 	runRulesTest("auxilary", "javascript_third_parties_datadog_test", t)
 }
 
 func TestSanitizer(t *testing.T) {
-	t.Parallel()
 	runRulesTest("sanitizer", "sanitizer_test", t)
 }
 
 func TestSimpleRuby(t *testing.T) {
-	t.Parallel()
 	runRulesTest("simple_ruby", "ruby_rails_insecure_communication_test", t)
 }
 
 func TestRubyRailsDefaultEncryptionStructure(t *testing.T) {
-	t.Parallel()
 	runRulesTest("ruby_rails_default_encryption_structure_sql", "ruby_rails_default_encryption", t)
 }
 
 func TestRubyRailsDefaultEncryptionSchema(t *testing.T) {
-	t.Parallel()
 	runRulesTest("ruby_rails_default_encryption_schema_rb", "ruby_rails_default_encryption", t)
 }

new/detector/composition/testhelper/testhelper.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	"github.com/bearer/bearer/pkg/commands"
-	"github.com/bearer/bearer/pkg/commands/process/balancer/filelist"
+	"github.com/bearer/bearer/pkg/commands/process/orchestrator/filelist"
 	"github.com/bearer/bearer/pkg/commands/process/settings"
 	"github.com/bearer/bearer/pkg/commands/process/worker"
 	"github.com/bearer/bearer/pkg/commands/process/worker/work"
@@ -109,7 +109,7 @@ func (runner *Runner) scanSingleFile(t *testing.T, testDataPath string, fileRela
 	}
 
 	err = runner.worker.Scan(work.ProcessRequest{
-		Files:      []work.File{fileRelativePath},
+		File:       fileRelativePath,
 		ReportPath: detectorsReportPath,
 		Repository: work.Repository{
 			Dir: testDataPath,

pkg/commands/artifact/run.go

@@ -21,7 +21,7 @@ import (
 	"golang.org/x/xerrors"
 
 	"github.com/bearer/bearer/cmd/bearer/build"
-	"github.com/bearer/bearer/pkg/commands/process/balancer"
+	"github.com/bearer/bearer/pkg/commands/process/orchestrator"
 	"github.com/bearer/bearer/pkg/commands/process/settings"
 	"github.com/bearer/bearer/pkg/commands/process/worker/work"
 	"github.com/bearer/bearer/pkg/flag"
@@ -63,16 +63,14 @@ type Runner interface {
 }
 
 type runner struct {
-	balancer       *balancer.Monitor
 	reportPath     string
 	reuseDetection bool
+	scanSettings   settings.Config
 }
 
 // NewRunner initializes Runner that provides scanning functionalities.
 func NewRunner(ctx context.Context, scanSettings settings.Config) Runner {
-	r := &runner{}
-
-	r.balancer = balancer.New(scanSettings)
+	r := &runner{scanSettings: scanSettings}
 
 	scanID, err := buildScanID(scanSettings)
 	if err != nil {
@@ -196,18 +194,16 @@ func (r *runner) ScanRepository(ctx context.Context, opts flag.Options) (types.R
 
 func (r *runner) scanArtifact(ctx context.Context, opts flag.Options) (types.Report, error) {
 	if !r.reuseDetection {
-		task := r.balancer.ScheduleTask(work.ProcessRequest{
-			Repository: work.Repository{
+		if err := orchestrator.Scan(
+			work.Repository{
 				Dir:               opts.Target,
 				PreviousCommitSHA: "",
 				CommitSHA:         "",
 			},
-			ReportPath: r.reportPath,
-		})
-		result := <-task.Done
-
-		if result.Error != nil {
-			return types.Report{}, result.Error
+			r.scanSettings,
+			r.reportPath,
+		); err != nil {
+			return types.Report{}, err
 		}
 	}
 
@@ -261,8 +257,8 @@ func Run(ctx context.Context, opts flag.Options, targetKind TargetKind) (err err
 	switch targetKind {
 	case TargetFilesystem:
 		if report, err = r.ScanFilesystem(ctx, opts); err != nil {
-			if errors.Is(err, balancer.ErrFileListEmpty) {
-				outputhandler.StdOutLogger().Msgf("directory empty: %s", err)
+			if errors.Is(err, orchestrator.ErrFileListEmpty) {
+				outputhandler.StdOutLogger().Msgf(err.Error())
 				os.Exit(0)
 				return
 			}