feat: Cleanup tasks bbolt entries during org deletion

Author: devanbenz

cmd/influxd/launcher/launcher.go

@@ -277,6 +277,7 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) {
 	}
 
 	m.kvService = kv.NewService(m.log.With(zap.String("store", "kv")), m.kvStore, ts, serviceConfig)
+	ts.Apply(tenant.WithTaskService(m.kvService))
 
 	var (
 		opLogSvc                                              = tenant.NewOpLogService(m.kvStore, m.kvService)

tenant/service.go

@@ -7,6 +7,7 @@ import (
 	"github.com/influxdata/influxdb/v2/kit/metric"
 	"github.com/influxdata/influxdb/v2/label"
 	"github.com/influxdata/influxdb/v2/secret"
+	"github.com/influxdata/influxdb/v2/task/taskmodel"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
@@ -35,6 +36,7 @@ type Service struct {
 	influxdb.UserResourceMappingService
 	influxdb.OrganizationService
 	influxdb.BucketService
+	taskmodel.TaskService
 }
 
 func (s *Service) RLock() {
@@ -62,6 +64,18 @@ func (s *Service) SetUserOptions(opts ...func(*UserSvc)) {
 	s.userSvc.SetOptions(opts...)
 }
 
+type ServiceOption func(*Service)
+
+func WithTaskService(ts taskmodel.TaskService) ServiceOption {
+	return func(s *Service) { s.TaskService = ts }
+}
+
+func (s *Service) Apply(opts ...ServiceOption) {
+	for _, opt := range opts {
+		opt(s)
+	}
+}
+
 // creates a new Service with logging and metrics middleware wrappers.
 func NewSystem(store *Store, log *zap.Logger, reg prometheus.Registerer, strongPasswords bool, metricOpts ...metric.ClientOptFn) *Service {
 	ts := NewService(store, WithPasswordChecking(strongPasswords))

tenant/service_org.go

@@ -7,6 +7,7 @@ import (
 	icontext "github.com/influxdata/influxdb/v2/context"
 	"github.com/influxdata/influxdb/v2/kit/platform"
 	"github.com/influxdata/influxdb/v2/kv"
+	"github.com/influxdata/influxdb/v2/task/taskmodel"
 )
 
 type OrgSvc struct {
@@ -212,6 +213,16 @@ func (s *OrgSvc) DeleteOrganization(ctx context.Context, id platform.ID) error {
 		return err
 	}
 
+	tasks, _, err := s.svc.FindTasks(ctx, taskmodel.TaskFilter{OrganizationID: &id})
+	if err != nil {
+		return err
+	}
+	for _, t := range tasks {
+		if err := s.svc.DeleteTask(ctx, t.ID); err != nil && err != taskmodel.ErrTaskNotFound {
+			return err
+		}
+	}
+
 	return s.removeResourceRelations(ctx, id)
 }