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) }