Merge pull request #495 from latchbio/rahuldesai1/nf-latch-log-dir

add nf log dir

CHANGELOG.md

@@ -16,6 +16,13 @@ Types of changes
 
 # Latch SDK Changelog
 
+## v2.53.3 - 2024-10-02
+
+### Added
+
+* Nextflow
+  - Add support for uploading command logs after task execution
+
 ## v2.53.2 - 2024-10-02
 
 ### Changed

latch/types/metadata.py

@@ -848,6 +848,10 @@ class NextflowMetadata(LatchMetadata):
     """
     Directory to dump Nextflow logs
     """
+    upload_command_logs: bool = False
+    """
+    Upload .command.* logs to Latch Data after each task execution
+    """
     about_page_path: Optional[Path] = None
     """
     Path to a markdown file containing information about the pipeline - rendered in the About page.

latch_cli/nextflow/workflow.py

@@ -78,6 +78,14 @@ def initialize() -> str:
 def nextflow_runtime(pvc_name: str, {param_signature}) -> None:
     shared_dir = Path("/nf-workdir")
 
+    exec_name = _get_execution_name()
+    if exec_name is None:
+        print("Failed to get execution name.")
+        exec_name = "unknown"
+
+    latch_log_dir = urljoins("{log_dir}", exec_name)
+    print(f"Log directory: {{latch_log_dir}}")
+
 {output_shortcuts}
 {samplesheet_constructors}
 
@@ -139,6 +147,10 @@ def nextflow_runtime(pvc_name: str, {param_signature}) -> None:
             "NXF_DISABLE_CHECK_LATEST": "true",
             "NXF_ENABLE_VIRTUAL_THREADS": "false",
         }}
+
+        if {upload_command_logs}:
+            env["LATCH_LOG_DIR"] = latch_log_dir
+
         subprocess.run(
             cmd,
             env=env,
@@ -152,13 +164,9 @@ def nextflow_runtime(pvc_name: str, {param_signature}) -> None:
 
         nextflow_log = shared_dir / ".nextflow.log"
         if nextflow_log.exists():
-            name = _get_execution_name()
-            if name is None:
-                print("Skipping logs upload, failed to get execution name")
-            else:
-                remote = LPath(urljoins("{log_dir}", name, "nextflow.log"))
-                print(f"Uploading .nextflow.log to {{remote.path}}")
-                remote.upload_from(nextflow_log)
+            remote = LPath(urljoins(latch_log_dir, "nextflow.log"))
+            print(f"Uploading .nextflow.log to {{remote.path}}")
+            remote.upload_from(nextflow_log)
 
         print("Computing size of workdir... ", end="")
         try:
@@ -442,6 +450,7 @@ def generate_nextflow_workflow(
         storage_gib=resources.storage_gib,
         storage_expiration_hours=resources.storage_expiration_hours,
         log_dir=log_dir,
+        upload_command_logs=metadata._nextflow_metadata.upload_command_logs,
         nextflow_version=get_nextflow_major_version(pkg_root),
     )
 

setup.py

@@ -13,7 +13,7 @@
 
 setup(
     name="latch",
-    version="v2.53.2",
+    version="v2.53.3",
     author_email="[email protected]",
     description="The Latch SDK",
     packages=find_packages(),