From aaff0600465e0c44192a6453aa76cb8942dfef2b Mon Sep 17 00:00:00 2001 From: Michael Wallace Date: Wed, 12 Jul 2023 11:02:30 +0100 Subject: [PATCH] Add support for deletion of AWS snapshot backups Adds support for deletion of AWS snapshot backups by implementing the delete_snapshot_backup function in AwsCloudSnapshotInterface. Closes BAR-25. --- barman/cloud_providers/aws_s3.py | 29 +++++++- doc/manual/28-snapshots.en.md | 4 +- tests/test_cloud_snapshot_interface.py | 93 ++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 3 deletions(-) diff --git a/barman/cloud_providers/aws_s3.py b/barman/cloud_providers/aws_s3.py index d5f879cbe..2e0960879 100644 --- a/barman/cloud_providers/aws_s3.py +++ b/barman/cloud_providers/aws_s3.py @@ -774,13 +774,40 @@ def take_snapshot_backup(self, backup_info, instance_identifier, volumes): account_id=snapshot_resp["OwnerId"], ) + def _delete_snapshot(self, snapshot_id): + """ + Delete the specified snapshot. + + :param str snapshot_id: The ID of the snapshot to be deleted. + """ + try: + self.ec2_client.delete_snapshot(SnapshotId=snapshot_id) + except ClientError as exc: + error_code = exc.response["Error"]["Code"] + # If the snapshot could not be found then deletion is considered successful + # otherwise we raise a CloudProviderError + if error_code == "InvalidSnapshot.NotFound": + logging.warning("Snapshot {} could not be found".format(snapshot_id)) + else: + raise CloudProviderError( + "Deletion of snapshot %s failed with error code %s: %s" + % (snapshot_id, error_code, exc.response["Error"]) + ) + logging.info("Snapshot %s deleted", snapshot_id) + def delete_snapshot_backup(self, backup_info): """ Delete all snapshots for the supplied backup. :param barman.infofile.LocalBackupInfo backup_info: Backup information. """ - raise NotImplementedError() + for snapshot in backup_info.snapshots_info.snapshots: + logging.info( + "Deleting snapshot '%s' for backup %s", + snapshot.identifier, + backup_info.backup_id, + ) + self._delete_snapshot(snapshot.identifier) def get_attached_volumes( self, instance_identifier, disks=None, fail_on_missing=True diff --git a/doc/manual/28-snapshots.en.md b/doc/manual/28-snapshots.en.md index ba26376be..629f530c4 100644 --- a/doc/manual/28-snapshots.en.md +++ b/doc/manual/28-snapshots.en.md @@ -92,6 +92,7 @@ The following additional prerequisites apply to snapshot backups on AWS: The following permissions are required: - `ec2:CreateSnapshot` +- `ec2:DeleteSnapshot` - `ec2:DescribeSnapshots` - `ec2:DescribeInstances` - `ec2:DescribeVolumes` @@ -105,8 +106,7 @@ backup_method = snapshot snapshot_provider = gcp ``` -Currently Google Cloud Platform (`gcp`) and Microsoft Azure (`azure`) are fully supported. -Snapshot backups are supported using AWS however *support for deletion of AWS snapshot backups is not yet implemented*. +Currently Google Cloud Platform (`gcp`), Microsoft Azure (`azure`) and AWS (`aws`) are supported. The following parameters must be set regardless of cloud provider: diff --git a/tests/test_cloud_snapshot_interface.py b/tests/test_cloud_snapshot_interface.py index ce38645dc..89d129656 100644 --- a/tests/test_cloud_snapshot_interface.py +++ b/tests/test_cloud_snapshot_interface.py @@ -3231,6 +3231,99 @@ def test_instance_exists_not_found(self, mock_ec2_client): # THEN it returns False assert resp is False + def test_delete_snapshot(self, mock_ec2_client, caplog): + """Verify that a snapshot can be deleted successfully.""" + # GIVEN a successful response from the delete snapshot request + mock_ec2_client.delete_snapshot.return_value = {} + # AND a mock snapshots interface + snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region) + # AND log level is info + caplog.set_level(logging.INFO) + + # WHEN a snapshot is deleted + snapshot_id = "snap-0123" + snapshot_interface._delete_snapshot(snapshot_id) + + # THEN delete was called on the client with the expected arguments + mock_ec2_client.delete_snapshot.assert_called_once_with(SnapshotId=snapshot_id) + # AND a success message was logged + assert "Snapshot {} deleted".format(snapshot_id) in caplog.text + + def test_delete_snapshot_not_found(self, mock_ec2_client, caplog): + """Verify that a snapshot ID which can't be found is success.""" + # GIVEN a successful response from the delete snapshot request + mock_ec2_client.delete_snapshot.side_effect = ( + ClientError({"Error": {"Code": "InvalidSnapshot.NotFound"}}, "message"), + ) + # AND a mock snapshots interface + snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region) + # AND log level is info + caplog.set_level(logging.INFO) + + # WHEN a snapshot is deleted + # THEN no exceptions are raised + snapshot_id = "snap-0123" + snapshot_interface._delete_snapshot(snapshot_id) + + # THEN delete was called on the client with the expected arguments + mock_ec2_client.delete_snapshot.assert_called_once_with(SnapshotId=snapshot_id) + # AND a success message was logged + assert "Snapshot {} deleted".format(snapshot_id) in caplog.text + # AND a warning message was logged + assert "Snapshot {} could not be found".format(snapshot_id) in caplog.text + + def test_delete_snapshot_failed(self, mock_ec2_client, caplog): + """Verify that a failed deletion results in a CloudProviderError.""" + # GIVEN an unexpected error from the delete snapshot request + mock_ec2_client.delete_snapshot.side_effect = ( + ClientError({"Error": {"Code": "Something.Bad"}}, "message"), + ) + # AND a mock snapshots interface + snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region) + + # WHEN a snapshot is deleted + # THEN a CloudProviderError is raised + snapshot_id = "snap-0123" + with pytest.raises(CloudProviderError) as exc: + snapshot_interface._delete_snapshot(snapshot_id) + + # AND the exception has the expected message + expected_message = "Deletion of snapshot {} failed with error code {}".format( + snapshot_id, "Something.Bad" + ) + assert expected_message in str(exc.value) + + @pytest.mark.parametrize( + "snapshots_list", + ( + [], + [mock.Mock(identifier="snap-0123")], + [mock.Mock(identifier="snap-0123"), mock.Mock(identifier="snap0124")], + ), + ) + def test_delete_snapshot_backup(self, snapshots_list, mock_ec2_client, caplog): + """Verify that all snapshots for a backup are deleted.""" + # GIVEN a backup_info specifying zero or more snapshots + backup_info = mock.Mock( + backup_id=self.backup_id, + snapshots_info=mock.Mock(snapshots=snapshots_list), + ) + # AND log level is info + caplog.set_level(logging.INFO) + # AND the snapshot delete requests are successful + mock_ec2_client.delete_snapshot.return_value = {} + # AND a new AwsCloudSnapshotInterface + snapshot_interface = AwsCloudSnapshotInterface(region=self.aws_region) + + # WHEN delete_snapshot_backup is called + snapshot_interface.delete_snapshot_backup(backup_info) + + # THEN delete_snapshot was called for each snapshot + expected_calls = [ + mock.call(SnapshotId=snapshot.identifier) for snapshot in snapshots_list + ] + mock_ec2_client.delete_snapshot.assert_has_calls(expected_calls) + class TestAwsVolumeMetadata(object): """Verify behaviour of AwsVolumeMetadata."""