Fix bucket switching for backup recovery

simplyblock_core/controllers/backup_controller.py

@@ -168,6 +168,41 @@ def _ensure_s3_bucket(backup_config, bucket_name):
         raise RuntimeError(f"Error ensuring S3 bucket {bucket_name} exists") from e
 
 
+def backup_bucket_name(backup_config, cluster_id, source_cluster_id=None):
+    """The S3 bucket holding the backups a given cluster produced.
+
+    A cluster may override the name of its own bucket in its backup config;
+    other clusters' buckets are only reachable under the default name.
+    """
+    if source_cluster_id in (None, "", cluster_id):
+        return (backup_config or {}).get("bucket_name", f"simplyblock-backup-{cluster_id}")
+    return f"simplyblock-backup-{source_cluster_id}"
+
+
+def register_recovery_buckets(rpc_client, node, cluster, backups) -> None:
+    """Tell a node's S3 bdev which foreign buckets the given backups live in.
+
+    Imported backups keep the s3_ids they had on the cluster that created
+    them, but their objects stay in that cluster's bucket. Without this the
+    data plane resolves them against the local bucket and every GET 404s.
+    """
+    backup_config = cluster.backup_config or {}
+    cluster_id = cluster.get_id()
+    local_bucket = backup_bucket_name(backup_config, cluster_id)
+
+    s3_ids_by_bucket: dict = {}
+    for backup in backups:
+        bucket_name = backup_bucket_name(backup_config, cluster_id, backup.source_cluster_id)
+        if bucket_name != local_bucket:
+            s3_ids_by_bucket.setdefault(bucket_name, []).append(backup.s3_id)
+
+    s3_bdev_name = f"s3_{node.lvstore}"
+    for bucket_name, s3_ids in s3_ids_by_bucket.items():
+        rpc_client.bdev_s3_register_recovery_bucket(s3_bdev_name, bucket_name, s3_ids)
+        logger.info(f"Recovery bucket {bucket_name} registered for s3_ids {s3_ids} "
+                    f"on {s3_bdev_name} of node {node.get_id()}")
+
+
 def create_s3_bdev(node, backup_config) -> None:
     """Create the S3 bdev and attach it to a node's lvstore.
     Called during cluster activate / node restart.
@@ -206,7 +241,7 @@ def create_s3_bdev(node, backup_config) -> None:
             s3_thread_pool_size=backup_config.get("s3_thread_pool_size", 0),
         )
 
-        bucket_name = backup_config.get("bucket_name", f"simplyblock-backup-{node.cluster_id}")
+        bucket_name = backup_bucket_name(backup_config, node.cluster_id)
         _ensure_s3_bucket(backup_config, bucket_name)
 
         rpc_client.bdev_s3_add_bucket_name(s3_bdev_name, bucket_name, allow_existing=True)
@@ -424,9 +459,6 @@ def restore_backup(backup_id: str, lvol_name: str, pool_id_or_name: str,
     except KeyError as e:
         raise PreconditionError(str(e)) from e
 
-    # Verify the backup's source matches the active S3 source.
-    # If the backup came from an external cluster, the S3 bdev must be
-    # switched to that cluster's bucket before restoring.
     backup_src = backup.source_cluster_id or backup.cluster_id
     active_src = cluster.backup_source or cluster.uuid
     if backup_src != active_src:
@@ -726,7 +758,10 @@ def switch_backup_source(cluster_id, source_cluster_id) -> None:
         source_cluster_id: The cluster ID whose S3 bucket to activate.
             Use the local cluster_id (or "local") to switch back.
 
-    Returns (success, error_message).
+    Restoring an imported backup no longer requires this — see
+    register_recovery_buckets.
+
+    Raises PreconditionError if the cluster or the bucket is unavailable.
     """
     try:
         cluster = db_controller.get_cluster_by_id(cluster_id)
@@ -736,13 +771,8 @@ def switch_backup_source(cluster_id, source_cluster_id) -> None:
     if source_cluster_id == "local":
         source_cluster_id = cluster_id
 
-    # Determine the bucket name for the source cluster
     backup_config = cluster.backup_config or {}
-    if source_cluster_id == cluster_id:
-        bucket_name = backup_config.get("bucket_name",
-                                        f"simplyblock-backup-{cluster_id}")
-    else:
-        bucket_name = f"simplyblock-backup-{source_cluster_id}"
+    bucket_name = backup_bucket_name(backup_config, cluster_id, source_cluster_id)
 
     # Verify the bucket exists
     try:

simplyblock_core/rpc_client.py

@@ -1930,6 +1930,31 @@ def bdev_s3_add_bucket_name(self, name, bucket_name, allow_existing: bool = Fals
                 return None
             raise
 
+    def bdev_s3_register_recovery_bucket(self, name, bucket_name, s3_ids) -> None:
+        """Point recovery reads for the given backups at an external bucket.
+
+        The mapping is per-s3_id and additive: backups not listed here keep
+        reading from — and all backup writes keep going to — the bucket
+        registered via bdev_s3_add_bucket_name. It lives in the S3 bdev's
+        memory only, so it must be re-registered after an SPDK restart.
+
+        Args:
+            name: S3 bdev name (e.g. 's3_LVS_1234')
+            bucket_name: bucket the backup objects actually live in
+            s3_ids: list of S3 backup IDs (uint32) stored in that bucket
+
+        The data plane answers with a bare `true` acknowledging that the
+        mapping was stored — it says nothing about the bucket existing or
+        holding those backups, so there is nothing for a caller to inspect.
+        A rejected registration raises RPCRemoteError instead.
+        """
+        self._request3(
+            "bdev_s3_register_recovery_bucket",
+            name=name,
+            bucket_name=bucket_name,
+            s3_ids=s3_ids,
+        )
+
     def bdev_lvol_s3_backup(self, s3_id, snapshot_names, cluster_batch=1):
         """Start an async backup of snapshots to S3.
         Args:

simplyblock_core/services/tasks_runner_backup.py

@@ -10,7 +10,7 @@
 import time
 
 from simplyblock_core import constants, db_controller, utils
-from simplyblock_core.controllers import backup_events
+from simplyblock_core.controllers import backup_controller, backup_events
 from simplyblock_core.models.backup import Backup
 from simplyblock_core.models.cluster import Cluster
 from simplyblock_core.models.job_schedule import JobSchedule
@@ -221,6 +221,19 @@ def _run_restore(task):
 
     if not recovery_started:
         try:
+            cluster = db.get_cluster_by_id(task.cluster_id)
+        except KeyError:
+            task.function_result = f"Cluster {task.cluster_id} not found"
+            task.status = JobSchedule.STATUS_DONE
+            task.write_to_db(db.kv_store)
+            return
+
+        # The bucket mapping only lives in the S3 bdev's memory, so it is
+        # re-registered on every recovery attempt rather than once per restore.
+        chain = [b for b in db.get_backup_chain(backup_id) if b.s3_id in chain_ids]
+
+        try:
+            backup_controller.register_recovery_buckets(rpc_client, snode, cluster, chain)
             ret = rpc_client.bdev_lvol_s3_recovery(lvol_name, chain_ids, cluster_batch=16)
             if not ret:
                 task.function_result = "bdev_lvol_s3_recovery RPC failed"

tests/unit/rpc/test_client.py

@@ -42,6 +42,28 @@ def test_get_bdevs_with_name_separate_from_all(self, mock_req):
         self.assertEqual(mock_req.call_count, 2)
 
 
+class TestRegisterRecoveryBucket(unittest.TestCase):
+
+    @patch.object(RPCClient, "_request3")
+    def test_sends_bucket_and_ids(self, mock_req):
+        mock_req.return_value = True
+        client = _make_client()
+
+        self.assertIsNone(
+            client.bdev_s3_register_recovery_bucket("s3_LVS_21", "bucket-c1", [1, 2]))
+        mock_req.assert_called_once_with(
+            "bdev_s3_register_recovery_bucket",
+            name="s3_LVS_21", bucket_name="bucket-c1", s3_ids=[1, 2])
+
+    @patch.object(RPCClient, "_request3")
+    def test_rejection_propagates(self, mock_req):
+        mock_req.side_effect = RPCRemoteError("no such bdev", code=-errno.ENODEV)
+        client = _make_client()
+
+        with self.assertRaises(RPCRemoteError):
+            client.bdev_s3_register_recovery_bucket("s3_LVS_21", "bucket-c1", [1])
+
+
 class TestSubsystem(unittest.TestCase):
 
     @patch.object(RPCClient, "_request3")

tests/unit/test_backup_recovery_bucket.py

@@ -0,0 +1,196 @@
+# coding=utf-8
+"""
+test_backup_recovery_bucket.py — unit tests for cross-cluster restore bucket
+resolution (SFAM-2797: a backup exported from C1 and restored on C2 was read
+from C2's own bucket, so every S3 GET returned 404 and the restore task
+crash-looped until it exhausted its retries).
+"""
+
+import unittest
+from unittest.mock import MagicMock, call, patch
+
+from simplyblock_core.controllers.backup_controller import (
+    backup_bucket_name, register_recovery_buckets,
+)
+from simplyblock_core.models.backup import Backup
+from simplyblock_core.models.cluster import Cluster
+from simplyblock_core.models.job_schedule import JobSchedule
+from simplyblock_core.models.storage_node import StorageNode
+from simplyblock_core.rpc_client import RPCClient
+
+C1 = "3359a4e8-e17b-4ef8-a5fb-1924e0745287"
+C2 = "27d33e2f-ba6e-44ad-bde4-06bdcc2197d0"
+
+
+def _cluster(cluster_id=C2, backup_config=None):
+    c = MagicMock(spec=Cluster)
+    c.uuid = cluster_id
+    c.backup_config = backup_config
+    c.get_id = MagicMock(return_value=cluster_id)
+    return c
+
+
+def _node(node_id="node-1", lvstore="LVS_21"):
+    n = MagicMock(spec=StorageNode)
+    n.uuid = node_id
+    n.status = StorageNode.STATUS_ONLINE
+    n.lvstore = lvstore
+    n.get_id = MagicMock(return_value=node_id)
+    return n
+
+
+def _backup(s3_id=1, source_cluster_id="", cluster_id=C2):
+    b = MagicMock(spec=Backup)
+    b.uuid = f"backup-{s3_id}"
+    b.s3_id = s3_id
+    b.cluster_id = cluster_id
+    b.source_cluster_id = source_cluster_id
+    b.status = Backup.STATUS_COMPLETED
+    return b
+
+
+class TestBackupBucketName(unittest.TestCase):
+
+    def test_own_bucket_defaults_to_cluster_id(self):
+        self.assertEqual(backup_bucket_name({}, C2), f"simplyblock-backup-{C2}")
+
+    def test_own_bucket_honours_config_override(self):
+        self.assertEqual(backup_bucket_name({"bucket_name": "custom"}, C2), "custom")
+
+    def test_own_bucket_by_explicit_source(self):
+        self.assertEqual(backup_bucket_name({"bucket_name": "custom"}, C2, C2), "custom")
+
+    def test_foreign_bucket_ignores_local_override(self):
+        self.assertEqual(
+            backup_bucket_name({"bucket_name": "custom"}, C2, C1),
+            f"simplyblock-backup-{C1}")
+
+    def test_missing_config_is_tolerated(self):
+        self.assertEqual(backup_bucket_name(None, C2), f"simplyblock-backup-{C2}")
+
+
+class TestRegisterRecoveryBuckets(unittest.TestCase):
+
+    def test_imported_backups_are_registered(self):
+        rpc_client = MagicMock(spec=RPCClient)
+        register_recovery_buckets(
+            rpc_client, _node(), _cluster(),
+            [_backup(s3_id=1, source_cluster_id=C1),
+             _backup(s3_id=2, source_cluster_id=C1)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once_with(
+            "s3_LVS_21", f"simplyblock-backup-{C1}", [1, 2])
+
+    def test_local_backups_are_not_registered(self):
+        rpc_client = MagicMock(spec=RPCClient)
+        register_recovery_buckets(
+            rpc_client, _node(), _cluster(),
+            [_backup(s3_id=1), _backup(s3_id=2, source_cluster_id=C2)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_not_called()
+
+    def test_local_backups_under_a_renamed_bucket_are_not_registered(self):
+        rpc_client = MagicMock(spec=RPCClient)
+        register_recovery_buckets(
+            rpc_client, _node(), _cluster(backup_config={"bucket_name": "custom"}),
+            [_backup(s3_id=1, source_cluster_id=C2)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_not_called()
+
+    def test_mixed_chain_registers_only_foreign_ids(self):
+        rpc_client = MagicMock(spec=RPCClient)
+        register_recovery_buckets(
+            rpc_client, _node(), _cluster(),
+            [_backup(s3_id=1, source_cluster_id=C1), _backup(s3_id=2)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once_with(
+            "s3_LVS_21", f"simplyblock-backup-{C1}", [1])
+
+    def test_one_call_per_source_cluster(self):
+        other = "11111111-2222-3333-4444-555555555555"
+        rpc_client = MagicMock(spec=RPCClient)
+        register_recovery_buckets(
+            rpc_client, _node(), _cluster(),
+            [_backup(s3_id=1, source_cluster_id=C1),
+             _backup(s3_id=2, source_cluster_id=other),
+             _backup(s3_id=3, source_cluster_id=C1)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_has_calls([
+            call("s3_LVS_21", f"simplyblock-backup-{C1}", [1, 3]),
+            call("s3_LVS_21", f"simplyblock-backup-{other}", [2]),
+        ], any_order=True)
+        self.assertEqual(rpc_client.bdev_s3_register_recovery_bucket.call_count, 2)
+
+
+class TestRestoreTaskRegistersBeforeRecovery(unittest.TestCase):
+    """The registration is what makes the data plane read from the source
+    cluster's bucket, so it has to reach the node before recovery starts and
+    again on every re-issue — the mapping lives only in the S3 bdev's memory.
+    """
+
+    def _task(self, recovery_started=False):
+        t = MagicMock(spec=JobSchedule)
+        t.uuid = "task-1"
+        t.cluster_id = C2
+        t.node_id = "node-1"
+        t.retry = 0
+        t.status = JobSchedule.STATUS_NEW
+        t.function_params = {
+            "backup_id": "backup-1",
+            "lvol_name": "LVS_21/LVOL_43",
+            "lvol_id": "",
+            "chain_ids": [1],
+            "recovery_started": recovery_started,
+        }
+        return t
+
+    def _run(self, task, chain):
+        from simplyblock_core.services import tasks_runner_backup
+
+        snode = _node()
+        rpc_client = MagicMock(spec=RPCClient)
+        snode.rpc_client = MagicMock(return_value=rpc_client)
+
+        db = MagicMock()
+        db.get_storage_node_by_id.return_value = snode
+        db.get_cluster_by_id.return_value = _cluster()
+        db.get_backup_chain.return_value = chain
+
+        with patch.object(tasks_runner_backup, "db", db):
+            tasks_runner_backup._run_restore(task)
+
+        return rpc_client
+
+    def test_foreign_bucket_registered_before_recovery(self):
+        rpc_client = self._run(self._task(), [_backup(s3_id=1, source_cluster_id=C1)])
+
+        self.assertEqual(
+            [c[0] for c in rpc_client.method_calls],
+            ["bdev_s3_register_recovery_bucket", "bdev_lvol_s3_recovery"])
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once_with(
+            "s3_LVS_21", f"simplyblock-backup-{C1}", [1])
+
+    def test_recovery_reissue_re_registers(self):
+        """"No process" resets recovery_started, so the node may have restarted
+        and lost the mapping."""
+        rpc_client = self._run(self._task(), [_backup(s3_id=1, source_cluster_id=C1)])
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once()
+
+        rpc_client = self._run(self._task(), [_backup(s3_id=1, source_cluster_id=C1)])
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once()
+
+    def test_local_restore_issues_no_registration(self):
+        rpc_client = self._run(self._task(), [_backup(s3_id=1)])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_not_called()
+        rpc_client.bdev_lvol_s3_recovery.assert_called_once_with(
+            "LVS_21/LVOL_43", [1], cluster_batch=16)
+
+    def test_backups_outside_the_restored_chain_are_ignored(self):
+        rpc_client = self._run(
+            self._task(),
+            [_backup(s3_id=1, source_cluster_id=C1),
+             _backup(s3_id=9, source_cluster_id="other-cluster")])
+
+        rpc_client.bdev_s3_register_recovery_bucket.assert_called_once_with(
+            "s3_LVS_21", f"simplyblock-backup-{C1}", [1])