feat(tablet_repair): handle scylla tablet repair task aborts

We need to make sure that leftover scylla tablet repair tasks are not running,
as scheduling new scylla tablet repair tasks on a table with an ongoing tablet repair
ends with an error. On the other hand, this is just best effort, because we can't
ensure that new scylla tablet repair tasks won't be created in the meantime or that
the task that we were trying to abort has just finished on its own.

Author: Michal-Leszczynski

pkg/scyllaclient/client_scylla.go

@@ -1547,3 +1547,30 @@ func (c *Client) ScyllaSetUserTaskTTL(ctx context.Context, host string, ttlSecon
 	})
 	return err
 }
+
+// ScyllaTaskModule describes scylla task modules.
+type ScyllaTaskModule string
+
+// ScyllaTaskModuleTablets describes tablets ScyllaTaskModule.
+const ScyllaTaskModuleTablets ScyllaTaskModule = "tablets"
+
+// ScyllaTaskType describes scylla task types.
+type ScyllaTaskType string
+
+// ScyllaTaskTypeUserRepair describes user repair ScyllaTaskType.
+const ScyllaTaskTypeUserRepair ScyllaTaskType = "user_repair"
+
+// ScyllaListTasks lists Scylla tasks of given module.
+func (c *Client) ScyllaListTasks(ctx context.Context, host string, module ScyllaTaskModule) ([]*models.TaskStats, error) {
+	if host != "" {
+		ctx = forceHost(ctx, host)
+	}
+	tasks, err := c.scyllaOps.TaskManagerListModuleTasksModuleGet(&operations.TaskManagerListModuleTasksModuleGetParams{
+		Context: ctx,
+		Module:  string(module),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return tasks.GetPayload(), nil
+}

pkg/service/repair/tablet/service_integration_test.go

@@ -357,4 +357,22 @@ func TestTabletRepairIntegration(t *testing.T) {
 			t.Fatal(err)
 		}
 	})
+
+	t.Run("Ongoing scylla tablet repair tasks", func(t *testing.T) {
+		testCtx, testCancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer testCancel()
+
+		h.Hrt.SetInterceptor(nil)
+		// Start lots of scylla tablet repair tasks
+		for ft := range allTables {
+			_, err = h.Client.TabletRepair(testCtx, ft.ks, ft.tab, "", nil, nil, scyllaclient.IncrementalModeDisabled)
+			if err != nil {
+				if _, _, ok := scyllaclient.IsColocatedTableErr(err); !ok {
+					t.Fatal(err)
+				}
+			}
+		}
+		// Verify that those scylla tablet repair tasks won't break SM tablet repair task
+		smokeTest(t, testCtx)
+	})
 }

pkg/service/repair/tablet/worker.go

@@ -56,6 +56,12 @@ func (s *Service) newWorker(ctx context.Context, clusterID, taskID, runID uuid.U
 
 func (w *worker) repairAll(ctx context.Context, target Target) error {
 	w.init(ctx, target)
+	// We need to make sure that leftover scylla tablet repair tasks are not running,
+	// as scheduling new scylla tablet repair tasks on a table with an ongoing tablet repair
+	// ends with an error. On the other hand, this is just best effort, because we can't
+	// ensure that new scylla tablet repair tasks won't be created in the meantime or that
+	// the task that we were trying to abort has just finished on its own.
+	w.abortAllRepairTasks(ctx)
 
 	for ks, tabs := range target.KsTabs {
 		for _, tab := range tabs {
@@ -78,6 +84,22 @@ func (w *worker) init(ctx context.Context, target Target) {
 	}
 }
 
+func (w *worker) abortAllRepairTasks(ctx context.Context) {
+	tasks, err := w.client.ScyllaListTasks(ctx, "", scyllaclient.ScyllaTaskModuleTablets)
+	if err != nil {
+		w.logger.Error(ctx, "Failed to list scylla tablet tasks", "error", err)
+		return
+	}
+	for _, t := range tasks {
+		if t == nil {
+			continue
+		}
+		if scyllaclient.ScyllaTaskType(t.Type) == scyllaclient.ScyllaTaskTypeUserRepair {
+			w.abortRepairTask(ctx, t.TaskID)
+		}
+	}
+}
+
 func (w *worker) repairTable(ctx context.Context, client *scyllaclient.Client, ks, tab string) (err error) {
 	pr := newRunProgress(w.clusterID, w.taskID, w.runID, ks, tab)
 
@@ -115,6 +137,7 @@ func (w *worker) repairTable(ctx context.Context, client *scyllaclient.Client, k
 
 	status, err := client.ScyllaWaitTask(ctx, "", id, 0)
 	if err != nil {
+		w.abortRepairTask(context.Background(), id)
 		return errors.Wrap(err, "get tablet repair task status")
 	}
 	switch scyllaclient.ScyllaTaskState(status.State) {
@@ -134,6 +157,17 @@ func (w *worker) incrementalRepairMode() scyllaclient.IncrementalMode {
 	return ""
 }
 
+func (w *worker) abortRepairTask(ctx context.Context, id string) {
+	if err := w.client.ScyllaAbortTask(ctx, "", id); err != nil {
+		w.logger.Error(context.Background(), "Failed to abort scylla repair task",
+			"id", id,
+			"error", err,
+		)
+	} else {
+		w.logger.Info(ctx, "Aborted scylla tablet repair task", "id", id)
+	}
+}
+
 func (w *worker) upsertTableProgress(ctx context.Context, pr RunProgress) {
 	q := table.TabletRepairRunProgress.InsertQuery(w.smSession)
 	defer q.Release()