Registering an Example Workflow

The goal of this tutorial is to provide an example of the steps required to get an existing Snakemake workflow running on Latch.

To get started, first clone the starter repository:

$ git clone git@github.com:latchbio/snakemake-v2-tutorial.git

and ensure that you have latch installed:

$ pip install latch==2.55.0.a6

1. Updating the Pipeline to use Latch Storage

For now we will not need to make any edits to the Snakefile to make it work with Latch Storage. We will revisit this later in the tutorial however.

2. Adding Resources + Containers to each Rule

The first actual step we need to do is to specify how large each job’s machine needs to be. Since this is a relatively low footprint pipeline, we can make each machine small and provide 1 core and 2 GiB of RAM.

Because every rule has the same resource requirments, we can use a profile to specify them all at once, instead of having to update every rule individually.

Create a directory called profiles/default and in it touch a file called config.yaml:

$ mkdir -p profiles/default
$ touch profiles/default/config.yaml

Then, add the following YAML content to the config.yaml:

default-resources:
  cpu: 1
  mem_mib: 2048

This will set the resources for every rule. Note that you can override these for any rule by updating the resources of that rule directly.

We will skip creating containers for each rule; since all rules use the same conda environment, we can just install that environment in the docker image we make during latch register and have each rule run in that container instead.

3. Writing Metadata

Now, we need to write a metadata file that our workflow will use to generate its parameter interface.

First, make a directory called latch_metadata and in it touch a file called __init__.py:

$ mkdir latch_metadata
$ touch latch_metadata/__init__.py

In latch_metadata/__init__.py, create a SnakemakeV2Metadata object as below:

from latch.types.directory import LatchDir, LatchOutputDir
from latch.types.metadata.latch import LatchAuthor
from latch.types.metadata.snakemake import SnakemakeParameter
from latch.types.metadata.snakemake_v2 import SnakemakeV2Metadata

metadata = SnakemakeV2Metadata(
    display_name="Snakemake Tutorial Workflow",
    author=LatchAuthor(),
    parameters={},
)

This object still doesn’t have any parameter metadata yet, so we need to add it. Looking at config.yaml (not the file in profiles/default), we see that the pipeline expects 3 config parameters: samples_dir, genome_dir, and results_dir. The former two are inputs to the pipeline and the latter is the location where outputs will be stored.

We want all three of these to be exposed in the UI, so we will add them to the parameters dict in latch_metadata/__init__.py:

from latch.types.directory import LatchDir, LatchOutputDir
from latch.types.metadata.latch import LatchAuthor
from latch.types.metadata.snakemake import SnakemakeParameter
from latch.types.metadata.snakemake_v2 import SnakemakeV2Metadata

metadata = SnakemakeV2Metadata(
    display_name="Snakemake Tutorial Workflow",
    author=LatchAuthor(),
    parameters={
        "samples_dir": SnakemakeParameter(
            display_name="Sample Directory",
            type=LatchDir,
        ),
        "genome_dir": SnakemakeParameter(
            display_name="Genome Directory",
            type=LatchDir,
        ),
        "results_dir": SnakemakeParameter(
            display_name="Output Directory",
            type=LatchOutputDir,
        ),
    },
)

In each parameter, we specified (1) a human-readable name to display in the UI, and (2) the type of parameter to accept. Since the workflow expects all of these to be directories, they are all LatchDirs (we made results_dir a LatchOutputDir because it is an output directory).

For now, this is all we need and we can move on, but if you like feel free to customize the metadata object further using the interface described here.

4. Generating the Entrypoint

Now we need to generate the entrypoint file containing the Latch workflow wrapping our Snakemake workflow. This is a simple command:

$ latch snakemake generate-entrypoint .

This should create a directory called wf containing a file called entrypoint.py. The file should have the following contents:

import json
import os
import shutil
import subprocess
import sys
import typing
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

import requests
import typing_extensions
from latch.resources.tasks import custom_task, snakemake_runtime_task
from latch.resources.workflow import workflow
from latch.types.directory import LatchDir, LatchOutputDir
from latch.types.file import LatchFile
from latch_cli.services.register.utils import import_module_by_path
from latch_cli.snakemake.v2.utils import get_config_val

import_module_by_path(Path("latch_metadata/__init__.py"))

import latch.types.metadata.snakemake_v2 as smv2


@custom_task(cpu=0.25, memory=0.5, storage_gib=1)
def initialize() -> str:
  token = os.environ.get("FLYTE_INTERNAL_EXECUTION_ID")
  if token is None:
      raise RuntimeError("failed to get execution token")

  headers = {"Authorization": f"Latch-Execution-Token {token}"}

  print("Provisioning shared storage volume... ", end="")
  resp = requests.post(
      "http://nf-dispatcher-service.flyte.svc.cluster.local/provision-storage-ofs",
      headers=headers,
      json={
          "storage_expiration_hours": 0,
          "version": 2,
          "snakemake": True,
      },
  )
  resp.raise_for_status()
  print("Done.")

  return resp.json()["name"]


@snakemake_runtime_task(cpu=1, memory=2, storage_gib=50)
def snakemake_runtime(
  pvc_name: str,
  samples_dir: LatchDir,
  genome_dir: LatchDir,
  results_dir: LatchOutputDir,
):
  print(f"Using shared filesystem: {pvc_name}")

  shared = Path("/snakemake-workdir")
  snakefile = shared / "Snakefile"

  config = {
      "samples_dir": get_config_val(samples_dir),
      "genome_dir": get_config_val(genome_dir),
      "results_dir": get_config_val(results_dir),
  }

  config_path = (shared / "__latch.config.json").resolve()
  config_path.write_text(json.dumps(config, indent=2))

  ignore_list = [
      "latch",
      ".latch",
      ".git",
      "nextflow",
      ".nextflow",
      ".snakemake",
      "results",
      "miniconda",
      "anaconda3",
      "mambaforge",
  ]

  shutil.copytree(
      Path("/root"),
      shared,
      ignore=lambda src, names: ignore_list,
      ignore_dangling_symlinks=True,
      dirs_exist_ok=True,
  )

  cmd = [
      "snakemake",
      "--snakefile",
      str(snakefile),
      "--configfile",
      str(config_path),
      "--executor",
      "latch",
      "--default-storage-provider",
      "latch",
      "--jobs",
      "1000",
  ]

  print("Launching Snakemake Runtime")
  print(" ".join(cmd), flush=True)

  failed = False
  try:
      subprocess.run(cmd, cwd=shared, check=True)
  except subprocess.CalledProcessError:
      failed = True
  finally:
      if not failed:
          return

      sys.exit(1)


@workflow(smv2._snakemake_v2_metadata)
def snakemake_v2_snakemake_tutorial_workflow(
  samples_dir: LatchDir, genome_dir: LatchDir, results_dir: LatchOutputDir
):
  """
  Sample Description
  """

  snakemake_runtime(
      pvc_name=initialize(),
      samples_dir=samples_dir,
      genome_dir=genome_dir,
      results_dir=results_dir,
  )

5. Generating the Dockerfile

The last step pre-registering is to generate the Dockerfile that will define the environment the runtime executes in. In particular, we want that environment to contain the conda environment defined by environment.yaml.

Again, we can accomplish this with a simple command:

$ latch dockerfile --snakemake -c environment.yaml . -f

This will generate a file called Dockerfile with the following contents:

# Prologue
# DO NOT CHANGE
from 812206152185.dkr.ecr.us-west-2.amazonaws.com/latch-base:fe0b-main

workdir /tmp/docker-build/work/

shell [ \
    "/usr/bin/env", "bash", \
    "-o", "errexit", \
    "-o", "pipefail", \
    "-o", "nounset", \
    "-o", "verbose", \
    "-o", "errtrace", \
    "-O", "inherit_errexit", \
    "-O", "shift_verbose", \
    "-c" \
]
env TZ='Etc/UTC'
env LANG='en_US.UTF-8'

arg DEBIAN_FRONTEND=noninteractive

# Install Mambaforge
run apt-get update --yes && \
    apt-get install --yes curl git && \
    curl \
        --location \
        --fail \
        --remote-name \
        https://github.com/conda-forge/miniforge/releases/latest/download/Miniforge3-Linux-x86_64.sh && \
    `# Docs for -b and -p flags: https://docs.anaconda.com/anaconda/install/silent-mode/#linux-macos` \
    bash Miniforge3-Linux-x86_64.sh -b -p /opt/conda -u && \
    rm Miniforge3-Linux-x86_64.sh

# Set conda PATH
env PATH=/opt/conda/bin:$PATH
run conda config --set auto_activate_base false

# Build conda environment
copy environment.yaml /opt/latch/environment.yaml
run mamba env create \
    --file /opt/latch/environment.yaml \
    --name environment
env PATH=/opt/conda/envs/environment/bin:$PATH

# Copy workflow data (use .dockerignore to skip files)
copy . /root/

# Epilogue

# Latch SDK
# DO NOT REMOVE
run pip install "latch[snakemake]"==2.55.0.a6

# Latch workflow registration metadata
# DO NOT CHANGE
arg tag
# DO NOT CHANGE
env FLYTE_INTERNAL_IMAGE $tag

workdir /root

6. Registering and Running your Pipeline

Finally, we get to upload our pipeline to Latch. Simply run

$ latch register -y .

To run on Latch, you will also need to upload the test data. This is straightforward using latch cp:

$ latch cp data latch:///snakemake-tutorial-data

This will upload the data to a folder called snakemake-tutorial-data in your account on Latch.

Finally, navigate to Workflows and click on “Snakemake Tutorial Workflow”, select the parameters from the data you just uploaded, and run the workflow!

Appendix 1. Getting Sample Names Dynamically

You may have noticed that in the Snakefile, the sample names are hardcoded. This is obviously not desirable - we should be able to infer the sample names based on the contents of the Sample directory.

In order to accomplish this, we will need to edit both the Snakefile, and the entrypoint itself. Since we need to know the contents of the Sample directory outside of a rule, we will need to stage it locally before the pipeline executes.

First, add the following import to the top of the wf/entrypoint.py file:

from latch.ldata.path import LPath

Next, edit the start of snakemake_runtime(...) so that it is the following:

@snakemake_runtime_task(cpu=1, memory=2, storage_gib=50)
def snakemake_runtime(
    pvc_name: str,
    samples_dir: LatchDir,
    genome_dir: LatchDir,
    results_dir: LatchOutputDir,
):
    print(f"Using shared filesystem: {pvc_name}")

    shared = Path("/snakemake-workdir")
    snakefile = shared / "Snakefile"

    # Staging samples_dir
    local_samples_dir = LPath(samples_dir.remote_path).download(shared / "samples")

    config = {
        "samples_dir": get_config_val(local_samples_dir),
        "genome_dir": get_config_val(genome_dir),
        "results_dir": get_config_val(results_dir),
    }

    ...

Here we explicitly download the samples_dir before calling snakemake - this way we will know the contents of the directory without needing to be in a rule.

Lastly, we will need to edit the Snakefile and remove the hardcoded samples:

# Replace SAMPLES = ["A", "B"] with the following:

SAMPLES = []
for sample in samples_dir.iterdir():
    SAMPLES.append(sample.stem)

Now just re-register and see all 3 samples be run through the pipeline.

Overview

Workflows

Python SDK

Nextflow

Snakemake

[ALPHA] Snakemake V2

Automation

Registering an Example Workflow

1. Updating the Pipeline to use Latch Storage

2. Adding Resources + Containers to each Rule

3. Writing Metadata

4. Generating the Entrypoint

5. Generating the Dockerfile

6. Registering and Running your Pipeline

Appendix 1. Getting Sample Names Dynamically

Overview

Workflows

Python SDK

Nextflow

Snakemake

[ALPHA] Snakemake V2

Automation

​1. Updating the Pipeline to use Latch Storage

​2. Adding Resources + Containers to each Rule

​3. Writing Metadata

​4. Generating the Entrypoint

​5. Generating the Dockerfile

​6. Registering and Running your Pipeline

​Appendix 1. Getting Sample Names Dynamically

1. Updating the Pipeline to use Latch Storage

2. Adding Resources + Containers to each Rule

3. Writing Metadata

4. Generating the Entrypoint

5. Generating the Dockerfile

6. Registering and Running your Pipeline

Appendix 1. Getting Sample Names Dynamically