Funnel 0.11.1 release 🎉

.github/workflows/build.yml

@@ -23,7 +23,6 @@ jobs:
         key: ${{ runner.os }}-funnel-bin-${{ hashFiles('**/go.sum') }}-${{ github.ref }}
         restore-keys: |
           ${{ runner.os }}-funnel-bin-${{ github.ref }}
-          ${{ runner.os }}-funnel-bin-
 
     - name: Build Funnel (if cache doesn't exist)
       run: |

.github/workflows/compliance.yaml

@@ -37,7 +37,9 @@ jobs:
       matrix:
         version: [1.0.0, 1.1.0]
         db: ["boltdb", "mongodb"]
-        compute: ["local", "kubernetes"]
+        # Temporary: Only test local compute backend until we bring up a K8s cluster in Github Actions
+        # compute: ["local", "kubernetes"]
+        compute: ["local"]
         storage: ["local", "s3"]
     needs: build
     runs-on: ubuntu-latest
@@ -70,6 +72,11 @@ jobs:
             cat `pwd`/tests/s3.config.yml >> config.yml
           fi
 
+          if [ ${{ matrix.compute }} = "kubernetes" ]; then
+            kubectl apply -f tests/k8s
+            cat `pwd`/tests/k8s.config.yml >> config.yml
+          fi
+
           chmod +x funnel
           ./funnel server run --config `pwd`/config.yml &> funnel.logs &
 

.github/workflows/hugo.yml

@@ -4,7 +4,9 @@ name: Deploy Hugo site to Pages
 on:
   # Runs on pushes targeting the default branch
   push:
-    branches: ["master"]
+    branches:
+      - main
+      - master
 
   # Allows you to run this workflow manually from the Actions tab
   workflow_dispatch:
@@ -58,7 +60,12 @@ jobs:
             --minify \
             --baseURL "${{ steps.pages.outputs.base_url }}/" \
             --source website \
-            --destination public
+            --destination docs
+
+          # Run pagefind to add search functionality
+          # Reference: https://pagefind.app/docs/
+          npx -y pagefind --site docs
+
       - name: Upload artifact
         uses: actions/upload-pages-artifact@v3
         with:

.gitignore

@@ -20,6 +20,5 @@ storage/_test_download
 docker/funnel
 deployments/kubernetes/**/funnel
 deployments/kubernetes/**/*.json
-funnel
 dist/
 .vscode

.goreleaser.yaml

@@ -28,10 +28,10 @@ archives:
     name_template: "{{.ProjectName}}-{{.Os}}-{{.Arch}}-{{.Version}}"
 
 brews:
-  - tap:
+  - repository:
       owner: ohsu-comp-bio
       name: homebrew-formula
-    folder: Formula
+    directory: Formula
     skip_upload: true
     description: "distributed task execution toolkit"
     homepage: "https://ohsu-comp-bio.github.io/funnel/"

Dockerfile

@@ -1,10 +1,14 @@
 # build stage
-FROM golang:1.21-alpine AS build-env
+FROM golang:1.23-alpine AS build-env
 RUN apk add make git bash build-base
 ENV GOPATH=/go
 ENV PATH="/go/bin:${PATH}"
-ADD ./ /go/src/github.com/ohsu-comp-bio/funnel
-RUN cd /go/src/github.com/ohsu-comp-bio/funnel && make build
+
+WORKDIR /go/src/github.com/ohsu-comp-bio/funnel
+COPY go.* .
+RUN go mod download
+COPY . .
+RUN --mount=type=cache,target=/root/.cache/go-build make build
 
 # final stage
 FROM alpine

Dockerfile.dind

@@ -1,13 +1,17 @@
 # build stage
-FROM golang:1.20-alpine AS build-env
+FROM golang:1.23-alpine AS build-env
 RUN apk add make git bash build-base
 ENV GOPATH=/go
 ENV PATH="/go/bin:${PATH}"
-ADD ./ /go/src/github.com/ohsu-comp-bio/funnel
-RUN cd /go/src/github.com/ohsu-comp-bio/funnel && make build
+
+WORKDIR /go/src/github.com/ohsu-comp-bio/funnel
+COPY go.* .
+RUN go mod download
+COPY . .
+RUN --mount=type=cache,target=/root/.cache/go-build make build
 
 # final stage
-FROM docker:stable-dind
+FROM docker:dind
 WORKDIR /opt/funnel
 VOLUME /opt/funnel/funnel-work-dir
 EXPOSE 8000 9090

Dockerfile.dind-rootless

@@ -1,13 +1,17 @@
 # build stage
-FROM golang:1.20-alpine AS build-env
+FROM golang:1.23-alpine AS build-env
 RUN apk add make git bash build-base
 ENV GOPATH=/go
 ENV PATH="/go/bin:${PATH}"
-ADD ./ /go/src/github.com/ohsu-comp-bio/funnel
-RUN cd /go/src/github.com/ohsu-comp-bio/funnel && make build
+
+WORKDIR /go/src/github.com/ohsu-comp-bio/funnel
+COPY go.* .
+RUN go mod download
+COPY . .
+RUN --mount=type=cache,target=/root/.cache/go-build make build
 
 # final stage
-FROM docker:stable-dind-rootless
+FROM docker:dind-rootless
 WORKDIR /opt/funnel
 VOLUME /opt/funnel/funnel-work-dir
 EXPOSE 8000 9090

Makefile

@@ -11,7 +11,7 @@ git_upstream := $(shell git remote get-url $(shell git config branch.$(shell git
 export GIT_BRANCH = $(git_branch)
 export GIT_UPSTREAM = $(git_upstream)
 
-export FUNNEL_VERSION=0.11.0
+export FUNNEL_VERSION=0.11.1-rc.5
 
 # LAST_PR_NUMBER is used by the release notes builder to generate notes
 # based on pull requests (PR) up until the last release.
@@ -226,15 +226,7 @@ snapshot: release-dep
 
 # build a docker container locally
 docker:
-	docker build -t ohsucompbio/funnel:latest ./
-
-# build a docker container that supports docker-in-docker
-docker-dind:
-	docker build -t ohsucompbio/funnel-dind:latest -f Dockerfile.dind .
-
-# build a docker container that supports rootless docker-in-docker
-docker-dind-rootless:
-	docker build -t ohsucompbio/funnel-dind-rootless:latest -f Dockerfile.dind-rootless .
+	docker build -t quay.io/ohsu-comp-bio/funnel:latest ./
 
 # Create a release on Github using GoReleaser 
 release:
@@ -268,12 +260,11 @@ website:
 	@cp ./config/*.txt ./website/static/funnel-config-examples/
 	@cp ./config/default-config.yaml ./website/static/funnel-config-examples/
 	hugo --source ./website
+	npx -y pagefind --site docs
 
 # Serve the Funnel website on localhost:1313
-website-dev:
-	@cp ./config/*.txt ./website/static/funnel-config-examples/
-	@cp ./config/default-config.yaml ./website/static/funnel-config-examples/
-	hugo --source ./website -w server
+website-dev: website
+	hugo --source ./website --watch server
 
 # Remove build/development files.
 clean:

README.md

@@ -4,14 +4,14 @@
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
 [![Godoc](https://img.shields.io/badge/godoc-ref-blue.svg)](http://godoc.org/github.com/ohsu-comp-bio/funnel)
 
-
-Funnel
-======
-
-https://ohsu-comp-bio.github.io/funnel/
+<a title="Funnel Homepage" href="https://ohsu-comp-bio.github.io/funnel">
+  <img title="Funnel Logo" src="https://github.com/user-attachments/assets/f51cf06b-d802-4e20-bde1-bcd1fc5657e6" />
+</a>
 
 Funnel is a toolkit for distributed, batch task execution, including a server, worker, and a set of compute, storage, and database backends. Given a task description, Funnel will find a worker to execute the task, download inputs, run a series of (Docker) containers, upload outputs, capture logs, and track the whole process.
 
 Funnel is an implementation of the [GA4GH Task Execution Schemas](https://github.com/ga4gh/task-execution-schemas), an effort to standardize the APIs used for task execution across many platforms.
 
 Funnel provides an API server, multiple storage backends (local FS, S3, Google Bucket, etc.), multiple compute backends (local, HTCondor, Google Cloud, etc.), and a web dashboard.
+
+https://ohsu-comp-bio.github.io/funnel/

cmd/aws/batch/config.go

@@ -116,7 +116,7 @@ func DefaultConfig() Config {
 		},
 		JobDef: JobDefinitionConfig{
 			Name:      "funnel-job-def",
-			Image:     "docker.io/ohsucompbio/funnel:latest",
+			Image:     "quay.io/ohsu-comp-bio/funnel:latest",
 			VCPUs:     1,
 			MemoryMiB: 128,
 		},

cmd/node/run.go

@@ -34,8 +34,8 @@ func Run(ctx context.Context, conf config.Config, log *logger.Logger) error {
 		return err
 	}
 
-	runctx, cancel := context.WithCancel(context.Background())
-	runctx = util.SignalContext(ctx, time.Nanosecond, syscall.SIGINT, syscall.SIGTERM)
+	_, cancel := context.WithCancel(context.Background())
+	runctx := util.SignalContext(ctx, time.Nanosecond, syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
 
 	hupsig := make(chan os.Signal, 1)

cmd/server/run.go

@@ -147,7 +147,11 @@ func NewServer(ctx context.Context, conf config.Config, log *logger.Logger) (*Se
 		case strings.ToLower(conf.Database):
 			continue
 		case "log":
-			continue
+			if conf.Compute == "slurm" {
+				writer = &events.Logger{Log: log}
+			} else {
+				continue
+			}
 		case "boltdb":
 			writer, err = boltdb.NewBoltDB(conf.BoltDB)
 		case "badger":
@@ -271,6 +275,7 @@ func NewServer(ctx context.Context, conf config.Config, log *logger.Logger) (*Se
 				Compute: compute,
 				Read:    reader,
 				Log:     log,
+				Config:  conf,
 			},
 			Events: &events.Service{Writer: writer},
 			Nodes:  nodes,

cmd/util/flags.go

@@ -100,7 +100,9 @@ func workerFlags(flagConf *config.Config) *pflag.FlagSet {
 	f.Var(&flagConf.Worker.LogUpdateRate, "Worker.LogUpdateRate", "How often to send stdout/stderr log updates")
 	f.Var(&flagConf.Worker.PollingRate, "Worker.PollingRate", "How often to poll for cancel signals")
 	f.StringVar(&flagConf.Worker.WorkDir, "Worker.WorkDir", flagConf.Worker.WorkDir, "Working directory")
+	f.StringVar(&flagConf.Worker.ScratchPath, "Worker.ScratchPath", flagConf.Worker.ScratchPath, "Scratch directory")
 	f.BoolVar(&flagConf.Worker.LeaveWorkDir, "Worker.LeaveWorkDir", flagConf.Worker.LeaveWorkDir, "Leave working directory after execution")
+	f.StringVar(&flagConf.Worker.DriverCommand, "Worker.DriverCommand", flagConf.Worker.DriverCommand, "Overrides the default command used to run containers.")
 
 	return f
 }
@@ -225,6 +227,8 @@ func computeFlags(flagConf *config.Config) *pflag.FlagSet {
 	f.Var(&flagConf.Scheduler.NodeDeadTimeout, "Scheduler.NodeDeadTimeout", "How long to wait before deleting a dead node from the DB")
 
 	// Kubernetes
+	f.StringVar(&flagConf.Kubernetes.Executor, "Kubernetes.Executor", flagConf.Kubernetes.Executor, "Executor to use for executing tasks (docker or kubernetes)")
+	f.StringVar(&flagConf.Kubernetes.ExecutorTemplateFile, "Kubernetes.ExecutorTemplateFile", flagConf.Kubernetes.ExecutorTemplateFile, "Path to executor job template file")
 	f.StringVar(&flagConf.Kubernetes.TemplateFile, "Kubernetes.TemplateFile", flagConf.Kubernetes.TemplateFile, "Path to job template file")
 	f.StringVar(&flagConf.Kubernetes.Namespace, "Kubernetes.Namespace", flagConf.Kubernetes.Namespace, "Namespace to spawn jobs within")
 	f.StringVar(&flagConf.Kubernetes.ConfigFile, "Kubernetes.ConfigFile", flagConf.Kubernetes.ConfigFile, "Path to kubernetes config file")

cmd/worker/run.go

@@ -3,6 +3,7 @@ package worker
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 
 	"github.com/ohsu-comp-bio/funnel/config"
@@ -70,7 +71,28 @@ func NewWorker(ctx context.Context, conf config.Config, log *logger.Logger, opts
 	}
 	store.AttachLogger(log)
 
+	if conf.Kubernetes.ExecutorTemplateFile != "" {
+		content, err := os.ReadFile(conf.Kubernetes.ExecutorTemplateFile)
+		if err != nil {
+			return nil, fmt.Errorf("reading template: %v", err)
+		}
+		conf.Kubernetes.ExecutorTemplate = string(content)
+	}
+
+	// The executor always defaults to docker, unless explicitly set to kubernetes.
+	var executor = worker.Executor{
+		Backend: "docker",
+	}
+
+	if conf.Kubernetes.Executor == "kubernetes" {
+		executor.Backend = "kubernetes"
+		executor.Template = conf.Kubernetes.ExecutorTemplate
+		executor.Namespace = conf.Kubernetes.Namespace
+		executor.ServiceAccount = conf.Kubernetes.ServiceAccount
+	}
+
 	return &worker.DefaultWorker{
+		Executor:    executor,
 		Conf:        conf.Worker,
 		Store:       store,
 		TaskReader:  reader,

compute/hpc_backend.go

@@ -48,7 +48,7 @@ type HPCBackend struct {
 func (b *HPCBackend) WriteEvent(ctx context.Context, ev *events.Event) error {
 	switch ev.Type {
 	case events.Type_TASK_CREATED:
-		return b.Submit(ev.GetTask())
+		return b.Submit(ctx, ev.GetTask())
 
 	case events.Type_TASK_STATE:
 		if ev.GetState() == tes.State_CANCELED {
@@ -61,10 +61,8 @@ func (b *HPCBackend) WriteEvent(ctx context.Context, ev *events.Event) error {
 func (b *HPCBackend) Close() {}
 
 // Submit submits a task via "qsub", "condor_submit", "sbatch", etc.
-func (b *HPCBackend) Submit(task *tes.Task) error {
-	ctx := context.Background()
-
-	submitPath, err := b.setupTemplatedHPCSubmit(task)
+func (b *HPCBackend) Submit(ctx context.Context, task *tes.Task) error {
+	submitPath, err := b.setupTemplatedHPCSubmit(ctx, task)
 	if err != nil {
 		return err
 	}
@@ -229,7 +227,7 @@ ReconcileLoop:
 // setupTemplatedHPCSubmit sets up a task submission in a HPC environment with
 // a shared file system. It generates a submission file based on a template for
 // schedulers such as SLURM, HTCondor, SGE, PBS/Torque, etc.
-func (b *HPCBackend) setupTemplatedHPCSubmit(task *tes.Task) (string, error) {
+func (b *HPCBackend) setupTemplatedHPCSubmit(ctx context.Context, task *tes.Task) (string, error) {
 	var err error
 
 	// TODO document that these working dirs need manual cleanup
@@ -264,13 +262,24 @@ func (b *HPCBackend) setupTemplatedHPCSubmit(task *tes.Task) (string, error) {
 		zone = zones[0]
 	}
 
+	var args string
+	if ctx.Value("Config") != nil {
+		conf := ctx.Value("Config").(config.Config)
+		configFile := filepath.Join(workdir, "config.yaml")
+		err = config.ToYamlFile(conf, configFile)
+		if err != nil {
+			return "", err
+		}
+		args = fmt.Sprintf("--config %v", configFile)
+	}
 	err = submitTpl.Execute(f, map[string]interface{}{
 		"TaskId":  task.Id,
 		"WorkDir": workdir,
 		"Cpus":    res.GetCpuCores(),
 		"RamGb":   res.GetRamGb(),
 		"DiskGb":  res.GetDiskGb(),
 		"Zone":    zone,
+		"Args":    args,
 	})
 	if err != nil {
 		return "", err

compute/hpc_backend_test.go

@@ -1,6 +1,7 @@
 package compute
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"testing"
@@ -58,7 +59,7 @@ funnel worker run --taskID {{.TaskId}}
 		Conf:      conf,
 	}
 
-	sf, err := b.setupTemplatedHPCSubmit(task)
+	sf, err := b.setupTemplatedHPCSubmit(context.Background(), task)
 	if err != nil {
 		t.Fatal(err)
 	}