Including Globus in Slurm workflows

Sometimes, a Slurm workflow may need input data from outside of the O2 active file systems. If the data is accessible by Globus, you can add Globus to the workflow, so that data transfer can be automatically started, and once the data transfer is finished, downstream data analysis will automatically start.

O2 active file systems include /home/, /n/data1, /n/data2, /n/data3, /n/groups and /n/scratch.

O2 can not directly access /n/files/ and /n/standby, but they are accessible by Globus.

Here is how you can set it up:

First, create conda env for Globus software, then login Globus to save the login token and approve data access permission:

# login to the cluster
ssh o2.hms.harvard.edu

# load conda module
module load conda/miniforge3/24.11.3-0

# create globus env is not done yet 
srun -p interactive --mem 4G -t 2:0:0 --pty bash
mamba create -n globusEnv python=3.11 globus-cli -y

# activate globus env
conda activate globusEnv

# log in to globus. The tocken will be saved in ~/.config/globus/tokens.json
globus login

# find globus endpoint uuid using globus.org
# this is good for O2 actitive storage and also good for /n/files and /n/standby
SRC_ENDPOINT="b0718922-7031-11e9-b7f8-0a37f382de32"

# grant consent for the Globus CLI to access the data.
# click the link and autherize the scrope. 
# If you see an error, refresh and try again. 
globus session consent "urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/${SRC_ENDPOINT}/data_access]"

Then prepare testing data, and start the workflow:

# login to the cluster
ssh o2.hms.harvard.edu

# load conda module
module load conda/miniforge3/24.11.3-0

# find globus endpoint uuid using globus.org
# this is good for O2 actitive storage and also good for /n/files and /n/standby
# so you can use this ID except for guest collection shared by your colaberators
SRC_ENDPOINT="b0718922-7031-11e9-b7f8-0a37f382de32"
DST_ENDPOINT="b0718922-7031-11e9-b7f8-0a37f382de32"

# make test data directory
mkdir -p /home/$USER/globus
echo "This is a test file for Globus transfer." > /home/$USER/globus/test_file.txt

# set source and destination paths. 
# you can use any path on O2 actitive storage, /n/files and /n/standby
SRC_PATH="/home/$USER/globus"
DST_PATH="/home/$USER/globus.copy"

# notes: globus works differently than rsync or cp
# 1. the ending '/' does not matter.
# 2 if the source does not exist, globus will fail.
# 3 if the destiation folder's parent folder does not exist, globus will fail.
# 4 if the destination folder does not exist, it will be created.
# 5 blobus does not copy the source folder itself, but only copy the contents of the source folder.

# remove the destinaton if it exists. 
[ -d "$DST_PATH" ] && rm -r "$DST_PATH"

# check source path and make sure it exists
globus ls "${SRC_ENDPOINT}:${SRC_PATH}" > /dev/null 2>&1 || {
    echo "Error: Source path does not exist: ${SRC_ENDPOINT}:${SRC_PATH}."
}

# check destination parent path and make sure it exists
globus ls "${DST_ENDPOINT}:$(dirname "$DST_PATH")" > /dev/null 2>&1 || {
    echo "Error: Destination parent path does not exist: ${DST_ENDPOINT}:$(dirname "$DST_PATH")."
}

# testing consent / permissions with dry-run
globus transfer "${SRC_ENDPOINT}:${SRC_PATH}" "${DST_ENDPOINT}:${DST_PATH}" \
    --recursive --dry-run > /dev/null 2>&1 || {
    echo "Error: Failed to test transfer permissions: ${SRC_ENDPOINT}:${SRC_PATH} -> ${DST_ENDPOINT}:${DST_PATH}."
}

# copy the three scripts (globusJob.sh, globusAfter.sh, globusWorkflow.sh with content as below): 
cp /n/groups/shared_databases/rcbio/globus/globus*.sh . 

# run workflow and submit jobs to Slurm
bash globusWorkflow.sh

Bellow are the three scripts:

globusJob.sh

#!/bin/bash

#SBATCH -p short
#SBATCH --job-name=globus_transfer
#SBATCH --output=globus_transfer_%j.out
#SBATCH --error=globus_transfer_%j.err
#SBATCH --time=01:00:00
#SBATCH --mem=512M

#set -x 

set -euo pipefail

usage() {
  echo "Usage: $0 <src_endpoint_id> <src_path> <dst_endpoint_id> <dst_path>"
}

log() {
  echo "[$(date +"%Y-%m-%d %H:%M:%S")] $*"
}

if [[ $# -ne 4 ]]; then
  usage
  exit 2
fi

if ! command -v globus >/dev/null 2>&1; then
  echo "Error: globus CLI is not available in PATH." >&2
  exit 127
fi

if ! command -v python3 >/dev/null 2>&1; then
  echo "Error: python3 is required to parse Globus JSON output." >&2
  exit 127
fi

readonly SRC="${1}:${2}"
readonly DST="${3}:${4}"
readonly WAIT_TIMEOUT="${GLOBUS_WAIT_TIMEOUT:-7200}"
readonly WAIT_POLL_INTERVAL="${GLOBUS_WAIT_POLL_INTERVAL:-60}"

log "Starting Globus transfer..."

transfer_json="$(globus transfer "$SRC" "$DST" --recursive --format json)"

TASK_ID="$(python3 -c 'import json, sys; print(json.load(sys.stdin)["task_id"])' <<< "$transfer_json")"

if [[ -z "$TASK_ID" ]]; then
  echo "Error: failed to extract task_id from Globus response." >&2
  printf '%s\n' "$transfer_json" >&2
  exit 1
fi

log "Globus task ID: $TASK_ID"
log "Waiting for Globus transfer to complete (timeout=${WAIT_TIMEOUT}s, poll=${WAIT_POLL_INTERVAL}s)..."

# refresh globus tokens every 10 hour to avoid expiration during long transfers
while true; do
  globus refresh --all
  sleep 36000
done &

globus task wait "$TASK_ID" --timeout "$WAIT_TIMEOUT" --polling-interval "$WAIT_POLL_INTERVAL"

log "Globus transfer completed successfully."

globusAfter.sh

#!/bin/bash
#SBATCH -p short 
#SBATCH --output=post_transfer_%j.out
#SBATCH --error=post_transfer_%j.err
#SBATCH --time=01:00:00
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=1
#SBATCH --mem=512M

#set -x 

set -euo pipefail

log() {
	echo "[$(date +"%Y-%m-%d %H:%M:%S")] $*"
}

on_error() {
	local exit_code="$1"
	local line_no="$2"
	echo "Error: post-transfer job failed at line ${line_no} (exit code ${exit_code})." >&2
}

trap 'on_error "$?" "$LINENO"' ERR

log "Running job after Globus transfer..."

# Add your post-transfer commands below.
# Example:
# bash /path/to/analysis.sh input1 input2

log "Done with job after Globus transfer."

globusWorkflow.sh:

#!/bin/bash

module load conda/miniforge3/24.11.3-0

# activate globus env
conda activate globusEnv

# find globus endpoint uuid using globus.org
SRC_ENDPOINT="b0718922-7031-11e9-b7f8-0a37f382de32"
DST_ENDPOINT="b0718922-7031-11e9-b7f8-0a37f382de32"

# set source and destination paths. 
SRC_PATH="/home/$USER/globus"
DST_PATH="/home/$USER/globus.copy"

# submit transfer job
TRANSFER_JOB_ID=$(sbatch --parsable globusJob.sh "${SRC_ENDPOINT}" "${SRC_PATH}" "${DST_ENDPOINT}" "${DST_PATH}")

# make sure the transfer job is submitted successfully 
echo "Submitted Globus transfer job: ${TRANSFER_JOB_ID}"

# submit dependent job that will run after the transfer job completes successfully
DEPENDENT_JOB_ID=$(sbatch --parsable --dependency=afterok:${TRANSFER_JOB_ID} globusAfter.sh)

echo "Submitted dependent job: ${DEPENDENT_JOB_ID}. Dependent job will start only if transfer job ${TRANSFER_JOB_ID} succeeds."