Merge branch 'main' of github.com:latchbio/latch into ayush/fix-lps

Author: ayushkamat

CHANGELOG.md

@@ -16,6 +16,25 @@ Types of changes
 
 # Latch SDK Changelog
 
+## 2.58.1 - 2024-04-03
+
+### Fixed
+
+* `latch register --workflow-module` treating `.` separated python paths as file paths
+
+## 2.58.0 - 2024-03-26
+
+### Fixed
+
+* `latch launch` error messages being suppressed
+* `g6e-12xlarge` GPU instance type incorrect GPU annotation (1 -> 4 L40s GPUs)
+
+## 2.57.3 - 2024-03-19
+
+### Fixed
+
+* Use Nextflow base image when generating Dockerfile for a Nextflow workflow.
+
 ## 2.57.2 - 2024-03-07
 
 ### Changed

pyproject.toml

@@ -12,7 +12,7 @@ include = ["src/**/*.py", "src/latch_cli/services/init/*"]
 
 [project]
 name = "latch"
-version = "2.57.2"
+version = "2.58.1"
 description = "The Latch SDK"
 authors = [{ name = "Kenny Workman", email = "[email protected]" }]
 maintainers = [

src/latch/resources/tasks.py

@@ -696,9 +696,9 @@ def _get_l40s_pod(instance_type: str, cpu: int, memory_gib: int, gpus: int) -> P
 
 g6e_12xlarge_task = functools.partial(
     task,
-    task_config=_get_l40s_pod("g6e-12xlarge", cpu=48, memory_gib=384, gpus=1)
+    task_config=_get_l40s_pod("g6e-12xlarge", cpu=48, memory_gib=384, gpus=4)
 )
-"""48 vCPUs, 384 GiB RAM, 1 L40s GPU"""
+"""48 vCPUs, 384 GiB RAM, 4 L40s GPUs"""
 
 g6e_16xlarge_task = functools.partial(
     task,

src/latch_cli/centromere/ctx.py

@@ -95,6 +95,15 @@ def __init__(
         self.disable_auto_version = disable_auto_version
         self.wf_module = wf_module if wf_module is not None else "wf"
 
+        if self.wf_module.startswith("."):
+            click.secho(
+                dedent(f"""\
+                Workflow module `{self.wf_module}` must be absolute (i.e. must not start with `.`)
+                """),
+                fg="red",
+            )
+            raise click.exceptions.Exit(1)
+
         try:
             self.token = retrieve_or_login()
             self.account_id = current_workspace()
@@ -173,17 +182,20 @@ def __init__(
             self.container_map: Dict[str, _Container] = {}
 
             if self.workflow_type == WorkflowType.latchbiosdk:
+                # fixme(ayush): this sucks
+                module_path = pkg_root / Path(self.wf_module.replace(".", "/"))
+
                 try:
-                    flyte_objects = get_flyte_objects(self.pkg_root / self.wf_module)
+                    flyte_objects = get_flyte_objects(module_path)
                 except ModuleNotFoundError as e:
                     click.secho(
                         dedent(
                             f"""
-                            Unable to locate workflow module `{self.wf_module}`. Check that:
+                            Unable to locate workflow module `{self.wf_module}` in `{self.pkg_root.resolve()}`. Check that:
 
-                            1. Package `{self.wf_module}` exists in {pkg_root.resolve()}.
-                            2. Package `{self.wf_module}` is an importable Python path (e.g. `workflows.my_workflow`).
-                            3. Any directories in `{self.wf_module}` contain an `__init__.py` file."""
+                            1. {module_path} exists.
+                            2. Package `{self.wf_module}` is an absolute importable Python path (e.g. `workflows.my_workflow`).
+                            3. All directories in `{module_path}` contain an `__init__.py` file."""
                         ),
                         fg="red",
                     )

src/latch_cli/docker_utils/__init__.py

@@ -9,7 +9,11 @@
 import yaml
 
 from latch_cli.utils import WorkflowType
-from latch_cli.workflow_config import LatchWorkflowConfig, get_or_create_workflow_config
+from latch_cli.workflow_config import (
+    BaseImageOptions,
+    LatchWorkflowConfig,
+    get_or_create_workflow_config,
+)
 
 
 class DockerCmdBlockOrder(str, Enum):
@@ -424,8 +428,12 @@ def generate_dockerignore(
 def get_default_dockerfile(pkg_root: Path, *, wf_type: WorkflowType):
     default_dockerfile = pkg_root / "Dockerfile"
 
-    # shitty: assumption that config is already there so base image will be ignored
-    config = get_or_create_workflow_config(pkg_root=pkg_root)
+    config = get_or_create_workflow_config(
+        pkg_root=pkg_root,
+        base_image_type=BaseImageOptions.nextflow
+        if wf_type == WorkflowType.nextflow
+        else BaseImageOptions.default,
+    )
 
     if not default_dockerfile.exists():
         default_dockerfile = pkg_root / ".latch" / "Dockerfile"

src/latch_cli/main.py

@@ -2,6 +2,7 @@
 
 import os
 import sys
+import traceback
 from pathlib import Path
 from textwrap import dedent
 from typing import Callable, List, Optional, Tuple, TypeVar, Union
@@ -647,12 +648,14 @@ def register(
 def launch(params_file: Path, version: Union[str, None] = None):
     """Launch a workflow using a python parameter map."""
 
-    crash_handler.message = f"Unable to launch workflow"
-    crash_handler.pkg_root = str(Path.cwd())
-
     from latch_cli.services.launch import launch
 
-    wf_name = launch(params_file, version)
+    try:
+        wf_name = launch(params_file, version)
+    except Exception as e:
+        traceback.print_exc()
+        raise click.exceptions.Exit(1) from e
+
     if version is None:
         version = "latest"
 

src/latch_cli/services/launch.py

@@ -192,10 +192,8 @@ def _get_workflow_interface(
 def _launch_workflow(token: str, wf_id: str, params: dict) -> bool:
     """Launch the workflow of given id with parameter map.
 
-    Return True if success
+    Return True if success, raises appropriate exceptions on failure.
     """
-
-    # TODO (kenny) - pull out to consolidated requests class
     # Server sometimes stalls on requests with python user-agent
     headers = {
         "Authorization": f"Bearer {token}",
@@ -213,17 +211,57 @@ def _launch_workflow(token: str, wf_id: str, params: dict) -> bool:
     url = config.api.execution.create
 
     response = requests.post(url, headers=headers, json=_interface_request)
+    response_data = response.json()
+
+    def extract_error_message(data: dict) -> str:
+        if "error" in data:
+            error = data["error"]
+            source = error.get("source", "unknown")
+
+            error_data = error.get("data", {})
+            message = (
+                error_data.get("stderr") or
+                error_data.get("message") or
+                str(error_data)
+            )
+
+            if isinstance(message, str):
+                error_lines = [line for line in message.split("\n") if "Error:" in line]
+                if error_lines:
+                    message = error_lines[-1].replace("Error:", "").strip()
+
+            return f"({source}): {message}"
+        return str(data)
+
+    if response.status_code != 200:
+        print("\nRaw server response:")
+        print(response_data)
 
     if response.status_code == 403:
         raise PermissionError(
             "You need access to the latch sdk beta ~ join the waitlist @"
             " https://latch.bio/sdk"
         )
-    elif response.status_code == 401:
+    if response.status_code == 401:
         raise ValueError(
             "your token has expired - please run latch login to refresh your token and"
             " try again."
         )
-    wf_interface_resp = response.json()
-
-    return wf_interface_resp.get("success") is True
+    if response.status_code == 429:
+        error_msg = extract_error_message(response_data)
+        print(f"\nFormatted error message: {error_msg}")
+        raise RuntimeError(f"Rate limit reached - {error_msg}")
+    if response.status_code == 400:
+        error_msg = extract_error_message(response_data)
+        print(f"\nFormatted error message: {error_msg}")
+        raise ValueError(f"Workflow launch failed - {error_msg}")
+    if response.status_code != 200:
+        error_msg = extract_error_message(response_data)
+        print(f"\nFormatted error message: {error_msg}")
+        raise RuntimeError(f"Server error (HTTP {response.status_code}) - {error_msg}")
+    if "error" in response_data or response_data.get("status") != "Successfully launched workflow":
+        error_msg = extract_error_message(response_data)
+        print(f"\nFormatted error message: {error_msg}")
+        raise RuntimeError(f"Workflow launch failed - {error_msg}")
+
+    return True

src/latch_cli/services/register/utils.py

@@ -110,9 +110,7 @@ def build_image(
 def upload_image(ctx: _CentromereCtx, image_name: str) -> List[str]:
     assert ctx.dkr_client is not None
     return ctx.dkr_client.push(
-        repository=f"{ctx.dkr_repo}/{image_name}",
-        stream=True,
-        decode=True,
+        repository=f"{ctx.dkr_repo}/{image_name}", stream=True, decode=True
     )
 
 
@@ -123,6 +121,8 @@ def serialize_pkg_in_container(
     wf_name_override: Optional[str] = None,
 ) -> Tuple[List[str], str]:
     assert ctx.dkr_client is not None
+    assert ctx.dkr_repo is not None
+    assert ctx.version is not None
 
     _env = {"LATCH_DKR_REPO": ctx.dkr_repo, "LATCH_VERSION": ctx.version}
     if wf_name_override is not None:
@@ -146,12 +146,7 @@ def serialize_pkg_in_container(
         volumes=[serialize_dir],
         environment=_env,
         host_config=ctx.dkr_client.create_host_config(
-            binds={
-                serialize_dir: {
-                    "bind": "/tmp/output",
-                    "mode": "rw",
-                },
-            }
+            binds={serialize_dir: {"bind": "/tmp/output", "mode": "rw"}}  # noqa: S108
         ),
     )
     container_id = typing.cast(str, container.get("Id"))
@@ -191,9 +186,7 @@ def register_serialized_pkg(
             serialize_files[fh.name] = fh
 
         response = requests.post(
-            latch_register_url,
-            headers=headers,
-            files=serialize_files,
+            latch_register_url, headers=headers, files=serialize_files
         )
         response.raise_for_status()
         return response.json()