Merge pull request #574 from geoadmin/fix-PB-1575-delete-external

PB-1575: make deletion batch size configurable.

Author: adk-swisstopo

app/config/settings_prod.py

@@ -209,6 +209,7 @@
 DELETE_EXPIRED_ITEMS_OLDER_THAN_HOURS = 24
 DELETE_EXPIRED_ITEMS_MAX = 110 * 1000
 DELETE_EXPIRED_ITEMS_MAX_PERCENTAGE = 50
+DELETE_EXPIRED_ITEMS_BATCH_SIZE = 10 * 1000
 
 # Media files (i.e. uploaded content=assets in this project)
 UPLOAD_FILE_CHUNK_SIZE = 1024 * 1024  # Size in Bytes

app/stac_api/management/commands/remove_expired_items.py

@@ -22,11 +22,49 @@ def __str__(self):
 
 class Handler(CommandHandler):
 
-    def delete(self, instance, object_type):
-        if self.options['dry_run']:
-            self.print_success(f'skipping deletion of {object_type} {instance}')
-        else:
-            instance.delete()
+    def delete_by_batch(self, queryset, object_type, batch_size):
+        # When many rows are involved, looping over each one is very slow.
+        # Running a single delete() against all of them consumes a lot of memory
+        # and does not delete anything if it fails mid-way. Hence, we batch.
+        #
+        # Django's delete() method already batches deletions in groups of 100
+        # rows. These batches are wrapped within transactions. It does not seem
+        # to be designed to allow disabling the transaction or tweaking the
+        # batch size.
+        # https://github.com/django/django/blob/main/django/db/models/sql/subqueries.py#L26
+        # https://github.com/django/django/blob/main/django/db/models/deletion.py#L454
+        # Also, it does not seem to do anything to reduce memory consumption.
+        #
+        # In our case, we don't need the deletions to be transactional. If we
+        # die in the middle, it's fine if some rows are deleted and some are
+        # not. We can remove the remaining rows next time we run. That's better
+        # than waiting forever, to fail and to have to start from scratch next
+        # time.
+        type_name = f'stac_api.{object_type.__name__}'
+        total = queryset.count()
+        deleted_count = 0
+        while deleted_count < total:
+            # We cannot just call queryset[:batch_size].delete() because DELETE
+            # does not support LIMIT/OFFSET. So instead we extract the ids
+            # then we'll build a new QuerySet to DELETE them.
+            ids = queryset.values('id')[:batch_size]
+            expected_deletions = len(ids)
+            if expected_deletions == 0:
+                break
+            dry_run_prefix = ''
+            if self.options['dry_run']:
+                dry_run_prefix = '[dry run]: '
+                deleted_objs = {}
+                actual_deletions = expected_deletions
+            else:
+                (_, deleted_objs) = object_type.objects.filter(id__in=ids).delete()
+                actual_deletions = deleted_objs.get(type_name, 0)
+            deleted_count += actual_deletions
+            self.print_success(
+                f'{dry_run_prefix}Deleted {deleted_count}/{total} {type_name}.'
+                f' In this batch: {actual_deletions}/{expected_deletions}.'
+                f' All objects in this batch: {deleted_objs}.'
+            )
 
     def _raise_if_too_many_deletions(self, max_deletions, max_deletions_pct, items_count):
         if items_count > max_deletions:
@@ -42,18 +80,20 @@ def _raise_if_too_many_deletions(self, max_deletions, max_deletions_pct, items_c
 
     def run(self):
         self.print_success('running command to remove expired items')
+        batch_size = self.options['batch_size']
         min_age_hours = self.options['min_age_hours']
         max_deletions = self.options['max_deletions']
         max_deletions_pct = self.options['max_deletions_percentage']
         self.print_warning(
             f"deleting no more than {max_deletions} or "
             f"{max_deletions_pct}%% items expired for longer"
-            f" than {min_age_hours} hours"
+            f" than {min_age_hours} hours, {batch_size} at a time"
         )
 
         expiration = timezone.now() - timedelta(hours=min_age_hours)
 
-        items = Item.objects.filter(properties_expires__lte=expiration)
+        items = Item.objects.filter(properties_expires__lte=expiration
+                                   ).order_by('properties_expires')
         items_count = items.count()
 
         self._raise_if_too_many_deletions(max_deletions, max_deletions_pct, items_count)
@@ -69,8 +109,8 @@ def run(self):
                 "WARNING: There were still pending asset uploads for expired items. "
                 "These were likely stale, so we aborted them"
             )
-        self.delete(assets, 'assets')
-        self.delete(items, 'items')
+        self.delete_by_batch(assets, Asset, batch_size)
+        self.delete_by_batch(items, Item, batch_size)
 
         if self.options['dry_run']:
             self.print_success(f'[dry run] would have removed {items_count} expired items')
@@ -105,6 +145,13 @@ def add_arguments(self, parser: CommandParser) -> None:
             action='store_true',
             help='Simulate deleting items, without actually deleting them'
         )
+        default_batch_size = settings.DELETE_EXPIRED_ITEMS_BATCH_SIZE
+        parser.add_argument(
+            '--batch-size',
+            type='positive_int',
+            default=default_batch_size,
+            help=f"How many rows to delete at a time ({default_batch_size})"
+        )
         default_min_age = settings.DELETE_EXPIRED_ITEMS_OLDER_THAN_HOURS
         parser.add_argument(
             '--min-age-hours',

app/tests/tests_10/test_remove_expired_items.py

@@ -230,11 +230,6 @@ def test_remove_item_dry_run(self):
             expected_stdout_patterns=[
                 "running command to remove expired items",
                 "deleting no more than 110000 or 50% items expired for longer than 24 hours",
-                (
-                    "skipping deletion of assets <QuerySet"
-                    " [<Asset: asset-0.tiff>, <Asset: asset-1.tiff>]>"
-                ),
-                "skipping deletion of items <ItemQuerySet [<Item: collection-1/item-0>]>",
                 "[dry run] would have removed 1 expired items",
             ]
         )
@@ -331,6 +326,12 @@ class RemoveExpiredItemsManyWithProfiling(RemoveExpiredItemsBase):
     def test_remove_item(self):
         self.run_test(command_args=["--max-deletions-percentage=99"], expected_stdout_patterns=[])
 
+    def test_multiple_batches(self):
+        self.run_test(
+            command_args=["--max-deletions-percentage=99", "--batch-size=7"],
+            expected_stdout_patterns=[]
+        )
+
     @staticmethod
     def _diff_memory(before, after):
         diff = after.compare_to(before, 'filename')