improve expiry performance (#2800)

- use elastic to get expired items
- only search for broadcast if enabled
- introduce `archive_internal` resource to avoid additional filtering

Author: petrjasek

apps/archive/__init__.py

@@ -18,12 +18,14 @@
 from superdesk.celery_app import celery
 
 from .archive import (
+    ArchiveInternalResource,
     ArchiveResource,
     ArchiveService,
     ArchiveVersionsResource,
     ArchiveVersionsService,
     AutoSaveResource,
     ArchiveSaveService,
+    archive_internal_service,
 )
 from .commands import RemoveExpiredContent, LOCK_EXPIRY
 from .ingest import IngestResource, AppIngestService
@@ -97,6 +99,9 @@ def init_app(app) -> None:
     service = NewsService(endpoint_name, backend=superdesk.get_backend())
     NewsResource(endpoint_name, app=app, service=service)
 
+    endpoint_name = "archive_internal"
+    ArchiveInternalResource(endpoint_name, app=app, service=archive_internal_service)
+
     from apps.item_autosave.components.item_autosave import ItemAutosave
     from apps.item_autosave.models.item_autosave import ItemAutosaveModel
 

apps/archive/archive.py

@@ -1178,33 +1178,47 @@ def get_expired_items(self, expiry_datetime, last_id=None, invalid_only=False):
         """
         for i in range(app.config["MAX_EXPIRY_LOOPS"]):  # avoid blocking forever just in case
             query = {
-                "$and": [
-                    {"expiry": {"$lte": expiry_datetime}},
-                    {"$or": [{"task.desk": {"$ne": None}}, {ITEM_STATE: CONTENT_STATE.SPIKED, "task.desk": None}]},
-                ]
+                "bool": {
+                    "must": [
+                        {"range": {"expiry": {"lte": expiry_datetime}}},
+                        {
+                            "bool": {
+                                "should": [
+                                    {"exists": {"field": "task.desk"}},
+                                    {"term": {ITEM_STATE: CONTENT_STATE.SPIKED}},
+                                ],
+                            }
+                        },
+                    ],
+                    "must_not": [],
+                }
             }
 
             if invalid_only:
-                query["$and"].append({"expiry_status": "invalid"})
+                query["bool"]["must"].append({"term": {"expiry_status": "invalid"}})
             else:
-                query["$and"].append({"expiry_status": {"$ne": "invalid"}})
+                query["bool"]["must_not"].append({"term": {"expiry_status": "invalid"}})
 
-            if last_id:
-                query["$and"].append({"_id": {"$gt": last_id}})
+            if last_id:  # elastic does not support range query on _id, so using guid
+                query["bool"]["must"].append({"range": {"guid": {"gt": last_id}}})
 
-            req = ParsedRequest()
-            req.sort = "_id"
-            req.max_results = app.config["MAX_EXPIRY_QUERY_LIMIT"]
-            req.where = json.dumps(query)
+            source = {
+                "query": query,
+                "sort": [{"guid": "asc"}, {"versioncreated": "asc"}],
+                "size": app.config["MAX_EXPIRY_QUERY_LIMIT"],
+            }
 
-            items = list(self.get_from_mongo(req=req, lookup={}))
+            items = list(archive_internal_service.search(source))
 
             yield items  # we need to yield the empty list too to signal it's the end
 
             if not len(items):
                 break
             else:
-                last_id = items[-1]["_id"]
+                try:
+                    last_id = items[-1]["guid"]
+                except KeyError:
+                    pass
 
         else:
             logger.warning("get_expired_items did not finish in %d loops", app.config["MAX_EXPIRY_LOOPS"])
@@ -1444,6 +1458,28 @@ def on_fetched_item(self, item):
         return item
 
 
+class ArchiveInternalResource(Resource):
+    """Archive Internal Resource without additional filtering."""
+
+    schema = item_schema()
+    datasource = {
+        "source": "archive",
+        "search_backend": "elastic",
+    }
+    resource_methods = []
+    item_methods = []
+    versioning = False
+    collation = False
+    internal_resource = True
+
+
+class ArchiveInternalService(BaseService):
+    pass
+
+
+archive_internal_service = ArchiveInternalService("archive_internal", backend=superdesk.get_backend())
+
+
 superdesk.workflow_state("in_progress")
 superdesk.workflow_action(
     name="save",

apps/archive/commands.py

@@ -125,10 +125,12 @@ def _remove_expired_items(self, expiry_datetime):
         config_service = get_resource_service("config")
         archive_service = get_resource_service(ARCHIVE)
         published_service = get_resource_service("published")
-        preserve_published_desks = {
-            desk.get(config.ID_FIELD): 1
-            for desk in get_resource_service("desks").find(where={"preserve_published_content": True})
-        }
+        preserve_published_desks = set(
+            [
+                str(desk.get(config.ID_FIELD))
+                for desk in get_resource_service("desks").find(where={"preserve_published_content": True})
+            ]
+        )
 
         last_id = config_service.get(LAST_ID_CONFIG)
         if last_id:
@@ -310,20 +312,23 @@ def _can_remove_item(self, item, now, processed_item=None, preserve_published_de
             # Get the item references for is package
             item_refs = package_service.get_residrefs(item)
 
-        if item.get(ITEM_TYPE) in [CONTENT_TYPE.TEXT, CONTENT_TYPE.PREFORMATTED]:
+        if item.get(ITEM_TYPE) in [CONTENT_TYPE.TEXT, CONTENT_TYPE.PREFORMATTED] and app.config.get(
+            "BROADCAST_ENABLED", True
+        ):
             broadcast_items = get_resource_service("archive_broadcast").get_broadcast_items_from_master_story(item)
             # If master story expires then check if broadcast item is included in a package.
             # If included in a package then check the package expiry.
             item_refs.extend([broadcast_item.get(config.ID_FIELD) for broadcast_item in broadcast_items])
 
+        if item.get(ITEM_TYPE) in [CONTENT_TYPE.TEXT, CONTENT_TYPE.PREFORMATTED]:
             if item.get("rewrite_of"):
                 item_refs.append(item.get("rewrite_of"))
 
             if item.get("rewritten_by"):
                 item_refs.append(item.get("rewritten_by"))
 
         # get the list of associated item ids
-        if item.get(ITEM_TYPE) in MEDIA_TYPES:
+        elif item.get(ITEM_TYPE) in MEDIA_TYPES:
             item_refs.extend(self._get_associated_items(item))
 
         # get item reference where this referred
@@ -337,7 +342,7 @@ def _can_remove_item(self, item, now, processed_item=None, preserve_published_de
         if (
             preserve_published_desks
             and item.get(ITEM_STATE) in {CONTENT_STATE.PUBLISHED, CONTENT_STATE.CORRECTED}
-            and item.get("task").get("desk") in preserve_published_desks
+            and str(item.get("task").get("desk")) in preserve_published_desks
         ):
             is_expired = False
             reason = "Desk config"

superdesk/default_settings.py

@@ -1140,3 +1140,9 @@ def local_to_utc_hour(hour):
 #: .. versionadded:: 2.8
 #:
 PICTURE_METADATA_MAPPING = {}
+
+#: Enable broadcast feature
+#:
+#: .. versionadded:: 2.8
+#:
+BROADCAST_ENABLED = strtobool(env("BROADCAST_ENABLED", "true"))

tests/archive/archive_test.py

@@ -459,7 +459,7 @@ def test_get_expired_items(self):
         for i in range(1000):
             items.append(
                 {
-                    "_id": generate_guid(type=GUID_TAG),
+                    "guid": generate_guid(type=GUID_TAG),
                     "type": "text",
                     "expiry": now - timedelta(days=1),
                     "task": {"desk": "foo"},
@@ -478,7 +478,7 @@ def test_get_expired_items(self):
         assert 1000 == counter
 
         counter = 0
-        ids = sorted([item["_id"] for item in items])
+        ids = sorted([item["guid"] for item in items])
         last_id = ids[499]
         for _items in service.get_expired_items(now, last_id=last_id):
             for item in _items: