Task concurrency not being respected influxdata#23582

redaianer · Nov 8, 2022 · f15410e · f15410e
1 parent de247ba
commit f15410e
Show file tree

Hide file tree

Showing 8 changed files with 50 additions and 49 deletions.
diff --git a/cmd/influxd/launcher/launcher.go b/cmd/influxd/launcher/launcher.go
@@ -451,15 +451,16 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 			query.QueryServiceBridge{AsyncQueryService: m.queryController},
 		)
 
-		executor, executorMetrics := executor.NewExecutor(
-			m.log.With(zap.String("service", "task-executor")),
+		exec, executorMetrics := executor.NewExecutor(
+			m.log.With(zap.String("service", "task-exec")),
 			query.QueryServiceBridge{AsyncQueryService: m.queryController},
 			ts.UserService,
 			combinedTaskService,
 			combinedTaskService,
 			executor.WithFlagger(m.flagger),
 		)
-		m.executor = executor
+		exec.SetLimitFunc(executor.ConcurrencyLimit(exec, fluxlang.DefaultService))
+		m.executor = exec
 		m.reg.MustRegister(executorMetrics.PrometheusCollectors()...)
 		schLogger := m.log.With(zap.String("service", "task-scheduler"))
 
@@ -470,7 +471,7 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 				err error
 			)
 			sch, sm, err = scheduler.NewScheduler(
-				executor,
+				exec,
 				taskbackend.NewSchedulableTaskService(m.kvService),
 				scheduler.WithOnErrorFn(func(ctx context.Context, taskID scheduler.ID, scheduledAt time.Time, err error) {
 					schLogger.Info(
@@ -499,7 +500,7 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 		taskCoord := coordinator.NewCoordinator(
 			coordLogger,
 			sch,
-			executor)
+			exec)
 
 		taskSvc = middleware.New(combinedTaskService, taskCoord)
 		if err := taskbackend.TaskNotifyCoordinatorOfExisting(
@@ -508,7 +509,7 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 			combinedTaskService,
 			taskCoord,
 			func(ctx context.Context, taskID platform2.ID, runID platform2.ID) error {
-				_, err := executor.ResumeCurrentRun(ctx, taskID, runID)
+				_, err := exec.ResumeCurrentRun(ctx, taskID, runID)
 				return err
 			},
 			coordLogger); err != nil {

diff --git a/influxql/query/internal/internal.pb.go b/influxql/query/internal/internal.pb.go
diff --git a/pkg/tracing/wire/binary.pb.go b/pkg/tracing/wire/binary.pb.go
diff --git a/storage/reads/datatypes/predicate.pb.go b/storage/reads/datatypes/predicate.pb.go
diff --git a/storage/reads/datatypes/storage_common.pb.go b/storage/reads/datatypes/storage_common.pb.go
diff --git a/tsdb/internal/fieldsindex.pb.go b/tsdb/internal/fieldsindex.pb.go
diff --git a/v1/services/meta/internal/meta.pb.go b/v1/services/meta/internal/meta.pb.go
diff --git a/v1/services/storage/source.pb.go b/v1/services/storage/source.pb.go