dagger parallel lint pipeline

dagger.json

{
  "name": "go-orb",
  "sdk": "go",
  "dependencies": [
    {
      "name": "go",
      "source": "https://github.com/sagikazarmark/daggerverse/go@07f08ab95be76586c798bc55424c684ad6d41497"
    },
    {
      "name": "golangci-lint",
      "source": "https://github.com/sagikazarmark/daggerverse/golangci-lint@d814d0d7c421348f51cdda96870a05ca2aa8e96a"
    }
  ],
  "source": "dagger",
  "engineVersion": "v0.12.6"
}

main.go

// A generated module for GoOrb functions
//
// This module has been generated via dagger init and serves as a reference to
// basic module structure as you get started with Dagger.
//
// Two functions have been pre-created. You can modify, delete, or add to them,
// as needed. They demonstrate usage of arguments and return types using simple
// echo and grep commands. The functions can be called from the dagger CLI or
// from one of the SDKs.
//
// The first line in this comment block is a short description line and the
// rest is a long description with more detail on the module's purpose or usage,
// if appropriate. All modules should have a short description.

package main

import (
	"context"
	"dagger/go-orb/internal/dagger"
	"runtime"
	"sync"

	"github.com/hashicorp/go-multierror"
)

type WorkerResult struct {
	Module string
	Logs   string
	Err    error
	Source *dagger.Directory
}

type AllResult struct {
	Logs   []string
	Source *dagger.Directory
}

func (m *GoOrb) runAll(ctx context.Context, root *dagger.Directory, worker func(ctx context.Context, wg *sync.WaitGroup, inpupt <-chan string, res chan<- *WorkerResult, root *dagger.Directory)) (*AllResult, error) {
	mods, err := m.Modules(ctx, root)
	if err != nil {
		return nil, err
	}

	input := make(chan string, len(mods))
	res := make(chan *WorkerResult, len(mods))

	var wg sync.WaitGroup

	// Start workers
	for i := 0; i < runtime.NumCPU()/2; i++ {
		wg.Add(1)
		go worker(ctx, &wg, input, res, root)
	}

	// Add jobs
	for _, mod := range mods {
		input <- mod
	}
	close(input)

	// Wait for all workers to complete
	wg.Wait()
	close(res)

	// Aggregate results
	result := &AllResult{Logs: []string{}, Source: dag.Directory()}
	for r := range res {
		if r.Err != nil {
			err = multierror.Append(err, r.Err)
		}

		if r.Source != nil {
			result.Source = result.Source.WithDirectory(r.Module, r.Source)
		}

		if len(r.Logs) > 0 {
			result.Logs = append(result.Logs, r.Module+"\n"+r.Logs)
		}
	}

	return result, err
}

type GoOrb struct{}

// Returns all modules in `root`
func (m *GoOrb) Modules(ctx context.Context, root *dagger.Directory) ([]string, error) {
	mods, err := root.Glob(ctx, "**/go.mod")
	if err != nil {
		return nil, err
	}

	for i, mod := range mods {
		if len(mod) > 7 {
			mods[i] = mod[0 : len(mod)-7]
		} else {
			mods[i] = "."
		}
	}

	return mods, err
}

// Lints all modules starting from `root` with golangci-lint
func (m *GoOrb) Lint(ctx context.Context, root *dagger.Directory) (*AllResult, error) {
	lintWorker := func(ctx context.Context, wg *sync.WaitGroup, input <-chan string, res chan<- *WorkerResult, root *dagger.Directory) {
		defer wg.Done()
		for dir := range input {
			select {
			case <-ctx.Done():
				return
			default:
			}

			out, err := dag.GolangciLint().Run(root.Directory(dir)).Stdout(ctx)
			res <- &WorkerResult{Module: dir, Logs: out, Err: err}
		}
	}

	return m.runAll(ctx, root, lintWorker)
}

// Runs `go get -u -t ./...` in all modules starting with `root`
func (m *GoOrb) Update(ctx context.Context, root *dagger.Directory) (*AllResult, error) {
	updateWorker := func(ctx context.Context, wg *sync.WaitGroup, input <-chan string, res chan<- *WorkerResult, root *dagger.Directory) {
		defer wg.Done()
		for dir := range input {
			select {
			case <-ctx.Done():
				return
			default:
			}

			c := dag.Go().
				WithSource(root.Directory(dir)).
				Exec([]string{"go", "get", "-u", "-t", "./..."})
			stdout, err := c.Stdout(ctx)
			if err != nil {
				res <- &WorkerResult{Module: dir, Source: c.Directory("/work/src"), Err: err}
				continue
			}

			res <- &WorkerResult{Module: dir, Logs: stdout, Source: c.Directory("/work/src"), Err: err}
		}
	}

	return m.runAll(ctx, root, updateWorker)
}

Usage:

dagger call modules --root=.
dagger call lint --root=.
dagger call update --root=. source export --path=.