Always purge all but one incomplete backups

Author: rzvoncek

medusa/purge.py

@@ -38,10 +38,16 @@ def main(config, max_backup_age=0, max_backup_count=0):
             logging.info('Listing backups for {}'.format(config.storage.fqdn))
             backup_index = storage.list_backup_index_blobs()
             backups = list(storage.list_node_backups(fqdn=config.storage.fqdn, backup_index_blobs=backup_index))
+
+            # split backups by completion
+            complete_backups, incomplete_to_purge = backups_to_purge_by_completion(backups)
+            backups_to_purge |= set(incomplete_to_purge)
+
             # list all backups to purge based on date conditions
-            backups_to_purge |= set(backups_to_purge_by_age(backups, max_backup_age))
+            backups_to_purge |= set(backups_to_purge_by_age(complete_backups, max_backup_age))
             # list all backups to purge based on count conditions
-            backups_to_purge |= set(backups_to_purge_by_count(backups, max_backup_count))
+            backups_to_purge |= set(backups_to_purge_by_count(complete_backups, max_backup_count))
+
             # purge all candidate backups
             object_counts = purge_backups(
                 storage, backups_to_purge, config.storage.backup_grace_period_in_days, config.storage.fqdn
@@ -62,6 +68,17 @@ def main(config, max_backup_age=0, max_backup_count=0):
         sys.exit(1)
 
 
+def backups_to_purge_by_completion(backups):
+    complete, incomplete = [], []
+    for backup in backups:
+        complete.append(backup) if backup.finished is not None else incomplete.append(backup)
+
+    # keep the most recent one because it might be in progress
+    incomplete_to_purge = set(incomplete[:-1]) if len(incomplete) > 1 else set()
+
+    return complete, incomplete_to_purge
+
+
 def backups_to_purge_by_age(backups, max_backup_age):
     if max_backup_age > 0:
         max_date = datetime.now() - timedelta(days=max_backup_age)

tests/purge_test.py

@@ -20,7 +20,9 @@
 
 from medusa.config import MedusaConfig, StorageConfig, _namedtuple_from_dict
 from medusa.storage import Storage
-from medusa.purge import backups_to_purge_by_age, backups_to_purge_by_count, backups_to_purge_by_name
+from medusa.purge import (
+    backups_to_purge_by_age, backups_to_purge_by_count, backups_to_purge_by_name, backups_to_purge_by_completion
+)
 from medusa.purge import filter_differential_backups, filter_files_within_gc_grace
 
 from tests.storage_test import make_node_backup, make_cluster_backup, make_blob
@@ -91,6 +93,43 @@ def test_purge_backups_by_count(self):
         obsolete_backups = backups_to_purge_by_count(backups, 40)
         assert len(obsolete_backups) == 0
 
+    def test_purge_backups_by_completion(self):
+        backups = list()
+
+        # Build a list of 40 bi-daily backups, making every second backup incomplete
+        complete = True
+        now = datetime.now()
+        for i in range(0, 80, 2):
+            file_time = now + timedelta(days=(i + 1) - 80)
+            backups.append(make_node_backup(self.storage, str(i), file_time, differential=True, complete=complete))
+            complete = not complete
+
+        self.assertEqual(40, len(backups))
+        complete_backup_names = {nb.name for nb in filter(lambda nb: nb.finished is not None, backups)}
+        self.assertEqual(len(complete_backup_names), 20, "The amount of complete backups is not correct")
+
+        # the base with all 40 backups
+        complete, incomplete_to_purge = backups_to_purge_by_completion(backups)
+        self.assertEqual(20, len(complete))  # 1 is kept because it might be in progress
+        self.assertEqual(19, len(incomplete_to_purge))  # 1 is kept because it might be in progress
+
+        # take all complete backups, but only half of the incomplete ones
+        test_backups = list()
+        for i in range(0, 40, 1):
+            # take each complete backup
+            if backups[i].finished is not None:
+                test_backups.append(backups[i])
+                continue
+            # but only first half of the incomplete ones
+            if i > 20:
+                continue
+            test_backups.append(backups[i])
+        self.assertEqual(20, len(list(filter(lambda b: b.finished is not None, test_backups))))
+        self.assertEqual(10, len(list(filter(lambda b: b.finished is None, test_backups))))
+        complete, incomplete_to_purge = backups_to_purge_by_completion(test_backups)
+        self.assertEqual(20, len(complete))  # 1 is kept because it might be in progress
+        self.assertEqual(9, len(incomplete_to_purge))  # 1 is kept because it might be in progress
+
     def test_filter_differential_backups(self):
         backups = list()
         backups.append(make_node_backup(self.storage, "one", datetime.now(), differential=True))

tests/storage_test.py

@@ -455,7 +455,7 @@ def test_get_table_prefix(self):
         self.assertEqual('prefix/localhost/data/', self.storage._get_table_prefix('prefix', 'localhost'))
 
 
-def make_node_backup(storage, name, backup_date, differential=False, fqdn="localhost"):
+def make_node_backup(storage, name, backup_date, differential=False, fqdn="localhost", complete=True):
     if differential is True:
         differential_blob = make_blob("localhost/{}/meta/differential".format(name), backup_date.timestamp())
     else:
@@ -464,9 +464,10 @@ def make_node_backup(storage, name, backup_date, differential=False, fqdn="local
     schema_blob = make_blob("localhost/{}/meta/schema.cql".format(name), backup_date.timestamp())
     manifest_blob = make_blob("localhost/{}/meta/manifest.json".format(name), backup_date.timestamp())
     return NodeBackup(storage=storage, fqdn=fqdn, name=str(name),
-                      differential_blob=differential_blob, manifest_blob=manifest_blob,
+                      differential_blob=differential_blob, manifest_blob=manifest_blob if complete else None,
                       tokenmap_blob=tokenmap_blob, schema_blob=schema_blob,
-                      started_timestamp=backup_date.timestamp(), finished_timestamp=backup_date.timestamp())
+                      started_timestamp=backup_date.timestamp(),
+                      finished_timestamp=backup_date.timestamp() if complete else None)
 
 
 def make_cluster_backup(storage, name, backup_date, nodes, differential=False):