Add API endpoint for background job progress

Author: tw4l

backend/btrixcloud/background_jobs.py

@@ -30,6 +30,7 @@
     User,
     SuccessResponse,
     SuccessResponseId,
+    JobProgress,
 )
 from .pagination import DEFAULT_PAGE_SIZE, paginated_format
 from .utils import dt_now
@@ -639,6 +640,52 @@ def _get_job_by_type_from_data(self, data: dict[str, object]):
 
         return DeleteOrgJob.from_dict(data)
 
+    async def get_job_progress(self, job_id: str) -> JobProgress:
+        """Return progress of background job for supported types"""
+        job = await self.get_background_job(job_id)
+
+        if job.type != BgJobType.COPY_BUCKET:
+            raise HTTPException(status_code=403, detail="job_type_not_supported")
+
+        if job.success is False:
+            raise HTTPException(status_code=400, detail="job_failed")
+
+        if job.finished:
+            return JobProgress(percentage=1.0)
+
+        log_tail = await self.crawl_manager.tail_background_job(job_id)
+        if not log_tail:
+            raise HTTPException(status_code=400, detail="job_log_not_available")
+
+        lines = log_tail.splitlines()
+        reversed_lines = list(reversed(lines))
+
+        progress = JobProgress(percentage=0.0)
+
+        # Parse lines in reverse order until we find one with latest stats
+        for line in reversed_lines:
+            try:
+                if "ETA" not in line:
+                    continue
+
+                stats_groups = line.split(",")
+                for group in stats_groups:
+                    group = group.strip()
+                    if "%" in group:
+                        progress.percentage = float(group.strip("%")) / 100
+                    if "ETA" in group:
+                        eta_str = group.strip("ETA ")
+                        # Split on white space to remove byte mark rclone sometimes
+                        # adds to end of stats line
+                        eta_list = eta_str.split(" ")
+                        progress.eta = eta_list[0]
+
+                break
+            except:
+                continue
+
+        return progress
+
     async def list_background_jobs(
         self,
         org: Optional[Organization] = None,
@@ -894,6 +941,17 @@ async def get_org_background_job(
         """Retrieve information for background job"""
         return await ops.get_background_job(job_id, org.id)
 
+    @router.get(
+        "/{job_id}/progress",
+        response_model=JobProgress,
+    )
+    async def get_job_progress(
+        job_id: str,
+        org: Organization = Depends(org_crawl_dep),
+    ):
+        """Return progress information for background job"""
+        return await ops.get_job_progress(job_id)
+
     @app.get("/orgs/all/jobs/{job_id}", response_model=AnyJob, tags=["jobs"])
     async def get_background_job_all_orgs(job_id: str, user: User = Depends(user_dep)):
         """Get background job from any org"""

backend/btrixcloud/crawlmanager.py

@@ -415,6 +415,22 @@ async def delete_crawl_config_by_id(self, cid: str) -> None:
         """Delete all crawl configs by id"""
         await self._delete_crawl_configs(f"btrix.crawlconfig={cid}")
 
+    async def tail_background_job(self, job_id: str) -> str:
+        """Tail running background job pod"""
+        pods = await self.core_api.list_namespaced_pod(
+            namespace=self.namespace,
+            label_selector=f"batch.kubernetes.io/job-name={job_id}",
+        )
+
+        if not pods.items:
+            return ""
+
+        pod_name = pods.items[0].metadata.name
+
+        return await self.core_api.read_namespaced_pod_log(
+            pod_name, self.namespace, tail_lines=10
+        )
+
     # ========================================================================
     # Internal Methods
     async def _delete_crawl_configs(self, label) -> None:

backend/btrixcloud/models.py

@@ -2651,6 +2651,14 @@ class CopyBucketJob(BackgroundJob):
 ]
 
 
+# ============================================================================
+class JobProgress(BaseModel):
+    """Model for reporting background job progress"""
+
+    percentage: float
+    eta: Optional[str] = None
+
+
 # ============================================================================
 
 ### GENERIC RESPONSE MODELS ###

chart/app-templates/copy_job.yaml

@@ -3,12 +3,13 @@ kind: Job
 metadata:
   name: "{{ id }}"
   labels:
+    job-id: "{{ id }}"
     role: "background-job"
     job_type: {{ job_type }}
     btrix.org: {{ oid }}
 
 spec:
-  ttlSecondsAfterFinished: 60
+  ttlSecondsAfterFinished: 30
   backoffLimit: 3
   template:
     spec:
@@ -86,7 +87,7 @@ spec:
         - name: RCLONE_CONFIG_NEW_ENDPOINT
           value: "{{ new_endpoint }}"
 
-        command: ["rclone", "-vv", "--progress", "copy", "--checksum", "--use-mmap", "--transfers=2", "--checkers=2", "prev:{{ prev_bucket }}{{ oid }}", "new:{{ new_bucket }}{{ oid }}"]
+        command: ["rclone", "-v", "--stats-one-line", "--stats", "10s", "copy", "--checksum", "--use-mmap", "--transfers=2", "--checkers=2", "prev:{{ prev_bucket }}{{ oid }}", "new:{{ new_bucket }}{{ oid }}"]
         resources:
           limits:
             memory: "350Mi"