Recreate developer environment for Gen3-Workflow, Nextflow, and Funnel

[lbeckman314]

Overview ⚙

Currently the "MVP" for Funnel deployments to Kubernetes is based off submitting tasks with Funnel alone.

This can lead to unexpected or unclear behaviors when Funnel is run as part of a larger workflow "setup" such as with Gen3-Workflow and Nextflow.

Current Setup ⚠️

sequenceDiagram
    title Current Funnel K8s Setup
    participant User
    participant Funnel Server
    participant Funnel Worker/Executor

    User-->>Funnel Server: TES Tasks
    Note over Funnel Server: Creates Worker/Executor
    Funnel Server-->>Funnel Worker/Executor: TES Tasks
    Note over Funnel Worker/Executor: Runs Tasks
    Funnel Worker/Executor-->>Funnel Server: Tasks Status
    Funnel Server-->>User: Tasks Status

Loading

Expected Setup ✅

sequenceDiagram
    title Expected Funnel K8s Setup
    participant User
    participant Nextflow
    participant Gen3-Workflow
    participant Funnel Server
    participant Funnel Worker/Executor

    User-->>Nextflow: Nextflow Workflow
    Note over Nextflow: Translates Workflow to TES Tasks
    Nextflow-->>Gen3-Workflow: TES Tasks
    Note over Gen3-Workflow: AuthZ/AuthN
    Gen3-Workflow-->>Funnel Server: TES Tasks
    Note over Funnel Server: Creates Worker/Executor
    Funnel Server-->>Funnel Worker/Executor: TES Tasks
    Note over Funnel Worker/Executor: Runs Tasks
    Funnel Worker/Executor-->>Funnel Server: Tasks Status
    Funnel Server-->>Gen3-Workflow: Tasks Status
    Gen3-Workflow-->>Nextflow: Tasks Status
    Nextflow-->>User: Workflow Status

Loading

Next Steps 🌀

• Deploy Gen3-Workflow (Helm Chart)
• Deploy Funnel (Helm Chart)
• Run Nextflow Workflows:

salmon-example.nf

reads_input = "$baseDir/input_data/gut_{1,2}.fq"
workflow {
    directory_output_test()
    read_pairs_ch = channel.fromFilePairs( reads_input, checkIfExists: true )
    salmon_quant(salmon_index.out, read_pairs_ch)
}

process directory_output_test {
    output:
    path 'mydirectory'    <=== doesn't work
    path('mydirectory/*') <=== works
    
    script:
    """
    mkdir mydirectory
    touch mydirectory/file1.txt
    touch mydirectory/file2.txt
    """
}

process salmon_quant {
    input:
    path index
    tuple val(pair_id), path(reads)

    output:
    path pair_id

    script:
    """
    salmon quant -i index -l A -1 ${reads[0]} -2 ${reads[1]} -o $pair_id
    """
}

reproducible-example.nf

input_path = "s3://cellxgene-census-public-us-west-2/cell-census/release.json"

process print_and_output {
    label 'tesStep'

    input:
    path input_files

    output:
    stdout emit: print_and_output_log
    path('*.json'), emit: output_files

    script:
    """
    echo \$PWD
    echo "  Listing:"
    ls
    echo "  Contents:"
    cat release.json | tee output.json
    """
}

workflow {
    input_files = Channel.fromPath(input_path)
    print_and_output(input_files)
}
Shared in

[lbeckman314]

Gen3Workflow Setup 🌀

1. Create jobs Namespace

➜ kubectl create namespace jobs

2. Update Helm

➜ helm repo update ohsu
Update Complete. ⎈Happy Helming!⎈

➜ helm search repo ohsu
NAME                    CHART VERSION   APP VERSION     DESCRIPTION
ohsu/funnel             0.1.45          2025-06-05-2    A toolkit for distributed task execution ⚙️
ohsu/gen3workflow       0.1.2           master          Gen3 service for GA4GH TES integration
ohsu/grip               0.1.2           1.16.0          GRaph Integration Platform (GRIP) 🛠️

3. Deploy Funnel

➜ helm upgrade --install funnel ohsu/funnel --namespace funnel --create-namespace -f values.yaml
STATUS: deployed

values.yaml

AmazonS3:
  Disabled: true

GenericS3:
  - Disabled: false
    Endpoint: s3.us-west-2.amazonaws.com
    Key: <KEY>
    Secret: <SECRET>

Kubernetes:
  Bucket: funnel-testing
  Region: us-west-2
  JobsNamespace: jobs

# Local MongoDB task database
mongodb:
  image:
    # ARM compatible image for running MongoDB locally on Apple Silicon
    registry: docker.io
    repository: dlavrenuek/bitnami-mongodb-arm
    tag: 8.0.4

Expected Pods

➜ kubectl get pods -n funnel
NAME             READY   STATUS
funnel-mongodb   1/1     Running  <--- Task Database
funnel-server    1/1     Running  <--- Funnel Server

4. Deploy Gen3Workflow

Note

MOCK_AUTH is enabled by default in order to run Gen3Workflow without calling to Arborist

➜ helm upgrade --install gen3workflow ./charts/gen3workflow --namespace funnel
STATUS: deployed

Expected Pods

➜ kubectl get pods -n funnel
NAME             READY   STATUS
funnel-mongodb   1/1     Running
funnel-server    1/1     Running
gen3workflow     1/1     Running  <--- Gen3 Workflow

5. Port-Forward Gen3Workflow

Tip

Gen3Workflow API (Swagger)

➜ kubectl port-forward deployment/gen3workflow 8000:8000 --namespace funnel
Forwarding from 127.0.0.1:8000 -> 8000
Forwarding from [::1]:8000 -> 8000

➜ curl localhost:8000
{
  "status": "OK"
}

➜ curl localhost:8000/ga4gh/tes/v1/service-info
{
  "contactUrl": "https://ohsu-comp-bio.github.io/funnel/",
  "createdAt": "2016-03-21T16:27:49-07:00",
  "description": "Funnel is a toolkit for distributed task execution via a simple, standard API.",
  "documentationUrl": "https://ohsu-comp-bio.github.io/funnel/",
  ...
}

➜ curl -X POST localhost:8000/ga4gh/tes/v1/tasks --data @hello-world.json
{
  "id": "d13ntgbik0mc739hk0hg"
}

[lbeckman314]

Nextflow Setup (WIP) 🌀

1. Install + Configure Nextflow

Tip

Nextflow Installation Steps

Nextflow Config (nextflow.config)

Gen3Workflow S3 interaction

➜ which nextflow
$HOME/.local/bin/nextflow

➜ nextflow info
  Version: 25.04.3 build 5949

nextflow.config

plugins {
	id 'nf-ga4gh'
}
process {
	executor = 'tes'
	container = 'quay.io/nextflow/bash'
}
tes {
	endpoint = 'http://localhost:8000/ga4gh/tes'
	oauthToken = "${GEN3_TOKEN}"
}
aws {
	accessKey = "${GEN3_TOKEN}"
	secretKey = 'N/A'
	region = 'us-east-1'
	client {
		s3PathStyleAccess = true
		endpoint = 'http://localhost:8000/s3'
	}
}
workDir = '<your working directory>'

2. Run Nextflow Workflows

➜ nextflow run salmon-example.nf

➜ nextflow run reproducible-example.nf

salmon-example.nf

reads_input = "$baseDir/input_data/gut_{1,2}.fq"
workflow {
    directory_output_test()
    read_pairs_ch = channel.fromFilePairs( reads_input, checkIfExists: true )
    salmon_quant(salmon_index.out, read_pairs_ch)
}

process directory_output_test {
    output:
    path 'mydirectory'    <=== doesn't work
    path('mydirectory/*') <=== works
    
    script:
    """
    mkdir mydirectory
    touch mydirectory/file1.txt
    touch mydirectory/file2.txt
    """
}

process salmon_quant {
    input:
    path index
    tuple val(pair_id), path(reads)

    output:
    path pair_id

    script:
    """
    salmon quant -i index -l A -1 ${reads[0]} -2 ${reads[1]} -o $pair_id
    """
}

reproducible-example.nf

input_path = "s3://cellxgene-census-public-us-west-2/cell-census/release.json"

process print_and_output {
    label 'tesStep'

    input:
    path input_files

    output:
    stdout emit: print_and_output_log
    path('*.json'), emit: output_files

    script:
    """
    echo \$PWD
    echo "  Listing:"
    ls
    echo "  Contents:"
    cat release.json | tee output.json
    """
}

workflow {
    input_files = Channel.fromPath(input_path)
    print_and_output(input_files)
}