Sync metadata

Author: ssjunnebo

dataflow_transfer/dataflow_transfer.py

@@ -34,6 +34,11 @@ def process_run(run_dir, sequencer, config):
         run.start_transfer(final=False)
         return
 
+    ## Sequencing finished. Copy metadata in the background if not already done.
+    if run.has_status("sequencing_finished"):
+        if not run.metadata_synced:
+            run.sync_metadata()
+
     ## Sequencing finished but transfer not complete. Start final transfer.
     if not run.final_sync_successful:
         if run.has_status("sequencing_finished"):

dataflow_transfer/run_classes/generic_runs.py

@@ -21,6 +21,14 @@ def __init__(self, run_dir, configuration):
         )
         self.final_file = ""
         self.transfer_details = self.configuration.get("transfer_details", {})
+        self.metadata_rsync_exitcode_file = os.path.join(
+            self.run_dir, ".metadata_rsync_exitcode"
+        )
+        self.metadata_destination = os.path.join(
+            self.configuration.get("metadata_archive"),
+            getattr(self, "run_type", None),
+            self.run_id,
+        )
         self.final_rsync_exitcode_file = os.path.join(
             self.run_dir, ".final_rsync_exitcode"
         )
@@ -42,36 +50,80 @@ def sequencing_ongoing(self):
             return False
         return True
 
-    def generate_rsync_command(self, is_final_sync=False):
-        """Generate an rsync command string."""
-        destination = (
-            self.transfer_details.get("user")
-            + "@"
-            + self.transfer_details.get("host")
-            + ":"
-            + self.miarka_destination
+    @property
+    def metadata_synced(self):
+        """Check if the metadata rsync was successful by reading the exit code file."""
+        return fs.check_exit_status(self.metadata_rsync_exitcode_file)
+
+    def sync_metadata(self):
+        """Start background rsync transfer for metadata files."""
+        # make metadata destination path if it doesn't exist
+        if not os.path.exists(self.metadata_destination):
+            os.makedirs(self.metadata_destination)
+        metadata_rsync_command = self.generate_rsync_command(
+            remote=False, with_exit_code_file=True
         )
+
+        if fs.rsync_is_running(src=self.run_dir, dst=self.metadata_destination):
+            logger.info(
+                f"Metadata rsync is already running for {self.run_dir} to destination {self.metadata_destination}. Skipping background metadata sync initiation."
+            )
+            return
+        try:
+            fs.submit_background_process(metadata_rsync_command)
+            logger.info(
+                f"{self.run_id}: Started metadata rsync to {self.metadata_destination}"
+                + f" with the following command: '{metadata_rsync_command}'"
+            )
+        except Exception as e:
+            logger.error(f"Failed to start metadata rsync for {self.run_id}: {e}")
+            raise e
+
+    def generate_rsync_command(self, remote=False, with_exit_code_file=False):
+        """Generate an rsync command string."""
+        if remote:
+            destination = (
+                self.transfer_details.get("user")
+                + "@"
+                + self.transfer_details.get("host")
+                + ":"
+                + self.miarka_destination
+            )
+            log_file_option = "--log-file=" + os.path.join(
+                self.run_dir, "rsync_remote_log.txt"
+            )
+            rsync_options = self.sequencer_config.get("remote_rsync_options", [])
+            exit_code_file = self.final_rsync_exitcode_file
+        else:
+            destination = self.metadata_destination
+            log_file_option = "--log-file=" + os.path.join(
+                self.run_dir, "rsync_metadata_log.txt"
+            )
+            rsync_options = self.sequencer_config.get("metadata_rsync_options", [])
+            exit_code_file = self.metadata_rsync_exitcode_file
         run_one_bin = self.configuration.get("run_one_path", "run-one")
         command = [
             run_one_bin,
             "rsync",
             "-au",
-            "--log-file=" + os.path.join(self.run_dir, "rsync_remote_log.txt"),
-            *(self.sequencer_config.get("rsync_options", [])),
+            log_file_option,
+            *(rsync_options),
             self.run_dir,
             destination,
         ]
         command_str = " ".join(command)
-        if is_final_sync:
-            command_str += f"; echo $? > {self.final_rsync_exitcode_file}"
+        if with_exit_code_file:
+            command_str += f"; echo $? > {exit_code_file}"
         return command_str
 
     def start_transfer(self, final=False):
         """Start background rsync transfer to storage."""
-        transfer_command = self.generate_rsync_command(is_final_sync=final)
-        if fs.rsync_is_running(src=self.run_dir):
+        transfer_command = self.generate_rsync_command(
+            remote=True, with_exit_code_file=final
+        )
+        if fs.rsync_is_running(src=self.run_dir, dst=self.miarka_destination):
             logger.info(
-                f"Rsync is already running for {self.run_dir}. Skipping background transfer initiation."
+                f"Rsync is already running for {self.run_dir} to destination {self.miarka_destination}. Skipping background transfer initiation."
             )
             return
         try:

dataflow_transfer/utils/filesystem.py

@@ -28,9 +28,9 @@ def find_runs(base_dir, ignore_folders=[]):
     return runs
 
 
-def rsync_is_running(src):
-    """Check if rsync is already running for given src."""
-    pattern = f"rsync.*{src}"
+def rsync_is_running(src, dst):
+    """Check if rsync is already running for given src and destination."""
+    pattern = f"rsync.*{src}.*{dst}"
     try:
         subprocess.check_output(["pgrep", "-f", pattern])
         return True