Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632package nixery
import ( "bufio" "context" "errors" "fmt" "io" "log/slog" "net" "path" "runtime" "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/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/models" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets")
const ( workspaceDir = "/tangled/workspace" homeDir = "/tangled/home" // how long a workflow's container gets to stop after the workflow ends, // docker's own stop grace included teardownTimeout = 30 * time.Second // how long the reader gets to notice the stream was closed closeGrace = 5 * time.Second)
type cleanupFunc func(context.Context) error
type Engine struct { dockerMu sync.Mutex docker client.APIClient l *slog.Logger cfg *config.Config
slotter engine.WorkflowSlotter seats *seatNetworks
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 Stepstype 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 mounts []mount.Mount}
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"` Dependencies map[string][]string `yaml:"dependencies"` Environment map[string]string `yaml:"environment"` Cache []models.CacheEntry `yaml:"cache"` }{} if err := engine.DescribeManifestError(twf.Raw, dwf); err != nil { return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, err) } if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil { return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, 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) } var ownerDID, repoDID string if md := tpl.TriggerMetadata; md != nil && md.Repo != nil { ownerDID = md.Repo.Did repoDID = ownerDID if md.Repo.RepoDid != nil && *md.Repo.RepoDid != "" { repoDID = *md.Repo.RepoDid } } if ownerDID == "" { return nil, engine.ClassifiedFailure( engine.FailureClassInfrastructure, engine.FailureReasonRuntimeFailed, fmt.Errorf("missing owner DID in pipeline trigger metadata"), ) } if repoDID == "" { return nil, engine.ClassifiedFailure( engine.FailureClassInfrastructure, engine.FailureReasonRuntimeFailed, fmt.Errorf("missing repository DID in pipeline trigger metadata"), ) } swf.OwnerDID = ownerDID swf.RepoDID = repoDID
swf.Name = twf.Name swf.Environment = dwf.Environment caches, err := models.ParseCacheEntries(twf.Raw) if err != nil { return nil, err } swf.Caches = caches addl.image = workflowImage(dwf.Dependencies, e.cfg.NixeryPipelines.Nixery)
if sock := e.cfg.Server.DockerSocket; sock != "" { addl.mounts = append(addl.mounts, mount.Mount{ Type: mount.TypeBind, Source: sock, Target: sock, ReadOnly: false, }) } setup := &setupSteps{}
setup.addStep(nixConfStep()) setup.addStep(models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev)) // this step could be empty if s := dependencyStep(dwf.Dependencies); s != nil { setup.addStep(*s) }
// 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.NixeryPipelines.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 workflowImage(deps map[string][]string, nixery string) string { var dependencies string for reg, ds := range deps { if reg == "nixpkgs" { dependencies = path.Join(ds...) } }
// load defaults from somewhere else dependencies = path.Join(dependencies, "bash", "git", "coreutils", "gnutar", "zstd", "nix")
if runtime.GOARCH == "arm64" { dependencies = path.Join("arm64", dependencies) }
return path.Join(nixery, dependencies)}
func New(ctx context.Context, cfg *config.Config) (*Engine, error) { l := log.FromContext(ctx).With("component", "spindle")
e := &Engine{ l: l, cfg: cfg, slotter: engine.NewSemaphoreSlotter(cfg.NixeryPipelines.MaxConcurrentWorkflows), }
namespace := networkNamespace(cfg) e.seats = newSeatNetworks(namespace, func(ctx context.Context, name string) error { return e.ensureSeatNetwork(ctx, name, namespace) })
e.cleanup = make(map[string][]cleanupFunc)
return e, nil}
func (e *Engine) ensureDocker() (client.APIClient, error) { e.dockerMu.Lock() defer e.dockerMu.Unlock()
if e.docker != nil { return e.docker, nil }
dcli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) if err != nil { return nil, err } e.docker = dcli return dcli, nil}
func (e *Engine) AcquireWorkflowSlot( ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode,) (engine.WorkflowSlot, error) { if e.slotter == nil { return engine.NoopSlot{}, nil }
return e.slotter.AcquireWorkflowSlot(ctx, wid, wf, mode)}
func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) (err error) { /// -------------------------INITIAL SETUP------------------------------------------ l := e.l.With("workflow", wid) l.Info("setting up workflow")
setupStep := Step{ name: "Pull image from Nixery", kind: models.StepKindSystem, } setupStepIdx := -1
wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusStart).Write([]byte{0}) defer func() { wfLogger.EndWriter(setupStepIdx, setupStep, models.OutcomeOf(err)).Write([]byte{0}) }()
defer func() { if err != nil { err = fmt.Errorf("Failed to setup container:\n%w", err) } }()
if _, err := e.ensureDocker(); err != nil { return err }
/// -------------------------WORKFLOW NETWORK--------------------------------------- seat, releaseSeat, err := e.seats.take(ctx) if err != nil { return fmt.Errorf("claiming workflow network: %w", err) }
e.registerCleanup(wid, func(context.Context) error { releaseSeat() return nil })
/// -------------------------IMAGE PULL--------------------------------------------- addl := wf.Data.(addlFields) l.Info("pulling image", "image", addl.image) fmt.Fprintf( wfLogger.DataWriter(setupStepIdx, "stdout"), "Pulling image: %s", addl.image, )
reader, err := e.docker.ImagePull(ctx, addl.image, image.PullOptions{}) if err != nil { l.Error("pipeline image pull failed!", "error", err.Error()) fmt.Fprintf(wfLogger.DataWriter(setupStepIdx, "stderr"), "image pull failed: %s", err) return fmt.Errorf("pulling image: %w", err) } defer reader.Close()
scanner := bufio.NewScanner(reader) for scanner.Scan() { line := scanner.Text() wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte(line)) l.Info("image pull progress", "stdout", line) }
/// -------------------------CONTAINER CREATION------------------------------------- l.Info("creating container", "network", seat) wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte("Creating container..."))
extraHosts := []string{"host.docker.internal:host-gateway"} for _, h := range e.cfg.Server.DevExtraHosts { extraHosts = append(extraHosts, h+":host-gateway") }
keepaliveTimeout := e.WorkflowTimeout() + 5*time.Minute if keepaliveTimeout <= 0 { keepaliveTimeout = 2 * time.Hour }
resp, err := e.docker.ContainerCreate(ctx, &container.Config{ Image: addl.image, OpenStdin: true, // so cat stays alive :3 // unless its eating too much :( Cmd: []string{"timeout", fmt.Sprintf("%ds", int64(keepaliveTimeout.Seconds())), "cat"}, Tty: false, Hostname: "spindle", WorkingDir: workspaceDir, Labels: map[string]string{ "sh.tangled.pipeline/workflow_id": wid.String(), }, // TODO(winter): investigate whether environment variables passed here // get propagated to ContainerExec processes }, &container.HostConfig{ NetworkMode: container.NetworkMode(seat), Mounts: append([]mount.Mount{ { Type: mount.TypeTmpfs, Target: "/tmp", ReadOnly: false, TmpfsOptions: &mount.TmpfsOptions{ Mode: 0o1777, // world-writable sticky bit Options: [][]string{ {"exec"}, }, }, }, }, addl.mounts...), ReadonlyRootfs: false, CapDrop: []string{"ALL"}, CapAdd: []string{"CAP_DAC_OVERRIDE", "CAP_CHOWN", "CAP_FOWNER", "CAP_SETUID", "CAP_SETGID"}, SecurityOpt: []string{"no-new-privileges"}, ExtraHosts: extraHosts, Resources: container.Resources{ Memory: e.cfg.NixeryPipelines.MaxJobMemoryMB * 1024 * 1024, }, }, nil, nil, "") if err != nil { fmt.Fprintf( wfLogger.DataWriter(setupStepIdx, "stderr"), "container creation failed: %s", err, ) return fmt.Errorf("creating container: %w", err) }
e.registerCleanup(wid, func(ctx context.Context) error { if err := e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{}); err != nil { return fmt.Errorf("stopping container: %w", err) }
err := e.docker.ContainerRemove(ctx, resp.ID, container.RemoveOptions{ RemoveVolumes: true, RemoveLinks: false, Force: false, }) if err != nil { return fmt.Errorf("removing container: %w", err) }
return nil })
/// -------------------------CONTAINER START---------------------------------------- wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte("Starting container...")) if err := e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil { return fmt.Errorf("starting container: %w", err) }
mkExecResp, err := e.docker.ContainerExecCreate(ctx, resp.ID, container.ExecOptions{ Cmd: []string{"mkdir", "-p", workspaceDir, homeDir}, AttachStdout: true, // NOTE(winter): pretty sure this will make it so that when stdout read is done below, mkdir is done. maybe?? AttachStderr: true, // for good measure, backed up by docker/cli ("If -d is not set, attach to everything by default") }) 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()
if err := awaitExec(ctx, execResp.Reader, execResp.Close); err != nil { return err }
/// -----------------------------------FINISH--------------------------------------- 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}
// the exec stream outlives the workflow context, so a canceled or timed out// workflow has to close it or setup blocks forever holding its seatfunc awaitExec(ctx context.Context, stream io.Reader, closeStream func()) error { done := make(chan error, 1) go func() { _, err := io.ReadAll(stream) done <- err }()
select { case err := <-done: return err case <-ctx.Done(): closeStream() select { case <-done: case <-time.After(closeGrace): } return ctx.Err() }}
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) // TODO(winter): should SetupWorkflow also have secret access? // IMO yes, but probably worth thinking on. for _, s := range secrets { workflowEnvs.AddEnv(s.Key, s.Value) }
step := w.Steps[idx]
select { case <-ctx.Done(): return ctx.Err() default: }
envs := append(EnvVars(nil), workflowEnvs...) if nixStep, ok := step.(Step); ok { for k, v := range nixStep.environment { envs.AddEnv(k, v) } }
envs = append(envs, e.baseEnv()...) if sock := e.cfg.Server.DockerSocket; sock != "" { envs.AddEnv("DOCKER_HOST", fmt.Sprintf("unix://%s", sock)) }
mkExecResp, err := e.docker.ContainerExecCreate(ctx, addl.container, container.ExecOptions{ Cmd: []string{"bash", "-c", step.Command()}, AttachStdout: true, AttachStderr: true, Env: envs, }) if err != nil { return engine.ClassifiedFailure( engine.FailureClassInfrastructure, engine.FailureReasonRuntimeFailed, 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, idx) }()
select { case <-tailDone:
case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { e.l.Warn("step timed out", "step", step.Name()) } select { case <-tailDone: case <-time.After(2 * time.Second): } if errors.Is(ctx.Err(), context.DeadlineExceeded) { return engine.ErrTimedOut } return ctx.Err() }
if ctx.Err() != nil { if errors.Is(ctx.Err(), context.DeadlineExceeded) { return engine.ErrTimedOut } return ctx.Err() }
execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID) if err != nil { return engine.ClassifiedFailure( engine.FailureClassInfrastructure, engine.FailureReasonRuntimeFailed, fmt.Errorf("inspecting exec: %w", err), ) }
if execInspectResp.ExitCode != 0 { inspectResp, err := e.docker.ContainerInspect(ctx, addl.container) if err != nil { return engine.ClassifiedFailure( engine.FailureClassInfrastructure, engine.FailureReasonRuntimeFailed, fmt.Errorf("inspecting workflow container: %w", err), ) }
e.l.Error("workflow failed!", "workflow_id", wid.String(), "exit_code", execInspectResp.ExitCode, "oom_killed", inspectResp.State.OOMKilled)
if inspectResp.State.OOMKilled { return engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonResourceExhausted, fmt.Errorf("User step error:\n%w", ErrOOMKilled)) } return engine.ClassifiedFailure( engine.FailureClassUser, engine.FailureReasonCommandFailed, fmt.Errorf("User step error: exited with code %d", execInspectResp.ExitCode), ) }
return nil}
func (e *Engine) tailStep(ctx context.Context, wfLogger models.WorkflowLogger, execID string, stepIdx int) error { if wfLogger == nil { return nil }
logs, err := e.docker.ContainerExecAttach(ctx, execID, container.ExecAttachOptions{}) if err != nil { return err } defer logs.Close()
stop := context.AfterFunc(ctx, func() { logs.Close() }) defer stop()
_, err = stdcopy.StdCopy( wfLogger.DataWriter(stepIdx, "stdout"), wfLogger.DataWriter(stepIdx, "stderr"), logs.Reader, ) if err != nil && err != io.EOF && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) && !errors.Is(err, net.ErrClosed) && ctx.Err() == nil { return fmt.Errorf("failed to copy logs: %w", err) }
return nil}
func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { fns := e.drainCleanups(wid)
// a canceled or timed out workflow still owns a container ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), teardownTimeout) defer cancel()
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 (e *Engine) drainCleanups(wid models.WorkflowId) []cleanupFunc { e.cleanupMu.Lock() key := wid.String()
fns := e.cleanup[key] delete(e.cleanup, key) e.cleanupMu.Unlock()
return fns}
func (e *Engine) QuotaResources(wf *models.Workflow) quota.Resources { return quota.Resources{ quota.ResourceWorkflows: 1, quota.ResourceMemoryMiB: e.cfg.NixeryPipelines.MaxJobMemoryMB, }}