WorkerPools always warn (and never exit on errors) (#1077)

apigee · Mar 8, 2023 · b9f77d1 · b9f77d1
1 parent c46b9de
commit b9f77d1
Show file tree

Hide file tree

Showing 6 changed files with 6 additions and 27 deletions.
diff --git a/cmd/registry/cmd/compute/score/score.go b/cmd/registry/cmd/compute/score/score.go
@@ -56,7 +56,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			inputPattern, err := patterns.ParseResourcePattern(args[0])

diff --git a/cmd/registry/cmd/compute/scorecard/scorecard.go b/cmd/registry/cmd/compute/scorecard/scorecard.go
@@ -56,7 +56,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			inputPattern, err := patterns.ParseResourcePattern(args[0])

diff --git a/cmd/registry/cmd/compute/vocabulary/vocabulary.go b/cmd/registry/cmd/compute/vocabulary/vocabulary.go
@@ -67,7 +67,7 @@ func Command() *cobra.Command {
 			if err != nil {
 				log.FromContext(ctx).WithError(err).Fatal("Failed to get jobs from flags")
 			}
-			taskQueue, wait := tasks.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 
 			parsed, err := names.ParseSpecRevision(path)

diff --git a/cmd/registry/cmd/resolve/resolve.go b/cmd/registry/cmd/resolve/resolve.go
@@ -108,7 +108,7 @@ func Command() *cobra.Command {
 			}
 
 			log.Debug(ctx, "Starting execution...")
-			taskQueue, wait := tasks.WorkerPoolWithWarnings(ctx, jobs)
+			taskQueue, wait := tasks.WorkerPool(ctx, jobs)
 			defer wait()
 			// Submit tasks to taskQueue
 			for i := 0; i < len(actions) && i < maxActions; i++ {

diff --git a/cmd/registry/tasks/tasks.go b/cmd/registry/tasks/tasks.go
@@ -49,23 +49,6 @@ func WorkerPool(ctx context.Context, n int) (chan<- Task, func()) {
 	return taskQueue, wait
 }
 
-// Similar to WorkerPool except it creates workers which log task errors as "Warnings"
-func WorkerPoolWithWarnings(ctx context.Context, n int) (chan<- Task, func()) {
-	var wg sync.WaitGroup
-	taskQueue := make(chan Task, 1024)
-	for i := 0; i < n; i++ {
-		wg.Add(1)
-		go worker(ctx, &wg, taskQueue, true)
-	}
-
-	wait := func() {
-		close(taskQueue)
-		wg.Wait()
-	}
-
-	return taskQueue, wait
-}
-
 // A worker which pulls tasks from the taskQueue, executes them and logs errors if any.
 func worker(ctx context.Context, wg *sync.WaitGroup, taskQueue <-chan Task, warnOnError bool) {
 	defer wg.Done()
@@ -75,11 +58,7 @@ func worker(ctx context.Context, wg *sync.WaitGroup, taskQueue <-chan Task, warn
 			return
 		default:
 			if err := task.Run(ctx); err != nil {
-				if warnOnError {
-					log.FromContext(ctx).WithError(err).Warnf("Task failed: %s", task)
-				} else {
-					log.FromContext(ctx).WithError(err).Fatalf("Task failed: %s", task)
-				}
+				log.FromContext(ctx).WithError(err).Warnf("Task failed: %s", task)
 			}
 		}
 	}

diff --git a/cmd/registry/tasks/tasks_test.go b/cmd/registry/tasks/tasks_test.go
@@ -49,7 +49,7 @@ func TestWorkerPoolWithWarnings(t *testing.T) {
 	ctx := context.Background()
 	jobs := 1
 
-	taskQueue, wait := WorkerPoolWithWarnings(ctx, jobs)
+	taskQueue, wait := WorkerPool(ctx, jobs)
 	defer wait()
 
 	for i := 0; i < 10; i++ {