feat(1-1-restore): validates if source and target clusters nodes are …

…equal

This adds a validation stage for 1-1-restore. The logic is as follows:
- Collect node information for the source cluster from backup manifests
- Collect node information for the target cluster from the Scylla API
- Apply node mappings to the source cluster nodes
- Compare each source node with its corresponding target node

Fixes: #4201

pkg/service/one2onerestore/model.go

@@ -32,6 +32,23 @@ type node struct {
 	HostID string `json:"host_id"`
 }
 
+// LocationInfo contains some basic information about Location
+// Intended to be used for simplifying access to the Location.
+type LocationInfo struct {
+	Location Location
+	// Hosts that have an access to the Location
+	Hosts []Host
+	// Manifests from the Location
+	Manifest []*ManifestInfo
+}
+
+// Host contains basic information about Scylla node.
+type Host struct {
+	ID   string
+	DC   string
+	Addr string
+}
+
 func (t *Target) validateProperties() error {
 	if len(t.Location) == 0 {
 		return errors.New("missing location")

pkg/service/one2onerestore/service.go

@@ -61,6 +61,21 @@ func (s *Service) One2OneRestore(ctx context.Context, clusterID, taskID, runID u
 		return errors.Wrap(err, "parse target")
 	}
 
+	w, err := s.newWorker(ctx, clusterID)
+	if err != nil {
+		return errors.Wrap(err, "new worker")
+	}
+
+	locations, err := w.getLocationInfo(ctx, target)
+	if err != nil {
+		return errors.Wrap(err, "get location info")
+	}
+
+	if err := w.validateClusters(ctx, locations, target.NodesMapping); err != nil {
+		return errors.Wrap(err, "validate cluster")
+	}
+	s.logger.Info(ctx, "Can proceed with 1-1-restore")
+
 	s.logger.Info(ctx, "Not yet implemented", "target", target)
 	return nil
 }
@@ -75,3 +90,23 @@ func (s *Service) parseTarget(properties json.RawMessage) (Target, error) {
 	}
 	return target, nil
 }
+
+func (s *Service) newWorker(ctx context.Context, clusterID uuid.UUID) (worker, error) {
+	client, err := s.scyllaClient(ctx, clusterID)
+	if err != nil {
+		return worker{}, errors.Wrap(err, "get client")
+	}
+	clusterSession, err := s.clusterSession(ctx, clusterID)
+	if err != nil {
+		return worker{}, errors.Wrap(err, "get CQL cluster session")
+	}
+
+	return worker{
+		managerSession: s.session,
+
+		client:         client,
+		clusterSession: clusterSession,
+
+		logger: s.logger,
+	}, nil
+}

pkg/service/one2onerestore/worker.go

@@ -0,0 +1,84 @@
+// Copyright (C) 2025 ScyllaDB
+
+package one2onerestore
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/scylladb/go-log"
+	"github.com/scylladb/gocqlx/v2"
+	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
+)
+
+type worker struct {
+	managerSession gocqlx.Session
+
+	client         *scyllaclient.Client
+	clusterSession gocqlx.Session
+
+	logger log.Logger
+}
+
+func getSourceMappings(mappings []nodeMapping) map[node]node {
+	sourceMappings := map[node]node{}
+	for _, m := range mappings {
+		sourceMappings[m.Source] = m.Target
+	}
+	return sourceMappings
+}
+
+func (w *worker) getLocationInfo(ctx context.Context, target Target) ([]LocationInfo, error) {
+	var result []LocationInfo
+
+	nodeStatus, err := w.client.Status(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "nodes status")
+	}
+	for _, location := range target.Location {
+		// Ignore location.DC because all mappings should be specified via nodes-mapping file
+		nodes, err := w.getNodesWithAccess(ctx, nodeStatus, location.RemotePath(""))
+		if err != nil {
+			return nil, err
+		}
+		manifests, err := w.getManifestInfo(ctx, nodes[0].Addr, target.SnapshotTag, location)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, LocationInfo{
+			Manifest: manifests,
+			Location: location,
+			Hosts:    nodesToHosts(nodes),
+		})
+	}
+
+	return result, nil
+}
+
+func (w *worker) getNodesWithAccess(ctx context.Context, nodesInfo scyllaclient.NodeStatusInfoSlice, remotePath string) (scyllaclient.NodeStatusInfoSlice, error) {
+	nodesWithAccess, err := w.client.GetNodesWithLocationAccess(ctx, nodesInfo, remotePath)
+	if err != nil {
+		if strings.Contains(err.Error(), "NoSuchBucket") {
+			return nil, errors.Errorf("specified bucket does not exist: %s", remotePath)
+		}
+		return nil, errors.Wrapf(err, "location %s is not accessible", remotePath)
+	}
+	if len(nodesWithAccess) == 0 {
+		return nil, fmt.Errorf("no nodes with location %s access", remotePath)
+	}
+	return nodesWithAccess, nil
+}
+
+func nodesToHosts(nodes scyllaclient.NodeStatusInfoSlice) []Host {
+	var hosts []Host
+	for _, n := range nodes {
+		hosts = append(hosts, Host{
+			ID:   n.HostID,
+			DC:   n.Datacenter,
+			Addr: n.Addr,
+		})
+	}
+	return hosts
+}

pkg/service/one2onerestore/worker_manifest.go

@@ -0,0 +1,44 @@
+// Copyright (C) 2025 ScyllaDB
+
+package one2onerestore
+
+import (
+	"context"
+	"path"
+	"slices"
+	"strings"
+
+	. "github.com/scylladb/scylla-manager/v3/pkg/service/backup/backupspec"
+
+	"github.com/scylladb/scylla-manager/v3/pkg/scyllaclient"
+)
+
+// getManifestInfo returns manifests with receiver's snapshot tag for all nodes in the location.
+func (w *worker) getManifestInfo(ctx context.Context, host, snapshotTag string, location Location) ([]*ManifestInfo, error) {
+	baseDir := path.Join("backup", string(MetaDirKind))
+	opts := scyllaclient.RcloneListDirOpts{
+		FilesOnly: true,
+		Recurse:   true,
+	}
+
+	var manifests []*ManifestInfo
+	err := w.client.RcloneListDirIter(ctx, host, location.RemotePath(baseDir), &opts, func(f *scyllaclient.RcloneListDirItem) {
+		m := new(ManifestInfo)
+		if err := m.ParsePath(path.Join(baseDir, f.Path)); err != nil {
+			return
+		}
+		m.Location = location
+		if m.SnapshotTag == snapshotTag {
+			manifests = append(manifests, m)
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Ensure deterministic order
+	slices.SortFunc(manifests, func(a, b *ManifestInfo) int {
+		return strings.Compare(a.NodeID, b.NodeID)
+	})
+	return manifests, nil
+}