Replace per-file replication with periodic cron job to copy entire bucket at once

backend/btrixcloud/basecrawls.py

@@ -344,47 +344,6 @@ async def update_usernames(self, userid: UUID, updated_name: str) -> None:
             {"userid": userid}, {"$set": {"userName": updated_name}}
         )
 
-    async def replicate_crawl_files(
-        self, crawl_id: str, org: Organization, type_: TYPE_CRAWL_TYPES
-    ):
-        """Replicate crawl files to configured replica locations"""
-        repl_logger = logger.bind(crawl_id=crawl_id, oid=org.id, type=type_)
-
-        try:
-            crawl = await self.get_base_crawl(crawl_id, org, type_)
-        # pylint: disable=broad-exception-caught
-        except Exception:
-            repl_logger.warning(
-                "crawl_replicate_skipped_not_found",
-                unstructured_message=f"Not replicating files for crawl {crawl_id}: crawl not found",
-            )
-            return
-
-        for crawl_file in crawl.files:
-            try:
-                await self.background_job_ops.create_replica_jobs(
-                    crawl.oid, crawl_file, crawl.id, type_
-                )
-            # pylint: disable=broad-exception-caught
-            except Exception as exc:
-                repl_logger.exception(
-                    "crawl_replicate_failed",
-                    unstructured_message=f"Replicate Exception {exc}",
-                )
-
-    async def add_crawl_file_replica(
-        self, crawl_id: str, filename: str, ref: StorageRef
-    ) -> dict[str, object]:
-        """Add replica StorageRef to existing CrawlFile"""
-        return await self.crawls.find_one_and_update(
-            {"_id": crawl_id, "files.filename": filename},
-            {
-                "$addToSet": {
-                    "files.$.replicas": {"name": ref.name, "custom": ref.custom}
-                }
-            },
-        )
-
     async def shutdown_crawl(self, crawl_id: str, org: Organization, graceful: bool):
         """placeholder, implemented in crawls, base version does nothing"""
 
@@ -675,7 +634,6 @@ async def bulk_presigned_files(
                                 hash=file["hash"],
                                 size=file["size"],
                                 crawlId=file["crawl_id"],
-                                numReplicas=len(file.get("replicas") or []),
                                 expireAt=date_to_str(
                                     presigned["signedAt"]
                                     + self.storage_ops.signed_duration_delta
@@ -717,7 +675,6 @@ async def bulk_presigned_files(
                         hash=file["hash"],
                         size=file["size"],
                         crawlId=file["crawl_id"],
-                        numReplicas=len(file.get("replicas") or []),
                         expireAt=date_to_str(expire_at),
                     )
                 )

backend/btrixcloud/crawlmanager.py

@@ -81,21 +81,54 @@ async def run_profile_browser(
 
         return browserid
 
-    async def run_replica_job(
+    async def run_copy_bucket_job(
+        self,
+        primary_storage: StorageRef,
+        replica_storage: StorageRef,
+        primary_endpoint: str,
+        primary_bucket_suffix: str,
+        replica_endpoint: str,
+        replica_bucket_suffix: str,
+        existing_job_id: str | None = None,
+    ) -> str:
+        """run job to replicate primary storage bucket to replica location"""
+        job_type = BgJobType.COPY_BUCKET.value
+
+        if existing_job_id:
+            job_id = existing_job_id
+        else:
+            job_id = f"{job_type}-{secrets.token_hex(5)}"
+
+        params: dict[str, object] = {
+            "id": job_id,
+            "primary_secret_name": primary_storage.get_storage_secret_name(),
+            "primary_file_path": primary_bucket_suffix,
+            "primary_endpoint": primary_endpoint,
+            "replica_secret_name": replica_storage.get_storage_secret_name(),
+            "replica_file_path": replica_bucket_suffix,
+            "replica_endpoint": replica_endpoint,
+            "BgJobType": BgJobType,
+        }
+
+        data = self.templates.env.get_template("copy_bucket_job.yaml").render(params)
+
+        await self.create_from_yaml(data)
+
+        return job_id
+
+    async def run_delete_replica_job(
         self,
         oid: str,
-        job_type: str,
         replica_storage: StorageRef,
         replica_file_path: str,
         replica_endpoint: str,
         delay_days: int = 0,
-        primary_storage: StorageRef | None = None,
-        primary_file_path: str | None = None,
-        primary_endpoint: str | None = None,
         existing_job_id: str | None = None,
     ) -> tuple[str, str | None]:
         """run job to replicate file from primary storage to replica storage"""
 
+        job_type = BgJobType.DELETE_REPLICA.value
+
         if existing_job_id:
             job_id = existing_job_id
         else:
@@ -109,23 +142,16 @@ async def run_replica_job(
             "replica_secret_name": replica_storage.get_storage_secret_name(oid),
             "replica_file_path": replica_file_path,
             "replica_endpoint": replica_endpoint,
-            "primary_secret_name": (
-                primary_storage.get_storage_secret_name(oid)
-                if primary_storage
-                else None
-            ),
-            "primary_file_path": primary_file_path if primary_file_path else None,
-            "primary_endpoint": primary_endpoint if primary_endpoint else None,
             "BgJobType": BgJobType,
         }
 
-        if job_type == BgJobType.DELETE_REPLICA.value and delay_days > 0:
+        if delay_days > 0:
             # If replica deletion delay is configured, schedule as cronjob
             return await self.create_replica_deletion_scheduled_job(
                 job_id, params, delay_days
             )
 
-        data = self.templates.env.get_template("replica_job.yaml").render(params)
+        data = self.templates.env.get_template("delete_replica_job.yaml").render(params)
 
         await self.create_from_yaml(data)
 
@@ -352,19 +378,52 @@ async def ensure_cleanup_seed_file_cron_job_exists(self):
             larger_resources=True,
         )
 
-    async def ensure_retry_stuck_uploads_cron_job_exists(self):
+    async def ensure_retry_stuck_uploads_cron_job_exists(
+        self, disable_job: bool = False
+    ):
         """ensure cron background job to retry stuck uploads exists"""
 
         default_schedule = "0 * * * *"
         job_schedule = os.environ.get(
             "RETRY_STUCK_UPLOADS_CRON_SCHEDULE", default_schedule
         )
 
-        await self._ensure_bg_cron_job_exists(
-            "retry-stuck-uploads-cron",
-            BgJobType.RETRY_STUCK_UPLOADS.value,
-            job_schedule,
-        )
+        job_id = "retry-stuck-uploads-cron"
+
+        if not disable_job:
+            return await self._ensure_bg_cron_job_exists(
+                job_id,
+                BgJobType.RETRY_STUCK_UPLOADS.value,
+                job_schedule,
+            )
+
+        # If no replica locations are configured, make sure no replication cron
+        # job exists, as one could have been previously configured.
+        logger.info("bg_cron_job_deleting", job_id=job_id, disable_job=True)
+        await self._delete_cron_job_if_exists(job_id)
+
+    async def ensure_file_replication_cron_job_exists(
+        self, replicas_configured: bool = False
+    ):
+        """ensure cron background job to replica default storages exists"""
+
+        # Default schedule is every 2 hours
+        default_schedule = "0 */2 * * *"
+        job_schedule = os.environ.get("REPLICATION_JOB_CRON_SCHEDULE", default_schedule)
+
+        job_id = "replicate-files-cron"
+
+        if replicas_configured:
+            return await self._ensure_bg_cron_job_exists(
+                job_id,
+                BgJobType.REPLICATE_FILES_CRON.value,
+                job_schedule,
+            )
+
+        # If no replica locations are configured, make sure no replication cron
+        # job exists, as one could have been previously configured.
+        logger.info("bg_cron_job_deleting", job_id=job_id, replicas_configured=False)
+        await self._delete_cron_job_if_exists(job_id)
 
     async def _ensure_bg_cron_job_exists(
         self,
@@ -413,6 +472,17 @@ async def _ensure_bg_cron_job_exists(
         cron_logger.info("bg_cron_job_creating")
         await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE)
 
+    async def _delete_cron_job_if_exists(self, job_id: str):
+        """Delete cron job by id if it exists"""
+        try:
+            await self.batch_api.delete_namespaced_cron_job(
+                name=job_id,
+                namespace=DEFAULT_NAMESPACE,
+            )
+        except ApiException as exc:
+            if exc.status != 404:
+                raise
+
     async def create_crawl_job(
         self,
         crawlconfig: CrawlConfig,

backend/btrixcloud/db.py

@@ -39,7 +39,7 @@
     ) = object
 
 
-CURR_DB_VERSION = "0058"
+CURR_DB_VERSION = "0059"
 
 MIN_DB_VERSION = 7.0
 

backend/btrixcloud/main_bg.py

@@ -60,7 +60,7 @@ async def main():
         coll_ops,
         _,
         _,
-        _,
+        bg_job_ops,
         _,
         user_manager,
         _,
@@ -107,6 +107,15 @@ async def main():
             )
             return ExitCode.ERROR
 
+    if job_type == BgJobType.REPLICATE_FILES_CRON:
+        try:
+            await bg_job_ops.create_copy_bucket_jobs()
+            return ExitCode.SUCCESS
+        # pylint: disable=broad-exception-caught
+        except Exception:
+            crawl_logger.exception("bg_job_failed")
+            return ExitCode.ERROR
+
     # Run job (org-specific)
     if not oid:
         crawl_logger.error(

backend/btrixcloud/migrations/migration_0052_profile_filenames.py

@@ -26,8 +26,6 @@ async def migrate_up(self) -> None:
         """Perform migration up.
 
         Add oid prefix to profile resource filenames that don't already have it.
-        For any profiles that match, also delete the database record for any
-        existing replicas and then spawn new replication jobs.
         """
         profiles_mdb = self.mdb["profiles"]
 
@@ -59,24 +57,9 @@ async def migrate_up(self) -> None:
                         {
                             "$set": {
                                 "resource.filename": new_filename,
-                                "resource.replicas": [],
                             }
                         },
                     )
-
-                    profile.resource.filename = new_filename
-                    profile.resource.replicas = []
-
-                    logger.info(
-                        "profile_replication_job_started",
-                        profile_id=profile.id,
-                        unstructured_message=(
-                            f"Starting background jobs to replicate profile {profile.id}"
-                        ),
-                    )
-                    await self.background_job_ops.create_replica_jobs(
-                        profile.oid, profile.resource, str(profile.id), "profile"
-                    )
                 # pylint: disable=broad-exception-caught
                 except Exception:
                     logger.exception(

backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py

@@ -0,0 +1,49 @@
+"""
+Migration 0059 - Remove per-file tracking of file replicas in database
+"""
+
+import structlog
+
+from btrixcloud.migrations import BaseMigration
+
+logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__)
+
+MIGRATION_VERSION = "0059"
+
+
+class Migration(BaseMigration):
+    """Migration class."""
+
+    # pylint: disable=unused-argument
+    def __init__(self, mdb, **kwargs):
+        super().__init__(mdb, migration_version=MIGRATION_VERSION)
+
+    async def migrate_up(self):
+        """Perform migration up.
+
+        Remove replicas array from all crawl and profile files.
+        """
+        crawls_mdb = self.mdb["crawls"]
+        profiles_mdb = self.mdb["profiles"]
+
+        try:
+            res = await crawls_mdb.update_many(
+                {"files": {"$nin": [None, []]}},
+                {"$unset": {"files.$[].replicas": 1}},
+            )
+            updated = res.modified_count
+            logger.info("updated_crawl_files", count=updated)
+        # pylint: disable=broad-exception-caught
+        except Exception:
+            logger.exception("failed_to_update_crawl_files")
+
+        try:
+            res = await profiles_mdb.update_many(
+                {"resource": {"$ne": None}},
+                {"$unset": {"resource.replicas": 1}},
+            )
+            updated = res.modified_count
+            logger.info("updated_profile_files", count=updated)
+        # pylint: disable=broad-exception-caught
+        except Exception:
+            logger.exception("failed_to_update_profile_files")