feat(tablet_repair): plug tablet repair into scheduler and api

Since it also changes svc interfaces, it also runs `make generate`.

Author: Michal-Leszczynski

pkg/cmd/scylla-manager/server.go

@@ -102,6 +102,7 @@ func (s *server) makeServices(ctx context.Context) error {
 		s.session,
 		s.config.Repair,
 		metrics.NewRepairMetrics().MustRegister(),
+		metrics.NewTabletRepairMetrics().MustRegister(),
 		s.clusterSvc.Client,
 		s.clusterSvc.GetSession,
 		s.configCacheSvc,
@@ -177,6 +178,8 @@ func (s *server) makeServices(ctx context.Context) error {
 	s.schedSvc.SetRunner(scheduler.HealthCheckTask, s.healthSvc.Runner())
 	s.schedSvc.SetRunner(scheduler.RepairTask,
 		scheduler.PolicyRunner{Policy: restoreExclusiveLock, Runner: s.repairSvc.Runner(), TaskType: scheduler.RepairTask})
+	s.schedSvc.SetRunner(scheduler.TabletRepairTask,
+		scheduler.PolicyRunner{Policy: restoreExclusiveLock, Runner: s.repairSvc.TabletService, TaskType: scheduler.TabletRepairTask})
 	s.schedSvc.SetRunner(scheduler.ValidateBackupTask, s.backupSvc.ValidationRunner())
 
 	// Add additional properties on task run.

pkg/restapi/mock_repairservice_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/scylladb/scylla-manager/v3/pkg/service/healthcheck"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/one2onerestore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/repair"
+	"github.com/scylladb/scylla-manager/v3/pkg/service/repair/tablet"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/restore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/scheduler"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/query"
@@ -57,6 +58,9 @@ type RepairService interface {
 	GetTarget(ctx context.Context, clusterID uuid.UUID, properties json.RawMessage) (repair.Target, error)
 	SetIntensity(ctx context.Context, runID uuid.UUID, intensity float64) error
 	SetParallel(ctx context.Context, runID uuid.UUID, parallel int) error
+
+	GetTabletTarget(ctx context.Context, clusterID uuid.UUID, properties json.RawMessage) (tablet.Target, error)
+	GetTabletProgress(ctx context.Context, clusterID, taskID, runID uuid.UUID) (tablet.Progress, error)
 }
 
 // BackupService service interface for the REST API handlers.

pkg/restapi/services.go

@@ -17,6 +17,7 @@ import (
 	"github.com/scylladb/scylla-manager/v3/pkg/service/backup"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/one2onerestore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/repair"
+	"github.com/scylladb/scylla-manager/v3/pkg/service/repair/tablet"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/restore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/scheduler"
 	"github.com/scylladb/scylla-manager/v3/pkg/util"
@@ -263,6 +264,10 @@ func (h *taskHandler) validateTask(ctx context.Context, newTask *scheduler.Task,
 		if _, err := h.Repair.GetTarget(ctx, newTask.ClusterID, p); err != nil {
 			return errors.Wrap(err, "create repair target")
 		}
+	case scheduler.TabletRepairTask:
+		if _, err := h.Repair.GetTabletTarget(ctx, newTask.ClusterID, p); err != nil {
+			return errors.Wrap(err, "create tablet repair target")
+		}
 	case scheduler.ValidateBackupTask:
 		if _, err := h.Backup.GetValidationTarget(ctx, newTask.ClusterID, p); err != nil {
 			return errors.Wrap(err, "create backup validation target")
@@ -469,6 +474,8 @@ func (h *taskHandler) taskRunProgress(w http.ResponseWriter, r *http.Request) {
 			switch t.Type {
 			case scheduler.RepairTask:
 				prog.Progress = repair.Progress{}
+			case scheduler.TabletRepairTask:
+				prog.Progress = tablet.Progress{}
 			case scheduler.BackupTask:
 				prog.Progress = backup.Progress{}
 			case scheduler.RestoreTask:
@@ -505,6 +512,8 @@ func (h *taskHandler) taskRunProgress(w http.ResponseWriter, r *http.Request) {
 	switch t.Type {
 	case scheduler.RepairTask:
 		pr, err = h.Repair.GetProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID)
+	case scheduler.TabletRepairTask:
+		pr, err = h.Repair.GetTabletProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID)
 	case scheduler.BackupTask:
 		pr, err = h.Backup.GetProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID)
 	case scheduler.RestoreTask:

pkg/restapi/task.go

@@ -154,6 +154,7 @@ func newRestoreSvc(t *testing.T, mgrSession gocqlx.Session, client *scyllaclient
 		mgrSession,
 		repair.DefaultConfig(),
 		metrics.NewRepairMetrics(),
+		metrics.NewTabletRepairMetrics(),
 		func(context.Context, uuid.UUID) (*scyllaclient.Client, error) {
 			return client, nil
 		},

pkg/service/one2onerestore/helpers_integration_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/cluster"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/configcache"
+	"github.com/scylladb/scylla-manager/v3/pkg/service/repair/tablet"
 	"github.com/scylladb/scylla-manager/v3/pkg/util"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/inexlist/dcfilter"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/inexlist/ksfilter"
@@ -35,6 +36,8 @@ import (
 
 // Service orchestrates cluster repairs.
 type Service struct {
+	TabletService *tablet.Service
+
 	session gocqlx.Session
 	config  Config
 	metrics metrics.RepairMetrics
@@ -48,7 +51,7 @@ type Service struct {
 	mu                sync.Mutex
 }
 
-func NewService(session gocqlx.Session, config Config, metrics metrics.RepairMetrics,
+func NewService(session gocqlx.Session, config Config, m metrics.RepairMetrics, tm metrics.TabletRepairMetrics,
 	scyllaClient scyllaclient.ProviderFunc, clusterSession cluster.SessionFunc, configCache configcache.ConfigCacher,
 	logger log.Logger,
 ) (*Service, error) {
@@ -61,9 +64,10 @@ func NewService(session gocqlx.Session, config Config, metrics metrics.RepairMet
 	}
 
 	return &Service{
+		TabletService:     tablet.NewService(session, tm, scyllaClient, logger),
 		session:           session,
 		config:            config,
-		metrics:           metrics,
+		metrics:           m,
 		scyllaClient:      scyllaClient,
 		clusterSession:    clusterSession,
 		configCache:       configCache,
@@ -613,3 +617,13 @@ func getAllPrevRunIDs(session gocqlx.Session, clusterID, taskID, runID uuid.UUID
 		runID = prevID
 	}
 }
+
+// GetTabletTarget returns tablet repair target.
+func (s *Service) GetTabletTarget(ctx context.Context, clusterID uuid.UUID, properties json.RawMessage) (tablet.Target, error) {
+	return s.TabletService.GetTarget(ctx, clusterID, properties)
+}
+
+// GetTabletProgress returns tablet repair progress.
+func (s *Service) GetTabletProgress(ctx context.Context, clusterID, taskID, runID uuid.UUID) (tablet.Progress, error) {
+	return s.TabletService.GetProgress(ctx, clusterID, taskID, runID)
+}

pkg/service/repair/service.go

@@ -311,6 +311,7 @@ func newTestService(t *testing.T, session gocqlx.Session, client *scyllaclient.C
 		session,
 		c,
 		metrics.NewRepairMetrics(),
+		metrics.NewTabletRepairMetrics(),
 		func(context.Context, uuid.UUID) (*scyllaclient.Client, error) {
 			return client, nil
 		},
@@ -334,6 +335,7 @@ func newTestServiceWithClusterSession(t *testing.T, session gocqlx.Session, clie
 		session,
 		c,
 		metrics.NewRepairMetrics(),
+		metrics.NewTabletRepairMetrics(),
 		func(context.Context, uuid.UUID) (*scyllaclient.Client, error) {
 			return client, nil
 		},

pkg/service/repair/service_repair_integration_test.go

@@ -165,6 +165,7 @@ func newRestoreSvc(t *testing.T, mgrSession gocqlx.Session, client *scyllaclient
 		mgrSession,
 		repair.DefaultConfig(),
 		metrics.NewRepairMetrics(),
+		metrics.NewTabletRepairMetrics(),
 		func(context.Context, uuid.UUID) (*scyllaclient.Client, error) {
 			return client, nil
 		},

pkg/service/restore/helper_integration_test.go

@@ -119,6 +119,7 @@ func newTestService(t *testing.T, session gocqlx.Session, client *scyllaclient.C
 		session,
 		repair.DefaultConfig(),
 		metrics.NewRepairMetrics(),
+		metrics.NewTabletRepairMetrics(),
 		func(context.Context, uuid.UUID) (*scyllaclient.Client, error) {
 			return client, nil
 		},

pkg/service/restore/service_restore_integration_test.go

@@ -30,6 +30,7 @@ const (
 	One2OneRestoreTask TaskType = "1_1_restore"
 	HealthCheckTask    TaskType = "healthcheck"
 	RepairTask         TaskType = "repair"
+	TabletRepairTask   TaskType = "tablet_repair"
 	SuspendTask        TaskType = "suspend"
 	ValidateBackupTask TaskType = "validate_backup"
 
@@ -58,6 +59,8 @@ func (t *TaskType) UnmarshalText(text []byte) error {
 		*t = HealthCheckTask
 	case RepairTask:
 		*t = RepairTask
+	case TabletRepairTask:
+		*t = TabletRepairTask
 	case SuspendTask:
 		*t = SuspendTask
 	case ValidateBackupTask: