Integrate the new changes to tasks

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Features
 1. [13423](https://github.com/influxdata/influxdb/pull/13423): Set autorefresh of dashboard to pause if absolute time range is selected
+1. [13473](https://github.com/influxdata/influxdb/pull/13473): Switch task back end to a more modular and flexible system
 1. [13493](https://github.com/influxdata/influxdb/pull/13493): Add org profile tab with ability to edit organization name
 
 ### Bug Fixes

cmd/influxd/launcher/launcher.go

@@ -37,7 +37,6 @@ import (
 	"github.com/influxdata/influxdb/storage/readservice"
 	"github.com/influxdata/influxdb/task"
 	taskbackend "github.com/influxdata/influxdb/task/backend"
-	taskbolt "github.com/influxdata/influxdb/task/backend/bolt"
 	"github.com/influxdata/influxdb/task/backend/coordinator"
 	taskexecutor "github.com/influxdata/influxdb/task/backend/executor"
 	"github.com/influxdata/influxdb/telemetry"
@@ -211,8 +210,8 @@ type Launcher struct {
 
 	natsServer *nats.Server
 
-	scheduler *taskbackend.TickScheduler
-	taskStore taskbackend.Store
+	scheduler          *taskbackend.TickScheduler
+	taskControlService taskbackend.TaskControlService
 
 	jaegerTracerCloser io.Closer
 	logger             *zap.Logger
@@ -504,35 +503,22 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 	var storageQueryService = readservice.NewProxyQueryService(m.queryController)
 	var taskSvc platform.TaskService
 	{
-		var (
-			store taskbackend.Store
-			err   error
-		)
-		store, err = taskbolt.New(m.boltClient.DB(), "tasks", taskbolt.NoCatchUp)
-		if err != nil {
-			m.logger.Error("failed opening task bolt", zap.Error(err))
-			return err
-		}
 
-		if m.storeType == "memory" {
-			store = taskbackend.NewInMemStore()
-		}
+		// create the task stack:
+		// validation(coordinator(analyticalstore(kv.Service)))
 
-		executor := taskexecutor.NewAsyncQueryServiceExecutor(m.logger.With(zap.String("service", "task-executor")), m.queryController, authSvc, nil)
+		// define the executor and build analytical storage middleware
+		combinedTaskService := taskbackend.NewAnalyticalStorage(m.kvService, m.kvService, pointsWriter, query.QueryServiceBridge{AsyncQueryService: m.queryController})
+		executor := taskexecutor.NewAsyncQueryServiceExecutor(m.logger.With(zap.String("service", "task-executor")), m.queryController, authSvc, combinedTaskService)
 
-		lw := taskbackend.NewPointLogWriter(pointsWriter)
-		queryService := query.QueryServiceBridge{AsyncQueryService: m.queryController}
-		lr := taskbackend.NewQueryLogReader(queryService)
-		taskControlService := taskbackend.TaskControlAdaptor(store, lw, lr)
-		m.scheduler = taskbackend.NewScheduler(taskControlService, executor, time.Now().UTC().Unix(), taskbackend.WithTicker(ctx, 100*time.Millisecond), taskbackend.WithLogger(m.logger))
+		// create the scheduler
+		m.scheduler = taskbackend.NewScheduler(combinedTaskService, executor, time.Now().UTC().Unix(), taskbackend.WithTicker(ctx, 100*time.Millisecond), taskbackend.WithLogger(m.logger))
 		m.scheduler.Start(ctx)
 		m.reg.MustRegister(m.scheduler.PrometheusCollectors()...)
 
-		taskSvc = task.PlatformAdapter(store, lr, m.scheduler, authSvc, userResourceSvc, orgSvc)
-		taskexecutor.AddTaskService(executor, taskSvc)
-		taskSvc = coordinator.New(m.logger.With(zap.String("service", "task-coordinator")), m.scheduler, taskSvc)
+		taskSvc = coordinator.New(m.logger.With(zap.String("service", "task-coordinator")), m.scheduler, combinedTaskService)
 		taskSvc = task.NewValidator(m.logger.With(zap.String("service", "task-authz-validator")), taskSvc, bucketSvc)
-		m.taskStore = store
+		m.taskControlService = combinedTaskService
 	}
 
 	// NATS streaming server
@@ -691,8 +677,8 @@ func (m *Launcher) TaskService() platform.TaskService {
 }
 
 // TaskStore returns the internal store service.
-func (m *Launcher) TaskStore() taskbackend.Store {
-	return m.taskStore
+func (m *Launcher) TaskControlService() taskbackend.TaskControlService {
+	return m.taskControlService
 }
 
 // TaskScheduler returns the internal scheduler service.

cmd/influxd/launcher/tasks_test.go

@@ -96,21 +96,17 @@ stuff f=-123.456,b=true,s="hello"
 }
 from(bucket:"my_bucket_in") |> range(start:-5m) |> to(bucket:"%s", org:"%s")`, bOut.Name, be.Org.Name),
 	}
+
 	created, err := be.TaskService().CreateTask(ctx, create)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// Find the next due run of the task we just created, so that we can accurately tick the scheduler to it.
-	m, err := be.TaskStore().FindTaskMetaByID(ctx, created.ID)
+	ndr, err := be.TaskControlService().NextDueRun(ctx, created.ID)
 	if err != nil {
 		t.Fatal(err)
 	}
-	ndr, err := m.NextDueRun()
-	if err != nil {
-		t.Fatal(err)
-	}
-
 	be.TaskScheduler().(*backend.TickScheduler).Tick(ndr + 1)
 
 	// Poll for the task to have started and finished.
@@ -210,7 +206,6 @@ from(bucket:"my_bucket_in") |> range(start:-5m) |> to(bucket:"%s", org:"%s")`, b
 	}
 
 	t.Run("showrun", func(t *testing.T) {
-		t.Skip("FindRunByID isn't returning the log")
 		// do a show run!
 		showRun, err := be.TaskService().FindRunByID(ctx, created.ID, targetRun.ID)
 		if err != nil {