Merge pull request #149 from ecmwf-projects/COPDS-2359-request-cleaner

request cleaner with bulk size for system requests

Author: alex75

cads_broker/entry_points.py

@@ -4,6 +4,7 @@
 import enum
 import os
 import random
+import time
 import uuid
 from pathlib import Path
 from typing import Any, Optional
@@ -65,7 +66,10 @@ def add_dummy_requests(
 
 @app.command()
 def requests_cleaner(
-    connection_string: Optional[str] = None, older_than_days: Optional[int] = 365
+    connection_string: Optional[str] = None,
+    older_than_days: Optional[int] = 365,
+    delete_bulk_size: Optional[int] = None,
+    delete_sleep_time: Optional[int] = None,
 ) -> None:
     """Remove records from the system_requests table older than `older_than_days`."""
     if not connection_string:
@@ -74,18 +78,29 @@ def requests_cleaner(
     engine = sa.create_engine(connection_string)
     time_delta = datetime.datetime.now() - datetime.timedelta(days=older_than_days)
     # clean system requests and (via cascading delete) events
-    with engine.begin() as conn:
-        database.logger.info("deleting old system_requests and events...")
+    database.logger.info("deleting old system_requests and events...")
+    curr_deleted = 1
+    subquery = sa.select(database.SystemRequest.request_uid).where(
+        database.SystemRequest.created_at <= time_delta
+    )
+    with engine.connect() as conn:
+        if delete_bulk_size is not None:
+            # delete in sized bulks to give time to db replicas for synch
+            subquery = subquery.limit(delete_bulk_size)
         stmt = sa.delete(database.SystemRequest).where(
-            database.SystemRequest.created_at <= time_delta
-        )
-        result = conn.execute(stmt)
-        conn.commit()
-        num_requests_deleted = result.rowcount
-        database.logger.info(
-            f"{num_requests_deleted} old system requests "
-            f"successfully removed from the broker database."
+            database.SystemRequest.request_uid.in_(subquery)
         )
+        while curr_deleted:
+            with conn.begin():
+                result = conn.execute(stmt)
+                conn.commit()
+                curr_deleted = result.rowcount
+            database.logger.info(
+                f"{curr_deleted} old system requests "
+                f"successfully removed from the broker database."
+            )
+            if delete_sleep_time is not None:
+                time.sleep(delete_sleep_time)
     # clean adaptor_properties
     with engine.begin() as conn:
         try:

tests/test_90_entry_points.py

@@ -1,8 +1,11 @@
 import datetime
+import json
+import logging
 import uuid
 from typing import Any
 
 import cacholote
+import pytest
 import sqlalchemy as sa
 from psycopg import Connection
 from typer.testing import CliRunner
@@ -140,7 +143,9 @@ def prepare_db(
         # not existing case of recent properties with old requests
 
 
-def test_requests_cleaner(session_obj: sa.orm.sessionmaker):
+def test_requests_cleaner(
+    session_obj: sa.orm.sessionmaker, caplog: pytest.LogCaptureFixture
+):
     connection_string = session_obj.kw["bind"].url
 
     # test remove nothing, older_than_days=365 by default
@@ -200,3 +205,45 @@ def test_requests_cleaner(session_obj: sa.orm.sessionmaker):
         assert len(all_requests) == 5
         assert len(all_events) == 5
         assert len(all_props) == 5
+
+    # test remove 10 requests in bulk size of 3 (all old props have old requests)
+    prepare_db(
+        session_obj,
+        num_old_props_used_by_old=10,
+        num_recent_props_used_by_recent=5,
+    )
+    caplog.clear()
+    caplog.set_level(logging.INFO)
+    result = runner.invoke(
+        entry_points.app,
+        [
+            "requests-cleaner",
+            "--connection-string",
+            connection_string,
+            "--delete-bulk-size",
+            3,
+            "--delete-sleep-time",
+            1,
+        ],
+    )
+    assert result.exit_code == 0
+    with session_obj() as session:
+        all_requests = session.query(database.SystemRequest).all()
+        all_events = session.query(database.Events).all()
+        all_props = session.query(database.AdaptorProperties).all()
+        assert len(all_requests) == 5
+        assert len(all_events) == 5
+        assert len(all_props) == 5
+    with caplog.at_level(logging.ERROR):
+        log_msgs = [json.loads(r.msg)["event"] for r in caplog.records]
+    assert log_msgs == [
+        "deleting old system_requests and events...",
+        "3 old system requests successfully removed from the broker database.",
+        "3 old system requests successfully removed from the broker database.",
+        "3 old system requests successfully removed from the broker database.",
+        "1 old system requests successfully removed from the broker database.",
+        "0 old system requests successfully removed from the broker database.",
+        "deleting old adaptor_properties...",
+        "10 old adaptor properties successfully removed from the broker database.",
+    ]
+    caplog.clear()