wip: spindle: docker engine

Signed-off-by: Anirudh Oppiliappan <[e-mail]>

spindle/config/config.go

···
 	WorkflowTimeout string `env:"WORKFLOW_TIMEOUT, default=5m"`
 }
 
+type DockerPipelines struct {
+	WorkflowTimeout string `env:"WORKFLOW_TIMEOUT, default=5m"`
+}
+
 type Config struct {
 	Server          Server          `env:",prefix=SPINDLE_SERVER_"`
 	NixeryPipelines NixeryPipelines `env:",prefix=SPINDLE_NIXERY_PIPELINES_"`
+	DockerPipelines DockerPipelines `env:",prefix=SPINDLE_DOCKER_PIPELINES_"`
 }
 
 func Load(ctx context.Context) (*Config, error) {

spindle/engines/docker/engine.go

···
+package docker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/mount"
+	"github.com/docker/docker/api/types/network"
+	"github.com/docker/docker/client"
+	"github.com/docker/docker/pkg/stdcopy"
+	"gopkg.in/yaml.v3"
+	"tangled.org/core/api/tangled"
+	"tangled.org/core/log"
+	"tangled.org/core/spindle/config"
+	"tangled.org/core/spindle/engine"
+	"tangled.org/core/spindle/engines/common"
+	engineerrors "tangled.org/core/spindle/engines/errors"
+	"tangled.org/core/spindle/models"
+	"tangled.org/core/spindle/secrets"
+)
+
+const (
+	workspaceDir = "/tangled/workspace"
+	homeDir      = "/tangled/home"
+	defaultImage = "ubuntu:22.04"
+)
+
+type cleanupFunc func(context.Context) error
+
+type Engine struct {
+	docker client.APIClient
+	l      *slog.Logger
+	cfg    *config.Config
+
+	cleanupMu sync.Mutex
+	cleanup   map[string][]cleanupFunc
+}
+
+type Step struct {
+	name        string
+	kind        models.StepKind
+	command     string
+	environment map[string]string
+}
+
+func (s Step) Name() string {
+	return s.name
+}
+
+func (s Step) Command() string {
+	return s.command
+}
+
+func (s Step) Kind() models.StepKind {
+	return s.kind
+}
+
+// setupSteps get added to start of Steps
+type setupSteps []models.Step
+
+// addStep adds a step to the beginning of the workflow's steps.
+func (ss *setupSteps) addStep(step models.Step) {
+	*ss = append(*ss, step)
+}
+
+type addlFields struct {
+	image     string
+	container string
+}
+
+func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
+	swf := &models.Workflow{}
+	addl := addlFields{}
+
+	dwf := &struct {
+		Steps []struct {
+			Command     string            `yaml:"command"`
+			Name        string            `yaml:"name"`
+			Environment map[string]string `yaml:"environment"`
+		} `yaml:"steps"`
+		Environment map[string]string `yaml:"environment"`
+		Image       string            `yaml:"image"`
+	}{}
+	err := yaml.Unmarshal([]byte(twf.Raw), &dwf)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, dstep := range dwf.Steps {
+		sstep := Step{}
+		sstep.environment = dstep.Environment
+		sstep.command = dstep.Command
+		sstep.name = dstep.Name
+		sstep.kind = models.StepKindUser
+		swf.Steps = append(swf.Steps, sstep)
+	}
+	swf.Name = twf.Name
+	swf.Environment = dwf.Environment
+
+	// Use specified image or default
+	if dwf.Image != "" {
+		addl.image = dwf.Image
+	} else {
+		addl.image = defaultImage
+	}
+
+	setup := &setupSteps{}
+
+	// Add clone step
+	setup.addStep(models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev))
+
+	// append setup steps in order to the start of workflow steps
+	swf.Steps = append(*setup, swf.Steps...)
+	swf.Data = addl
+
+	return swf, nil
+}
+
+func (e *Engine) WorkflowTimeout() time.Duration {
+	workflowTimeoutStr := e.cfg.DockerPipelines.WorkflowTimeout
+	workflowTimeout, err := time.ParseDuration(workflowTimeoutStr)
+	if err != nil {
+		e.l.Error("failed to parse workflow timeout", "error", err, "timeout", workflowTimeoutStr)
+		workflowTimeout = 5 * time.Minute
+	}
+
+	return workflowTimeout
+}
+
+func New(ctx context.Context, cfg *config.Config) (*Engine, error) {
+	dcli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	if err != nil {
+		return nil, err
+	}
+
+	l := log.FromContext(ctx).With("component", "spindle-docker")
+
+	e := &Engine{
+		docker: dcli,
+		l:      l,
+		cfg:    cfg,
+	}
+
+	e.cleanup = make(map[string][]cleanupFunc)
+
+	return e, nil
+}
+
+func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) error {
+	e.l.Info("setting up workflow", "workflow", wid)
+
+	_, err := e.docker.NetworkCreate(ctx, networkName(wid), network.CreateOptions{
+		Driver: "bridge",
+	})
+	if err != nil {
+		return err
+	}
+	e.registerCleanup(wid, func(ctx context.Context) error {
+		return e.docker.NetworkRemove(ctx, networkName(wid))
+	})
+
+	addl := wf.Data.(addlFields)
+
+	reader, err := e.docker.ImagePull(ctx, addl.image, image.PullOptions{})
+	if err != nil {
+		e.l.Error("pipeline image pull failed!", "image", addl.image, "workflowId", wid, "error", err.Error())
+
+		return fmt.Errorf("pulling image: %w", err)
+	}
+	defer reader.Close()
+	io.Copy(os.Stdout, reader)
+
+	resp, err := e.docker.ContainerCreate(ctx, &container.Config{
+		Image:      addl.image,
+		Cmd:        []string{"sleep", "infinity"}, // Keep container running
+		OpenStdin:  true,
+		Tty:        false,
+		Hostname:   "spindle",
+		WorkingDir: workspaceDir,
+		Labels: map[string]string{
+			"sh.tangled.pipeline/workflow_id": wid.String(),
+		},
+	}, &container.HostConfig{
+		Mounts: []mount.Mount{
+			{
+				Type:     mount.TypeTmpfs,
+				Target:   "/tmp",
+				ReadOnly: false,
+				TmpfsOptions: &mount.TmpfsOptions{
+					Mode: 0o1777, // world-writeable sticky bit
+					Options: [][]string{
+						{"exec"},
+					},
+				},
+			},
+		},
+		ReadonlyRootfs: false,
+		CapDrop:        []string{"ALL"},
+		CapAdd:         []string{"CAP_DAC_OVERRIDE", "CAP_CHOWN", "CAP_FOWNER", "CAP_SETUID", "CAP_SETGID"},
+		SecurityOpt:    []string{"no-new-privileges"},
+		ExtraHosts:     []string{"host.docker.internal:host-gateway"},
+	}, nil, nil, "")
+	if err != nil {
+		return fmt.Errorf("creating container: %w", err)
+	}
+	e.registerCleanup(wid, func(ctx context.Context) error {
+		err = e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{})
+		if err != nil {
+			return err
+		}
+
+		return e.docker.ContainerRemove(ctx, resp.ID, container.RemoveOptions{
+			RemoveVolumes: true,
+			RemoveLinks:   false,
+			Force:         false,
+		})
+	})
+
+	err = e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{})
+	if err != nil {
+		return fmt.Errorf("starting container: %w", err)
+	}
+
+	// Create necessary directories
+	mkExecResp, err := e.docker.ContainerExecCreate(ctx, resp.ID, container.ExecOptions{
+		Cmd:          []string{"mkdir", "-p", workspaceDir, homeDir},
+		AttachStdout: true,
+		AttachStderr: true,
+	})
+	if err != nil {
+		return err
+	}
+
+	// This actually *starts* the command. Thanks, Docker!
+	execResp, err := e.docker.ContainerExecAttach(ctx, mkExecResp.ID, container.ExecAttachOptions{})
+	if err != nil {
+		return err
+	}
+	defer execResp.Close()
+
+	// This is apparently best way to wait for the command to complete.
+	_, err = io.ReadAll(execResp.Reader)
+	if err != nil {
+		return err
+	}
+
+	execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
+	if err != nil {
+		return err
+	}
+
+	if execInspectResp.ExitCode != 0 {
+		return fmt.Errorf("mkdir exited with exit code %d", execInspectResp.ExitCode)
+	} else if execInspectResp.Running {
+		return errors.New("mkdir is somehow still running??")
+	}
+
+	addl.container = resp.ID
+	wf.Data = addl
+
+	return nil
+}
+
+func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger *models.WorkflowLogger) error {
+	addl := w.Data.(addlFields)
+	workflowEnvs := common.ConstructEnvs(w.Environment)
+	for _, s := range secrets {
+		workflowEnvs.AddEnv(s.Key, s.Value)
+	}
+
+	step := w.Steps[idx]
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	envs := append(common.EnvVars(nil), workflowEnvs...)
+	if dockerStep, ok := step.(Step); ok {
+		for k, v := range dockerStep.environment {
+			envs.AddEnv(k, v)
+		}
+	}
+	envs.AddEnv("HOME", homeDir)
+
+	mkExecResp, err := e.docker.ContainerExecCreate(ctx, addl.container, container.ExecOptions{
+		Cmd:          []string{"sh", "-c", step.Command()},
+		AttachStdout: true,
+		AttachStderr: true,
+		Env:          envs,
+	})
+	if err != nil {
+		return fmt.Errorf("creating exec: %w", err)
+	}
+
+	// start tailing logs in background
+	tailDone := make(chan error, 1)
+	go func() {
+		tailDone <- e.tailStep(ctx, wfLogger, mkExecResp.ID, wid, idx, step)
+	}()
+
+	select {
+	case <-tailDone:
+
+	case <-ctx.Done():
+		e.l.Warn("step timed out", "step", step.Name())
+		<-tailDone
+		return engine.ErrTimedOut
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
+	if err != nil {
+		return err
+	}
+
+	if execInspectResp.ExitCode != 0 {
+		inspectResp, err := e.docker.ContainerInspect(ctx, addl.container)
+		if err != nil {
+			return err
+		}
+
+		e.l.Error("workflow failed!", "workflow_id", wid.String(), "exit_code", execInspectResp.ExitCode, "oom_killed", inspectResp.State.OOMKilled)
+
+		if inspectResp.State.OOMKilled {
+			return engineerrors.ErrOOMKilled
+		}
+		return engine.ErrWorkflowFailed
+	}
+
+	return nil
+}
+
+func (e *Engine) tailStep(ctx context.Context, wfLogger *models.WorkflowLogger, execID string, wid models.WorkflowId, stepIdx int, step models.Step) error {
+	if wfLogger == nil {
+		return nil
+	}
+
+	// This actually *starts* the command. Thanks, Docker!
+	logs, err := e.docker.ContainerExecAttach(ctx, execID, container.ExecAttachOptions{})
+	if err != nil {
+		return err
+	}
+	defer logs.Close()
+
+	_, err = stdcopy.StdCopy(
+		wfLogger.DataWriter(stepIdx, "stdout"),
+		wfLogger.DataWriter(stepIdx, "stderr"),
+		logs.Reader,
+	)
+	if err != nil && err != io.EOF && !errors.Is(err, context.DeadlineExceeded) {
+		return fmt.Errorf("failed to copy logs: %w", err)
+	}
+
+	return nil
+}
+
+func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
+	e.cleanupMu.Lock()
+	key := wid.String()
+
+	fns := e.cleanup[key]
+	delete(e.cleanup, key)
+	e.cleanupMu.Unlock()
+
+	for _, fn := range fns {
+		if err := fn(ctx); err != nil {
+			e.l.Error("failed to cleanup workflow resource", "workflowId", wid, "error", err)
+		}
+	}
+	return nil
+}
+
+func (e *Engine) registerCleanup(wid models.WorkflowId, fn cleanupFunc) {
+	e.cleanupMu.Lock()
+	defer e.cleanupMu.Unlock()
+
+	key := wid.String()
+	e.cleanup[key] = append(e.cleanup[key], fn)
+}
+
+func networkName(wid models.WorkflowId) string {
+	return fmt.Sprintf("workflow-network-%s", wid)
+}

spindle/engines/nixery/engine.go

···
 	"tangled.org/core/log"
 	"tangled.org/core/spindle/config"
 	"tangled.org/core/spindle/engine"
+	"tangled.org/core/spindle/engines/common"
+	engineerrors "tangled.org/core/spindle/engines/errors"
 	"tangled.org/core/spindle/models"
 	"tangled.org/core/spindle/secrets"
 )
···
 
 func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger *models.WorkflowLogger) error {
 	addl := w.Data.(addlFields)
-	workflowEnvs := ConstructEnvs(w.Environment)
+	workflowEnvs := common.ConstructEnvs(w.Environment)
 	// TODO(winter): should SetupWorkflow also have secret access?
 	// IMO yes, but probably worth thinking on.
 	for _, s := range secrets {
···
 	default:
 	}
 
-	envs := append(EnvVars(nil), workflowEnvs...)
+	envs := append(common.EnvVars(nil), workflowEnvs...)
 	if nixStep, ok := step.(Step); ok {
 		for k, v := range nixStep.environment {
 			envs.AddEnv(k, v)
···
 		e.l.Error("workflow failed!", "workflow_id", wid.String(), "exit_code", execInspectResp.ExitCode, "oom_killed", inspectResp.State.OOMKilled)
 
 		if inspectResp.State.OOMKilled {
-			return ErrOOMKilled
+			return engineerrors.ErrOOMKilled
 		}
 		return engine.ErrWorkflowFailed
 	}

spindle/engines/nixery/envs.go spindle/engines/common/envs.go

···
-package nixery
+package common
 
 import (
 	"fmt"
···
 
 type EnvVars []string
 
-// ConstructEnvs converts a tangled.Pipeline_Step_Environment_Elem.{Key,Value}
-// representation into a docker-friendly []string{"KEY=value", ...} slice.
+// ConstructEnvs converts a map[string]string representation into a 
+// docker-friendly []string{"KEY=value", ...} slice.
 func ConstructEnvs(envs map[string]string) EnvVars {
 	var dockerEnvs EnvVars
 	for k, v := range envs {
···
 // AddEnv adds a key=value string to the EnvVar.
 func (ev *EnvVars) AddEnv(key, value string) {
 	*ev = append(*ev, fmt.Sprintf("%s=%s", key, value))
-}
+}

spindle/engines/nixery/envs_test.go spindle/engines/common/envs_test.go

···
-package nixery
+package common
 
 import (
 	"testing"
···
 	ev.AddEnv("BAZ", "qux")
 	want := EnvVars{"FOO=bar", "BAZ=qux"}
 	assert.ElementsMatch(t, want, ev)
-}
+}

spindle/engines/nixery/errors.go spindle/engines/errors/errors.go

···
-package nixery
+package errors
 
 import "errors"
 
 var (
 	ErrOOMKilled = errors.New("oom killed")
-)
+)

spindle/server.go

···
 	"tangled.org/core/spindle/config"
 	"tangled.org/core/spindle/db"
 	"tangled.org/core/spindle/engine"
+	"tangled.org/core/spindle/engines/docker"
 	"tangled.org/core/spindle/engines/nixery"
 	"tangled.org/core/spindle/models"
 	"tangled.org/core/spindle/queue"
···
 		return err
 	}
 
+	dockerEng, err := docker.New(ctx, cfg)
+	if err != nil {
+		return err
+	}
+
 	s, err := New(ctx, cfg, map[string]models.Engine{
 		"nixery": nixeryEng,
+		"docker": dockerEng,
 	})
 	if err != nil {
 		return err