Pause / Resume Crawls Initial Implmentation. (#2572)

- add 'pause' crawl state (fixes #2567)
- gracefully shut down crawler pods, and then redis pod when paused
- crawler uploads WACZ before shutting down (dependent on
webrecorder/browsertrix-crawler#824, supported in 1.6.1+)
- add 'paused_at' on crawl spec to indicate when crawl is paused
- support max pause time limit, after which crawl becomes automatically
stopped.
- add 'stopped_pause_expired' when pause automatically expires and crawl
is stopped
- /crawl/<id>/{pause,resume} apis to toggle 'paused' on crawl spec
- ui: add pause/resume button, paused state (partially addresses #2568)
- ui: add pausing/resuming derivative states when crawl is running and
pausing, or paused and not pausing (partially addresses #2569)
- Designed to work with crawler 1.6.1+ which support pausing + uploading on pause

Work on #2566, Fixes #2576 

---------
Co-authored-by: sua yoo <[email protected]>
Co-authored-by: Tessa Walsh <[email protected]>
Co-authored-by: sua yoo <[email protected]>

Author: 3 people

backend/btrixcloud/basecrawls.py

@@ -35,7 +35,7 @@
     User,
     StorageRef,
     RUNNING_AND_WAITING_STATES,
-    SUCCESSFUL_STATES,
+    SUCCESSFUL_AND_PAUSED_STATES,
     QARun,
     UpdatedResponse,
     DeletedResponseQuota,
@@ -460,7 +460,7 @@ async def _resolve_crawl_refs(
 
         if (
             files
-            and crawl.state in SUCCESSFUL_STATES
+            and crawl.state in SUCCESSFUL_AND_PAUSED_STATES
             and isinstance(crawl, CrawlOutWithResources)
         ):
             crawl.resources = await self._files_to_resources(files, org, crawl.id)

backend/btrixcloud/crawlconfigs.py

@@ -11,7 +11,7 @@
 import re
 import os
 import traceback
-from datetime import datetime
+from datetime import datetime, timedelta
 from uuid import UUID, uuid4
 import urllib.parse
 
@@ -95,6 +95,8 @@ class CrawlConfigOps:
     crawler_images_map: dict[str, str]
     crawler_image_pull_policy_map: dict[str, str]
 
+    paused_expiry_delta: timedelta
+
     def __init__(
         self,
         dbclient,
@@ -121,6 +123,10 @@ def __init__(
             "DEFAULT_CRAWLER_IMAGE_PULL_POLICY", "IfNotPresent"
         )
 
+        self.paused_expiry_delta = timedelta(
+            minutes=int(os.environ.get("PAUSED_CRAWL_LIMIT_MINUTES", "10080"))
+        )
+
         self.router = APIRouter(
             prefix="/crawlconfigs",
             tags=["crawlconfigs"],
@@ -765,6 +771,13 @@ async def _add_running_curr_crawl_stats(self, crawlconfig: CrawlConfigOut):
         crawlconfig.lastCrawlState = crawl.state
         crawlconfig.lastCrawlSize = crawl.stats.size if crawl.stats else 0
         crawlconfig.lastCrawlStopping = crawl.stopping
+        crawlconfig.lastCrawlShouldPause = crawl.shouldPause
+        crawlconfig.lastCrawlPausedAt = crawl.pausedAt
+        crawlconfig.lastCrawlPausedExpiry = None
+        if crawl.pausedAt:
+            crawlconfig.lastCrawlPausedExpiry = (
+                crawl.pausedAt + self.paused_expiry_delta
+            )
         crawlconfig.isCrawlRunning = True
 
     async def get_crawl_config_out(self, cid: UUID, org: Organization):

backend/btrixcloud/crawlmanager.py

@@ -4,7 +4,7 @@
 import secrets
 
 from typing import Optional, Dict, Tuple
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from fastapi import HTTPException
 
@@ -386,6 +386,14 @@ async def shutdown_crawl(self, crawl_id: str, graceful=True) -> dict:
 
         return await self.delete_crawl_job(crawl_id)
 
+    async def pause_resume_crawl(
+        self, crawl_id: str, paused_at: Optional[datetime] = None
+    ) -> dict:
+        """pause or resume a crawl"""
+        return await self._patch_job(
+            crawl_id, {"pausedAt": date_to_str(paused_at) if paused_at else ""}
+        )
+
     async def delete_crawl_configs_for_org(self, org: str) -> None:
         """Delete all crawl configs for given org"""
         await self._delete_crawl_configs(f"btrix.org={org}")

backend/btrixcloud/crawls.py

@@ -769,6 +769,39 @@ async def get_crawl_stats(
 
         return crawls_data
 
+    async def pause_crawl(
+        self, crawl_id: str, org: Organization, pause: bool
+    ) -> Dict[str, bool]:
+        """pause or resume a crawl temporarily"""
+        crawl = await self.get_base_crawl(crawl_id, org)
+        if crawl and crawl.type != "crawl":
+            raise HTTPException(status_code=400, detail="not_a_crawl")
+
+        result = None
+
+        if pause:
+            paused_at = dt_now()
+        else:
+            paused_at = None
+
+        try:
+            result = await self.crawl_manager.pause_resume_crawl(
+                crawl_id, paused_at=paused_at
+            )
+
+            if result.get("success"):
+                await self.crawls.find_one_and_update(
+                    {"_id": crawl_id, "type": "crawl", "oid": org.id},
+                    {"$set": {"shouldPause": pause, "pausedAt": paused_at}},
+                )
+
+                return {"success": True}
+        # pylint: disable=bare-except
+        except:
+            pass
+
+        raise HTTPException(status_code=404, detail="crawl_not_found")
+
     async def shutdown_crawl(
         self, crawl_id: str, org: Organization, graceful: bool
     ) -> Dict[str, bool]:
@@ -1242,6 +1275,22 @@ async def crawl_cancel_immediately(
     async def crawl_graceful_stop(crawl_id, org: Organization = Depends(org_crawl_dep)):
         return await ops.shutdown_crawl(crawl_id, org, graceful=True)
 
+    @app.post(
+        "/orgs/{oid}/crawls/{crawl_id}/pause",
+        tags=["crawls"],
+        response_model=SuccessResponse,
+    )
+    async def pause_crawl(crawl_id, org: Organization = Depends(org_crawl_dep)):
+        return await ops.pause_crawl(crawl_id, org, pause=True)
+
+    @app.post(
+        "/orgs/{oid}/crawls/{crawl_id}/resume",
+        tags=["crawls"],
+        response_model=SuccessResponse,
+    )
+    async def resume_crawl(crawl_id, org: Organization = Depends(org_crawl_dep)):
+        return await ops.pause_crawl(crawl_id, org, pause=False)
+
     @app.post(
         "/orgs/{oid}/crawls/delete",
         tags=["crawls"],

backend/btrixcloud/main.py

@@ -124,6 +124,8 @@ class SettingsResponse(BaseModel):
 
     localesEnabled: Optional[List[str]]
 
+    pausedExpiryMinutes: int
+
 
 # ============================================================================
 # pylint: disable=too-many-locals, duplicate-code
@@ -158,6 +160,7 @@ def main() -> None:
             if os.environ.get("LOCALES_ENABLED")
             else None
         ),
+        pausedExpiryMinutes=int(os.environ.get("PAUSED_CRAWL_LIMIT_MINUTES", 10080)),
     )
 
     invites = init_invites(mdb, email)

backend/btrixcloud/models.py

@@ -222,7 +222,9 @@ class UserOrgInfoOut(BaseModel):
 ]
 RUNNING_STATES = get_args(TYPE_RUNNING_STATES)
 
-TYPE_WAITING_STATES = Literal["starting", "waiting_capacity", "waiting_org_limit"]
+TYPE_WAITING_STATES = Literal[
+    "starting", "waiting_capacity", "waiting_org_limit", "paused"
+]
 WAITING_STATES = get_args(TYPE_WAITING_STATES)
 
 TYPE_FAILED_STATES = Literal[
@@ -236,11 +238,13 @@ class UserOrgInfoOut(BaseModel):
 TYPE_SUCCESSFUL_STATES = Literal[
     "complete",
     "stopped_by_user",
+    "stopped_pause_expired",
     "stopped_storage_quota_reached",
     "stopped_time_quota_reached",
     "stopped_org_readonly",
 ]
 SUCCESSFUL_STATES = get_args(TYPE_SUCCESSFUL_STATES)
+SUCCESSFUL_AND_PAUSED_STATES = ["paused", *SUCCESSFUL_STATES]
 
 TYPE_RUNNING_AND_WAITING_STATES = Literal[TYPE_WAITING_STATES, TYPE_RUNNING_STATES]
 RUNNING_AND_WAITING_STATES = [*WAITING_STATES, *RUNNING_STATES]
@@ -478,6 +482,9 @@ class CrawlConfigOut(CrawlConfigCore, CrawlConfigAdditional):
     id: UUID
 
     lastCrawlStopping: Optional[bool] = False
+    lastCrawlShouldPause: Optional[bool] = False
+    lastCrawlPausedAt: Optional[datetime] = None
+    lastCrawlPausedExpiry: Optional[datetime] = None
     profileName: Optional[str] = None
     firstSeed: Optional[str] = None
     seedCount: int = 0
@@ -863,6 +870,8 @@ class CrawlOut(BaseMongoModel):
     seedCount: Optional[int] = None
     profileName: Optional[str] = None
     stopping: Optional[bool] = False
+    shouldPause: Optional[bool] = False
+    pausedAt: Optional[datetime] = None
     manual: bool = False
     cid_rev: Optional[int] = None
     scale: Scale = 1
@@ -1017,6 +1026,7 @@ class Crawl(BaseCrawl, CrawlConfigCore):
     manual: bool = False
 
     stopping: Optional[bool] = False
+    shouldPause: Optional[bool] = False
 
     qaCrawlExecSeconds: int = 0