Fix profile filenames in database and re-replicate (#2892)

Fixes #2891 

- Fix profile filenames in database to be full path in bucket (including
org id prefix) for new profiles
- Add migration to fix profile filenames that don't already have org id
prefix, and for each found, delete existing replicas from database
(files were never pushed to s3) and then run background jobs to
replicate
- Add `--error-on-no-transfer` flag to rclone replication job to fail
with non-0 exit code if source file isn't found or another issue
prevents copy from being successful

---------

Co-authored-by: Ilya Kreymer <ikreymer@gmail.com>

Author: tw4l

backend/btrixcloud/db.py

@@ -34,7 +34,7 @@
     ) = PageOps = BackgroundJobOps = FileUploadOps = CrawlLogOps = object
 
 
-CURR_DB_VERSION = "0051"
+CURR_DB_VERSION = "0052"
 
 
 # ============================================================================

backend/btrixcloud/migrations/migration_0052_profile_filenames.py

@@ -0,0 +1,76 @@
+"""
+Migration 0052 - Fix profile filenames in db to be full path from bucket
+"""
+
+from btrixcloud.migrations import BaseMigration
+from btrixcloud.models import Profile
+
+
+MIGRATION_VERSION = "0052"
+
+
+# pylint: disable=duplicate-code
+class Migration(BaseMigration):
+    """Migration class."""
+
+    # pylint: disable=unused-argument
+    def __init__(self, mdb, **kwargs):
+        super().__init__(mdb, migration_version=MIGRATION_VERSION)
+
+        self.background_job_ops = kwargs.get("background_job_ops")
+
+    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"]
+
+        match_query = {"resource.filename": {"$regex": r"^profiles"}}
+
+        if self.background_job_ops is None:
+            print(
+                f"Unable to start migration {MIGRATION_VERSION}, ops class missing",
+                flush=True,
+            )
+            return
+
+        async for profile_res in profiles_mdb.find(match_query):
+            profile = Profile.from_dict(profile_res)
+            if not profile.resource:
+                continue
+
+            existing_filename = profile.resource.filename
+            oid = str(profile.oid)
+            new_filename = f"{oid}/{existing_filename}"
+
+            if not existing_filename.startswith(oid):
+                try:
+                    await profiles_mdb.find_one_and_update(
+                        {"_id": profile.id},
+                        {
+                            "$set": {
+                                "resource.filename": new_filename,
+                                "resource.replicas": [],
+                            }
+                        },
+                    )
+
+                    profile.resource.filename = new_filename
+                    profile.resource.replicas = []
+
+                    print(
+                        f"Starting background jobs to replicate profile {profile.id}",
+                        flush=True,
+                    )
+                    await self.background_job_ops.create_replica_jobs(
+                        profile.oid, profile.resource, str(profile.id), "profile"
+                    )
+                # pylint: disable=broad-exception-caught
+                except Exception as err:
+                    print(
+                        f"Error fixing filename and replicas for profile {profile.id}: {err}",
+                        flush=True,
+                    )

backend/btrixcloud/profiles.py

@@ -247,13 +247,14 @@ async def do_commit_to_profile(
                 created_by = user.id
                 created_by_name = user.name if user.name else user.email
 
-            filename_data = {"filename": f"profiles/profile-{profileid}.tar.gz"}
+            relative_filename = f"profiles/profile-{profileid}.tar.gz"
+            full_filename = f"{str(org.id)}/{relative_filename}"
 
             json = await self._send_browser_req(
                 browser_commit.browserid,
                 "/createProfileJS",
                 "POST",
-                json=filename_data,
+                json={"filename": relative_filename},
                 committing="committing",
             )
             resource = json["resource"]
@@ -264,7 +265,7 @@ async def do_commit_to_profile(
             profile_file = ProfileFile(
                 hash=resource["hash"],
                 size=file_size,
-                filename=resource["path"],
+                filename=full_filename,
                 storage=org.storage,
             )
 

backend/test/test_profiles.py

@@ -49,7 +49,11 @@ def test_get_profile(admin_auth_headers, default_org_id, profile_id, profile_con
 
             resource = data["resource"]
             assert resource
-            assert resource["filename"]
+
+            assert (
+                resource["filename"]
+                == f"{default_org_id}/profiles/profile-{profile_id}.tar.gz"
+            )
             assert resource["hash"]
             assert resource["size"]
             assert resource["storage"]

chart/app-templates/replica_job.yaml

@@ -90,7 +90,7 @@ spec:
           value: "{{ replica_endpoint }}"
 
 {% if job_type == BgJobType.CREATE_REPLICA %}
-        command: ["rclone", "-vv", "copyto", "--checksum", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"]
+        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 %}