fix: call raft read barrier before describe schema

This adds call to raft read barrier to ensure schema consistency before
getting schema from the host.

Author: VAveryanov8

pkg/service/one2onerestore/model.go

@@ -40,10 +40,11 @@ type node struct {
 
 // Host contains basic information about Scylla node.
 type Host struct {
-	ID         string
-	DC         string
-	Addr       string
-	ShardCount int
+	ID                 string
+	DC                 string
+	Addr               string
+	ShardCount         int
+	SafeDescribeMethod scyllaclient.SafeDescribeMethod
 }
 
 // ViewType either Materialized View or Secondary Index.

pkg/service/one2onerestore/progress.go

@@ -20,8 +20,7 @@ import (
 // initProgress adds entries to RunTableProgress and RunViewProgress, so that API call
 // to show progress can return some information at this point.
 func (w *worker) initProgress(ctx context.Context, workload []hostWorkload) error {
-	tablesToRestore := getTablesToRestore(workload)
-	views, err := w.getViews(ctx, tablesToRestore)
+	views, err := w.getViews(ctx, workload)
 	if err != nil {
 		return errors.Wrap(err, "get views")
 	}

pkg/service/one2onerestore/service.go

@@ -122,6 +122,7 @@ func (s *Service) newWorker(ctx context.Context, clusterID, taskID, runID uuid.U
 
 		client:         client,
 		clusterSession: clusterSession,
+		sessionFunc:    s.clusterSession,
 
 		logger: s.logger,
 

pkg/service/one2onerestore/worker.go

@@ -15,6 +15,7 @@ import (
 	"github.com/scylladb/gocqlx/v2"
 	"github.com/scylladb/scylla-manager/backupspec"
 	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
+	"github.com/scylladb/scylla-manager/v3/pkg/service/cluster"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/inexlist/ksfilter"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/parallel"
 	"github.com/scylladb/scylla-manager/v3/pkg/util/query"
@@ -30,6 +31,8 @@ type worker struct {
 	client         *scyllaclient.Client
 	clusterSession gocqlx.Session
 
+	sessionFunc cluster.SessionFunc // is needed to create cql session to single host
+
 	logger log.Logger
 
 	runInfo struct {
@@ -177,12 +180,40 @@ func (w *worker) prepareHostWorkload(ctx context.Context, manifests []*backupspe
 			return errors.Wrap(err, "read manifest content")
 		}
 
+		nodeInfo, err := w.client.NodeInfo(ctx, h.Addr)
+		if err != nil {
+			return errors.Wrapf(err, "get node %s info", h.Addr)
+		}
+		method, err := nodeInfo.SupportsSafeDescribeSchemaWithInternals()
+		if err != nil {
+			return errors.Wrapf(err, "node %s safe describe method", h.Addr)
+		}
+		hw.host.SafeDescribeMethod = method
+
 		result[i] = hw
 
 		return nil
 	}, parallel.NopNotify)
 }
 
+func (w *worker) singleHostCQLSession(ctx context.Context, clusterID uuid.UUID, host string) (gocqlx.Session, error) {
+	session, err := w.sessionFunc(ctx, clusterID, cluster.SingleHostSessionConfigOption(host))
+	if err != nil {
+		return gocqlx.Session{}, errors.Wrap(err, "create cql session")
+	}
+	return session, nil
+}
+
+func (w *worker) raftReadBarrier(ctx context.Context, session gocqlx.Session, host Host) error {
+	switch host.SafeDescribeMethod {
+	case scyllaclient.SafeDescribeMethodReadBarrierAPI:
+		return w.client.RaftReadBarrier(ctx, host.Addr, "")
+	case scyllaclient.SafeDescribeMethodReadBarrierCQL:
+		return query.RaftReadBarrier(session)
+	}
+	return errors.Errorf("unsupported method: %s", host.SafeDescribeMethod)
+}
+
 // alterSchemaRetryWrapper is useful when executing many statements altering schema,
 // as it might take more time for Scylla to process them one after another.
 // This wrapper exits on: success, context cancel, op returned non-timeout error or after maxTotalTime has passed.

pkg/service/one2onerestore/worker_tgc.go

@@ -30,16 +30,15 @@ func (w *worker) setTombstoneGCModeRepair(ctx context.Context, workload []hostWo
 		name     string
 		isView   bool
 	}
-	tablesToRestore := getTablesToRestore(workload)
 	var targets []alterTGCTarget
-	for table := range tablesToRestore {
+	for table := range getTablesToRestore(workload) {
 		targets = append(targets, alterTGCTarget{
 			keyspace: table.keyspace,
 			name:     table.table,
 			isView:   false,
 		})
 	}
-	views, err := w.getViews(ctx, tablesToRestore)
+	views, err := w.getViews(ctx, workload)
 	if err != nil {
 		return errors.Wrap(err, "get views")
 	}

pkg/service/one2onerestore/worker_views.go

@@ -23,7 +23,7 @@ func (w *worker) dropViews(ctx context.Context, workload []hostWorkload) ([]View
 		w.logger.Info(ctx, "Drop views", "took", timeutc.Since(start))
 	}()
 
-	views, err := w.getViews(ctx, getTablesToRestore(workload))
+	views, err := w.getViews(ctx, workload)
 	if err != nil {
 		return nil, errors.Wrap(err, "get views")
 	}
@@ -69,17 +69,19 @@ func (w *worker) dropView(ctx context.Context, view View) error {
 	return alterSchemaRetryWrapper(ctx, op, notify)
 }
 
-func (w *worker) getViews(ctx context.Context, tablesToRestore map[scyllaTable]struct{}) ([]View, error) {
+func (w *worker) getViews(ctx context.Context, workload []hostWorkload) ([]View, error) {
 	keyspaces, err := w.client.Keyspaces(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "get keyspaces")
 	}
 
-	describedViews, err := w.viewsSchemaByName()
+	describedViews, err := w.viewsSchemaByName(ctx, workload)
 	if err != nil {
 		return nil, err
 	}
 
+	tablesToRestore := getTablesToRestore(workload)
+
 	var views []View
 	for _, ks := range keyspaces {
 		meta, err := w.clusterSession.KeyspaceMetadata(ks)
@@ -233,8 +235,30 @@ func (w *worker) waitForViewBuilding(ctx context.Context, view View, pr *RunView
 	return nil
 }
 
-func (w *worker) viewsSchemaByName() (map[scyllaTable]string, error) {
-	describedSchema, err := query.DescribeSchemaWithInternals(w.clusterSession)
+func (w *worker) viewsSchemaByName(ctx context.Context, workload []hostWorkload) (map[scyllaTable]string, error) {
+	// In order to ensure schema consistency, we need to call raft read barrier on a host
+	// from which we are going to read the schema.
+	host := workload[0].host
+	if host.SafeDescribeMethod != scyllaclient.SafeDescribeMethodReadBarrierAPI {
+		// Let's try to find the host that supports raft read barrier api.
+		for _, w := range workload {
+			if w.host.SafeDescribeMethod == scyllaclient.SafeDescribeMethodReadBarrierAPI {
+				host = w.host
+				break
+			}
+		}
+	}
+
+	hostSession, err := w.singleHostCQLSession(ctx, w.runInfo.ClusterID, host.Addr)
+	if err != nil {
+		return nil, errors.Wrap(err, "single host cql session")
+	}
+
+	if err := w.raftReadBarrier(ctx, hostSession, host); err != nil {
+		return nil, errors.Wrap(err, "raft read barrier")
+	}
+
+	describedSchema, err := query.DescribeSchemaWithInternals(hostSession)
 	if err != nil {
 		return nil, errors.Wrap(err, "describe schema")
 	}