feat(1-1-restore): adds progress tracking of 1-1-restore (#4296)

This adds progress tracking of various 1-1-restore stage.
Each progress update represented by RunProgress model saved into db.
w.getProgress is used to aggregate current task progress by Keyspaces,
Tables and Hosts.

Author: VAveryanov8

pkg/cmd/scylla-manager/server.go

@@ -221,12 +221,13 @@ func (s *server) onClusterChange(ctx context.Context, c cluster.Change) error {
 
 func (s *server) makeServers(ctx context.Context) error {
 	services := restapi.Services{
-		Cluster:     s.clusterSvc,
-		HealthCheck: s.healthSvc,
-		Repair:      s.repairSvc,
-		Backup:      s.backupSvc,
-		Restore:     s.restoreSvc,
-		Scheduler:   s.schedSvc,
+		Cluster:        s.clusterSvc,
+		HealthCheck:    s.healthSvc,
+		Repair:         s.repairSvc,
+		Backup:         s.backupSvc,
+		Restore:        s.restoreSvc,
+		Scheduler:      s.schedSvc,
+		One2OneRestore: s.one2OneRestoreSvc,
 	}
 	h := restapi.New(services, s.logger.Named("http"))
 

pkg/restapi/services.go

@@ -10,6 +10,7 @@ import (
 	"github.com/scylladb/scylla-manager/v3/pkg/service/backup"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/cluster"
 	"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/restore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/scheduler"
@@ -19,12 +20,13 @@ import (
 
 // Services contains REST API services.
 type Services struct {
-	Cluster     ClusterService
-	HealthCheck HealthCheckService
-	Repair      RepairService
-	Backup      BackupService
-	Restore     RestoreService
-	Scheduler   SchedService
+	Cluster        ClusterService
+	HealthCheck    HealthCheckService
+	Repair         RepairService
+	Backup         BackupService
+	Restore        RestoreService
+	Scheduler      SchedService
+	One2OneRestore One2OneRestoreService
 }
 
 // ClusterService service interface for the REST API handlers.
@@ -93,3 +95,8 @@ type SchedService interface {
 	Suspend(ctx context.Context, clusterID uuid.UUID) error
 	Resume(ctx context.Context, clusterID uuid.UUID, startTasks bool) error
 }
+
+// One2OneRestoreService service interface for the 1-1-restore REST API handlers.
+type One2OneRestoreService interface {
+	GetProgress(ctx context.Context, clusterID, taskID, runID uuid.UUID, properties json.RawMessage) (one2onerestore.Progress, error)
+}

pkg/restapi/task.go

@@ -15,6 +15,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
 	"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/restore"
 	"github.com/scylladb/scylla-manager/v3/pkg/service/scheduler"
@@ -469,6 +470,8 @@ func (h *taskHandler) taskRunProgress(w http.ResponseWriter, r *http.Request) {
 				prog.Progress = restore.Progress{}
 			case scheduler.ValidateBackupTask:
 				prog.Progress = backup.ValidationHostProgress{}
+			case scheduler.One2OneRestoreTask:
+				prog.Progress = one2onerestore.Progress{}
 			}
 			render.Respond(w, r, prog)
 			return
@@ -503,6 +506,8 @@ func (h *taskHandler) taskRunProgress(w http.ResponseWriter, r *http.Request) {
 		pr, err = h.Restore.GetProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID)
 	case scheduler.ValidateBackupTask:
 		pr, err = h.Backup.GetValidationProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID)
+	case scheduler.One2OneRestoreTask:
+		pr, err = h.One2OneRestore.GetProgress(r.Context(), t.ClusterID, t.ID, prog.Run.ID, t.Properties)
 	default:
 		respondBadRequest(w, r, errors.Errorf("unsupported task type %s", t.Type))
 		return

pkg/schema/table/table.go

@@ -27,10 +27,11 @@ import (
 )
 
 type testHelper struct {
-	client     *scyllaclient.Client
-	clusterID  uuid.UUID
-	backupSvc  *backup.Service
-	restoreSvc Servicer
+	client                   *scyllaclient.Client
+	clusterID, taskID, runID uuid.UUID
+	props                    []byte
+	backupSvc                *backup.Service
+	restoreSvc               Servicer
 }
 
 func newTestHelper(t *testing.T, hosts []string) *testHelper {
@@ -40,13 +41,17 @@ func newTestHelper(t *testing.T, hosts []string) *testHelper {
 	session := CreateScyllaManagerDBSession(t)
 
 	clusterID := uuid.NewTime()
+	taskID := uuid.NewTime()
+	runID := uuid.NewTime()
 
 	backupSvc := newBackupSvc(t, session, sc, clusterID)
 	restoreSvc := newRestoreSvc(t, session, sc, clusterID, "", "")
 
 	return &testHelper{
 		client:     sc,
 		clusterID:  clusterID,
+		taskID:     taskID,
+		runID:      runID,
 		backupSvc:  backupSvc,
 		restoreSvc: restoreSvc,
 	}
@@ -86,15 +91,16 @@ func (h *testHelper) runRestore(t *testing.T, props map[string]any) {
 	t.Helper()
 	Printf("Run 1-1-restore with properties: %v", props)
 	ctx := context.Background()
-	taskID := uuid.NewTime()
-	runID := uuid.NewTime()
+	h.taskID = uuid.NewTime()
+	h.runID = uuid.NewTime()
 
 	rawProps, err := json.Marshal(props)
 	if err != nil {
 		t.Fatal(errors.Wrap(err, "marshal properties"))
 	}
+	h.props = rawProps
 
-	err = h.restoreSvc.One2OneRestore(ctx, h.clusterID, taskID, runID, rawProps)
+	err = h.restoreSvc.One2OneRestore(ctx, h.clusterID, h.taskID, h.runID, h.props)
 	if err != nil {
 		t.Fatal(errors.Wrap(err, "run 1-1-restore"))
 	}

pkg/service/one2onerestore/helpers_integration_test.go

@@ -3,8 +3,11 @@
 package one2onerestore
 
 import (
+	"time"
+
 	"github.com/pkg/errors"
 	"github.com/scylladb/go-set/strset"
+	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
 	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/uuid"
 )
@@ -37,9 +40,10 @@ type node struct {
 
 // Host contains basic information about Scylla node.
 type Host struct {
-	ID   string
-	DC   string
-	Addr string
+	ID         string
+	DC         string
+	Addr       string
+	ShardCount int
 }
 
 // ViewType either Materialized View or Secondary Index.
@@ -53,11 +57,12 @@ const (
 
 // View represents statement used for recreating restored (dropped) views.
 type View struct {
-	Keyspace   string   `json:"keyspace" db:"keyspace_name"`
-	View       string   `json:"view" db:"view_name"`
-	Type       ViewType `json:"type" db:"view_type"`
-	BaseTable  string   `json:"base_table"`
-	CreateStmt string   `json:"create_stmt"`
+	Keyspace    string                       `json:"keyspace" db:"keyspace_name"`
+	View        string                       `json:"view" db:"view_name"`
+	Type        ViewType                     `json:"type" db:"view_type"`
+	BaseTable   string                       `json:"base_table"`
+	CreateStmt  string                       `json:"create_stmt,omitempty"`
+	BuildStatus scyllaclient.ViewBuildStatus `json:"status"`
 }
 
 // hostWorkload represents what data (manifest) from the backup should be handled
@@ -72,6 +77,10 @@ type hostWorkload struct {
 
 type scyllaTable struct{ keyspace, table string }
 
+func (st scyllaTable) String() string {
+	return st.keyspace + "." + st.table
+}
+
 func getTablesToRestore(workload []hostWorkload) map[scyllaTable]struct{} {
 	tablesToRestore := map[scyllaTable]struct{}{}
 	for _, wl := range workload {
@@ -185,3 +194,95 @@ func checkHostMapping(hostMap map[string]struct{}, hostID string) error {
 	}
 	return errors.Errorf("host is already mapped: %s", hostID)
 }
+
+// RunProgress describes progress of various 1-1-restore stages.
+type RunProgress struct {
+	ClusterID uuid.UUID
+	TaskID    uuid.UUID
+	RunID     uuid.UUID
+
+	KeyspaceName     string
+	TableName        string
+	TableSize        int64
+	RemoteSSTableDir string `db:"remote_sstable_dir"`
+	TombstoneGC      string
+
+	Host     string // IP of the node to which SSTables are downloaded.
+	ShardCnt int    // Host shard count used for bandwidth per shard calculation.
+
+	VersionedProgress int64
+
+	StartedAt   *time.Time
+	CompletedAt *time.Time
+	// RClone job info fields
+	AgentJobID   int64
+	ScyllaTaskID string // reserved for future use
+
+	Downloaded int64
+	Skipped    int64
+	Failed     int64
+	Error      string
+
+	ViewName        string
+	ViewType        ViewType
+	ViewBuildStatus scyllaclient.ViewBuildStatus
+
+	Stage Stage
+}
+
+// Stage specifies the restore stage.
+type Stage string
+
+// Stage enumeration.
+const (
+	StageDropViews     Stage = "DROP_VIEWS"
+	StageAlterTGC      Stage = "ALTER_TGC"
+	StageData          Stage = "DATA"
+	StageRecreateViews Stage = "RECREATE_VIEWS"
+	StageDone          Stage = "DONE"
+)
+
+// Progress groups restore progress for all restored keyspaces.
+type Progress struct {
+	progress
+
+	SnapshotTag string             `json:"snapshot_tag"`
+	Keyspaces   []KeyspaceProgress `json:"keyspaces,omitempty"`
+	Hosts       []HostProgress     `json:"hosts,omitempty"`
+	Views       []View             `json:"views,omitempty"`
+	Stage       Stage              `json:"stage"`
+}
+
+// KeyspaceProgress groups restore progress for the tables belonging to this keyspace.
+type KeyspaceProgress struct {
+	progress
+
+	Keyspace string          `json:"keyspace"`
+	Tables   []TableProgress `json:"tables,omitempty"`
+}
+
+// TableProgress defines restore progress for the table.
+type TableProgress struct {
+	progress
+
+	Table       string          `json:"table"`
+	TombstoneGC tombstoneGCMode `json:"tombstone_gc"`
+	Error       string          `json:"error,omitempty"`
+}
+
+// HostProgress groups restore progress for the host.
+type HostProgress struct {
+	Host             string `json:"host"`
+	ShardCnt         int    `json:"shard_cnt"`
+	DownloadedBytes  int64  `json:"downloaded_bytes"`
+	DownloadDuration int64  `json:"download_duration"`
+}
+
+type progress struct {
+	Size        int64      `json:"size"`
+	Restored    int64      `json:"restored"`
+	Downloaded  int64      `json:"downloaded"`
+	Failed      int64      `json:"failed"`
+	StartedAt   *time.Time `json:"started_at"`
+	CompletedAt *time.Time `json:"completed_at"`
+}