diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py
index 732811085e..3cd080be38 100644
--- a/backend/btrixcloud/background_jobs.py
+++ b/backend/btrixcloud/background_jobs.py
@@ -14,12 +14,12 @@
from .crawlmanager import CrawlManager
from .models import (
- CRAWL_TYPES,
AnyJob,
BackgroundJob,
BaseFile,
BgJobType,
CleanupSeedFilesJob,
+ CopyBucketJob,
CreateReplicaJob,
DeleteOrgJob,
DeleteReplicaJob,
@@ -29,6 +29,7 @@
PostProcessUploadJob,
ReAddOrgPagesJob,
RecalculateOrgStatsJob,
+ ReplicateFilesCronJob,
RetryStuckUploadsJob,
StorageRef,
SuccessResponse,
@@ -99,144 +100,18 @@ def strip_bucket(self, endpoint_url: str) -> tuple[str, str]:
parts.path[1:], ""
)
- async def handle_replica_job_succeeded(self, job: CreateReplicaJob) -> None:
- """Update replicas in corresponding file objects, based on type"""
- res = None
- if job.object_type in CRAWL_TYPES:
- res = await self.base_crawl_ops.add_crawl_file_replica(
- job.object_id, job.file_path, job.replica_storage
- )
- elif job.object_type == "profile":
- res = await self.profile_ops.add_profile_file_replica(
- UUID(job.object_id), job.file_path, job.replica_storage
- )
- if not res:
- logger.debug(
- "file_deleted_before_replication",
- object_id=job.object_id,
- file_path=job.file_path,
- replica_storage=job.replica_storage,
- oid=job.oid,
- unstructured_message="File deleted before replication job started, ignoring",
- )
-
async def handle_delete_replica_job_finished(self, job: DeleteReplicaJob) -> None:
"""After successful replica deletion, delete cronjob if scheduled"""
if job.schedule:
await self.crawl_manager.delete_replica_deletion_scheduled_job(job.id)
- async def create_replica_jobs(
- self, oid: UUID, file: BaseFile, object_id: str, object_type: str
- ) -> dict[str, bool | list[str]]:
- """Create k8s background job to replicate a file to all replica storage locations."""
- org = await self.org_ops.get_org_by_id(oid)
-
- primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage)
- primary_endpoint, bucket_suffix = self.strip_bucket(
- primary_storage.endpoint_url
- )
-
- primary_file_path = bucket_suffix + file.filename
-
- ids = []
-
- for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org):
- job_id = await self.create_replica_job(
- org,
- file,
- object_id,
- object_type,
- replica_ref,
- primary_file_path,
- primary_endpoint,
- )
- ids.append(job_id)
-
- return {"added": True, "ids": ids}
-
- async def create_replica_job(
- self,
- org: Organization,
- file: BaseFile,
- object_id: str,
- object_type: str,
- replica_ref: StorageRef,
- primary_file_path: str,
- primary_endpoint: str,
- existing_job_id: str | None = None,
- ) -> str:
- """Create k8s background job to replicate a file to a specific replica storage location."""
- replica_storage = self.storage_ops.get_org_storage_by_ref(org, replica_ref)
- replica_endpoint, bucket_suffix = self.strip_bucket(
- replica_storage.endpoint_url
- )
- replica_file_path = bucket_suffix + file.filename
-
- job_type = BgJobType.CREATE_REPLICA.value
-
- try:
- job_id, _ = await self.crawl_manager.run_replica_job(
- oid=str(org.id),
- job_type=job_type,
- primary_storage=file.storage,
- primary_file_path=primary_file_path,
- primary_endpoint=primary_endpoint,
- replica_storage=replica_ref,
- replica_file_path=replica_file_path,
- replica_endpoint=replica_endpoint,
- delay_days=0,
- existing_job_id=existing_job_id,
- )
- if existing_job_id:
- replication_job = await self.get_background_job(existing_job_id, org.id)
- previous_attempt = {
- "started": replication_job.started,
- "finished": replication_job.finished,
- }
- if replication_job.previousAttempts:
- replication_job.previousAttempts.append(previous_attempt)
- else:
- replication_job.previousAttempts = [previous_attempt]
- replication_job.started = dt_now()
- replication_job.finished = None
- replication_job.success = None
- else:
- replication_job = CreateReplicaJob(
- id=job_id,
- oid=org.id,
- started=dt_now(),
- file_path=file.filename,
- object_type=object_type,
- object_id=object_id,
- primary=file.storage,
- replica_storage=replica_ref,
- )
-
- await self.jobs.find_one_and_update(
- {"_id": job_id}, {"$set": replication_job.to_dict()}, upsert=True
- )
-
- return job_id
- # pylint: disable=broad-exception-caught
- except Exception as exc:
- logger.warning(
- "replica_job_start_failed",
- object_type=object_type,
- oid=org.id,
- file=file,
- exc_info=True,
- unstructured_message=f"warning: replica job could not be started "
- f"for {object_type} {file}: {exc}",
- )
- return ""
-
async def create_delete_replica_jobs(
self, org: Organization, file: BaseFile, object_id: str, object_type: str
) -> dict[str, bool | list[str]]:
"""Create a job to delete each replica for the given file"""
ids = []
- for replica_ref in file.replicas or []:
+ for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org):
job_id = await self.create_delete_replica_job(
org, file, object_id, object_type, replica_ref
)
@@ -263,15 +138,12 @@ async def create_delete_replica_job(
)
replica_file_path = bucket_suffix + file.filename
- job_type = BgJobType.DELETE_REPLICA.value
-
delay_days = int(os.environ.get("REPLICA_DELETION_DELAY_DAYS", 0))
if force_start_immediately:
delay_days = 0
- job_id, schedule = await self.crawl_manager.run_replica_job(
+ job_id, schedule = await self.crawl_manager.run_delete_replica_job(
oid=str(org.id),
- job_type=job_type,
replica_storage=replica_ref,
replica_file_path=replica_file_path,
replica_endpoint=replica_endpoint,
@@ -325,6 +197,95 @@ async def create_delete_replica_job(
)
return ""
+ async def create_copy_bucket_jobs(self):
+ """Create background jobs to copy primary storage to each default replica location
+
+ Note that this replicates default storages only, and not any org-specific
+ custom storage, which is not yet fully supported by Browsertrix. When custom
+ storage support is added, we will need to spin up additional copy jobs.
+
+ Because rclone copy is only additive, these copy bucket jobs will ensure
+ that all files in primary storage also exist in the configured replica
+ locations without being able to delete any files from the replica location.
+ """
+ primary_storage_ref = self.storage_ops.get_default_primary()
+ primary_storage = self.storage_ops.get_default_s3_storage(primary_storage_ref)
+ primary_endpoint, primary_bucket_suffix = self.strip_bucket(
+ primary_storage.endpoint_url
+ )
+
+ for default_replica_ref in self.storage_ops.get_default_replicas():
+ await self.create_copy_bucket_job(
+ primary_storage_ref,
+ primary_endpoint,
+ primary_bucket_suffix,
+ default_replica_ref,
+ )
+
+ async def create_copy_bucket_job(
+ self,
+ primary_storage_ref: StorageRef,
+ primary_endpoint: str,
+ primary_bucket_suffix: str,
+ replica_ref: StorageRef,
+ existing_job_id: str | None = None,
+ ) -> str:
+ """Create background job to copy contents of bucket to replica location"""
+ replica_logger = logger.bind(
+ primary_storage=primary_storage_ref, replica_storage=replica_ref
+ )
+
+ try:
+ replica_storage = self.storage_ops.get_default_s3_storage(replica_ref)
+ replica_endpoint, replica_bucket_suffix = self.strip_bucket(
+ replica_storage.endpoint_url
+ )
+
+ job_id = await self.crawl_manager.run_copy_bucket_job(
+ primary_storage=primary_storage_ref,
+ replica_storage=replica_ref,
+ primary_endpoint=primary_endpoint,
+ primary_bucket_suffix=primary_bucket_suffix,
+ replica_endpoint=replica_endpoint,
+ replica_bucket_suffix=replica_bucket_suffix,
+ existing_job_id=existing_job_id,
+ )
+ if existing_job_id:
+ copy_bucket_job = await self.get_background_job(existing_job_id)
+ previous_attempt = {
+ "started": copy_bucket_job.started,
+ "finished": copy_bucket_job.finished,
+ }
+ if copy_bucket_job.previousAttempts:
+ copy_bucket_job.previousAttempts.append(previous_attempt)
+ else:
+ copy_bucket_job.previousAttempts = [previous_attempt]
+ copy_bucket_job.started = dt_now()
+ copy_bucket_job.finished = None
+ copy_bucket_job.success = None
+ else:
+ copy_bucket_job = CopyBucketJob(
+ id=job_id,
+ started=dt_now(),
+ replica_storage=replica_ref,
+ )
+
+ await self.jobs.find_one_and_update(
+ {"_id": job_id}, {"$set": copy_bucket_job.to_dict()}, upsert=True
+ )
+
+ replica_logger.info("copy_bucket_job_started", job_id=job_id)
+
+ return job_id
+ # pylint: disable=broad-exception-caught
+ except Exception:
+ replica_logger.warning(
+ "copy_bucket_job_start_failed",
+ exc_info=True,
+ existing_job_id=existing_job_id,
+ )
+ return ""
+
async def create_delete_org_job(
self,
org: Organization,
@@ -638,7 +599,12 @@ async def create_postprocess_upload_job(
async def ensure_cron_jobs_exist(self):
"""Ensure periodic background cron jobs exist"""
await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists()
- await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists()
+ await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists(
+ bool(os.environ.get("DISABLE_STUCK_UPLOADS_CRON", False))
+ )
+ await self.crawl_manager.ensure_file_replication_cron_job_exists(
+ bool(self.storage_ops.get_default_replicas())
+ )
async def job_finished(
self,
@@ -654,7 +620,11 @@ async def job_finished(
# For periodic cron jobs, no database record will exist for each
# run before this point, so create it here
- if job_type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.RETRY_STUCK_UPLOADS):
+ if job_type in (
+ BgJobType.CLEANUP_SEED_FILES,
+ BgJobType.RETRY_STUCK_UPLOADS,
+ BgJobType.REPLICATE_FILES_CRON,
+ ):
if not started:
started = finished
if job_type == BgJobType.CLEANUP_SEED_FILES:
@@ -664,6 +634,13 @@ async def job_finished(
finished=finished,
success=success,
)
+ elif job_type == BgJobType.REPLICATE_FILES_CRON:
+ cron_job = ReplicateFilesCronJob(
+ id=f"replicate-cron-{secrets.token_hex(5)}",
+ started=started,
+ finished=finished,
+ success=success,
+ )
else:
cron_job = RetryStuckUploadsJob(
id=f"stuck-uploads-{secrets.token_hex(5)}",
@@ -688,9 +665,6 @@ async def job_finished(
if job.type != job_type:
raise HTTPException(status_code=400, detail="invalid_job_type")
- if success and job_type == BgJobType.CREATE_REPLICA:
- await self.handle_replica_job_succeeded(cast(CreateReplicaJob, job))
-
if job_type == BgJobType.DELETE_REPLICA:
await self.handle_delete_replica_job_finished(cast(DeleteReplicaJob, job))
@@ -740,6 +714,8 @@ async def get_background_job(
| UpdateCollStatsJob
| PostProcessUploadJob
| RetryStuckUploadsJob
+ | ReplicateFilesCronJob
+ | CopyBucketJob
):
"""Get background job"""
query: dict[str, object] = {"_id": job_id}
@@ -782,6 +758,12 @@ def _get_job_by_type_from_data(self, data: dict[str, object]):
if data["type"] == BgJobType.RETRY_STUCK_UPLOADS:
return RetryStuckUploadsJob.from_dict(data)
+ if data["type"] == BgJobType.REPLICATE_FILES_CRON:
+ return ReplicateFilesCronJob.from_dict(data)
+
+ if data["type"] == BgJobType.COPY_BUCKET:
+ return CopyBucketJob.from_dict(data)
+
if data["type"] == BgJobType.DELETE_ORG:
return DeleteOrgJob.from_dict(data)
@@ -895,6 +877,26 @@ async def retry_background_job(self, job_id: str, org: Organization | None = Non
)
return {"success": True}
+ if job.type == BgJobType.COPY_BUCKET:
+ job = cast(CopyBucketJob, job)
+
+ primary_storage_ref = self.storage_ops.get_default_primary()
+ primary_storage = self.storage_ops.get_default_s3_storage(
+ primary_storage_ref
+ )
+ primary_endpoint, primary_bucket_suffix = self.strip_bucket(
+ primary_storage.endpoint_url
+ )
+
+ await self.create_copy_bucket_job(
+ primary_storage_ref,
+ primary_endpoint,
+ primary_bucket_suffix,
+ job.replica_storage,
+ existing_job_id=job_id,
+ )
+ return {"success": True}
+
return {"success": False}
async def retry_org_background_job(
@@ -902,24 +904,17 @@ async def retry_org_background_job(
) -> dict[str, bool | str | None]:
"""Retry background job specific to one org"""
if job.type == BgJobType.CREATE_REPLICA:
- job = cast(CreateReplicaJob, job)
- file = await self.get_replica_job_file(job, org)
- primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage)
- primary_endpoint, bucket_suffix = self.strip_bucket(
- primary_storage.endpoint_url
+ raise HTTPException(
+ status_code=400, detail="create_replica_job_retry_no_longer_supported"
)
- primary_file_path = bucket_suffix + file.filename
- await self.create_replica_job(
- org,
- file,
- job.object_id,
- job.object_type,
- job.replica_storage,
- primary_file_path,
- primary_endpoint,
- existing_job_id=job.id,
+
+ if job.type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.REPLICATE_FILES_CRON):
+ raise HTTPException(status_code=400, detail="cron_job_retry_not_supported")
+
+ if job.type in (BgJobType.COPY_BUCKET, BgJobType.OPTIMIZE_PAGES):
+ raise HTTPException(
+ status_code=400, detail="non_org_specific_job_retry_not_supported"
)
- return {"success": True}
if job.type == BgJobType.DELETE_REPLICA:
job = cast(DeleteReplicaJob, job)
@@ -979,7 +974,7 @@ async def retry_org_background_job(
)
return {"success": True}
- if job.type == BgJobType.CLEANUP_SEED_FILES:
+ if job.type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.REPLICATE_FILES_CRON):
raise HTTPException(status_code=400, detail="cron_job_retry_not_supported")
return {"success": False}
diff --git a/backend/btrixcloud/basecrawls.py b/backend/btrixcloud/basecrawls.py
index f56461786d..fd5e1674a8 100644
--- a/backend/btrixcloud/basecrawls.py
+++ b/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),
)
)
diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py
index e71963d085..40d8b658f0 100644
--- a/backend/btrixcloud/crawlmanager.py
+++ b/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,7 +378,9 @@ 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 * * * *"
@@ -360,11 +388,42 @@ async def ensure_retry_stuck_uploads_cron_job_exists(self):
"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,
diff --git a/backend/btrixcloud/db.py b/backend/btrixcloud/db.py
index e8df8c8b00..32d9eb0e0e 100644
--- a/backend/btrixcloud/db.py
+++ b/backend/btrixcloud/db.py
@@ -39,7 +39,7 @@
) = object
-CURR_DB_VERSION = "0058"
+CURR_DB_VERSION = "0059"
MIN_DB_VERSION = 7.0
diff --git a/backend/btrixcloud/main_bg.py b/backend/btrixcloud/main_bg.py
index 1930e86d09..dd96d3ee9a 100644
--- a/backend/btrixcloud/main_bg.py
+++ b/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(
diff --git a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py b/backend/btrixcloud/migrations/migration_0052_profile_filenames.py
index 04876450e9..e845c0e632 100644
--- a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py
+++ b/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(
diff --git a/backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py b/backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py
new file mode 100644
index 0000000000..d2c0460f6e
--- /dev/null
+++ b/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")
diff --git a/backend/btrixcloud/models.py b/backend/btrixcloud/models.py
index ec5e78cb87..ed6aefd83a 100644
--- a/backend/btrixcloud/models.py
+++ b/backend/btrixcloud/models.py
@@ -863,7 +863,7 @@ def __str__(self):
return self.name
return "cs-" + self.name
- def get_storage_secret_name(self, oid: str) -> str:
+ def get_storage_secret_name(self, oid: str = "") -> str:
"""get k8s secret name for this storage and oid"""
if not self.custom:
return "storage-" + self.name
@@ -896,8 +896,6 @@ class BaseFile(BaseModel):
size: int
storage: StorageRef
- replicas: list[StorageRef] | None = []
-
# ============================================================================
class CrawlFile(BaseFile):
@@ -914,7 +912,6 @@ class CrawlFileOut(BaseModel):
size: int
crawlId: str | None = None
- numReplicas: int = 0
expireAt: str | None = None
fromDependency: bool = False
@@ -3239,7 +3236,6 @@ class WebhookNotification(BaseMongoModel):
class BgJobType(StrEnum):
"""Background Job Types"""
- CREATE_REPLICA = "create-replica"
DELETE_REPLICA = "delete-replica"
DELETE_ORG = "delete-org"
RECALCULATE_ORG_STATS = "recalculate-org-stats"
@@ -3249,6 +3245,12 @@ class BgJobType(StrEnum):
UPDATE_COLL_STATS = "update-coll-stats"
POSTPROCESS_UPLOAD = "postprocess-upload"
RETRY_STUCK_UPLOADS = "retry-stuck-uploads"
+ # Deprecated per-file replication jobs
+ CREATE_REPLICA = "create-replica"
+ # New replication job that spins up rclone copy jobs as needed
+ REPLICATE_FILES_CRON = "replicate-files-cron"
+ # Per-bucket replication job
+ COPY_BUCKET = "copy-bucket"
# ============================================================================
@@ -3334,6 +3336,21 @@ class CleanupSeedFilesJob(BackgroundJob):
type: Literal[BgJobType.CLEANUP_SEED_FILES] = BgJobType.CLEANUP_SEED_FILES
+# ============================================================================
+class ReplicateFilesCronJob(BackgroundJob):
+ """Model for tracking cron jobs to replicate files"""
+
+ type: Literal[BgJobType.REPLICATE_FILES_CRON] = BgJobType.REPLICATE_FILES_CRON
+
+
+# ============================================================================
+class CopyBucketJob(BackgroundJob):
+ """Model for tracking job to copy primary storage bucket to replica storage"""
+
+ type: Literal[BgJobType.COPY_BUCKET] = BgJobType.COPY_BUCKET
+ replica_storage: StorageRef
+
+
# ============================================================================
class UpdateCollStatsJob(BackgroundJob):
"""Model for tracking jobs to readd pages for an org or single crawl"""
@@ -3374,6 +3391,8 @@ class RetryStuckUploadsJob(BackgroundJob):
| UpdateCollStatsJob
| PostProcessUploadJob
| RetryStuckUploadsJob
+ | ReplicateFilesCronJob
+ | CopyBucketJob
]
diff --git a/backend/btrixcloud/operator/crawls.py b/backend/btrixcloud/operator/crawls.py
index 3afc76d4c1..f70cb41c42 100644
--- a/backend/btrixcloud/operator/crawls.py
+++ b/backend/btrixcloud/operator/crawls.py
@@ -2202,7 +2202,6 @@ async def do_crawl_finished_tasks(
await self.coll_ops.add_successful_crawl_to_collections(
crawl.id, crawl.cid, crawl.oid
)
- await self.crawl_ops.replicate_crawl_files(crawl.id, crawl.org, "crawl")
if stats and stats.profile_update and crawl.profileid:
await self.crawl_config_ops.profiles.update_profile_from_crawl_upload(
diff --git a/backend/btrixcloud/profiles.py b/backend/btrixcloud/profiles.py
index 49ce65d4fc..a6ba151dac 100644
--- a/backend/btrixcloud/profiles.py
+++ b/backend/btrixcloud/profiles.py
@@ -30,7 +30,6 @@
ProfilePingResponse,
ProfileSearchValuesResponse,
ProfileUpdate,
- StorageRef,
SuccessResponse,
SuccessResponseStorageQuota,
TagsResponse,
@@ -351,10 +350,6 @@ async def do_commit_to_profile(
{"_id": profile.id}, {"$set": profile.to_dict()}, upsert=True
)
- await self.background_job_ops.create_replica_jobs(
- org.id, profile_file, str(profileid), "profile"
- )
-
await self.orgs.inc_org_bytes_stored(
org.id, file_size - prev_file_size, "profile"
)
@@ -450,14 +445,6 @@ async def update_profile_from_crawl_upload(
if profile_file.hash == hash_:
return True
- profile_file.size = size
- profile_file.hash = hash_
-
- # update replica
- await self.background_job_ops.create_replica_jobs(
- org_id, profile_file, str(profileid), "profile"
- )
-
# update size stats, if changed
if size != prev_file_size:
await self.orgs.inc_org_bytes_stored(
@@ -660,15 +647,6 @@ async def _send_browser_req(
return data or {}
- async def add_profile_file_replica(
- self, profileid: UUID, filename: str, ref: StorageRef
- ) -> dict[str, object]:
- """Add replica StorageRef to existing ProfileFile"""
- return await self.profiles.find_one_and_update(
- {"_id": profileid, "resource.filename": filename},
- {"$push": {"resource.replicas": {"name": ref.name, "custom": ref.custom}}},
- )
-
async def calculate_org_profile_file_storage(self, oid: UUID) -> int:
"""Calculate and return total size of profile files in org"""
total_size = 0
diff --git a/backend/btrixcloud/storages.py b/backend/btrixcloud/storages.py
index fc91e8b782..4a149cb282 100644
--- a/backend/btrixcloud/storages.py
+++ b/backend/btrixcloud/storages.py
@@ -287,6 +287,23 @@ def get_available_storages(self, org: Organization) -> list[StorageRef]:
refs.append(StorageRef(name=name, custom=True))
return refs
+ def get_default_primary(self) -> StorageRef:
+ """return default primary storageref"""
+ if not self.default_primary:
+ raise HTTPException(status_code=400, detail="no_primary_storage")
+ return self.default_primary
+
+ def get_default_replicas(self) -> list[StorageRef]:
+ """return default replica storage locations"""
+ return self.default_replicas
+
+ def get_default_s3_storage(self, ref: StorageRef) -> S3Storage:
+ """return s3storage object for default storage ref (not org specific)"""
+ storage = self.default_storages.get(ref.name)
+ if not storage:
+ raise KeyError(f"Storage ref {ref.name} not found")
+ return storage
+
@asynccontextmanager
async def get_s3_client(
self, storage: S3Storage, for_presign=False
diff --git a/backend/btrixcloud/uploads.py b/backend/btrixcloud/uploads.py
index 1f8d0792dd..4ea5c19c79 100644
--- a/backend/btrixcloud/uploads.py
+++ b/backend/btrixcloud/uploads.py
@@ -280,6 +280,7 @@ async def _create_upload(
upload_logger.debug(
"upload_create", state="completed", quota_reached=quota_reached
)
+
return {"id": crawl_id, "added": True, "storageQuotaReached": quota_reached}
async def post_process_upload(
@@ -393,9 +394,6 @@ async def _check_and_maybe_split(file: CrawlFile) -> None:
{"_id": crawl_id}, {"$set": {"state": "complete"}}
)
- pp_logger.debug("post_process_upload", state="replicate_crawl_files")
- await self.replicate_crawl_files(crawl_id, org, "upload")
-
pp_logger.debug(
"post_process_upload", state="finished_processing_dispatching_webhook"
)
diff --git a/backend/test/conftest.py b/backend/test/conftest.py
index 760544b56e..1c76246eb0 100644
--- a/backend/test/conftest.py
+++ b/backend/test/conftest.py
@@ -871,7 +871,6 @@ def profile_config_id(admin_auth_headers, default_org_id, profile_id):
assert resource["size"]
assert resource["storage"]
assert resource["storage"]["name"]
- assert resource.get("replicas") or resource.get("replicas") == []
# Use profile in a workflow
r = requests.post(
diff --git a/backend/test/test_profiles.py b/backend/test/test_profiles.py
index acdcad153e..dcef5769ce 100644
--- a/backend/test/test_profiles.py
+++ b/backend/test/test_profiles.py
@@ -62,7 +62,6 @@ def test_get_profile(admin_auth_headers, default_org_id, profile_id, profile_con
assert resource["size"]
assert resource["storage"]
assert resource["storage"]["name"]
- assert "replicas" in resource and isinstance(resource["replicas"], list)
assert "crawlconfigs" not in data
assert data["inUse"] == True
@@ -118,7 +117,6 @@ def test_list_profiles(admin_auth_headers, default_org_id, profile_id, profile_2
assert resource["size"]
assert resource["storage"]
assert resource["storage"]["name"]
- assert "replicas" in resource and isinstance(resource["replicas"], list)
# First profile should be listed second by default because it was
# modified less recently
@@ -145,7 +143,6 @@ def test_list_profiles(admin_auth_headers, default_org_id, profile_id, profile_2
assert resource["size"]
assert resource["storage"]
assert resource["storage"]["name"]
- assert "replicas" in resource and isinstance(resource["replicas"], list)
break
except:
diff --git a/backend/test/test_unit_background_jobs.py b/backend/test/test_unit_background_jobs.py
index 9728c317f8..8504648d19 100644
--- a/backend/test/test_unit_background_jobs.py
+++ b/backend/test/test_unit_background_jobs.py
@@ -12,6 +12,7 @@
from btrixcloud.models import (
BgJobType,
CleanupSeedFilesJob,
+ CopyBucketJob,
CreateReplicaJob,
DeleteOrgJob,
DeleteReplicaJob,
@@ -19,6 +20,7 @@
PostProcessUploadJob,
ReAddOrgPagesJob,
RecalculateOrgStatsJob,
+ ReplicateFilesCronJob,
RetryStuckUploadsJob,
UpdateCollStatsJob,
)
@@ -34,6 +36,8 @@
BgJobType.UPDATE_COLL_STATS: UpdateCollStatsJob,
BgJobType.POSTPROCESS_UPLOAD: PostProcessUploadJob,
BgJobType.RETRY_STUCK_UPLOADS: RetryStuckUploadsJob,
+ BgJobType.REPLICATE_FILES_CRON: ReplicateFilesCronJob,
+ BgJobType.COPY_BUCKET: CopyBucketJob,
}
_TYPE_EXTRAS = {
@@ -49,6 +53,9 @@
"object_id": "test-object",
"replica_storage": {"name": "test-storage"},
},
+ BgJobType.COPY_BUCKET: {
+ "replica_storage": {"name": "test-storage"},
+ },
}
diff --git a/backend/test_nightly/conftest.py b/backend/test_nightly/conftest.py
index 7234e18d1e..772ec36218 100644
--- a/backend/test_nightly/conftest.py
+++ b/backend/test_nightly/conftest.py
@@ -376,16 +376,8 @@ def deleted_crawl_id(admin_auth_headers, default_org_id):
break
time.sleep(5)
- # Wait until replica background job completes
- while True:
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=create-replica&success=True",
- headers=admin_auth_headers,
- )
- assert r.status_code == 200
- if r.json()["total"] == 1:
- break
- time.sleep(5)
+ # Wait for it to replicate
+ time.sleep(300)
# Delete crawl
r = requests.post(
@@ -395,15 +387,4 @@ def deleted_crawl_id(admin_auth_headers, default_org_id):
)
assert r.status_code == 200
- # Wait until delete replica background job completes
- while True:
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=delete-replica&success=True",
- headers=admin_auth_headers,
- )
- assert r.status_code == 200
- if r.json()["total"] == 1:
- break
- time.sleep(5)
-
return crawl_id
diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py
index ce1e430e10..5651c05ce1 100644
--- a/backend/test_nightly/test_crawl_timeout.py
+++ b/backend/test_nightly/test_crawl_timeout.py
@@ -1,7 +1,10 @@
import time
+import pytest
import requests
+from btrixcloud.utils import dt_now
+
from .conftest import API_PREFIX
from .utils import verify_file_replicated
@@ -32,52 +35,48 @@ def test_crawl_timeout(admin_auth_headers, default_org_id, timeout_crawl):
attempts += 1
+@pytest.mark.timeout(1800)
def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_crawl):
- time.sleep(20)
+ crawl_complete = dt_now()
- # Verify replication job was successful
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=create-replica",
- headers=admin_auth_headers,
- )
- assert r.status_code == 200
- latest_job = r.json()["items"][0]
- assert latest_job["type"] == "create-replica"
- job_id = latest_job["id"]
+ # Verify copy bucket job has run and succeeded since crawl completed
+ job_id = None
+ # Give copy bucket job (which is kicked off by cron replication job)
+ # up to 20 minutes to start and then complete
attempts = 0
- while attempts < 5:
+ while attempts < 20:
r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}",
+ f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket",
headers=admin_auth_headers,
)
assert r.status_code == 200
- job = r.json()
- finished = latest_job.get("finished")
- if not finished:
- attempts += 1
- time.sleep(10)
- continue
+ jobs = r.json().get("items", [])
+ for job in jobs:
+ assert job["type"] == "copy-bucket"
+ if job.get("started") >= crawl_complete and job.get("finished"):
+ job_id = job["id"]
+ break
+
+ attempts += 1
+ time.sleep(60)
- assert job["success"]
- break
+ assert job_id
- # Assert file was updated
+ # Verify crawlfiles are stored in replica location
r = requests.get(
f"{API_PREFIX}/orgs/{default_org_id}/crawls/{timeout_crawl}/replay.json",
headers=admin_auth_headers,
)
assert r.status_code == 200
data = r.json()
+
+ oid = data["oid"]
+ assert oid
+
files = data.get("resources")
assert files
for file_ in files:
- assert file_["numReplicas"] == 1
-
- # Verify replica is stored
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers
- )
- assert r.status_code == 200
- data = r.json()
- verify_file_replicated(data["file_path"])
+ filename = file_["name"]
+ file_path = f"{oid}/{filename}"
+ verify_file_replicated(file_path)
diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py
index 3f9c1337b5..a5a08711d7 100644
--- a/backend/test_nightly/test_upload_replicas.py
+++ b/backend/test_nightly/test_upload_replicas.py
@@ -2,8 +2,10 @@
import time
import structlog
+import pytest
import requests
+from btrixcloud.utils import dt_now
from test.utils import read_in_chunks
from .conftest import API_PREFIX
@@ -16,6 +18,8 @@
curr_dir = os.path.dirname(os.path.realpath(__file__))
+upload_file_path = None
+
def test_upload_stream(admin_auth_headers, default_org_id):
with open(os.path.join(curr_dir, "..", "test", "data", "example.wacz"), "rb") as fh:
@@ -32,62 +36,55 @@ def test_upload_stream(admin_auth_headers, default_org_id):
upload_id = r.json()["id"]
+@pytest.mark.timeout(1800)
def test_upload_file_replicated(admin_auth_headers, default_org_id):
- time.sleep(20)
+ upload_complete = dt_now()
- # Verify replication job was successful
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=create-replica",
- headers=admin_auth_headers,
- )
- assert r.status_code == 200
- latest_job = r.json()["items"][0]
- assert latest_job["type"] == "create-replica"
- job_id = latest_job["id"]
+ # Verify copy bucket job has run and succeeded since crawl completed
+ job_id = None
+ # Give copy bucket job (which is kicked off by cron replication job)
+ # up to 20 minutes to start and then complete
attempts = 0
- while attempts < 5:
+ while attempts < 20:
r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}",
+ f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket",
headers=admin_auth_headers,
)
assert r.status_code == 200
- job = r.json()
- finished = latest_job.get("finished")
- if not finished:
- attempts += 1
- time.sleep(10)
- continue
+ jobs = r.json().get("items", [])
+ for job in jobs:
+ assert job["type"] == "copy-bucket"
+ if job.get("started") >= upload_complete and job.get("finished"):
+ job_id = job["id"]
+ break
- assert job["success"]
- break
+ attempts += 1
+ time.sleep(60)
+
+ assert job_id
- # Verify file updated
+ # Verify upload file is stored
r = requests.get(
f"{API_PREFIX}/orgs/{default_org_id}/uploads/{upload_id}/replay.json",
headers=admin_auth_headers,
)
assert r.status_code == 200
data = r.json()
+
files = data.get("resources")
assert files
- for file_ in files:
- assert file_["numReplicas"] == 1
- # Verify replica is stored
- r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers
- )
- assert r.status_code == 200
- job = r.json()
- logger.info(
- "upload_file_path",
- file_path=job["file_path"],
- unstructured_message=f"{job['file_path']}",
- )
- verify_file_replicated(job["file_path"])
+ file_ = files[0]
+ filename = file_["name"]
+
+ global upload_file_path
+ upload_file_path = f"{default_org_id}/{filename}"
+ verify_file_replicated(upload_file_path)
+
+@pytest.mark.timeout(1800)
def test_delete_upload_and_replicas(admin_auth_headers, default_org_id):
r = requests.post(
f"{API_PREFIX}/orgs/{default_org_id}/uploads/delete",
@@ -145,4 +142,8 @@ def test_delete_upload_and_replicas(admin_auth_headers, default_org_id):
)
assert r.status_code == 200
job = r.json()
- verify_file_and_replica_deleted(job["file_path"])
+
+ job_file_path = job["file_path"]
+ assert job_file_path == upload_file_path
+
+ verify_file_and_replica_deleted(job_file_path)
diff --git a/backend/test_nightly/test_z_background_jobs.py b/backend/test_nightly/test_z_background_jobs.py
index 28cd4d4b43..81b4994730 100644
--- a/backend/test_nightly/test_z_background_jobs.py
+++ b/backend/test_nightly/test_z_background_jobs.py
@@ -8,6 +8,7 @@
job_id = None
+@pytest.mark.timeout(1800)
def test_background_jobs_list(admin_auth_headers, default_org_id, deleted_crawl_id):
r = requests.get(
f"{API_PREFIX}/orgs/{default_org_id}/jobs/", headers=admin_auth_headers
@@ -33,12 +34,11 @@ def test_background_jobs_list(admin_auth_headers, default_org_id, deleted_crawl_
assert job_id
-@pytest.mark.parametrize("job_type", [("create-replica"), ("delete-replica")])
def test_background_jobs_list_filter_by_type(
- admin_auth_headers, default_org_id, deleted_crawl_id, job_type
+ admin_auth_headers, default_org_id, deleted_crawl_id
):
r = requests.get(
- f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType={job_type}",
+ f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=delete-replica",
headers=admin_auth_headers,
)
assert r.status_code == 200
@@ -49,7 +49,7 @@ def test_background_jobs_list_filter_by_type(
assert len(items) == data["total"]
for item in items:
- assert item["type"] == job_type
+ assert item["type"] == "delete-replica"
def test_background_jobs_list_filter_by_success(
@@ -91,7 +91,12 @@ def test_get_background_job(admin_auth_headers, default_org_id, deleted_crawl_id
data = r.json()
assert data["id"]
- assert data["type"] in ("create-replica", "delete-replica")
+ assert data["type"] in (
+ "replicate-files-cron",
+ "copy-bucket",
+ "delete-replica",
+ "cleanup-seed-files",
+ )
assert data["oid"] == default_org_id
assert data["success"]
assert data["started"]
@@ -99,7 +104,8 @@ def test_get_background_job(admin_auth_headers, default_org_id, deleted_crawl_id
assert data["file_path"]
assert data["object_type"]
assert data["object_id"]
- assert data["replica_storage"]
+ if data["type"] in ("delete-replica", "copy-bucket"):
+ assert data["replica_storage"]
def test_retry_all_failed_bg_jobs_not_superuser(crawler_auth_headers, deleted_crawl_id):
diff --git a/chart/app-templates/background_cron_job.yaml b/chart/app-templates/background_cron_job.yaml
index e4ae98cc6c..dbe79b7181 100644
--- a/chart/app-templates/background_cron_job.yaml
+++ b/chart/app-templates/background_cron_job.yaml
@@ -8,7 +8,7 @@ metadata:
spec:
concurrencyPolicy: Forbid
- successfulJobsHistoryLimit: 0
+ successfulJobsHistoryLimit: 1
failedJobsHistoryLimit: 2
schedule: "{{ schedule }}"
diff --git a/chart/app-templates/replica_job.yaml b/chart/app-templates/copy_bucket_job.yaml
similarity index 74%
rename from chart/app-templates/replica_job.yaml
rename to chart/app-templates/copy_bucket_job.yaml
index f902ed873c..a9025a6ec3 100644
--- a/chart/app-templates/replica_job.yaml
+++ b/chart/app-templates/copy_bucket_job.yaml
@@ -6,29 +6,19 @@ metadata:
- metacontroller.io/decoratorcontroller-background-job-operator
labels:
role: "background-job"
- job_type: {{ job_type }}
- btrix.org: {{ oid }}
+ job_type: "{{ job_type }}"
spec:
- ttlSecondsAfterFinished: 0
- backoffLimit: 3
+ ttlSecondsAfterFinished: 300
+ backoffLimit: 6
template:
spec:
restartPolicy: Never
priorityClassName: bg-job
- podFailurePolicy:
- rules:
- - action: FailJob
- onExitCodes:
- containerName: rclone
- operator: NotIn
- values: [0]
containers:
- name: rclone
image: rclone/rclone:latest
env:
-
-{% if job_type == BgJobType.CREATE_REPLICA %}
- name: RCLONE_CONFIG_PRIMARY_TYPE
value: "s3"
@@ -59,8 +49,6 @@ spec:
- name: RCLONE_CONFIG_PRIMARY_ENDPOINT
value: "{{ primary_endpoint }}"
-{% endif %}
-
- name: RCLONE_CONFIG_REPLICA_TYPE
value: "s3"
@@ -91,15 +79,12 @@ spec:
- name: RCLONE_CONFIG_REPLICA_ENDPOINT
value: "{{ replica_endpoint }}"
-{% if job_type == BgJobType.CREATE_REPLICA %}
- command: ["rclone", "-vv", "copyto", "--checksum", "--error-on-no-transfer", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"]
-{% elif job_type == BgJobType.DELETE_REPLICA %}
- command: ["rclone", "-vv", "delete", "replica:{{ replica_file_path }}"]
-{% endif %}
+ command: ["rclone", "-vv", "copy", "--checksum", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"]
+
resources:
limits:
- memory: "200Mi"
+ memory: "1200Mi"
requests:
- memory: "200Mi"
- cpu: "50m"
+ memory: "500Mi"
+ cpu: "200m"
diff --git a/chart/app-templates/delete_replica_job.yaml b/chart/app-templates/delete_replica_job.yaml
new file mode 100644
index 0000000000..05259a4d90
--- /dev/null
+++ b/chart/app-templates/delete_replica_job.yaml
@@ -0,0 +1,61 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+ name: "{{ id }}"
+ finalizers:
+ - metacontroller.io/decoratorcontroller-background-job-operator
+ labels:
+ role: "background-job"
+ job_type: "{{ job_type }}"
+ btrix.org: "{{ oid }}"
+
+spec:
+ ttlSecondsAfterFinished: 0
+ backoffLimit: 6
+ template:
+ spec:
+ restartPolicy: Never
+ priorityClassName: bg-job
+ containers:
+ - name: rclone
+ image: rclone/rclone:latest
+ env:
+ - name: RCLONE_CONFIG_REPLICA_TYPE
+ value: "s3"
+
+ - name: RCLONE_CONFIG_REPLICA_ACCESS_KEY_ID
+ valueFrom:
+ secretKeyRef:
+ name: "{{ replica_secret_name }}"
+ key: STORE_ACCESS_KEY
+
+ - name: RCLONE_CONFIG_REPLICA_SECRET_ACCESS_KEY
+ valueFrom:
+ secretKeyRef:
+ name: "{{ replica_secret_name }}"
+ key: STORE_SECRET_KEY
+
+ - name: RCLONE_CONFIG_REPLICA_REGION
+ valueFrom:
+ secretKeyRef:
+ name: "{{ replica_secret_name }}"
+ key: STORE_REGION
+
+ - name: RCLONE_CONFIG_REPLICA_PROVIDER
+ valueFrom:
+ secretKeyRef:
+ name: "{{ replica_secret_name }}"
+ key: STORE_S3_PROVIDER
+
+ - name: RCLONE_CONFIG_REPLICA_ENDPOINT
+ value: "{{ replica_endpoint }}"
+
+ command: ["rclone", "-vv", "delete", "replica:{{ replica_file_path }}"]
+
+ resources:
+ limits:
+ memory: "200Mi"
+
+ requests:
+ memory: "200Mi"
+ cpu: "50m"
diff --git a/chart/templates/configmap.yaml b/chart/templates/configmap.yaml
index ebe29cbbab..5280b73482 100644
--- a/chart/templates/configmap.yaml
+++ b/chart/templates/configmap.yaml
@@ -121,10 +121,14 @@ data:
CLEANUP_JOB_CRON_SCHEDULE: "{{ .Values.cleanup_job_cron_schedule }}"
+ REPLICATION_JOB_CRON_SCHEDULE: "{{ .Values.replication_job_cron_schedule }}"
+
CLEANUP_FILES_AFTER_MINUTES: "{{ .Values.cleanup_files_after_minutes | default 1440 }}"
RETRY_STUCK_UPLOADS_CRON_SCHEDULE: "{{ .Values.stuck_uploads_cron_schedule }}"
+ DISABLE_STUCK_UPLOADS_CRON: "{{ .Values.disable_stuck_uploads_cron }}"
+
DEDUPE_IMPORTER_CHANNEL: "{{ .Values.dedupe.importer_channel }}"
ENABLE_AUTO_RESIZE_INDEX_STORAGE: "{{ .Values.dedupe.enable_auto_resize }}"
diff --git a/chart/test/test-nightly-addons.yaml b/chart/test/test-nightly-addons.yaml
index 028c872137..7fc4380744 100644
--- a/chart/test/test-nightly-addons.yaml
+++ b/chart/test/test-nightly-addons.yaml
@@ -10,6 +10,12 @@ cleanup_job_cron_schedule: "* * * * *"
# Clean up files > 1 minute old in testing
cleanup_files_after_minutes: 1
+# Every 2 minutes, for use in testing file replication
+replication_job_cron_schedule: "*/2 * * * *"
+
+# Disable retry uploads cron job to ease resource usage on CI workers
+disable_stuck_uploads_cron: true
+
# max crawl queue size x concurrent crawls
crawl_queue_limit_scale: 1
diff --git a/chart/values.yaml b/chart/values.yaml
index 5abd71e493..e1ec5c0fc5 100644
--- a/chart/values.yaml
+++ b/chart/values.yaml
@@ -192,6 +192,14 @@ cleanup_files_after_minutes: 1440
# without a running background job and retrying them
stuck_uploads_cron_schedule: "0 * * * *"
+# Option to disable stuck uploads cron job
+# Generally should only be set true for testing in CI
+disable_stuck_uploads_cron: false
+
+# Cron schedule for periodically syncing replica storage from primary
+# Defaults to every two hours
+replication_job_cron_schedule: "0 */2 * * *"
+
# Emails Image
# =========================================
emails_image: "docker.io/webrecorder/browsertrix-emails:1.25.1"
diff --git a/frontend/src/pages/org/browser-profiles/profile.ts b/frontend/src/pages/org/browser-profiles/profile.ts
index 0053837677..0a815531a6 100644
--- a/frontend/src/pages/org/browser-profiles/profile.ts
+++ b/frontend/src/pages/org/browser-profiles/profile.ts
@@ -24,7 +24,6 @@ import {
import type { ProfileUpdatedEvent } from "@/features/browser-profiles/types";
import type { WorkflowColumnName } from "@/features/crawl-workflows/workflow-list";
import { emptyMessage } from "@/layouts/emptyMessage";
-import { labelWithIcon } from "@/layouts/labelWithIcon";
import { pageHeader, pageNav } from "@/layouts/pageHeader";
import { panel, panelBody } from "@/layouts/panel";
import { OrgTab, WorkflowTab } from "@/routes";
@@ -548,19 +547,6 @@ export class BrowserProfilesProfilePage extends BtrixElement {
: noData,
)}
-
- ${this.renderDetail((profile) => {
- const isBackedUp =
- profile.resource?.replicas &&
- profile.resource.replicas.length > 0;
- return labelWithIcon({
- label: isBackedUp ? msg("Backed Up") : msg("Not Backed Up"),
- icon: html``,
- });
- })}
-
`,
});
diff --git a/frontend/src/types/collection.ts b/frontend/src/types/collection.ts
index 692b83aea6..1763917b61 100644
--- a/frontend/src/types/collection.ts
+++ b/frontend/src/types/collection.ts
@@ -40,7 +40,6 @@ export const publicCollectionSchema = z.object({
hash: z.string(),
size: z.number(),
crawlId: z.string().nullable(),
- numReplicas: z.number(),
expireAt: z.string().datetime().nullable(),
fromDependency: z.boolean(),
}),
diff --git a/frontend/src/types/crawler.ts b/frontend/src/types/crawler.ts
index 4d870d41c7..cbb123bf30 100644
--- a/frontend/src/types/crawler.ts
+++ b/frontend/src/types/crawler.ts
@@ -131,11 +131,6 @@ export type ListWorkflow = Omit & {
config: Workflow["config"] | null;
};
-export type ProfileReplica = {
- name: string;
- custom?: boolean;
-};
-
export type Profile = {
id: string;
name: string;
@@ -160,7 +155,6 @@ export type Profile = {
path: string;
hash: string;
size: number;
- replicas: ProfileReplica[] | null;
};
crawlerChannel?: CrawlerChannelImage | AnyString;
proxyId?: string;