Add seed file support to Browsertrix backend (#2710)

Fixes #2673 

Changes in this PR:

- Adds a new `file_uploads.py` module and corresponding `/files` API
prefix with methods/endpoints for uploading, GETing, and deleting seed
files (can be extended to other types of files moving forward)
- Seed files are supported via `CrawlConfig.config.seedFileId` on POST
and PATCH endpoints. This seedFileId is replaced by a presigned url when
passed to the crawler by the operator
- Seed files are read when first uploaded to calculate `firstSeed` and
`seedCount` and store them in the database, and this is copied into the
workflow and crawl documents when they are created.
- Logic is added to store `firstSeed` and `seedCount` for other
workflows as well, and a migration added to backfill data, to maintain
consistency and fix some of the pymongo aggregations that previously
assumed all workflows would have at least one `Seed` object in
`CrawlConfig.seeds`
- Seed file and thumbnail storage stats are added to org stats
- Seed file and thumbnail uploads first check that the org's storage
quota has not been exceeded and return a 400 if so
- A cron background job (run weekly each Sunday at midnight by default,
but configurable) is added to look for seed files at least x minutes old
(1440 minutes, or 1 day, by default, but configurable) that are not in
use in any workflows, and to delete them when they are found. The
backend pods will ensure this k8s batch job exists when starting up and
create it if it does not already exist. A database entry for each run of
the job is created in the operator on job completion so that it'll
appear in the `/jobs` API endpoints, but retrying of this type of
regularly scheduled background job is not supported as we don't want to
accidentally create multiple competing scheduled jobs.
- Adds a `min_seed_file_crawler_image` value to the Helm chart that is
checked before creating a crawl from a workflow if set. If a workflow
cannot be run, return the detail of the exception in
`CrawlConfigAddedResponse.errorDetail` so that we can display the reason
in the frontend
- Add SeedFile model from base UserFile (former ImageFIle), ensure all APIs
returning uploaded files return an absolute pre-signed URL (either with external origin or internal service origin)

---------
Co-authored-by: Ilya Kreymer <[email protected]>

Author: tw4l

backend/btrixcloud/background_jobs.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import os
+import secrets
 from datetime import datetime
 from typing import Optional, Tuple, Union, List, Dict, TYPE_CHECKING, cast
 from uuid import UUID
@@ -24,6 +25,7 @@
     RecalculateOrgStatsJob,
     ReAddOrgPagesJob,
     OptimizePagesJob,
+    CleanupSeedFilesJob,
     PaginatedBackgroundJobResponse,
     AnyJob,
     StorageRef,
@@ -43,7 +45,7 @@
 
 
 # ============================================================================
-# pylint: disable=too-many-instance-attributes
+# pylint: disable=too-many-instance-attributes, too-many-public-methods
 class BackgroundJobOps:
     """k8s background job management"""
 
@@ -473,17 +475,39 @@ async def create_optimize_crawl_pages_job(
             print(f"warning: optimize pages job could not be started: {exc}")
             return None
 
+    async def ensure_cron_cleanup_jobs_exist(self):
+        """Ensure background job to clean up unused seed files weekly exists"""
+        await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists()
+
     async def job_finished(
         self,
         job_id: str,
         job_type: str,
         success: bool,
         finished: datetime,
+        started: Optional[datetime] = None,
         oid: Optional[UUID] = None,
     ) -> None:
         """Update job as finished, including
         job-specific task handling"""
 
+        # For seed file cleanup jobs, no database record will exist for each
+        # run before this point, so create it here
+        if job_type == BgJobType.CLEANUP_SEED_FILES:
+            if not started:
+                started = finished
+            cleanup_job = CleanupSeedFilesJob(
+                id=f"seed-files-{secrets.token_hex(5)}",
+                type=BgJobType.CLEANUP_SEED_FILES,
+                started=started,
+                finished=finished,
+                success=success,
+            )
+            await self.jobs.insert_one(cleanup_job.to_dict())
+            if not success:
+                await self._send_bg_job_failure_email(cleanup_job, finished)
+            return
+
         job = await self.get_background_job(job_id)
         if job.finished:
             return
@@ -499,28 +523,31 @@ async def job_finished(
                     cast(DeleteReplicaJob, job)
                 )
         else:
-            print(
-                f"Background job {job.id} failed, sending email to superuser",
-                flush=True,
-            )
-            superuser = await self.user_manager.get_superuser()
-            org = None
-            if job.oid:
-                org = await self.org_ops.get_org_by_id(job.oid)
-            await asyncio.get_event_loop().run_in_executor(
-                None,
-                self.email.send_background_job_failed,
-                job,
-                finished,
-                superuser.email,
-                org,
-            )
+            await self._send_bg_job_failure_email(job, finished)
 
         await self.jobs.find_one_and_update(
             {"_id": job_id, "oid": oid},
             {"$set": {"success": success, "finished": finished}},
         )
 
+    async def _send_bg_job_failure_email(self, job: BackgroundJob, finished: datetime):
+        print(
+            f"Background job {job.id} failed, sending email to superuser",
+            flush=True,
+        )
+        superuser = await self.user_manager.get_superuser()
+        org = None
+        if job.oid:
+            org = await self.org_ops.get_org_by_id(job.oid)
+        await asyncio.get_event_loop().run_in_executor(
+            None,
+            self.email.send_background_job_failed,
+            job,
+            finished,
+            superuser.email,
+            org,
+        )
+
     async def get_background_job(
         self, job_id: str, oid: Optional[UUID] = None
     ) -> Union[
@@ -530,6 +557,7 @@ async def get_background_job(
         RecalculateOrgStatsJob,
         ReAddOrgPagesJob,
         OptimizePagesJob,
+        CleanupSeedFilesJob,
     ]:
         """Get background job"""
         query: dict[str, object] = {"_id": job_id}
@@ -542,6 +570,7 @@ async def get_background_job(
 
         return self._get_job_by_type_from_data(res)
 
+    # pylint: disable=too-many-return-statements
     def _get_job_by_type_from_data(self, data: dict[str, object]):
         """convert dict to propert background job type"""
         if data["type"] == BgJobType.CREATE_REPLICA:
@@ -559,6 +588,9 @@ def _get_job_by_type_from_data(self, data: dict[str, object]):
         if data["type"] == BgJobType.OPTIMIZE_PAGES:
             return OptimizePagesJob.from_dict(data)
 
+        if data["type"] == BgJobType.CLEANUP_SEED_FILES:
+            return CleanupSeedFilesJob.from_dict(data)
+
         return DeleteOrgJob.from_dict(data)
 
     async def list_background_jobs(
@@ -736,6 +768,9 @@ async def retry_org_background_job(
             )
             return {"success": True}
 
+        if job.type == BgJobType.CLEANUP_SEED_FILES:
+            raise HTTPException(status_code=400, detail="cron_job_retry_not_supported")
+
         return {"success": False}
 
     async def retry_failed_org_background_jobs(

backend/btrixcloud/basecrawls.py

@@ -432,27 +432,14 @@ async def _resolve_crawl_refs(
         crawl: Union[CrawlOut, CrawlOutWithResources],
         org: Optional[Organization],
         files: Optional[list[dict]],
-        add_first_seed: bool = True,
     ):
         """Resolve running crawl data"""
         # pylint: disable=too-many-branches
-        config = None
-        if crawl.cid:
-            config = await self.crawl_configs.get_crawl_config(
-                crawl.cid, org.id if org else None, active_only=False
-            )
-
         if not org:
             org = await self.orgs.get_org_by_id(crawl.oid)
             if not org:
                 raise HTTPException(status_code=400, detail="missing_org")
 
-        if config and config.config.seeds:
-            if add_first_seed:
-                first_seed = config.config.seeds[0]
-                crawl.firstSeed = first_seed.url
-            crawl.seedCount = len(config.config.seeds)
-
         if hasattr(crawl, "profileid") and crawl.profileid:
             crawl.profileName = await self.crawl_configs.profiles.get_profile_name(
                 crawl.profileid, org
@@ -685,9 +672,7 @@ async def list_all_base_crawls(
 
         aggregate = [
             {"$match": query},
-            {"$set": {"firstSeedObject": {"$arrayElemAt": ["$config.seeds", 0]}}},
-            {"$set": {"firstSeed": "$firstSeedObject.url"}},
-            {"$unset": ["firstSeedObject", "errors", "behaviorLogs", "config"]},
+            {"$unset": ["errors", "behaviorLogs", "config"]},
             {"$set": {"activeQAStats": "$qa.stats"}},
             {
                 "$set": {