From f1dcea58bcc91310a4131e45188edab01585efe8 Mon Sep 17 00:00:00 2001 From: Max Schettler Date: Thu, 13 Aug 2026 15:35:31 +0200 Subject: [PATCH] Fix bucket switching for backup recovery --- .../controllers/backup_controller.py | 52 ++++- simplyblock_core/rpc_client.py | 25 +++ .../services/tasks_runner_backup.py | 15 +- tests/unit/rpc/test_client.py | 22 ++ tests/unit/test_backup_recovery_bucket.py | 196 ++++++++++++++++++ 5 files changed, 298 insertions(+), 12 deletions(-) create mode 100644 tests/unit/test_backup_recovery_bucket.py diff --git a/simplyblock_core/controllers/backup_controller.py b/simplyblock_core/controllers/backup_controller.py index 60409a9db..67e181d3f 100644 --- a/simplyblock_core/controllers/backup_controller.py +++ b/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: diff --git a/simplyblock_core/rpc_client.py b/simplyblock_core/rpc_client.py index 1bba9177e..71324c2a9 100644 --- a/simplyblock_core/rpc_client.py +++ b/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: diff --git a/simplyblock_core/services/tasks_runner_backup.py b/simplyblock_core/services/tasks_runner_backup.py index 8f0ed8af0..885570b62 100644 --- a/simplyblock_core/services/tasks_runner_backup.py +++ b/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" diff --git a/tests/unit/rpc/test_client.py b/tests/unit/rpc/test_client.py index 1a146cfac..cecd32fd6 100644 --- a/tests/unit/rpc/test_client.py +++ b/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") diff --git a/tests/unit/test_backup_recovery_bucket.py b/tests/unit/test_backup_recovery_bucket.py new file mode 100644 index 000000000..d033ce6f5 --- /dev/null +++ b/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])