syz-cluster/controller: make parallelization configurable

Configure the number of patch series processed in parallel via an env
variable.

Author: a-nogikh

syz-cluster/controller/processor.go

@@ -8,6 +8,8 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os"
+	"strconv"
 	"sync"
 	"time"
 
@@ -19,31 +21,37 @@ import (
 )
 
 type SeriesProcessor struct {
-	blobStorage    blob.Storage
-	seriesRepo     *db.SeriesRepository
-	sessionRepo    *db.SessionRepository
-	workflows      workflow.Service
-	dbPollInterval time.Duration
+	blobStorage     blob.Storage
+	seriesRepo      *db.SeriesRepository
+	sessionRepo     *db.SessionRepository
+	workflows       workflow.Service
+	dbPollInterval  time.Duration
+	parallelWorkers int
 }
 
 func NewSeriesProcessor(env *app.AppEnvironment) *SeriesProcessor {
 	workflows, err := workflow.NewArgoService()
 	if err != nil {
 		app.Fatalf("failed to initialize workflows: %v", err)
 	}
+	parallelWorkers := 1
+	if val := os.Getenv("PARALLEL_WORKERS"); val != "" {
+		var err error
+		parallelWorkers, err = strconv.Atoi(val)
+		if err != nil || parallelWorkers < 1 {
+			app.Fatalf("invalid PARALLEL_WORKERS value")
+		}
+	}
 	return &SeriesProcessor{
-		blobStorage:    env.BlobStorage,
-		seriesRepo:     db.NewSeriesRepository(env.Spanner),
-		sessionRepo:    db.NewSessionRepository(env.Spanner),
-		dbPollInterval: time.Minute,
-		workflows:      workflows,
+		blobStorage:     env.BlobStorage,
+		seriesRepo:      db.NewSeriesRepository(env.Spanner),
+		sessionRepo:     db.NewSessionRepository(env.Spanner),
+		dbPollInterval:  time.Minute,
+		workflows:       workflows,
+		parallelWorkers: parallelWorkers,
 	}
 }
 
-// Do not run more than this number of sessions in parallel.
-// TODO: it'd be different for dev and prod, make it configurable.
-const parallelWorkers = 1
-
 func (sp *SeriesProcessor) Loop(ctx context.Context) error {
 	var wg sync.WaitGroup
 	defer wg.Wait()
@@ -107,7 +115,7 @@ func (sp *SeriesProcessor) seriesRunner(ctx context.Context, ch <-chan *db.Sessi
 	var eg errgroup.Group
 	defer eg.Wait()
 
-	eg.SetLimit(parallelWorkers)
+	eg.SetLimit(sp.parallelWorkers)
 	for {
 		var session *db.Session
 		select {

syz-cluster/controller/processor_test.go

@@ -142,9 +142,10 @@ func newMockedWorkflows() *mockedWorkflows {
 func prepareProcessorTest(t *testing.T, workflows workflow.Service) (*SeriesProcessor, context.Context) {
 	client, ctx := db.NewTransientDB(t)
 	return &SeriesProcessor{
-		seriesRepo:     db.NewSeriesRepository(client),
-		sessionRepo:    db.NewSessionRepository(client),
-		workflows:      workflows,
-		dbPollInterval: time.Second / 10,
+		seriesRepo:      db.NewSeriesRepository(client),
+		sessionRepo:     db.NewSessionRepository(client),
+		workflows:       workflows,
+		dbPollInterval:  time.Second / 10,
+		parallelWorkers: 2,
 	}, ctx
 }

syz-cluster/overlays/dev/global-config.yaml

@@ -9,3 +9,4 @@ data:
   SPANNER_EMULATOR_HOST: "cloud-spanner-emulator:9010"
   SPANNER_DATABASE_URI: "projects/my-project/instances/my-instance/databases/db"
   LOCAL_BLOB_STORAGE_PATH: "/blob-storage"
+  PARALLEL_WORKERS: "1" # Process only one series at a time.