package dap import ( "context" "path/filepath" "slices" "strings" "sync" "github.com/docker/buildx/build" "github.com/docker/buildx/dap/common" "github.com/google/go-dap" "github.com/moby/buildkit/client/llb" gateway "github.com/moby/buildkit/frontend/gateway/client" gwpb "github.com/moby/buildkit/frontend/gateway/pb" "github.com/moby/buildkit/solver/errdefs" "github.com/moby/buildkit/solver/pb" "github.com/opencontainers/go-digest" "github.com/pkg/errors" "github.com/tonistiigi/fsutil/types" "golang.org/x/sync/errgroup" ) type thread struct { // Persistent data. id int name string // Persistent state from the adapter. sharedState variables *variableReferences // Inputs to the evaluate call. c gateway.Client ref gateway.Reference meta map[string][]byte sourceInfoMap func(*pb.Source) *pb.Source // LLB state for the evaluate call. def *llb.Definition ops map[digest.Digest]*pb.Op head digest.Digest bps map[digest.Digest]int frames map[int32]*frame // Runtime state for the evaluate call. entrypoint *step // Controls pause. paused chan stepType mu sync.Mutex // Attributes set when a thread is paused. cancel context.CancelCauseFunc // invoked when the thread is resumed rCtx *build.ResultHandle stackTrace []int32 } type stepType int const ( stepContinue stepType = iota stepNext stepIn stepOut ) func (t *thread) Evaluate(ctx Context, c gateway.Client, headRef gateway.Reference, meta map[string][]byte, inputs build.Inputs, cfg common.Config) error { if err := t.init(ctx, c, headRef, meta, inputs); err != nil { return err } defer t.reset() var next *step action := stepContinue if cfg.StopOnEntry { // If we are stopping on entry, automatically advance to the // entrypoint. action, next = stepNext, t.entrypoint } var ( k string refs map[string]gateway.Reference err error ) for { event := t.needsDebug(next, action, err) if event.Reason != "" { select { case action = <-t.pause(ctx, k, refs, err, next, event): // do nothing here case <-ctx.Done(): return context.Cause(ctx) } } if err != nil { return err } t.setBreakpoints(ctx) if k, next, refs, err = t.seekNext(ctx, next, action); next == nil { break } } return nil } func (t *thread) init(ctx Context, c gateway.Client, ref gateway.Reference, meta map[string][]byte, inputs build.Inputs) error { t.c = c t.ref = ref t.meta = meta t.sourceInfoMap = func(s *pb.Source) *pb.Source { s = s.CloneVT() for _, sinfo := range s.Infos { // Map the filename from the source info from the frontend location to the // client location. fname := strings.Replace(sinfo.Filename, inputs.DockerfileMappingDst, inputs.DockerfileMappingSrc, 1) // Convert to an absolute path. if abspath, err := filepath.Abs(fname); err == nil { fname = abspath } sinfo.Filename = fname } return s } if err := t.getLLBState(ctx); err != nil { return err } return t.createProgram() } type step struct { // dgst holds the digest associated with this step. This is used for // breakpoint resolution. dgst digest.Digest // in holds the next target when step in is used. in *step // out holds the next target when step out is used. out *step // next holds the next target when next is used. next *step // frame will hold the stack frame associated with this step. frame *frame // parent holds the index of the parent step. parent int } func (t *thread) createProgram() error { t.frames = make(map[int32]*frame) // Create the entrypoint by using the last node. // We will build on top of that. t.entrypoint = t.createBranch(t.head, nil) return nil } func (t *thread) createBranch(dgst digest.Digest, exitpoint *step) (entrypoint *step) { // Construct the final two steps in this branch. The final steps // both point to the same line. The difference between them is one // step is before the execution of the digest and the other is // after the execution of that digest. returnpoint := &step{ in: exitpoint, next: exitpoint, out: exitpoint, parent: -1, } entrypoint = &step{ dgst: dgst, in: returnpoint, next: returnpoint, out: exitpoint, frame: t.getStackFrame(dgst, nil), parent: -1, } // The entrypoint doesn't have a source entry. Just skip this // branch. if entrypoint.frame.Source == nil { return nil } // Create a pseudo-frame and attach it to the return point. // This is mostly used for getting the correct inputs utilized // by this frame. // // We don't save this frame or assign it a unique ID as it should // never be returned. returnpoint.frame = &frame{ StackFrame: entrypoint.frame.StackFrame, op: &pb.Op{ Inputs: []*pb.Input{ {Digest: string(dgst), Index: 0}, }, }, } for { // Construct the input step for this digest based on the inputs. op := t.ops[entrypoint.dgst] if len(op.Inputs) == 0 { return entrypoint } entrypoint.parent = t.determineParent(op) for i := len(op.Inputs) - 1; i >= 0; i-- { if i == entrypoint.parent { // Skip the direct parent. continue } // When we find inputs that aren't the direct parent, // we want to add them as a step before the current step. // We have to do a few things when inserting this. // // 1. We move the digest from the old entrypoint to this node. // This is so the breakpoint happens before these inputs // are evaluated. // 2. We keep the next/out pointers the same but redirect in // to point to the new branch. // 3. The direct parent is excluded from this logic. We handle // that later. inp := op.Inputs[i] head := *entrypoint // Create the routine associated with this input. // Associate it with the entrypoint in step. head.in = t.createBranch(digest.Digest(inp.Digest), entrypoint) // If this branch is empty (signified by a nil return value) then // skip it. if head.in == nil { continue } entrypoint.dgst = "" entrypoint = &head } // If we have no direct parent, return the current entrypoint // as the beginning. if entrypoint.parent < 0 { return entrypoint } // Create a new step that refers to the direct parent. head := &step{ dgst: digest.Digest(op.Inputs[entrypoint.parent].Digest), in: entrypoint, next: entrypoint, out: entrypoint.out, parent: -1, } head.frame = t.getStackFrame(head.dgst, entrypoint) entrypoint = head } } func (t *thread) getStackFrame(dgst digest.Digest, next *step) *frame { f := &frame{ op: t.ops[dgst], } f.Id = int(t.idPool.Get()) if meta, ok := t.def.Metadata[dgst]; ok { f.setNameFromMeta(meta) } if loc, ok := t.def.Source.Locations[string(dgst)]; ok { f.fillLocation(t.def, loc, next) } t.frames[int32(f.Id)] = f return f } func (t *thread) determineParent(op *pb.Op) int { // Another section should have already checked this but // double check here just in case we forget somewhere else. // The rest of this method assumes there's at least one parent // at index zero. n := len(op.Inputs) if n == 0 { return -1 } switch op := op.Op.(type) { case *pb.Op_Exec: for _, m := range op.Exec.Mounts { if m.Dest == "/" { return int(m.Input) } } return -1 case *pb.Op_File: // Use the first input where the index is from one of the inputs. for _, action := range op.File.Actions { if input := int(action.Input); input >= 0 && input < n { return input } } // Default to having no parent. return -1 default: // Default to index zero. return 0 } } func (t *thread) reset() { t.c = nil t.ref = nil t.meta = nil t.ops = nil } func (t *thread) needsDebug(cur *step, step stepType, err error) (e dap.StoppedEventBody) { if err != nil { e.Reason = "exception" e.Description = "Encountered an error during result evaluation" } else if cur != nil { if step != stepContinue { e.Reason = "step" } else if id, ok := t.bps[cur.dgst]; ok { e.Reason = "breakpoint" e.Description = "Paused on breakpoint" e.HitBreakpointIds = []int{id} } } return } func (t *thread) pause(c Context, k string, refs map[string]gateway.Reference, err error, pos *step, event dap.StoppedEventBody) <-chan stepType { t.mu.Lock() defer t.mu.Unlock() if t.paused != nil { return t.paused } t.paused = make(chan stepType, 1) t.prepareResultHandle(c, k, refs, err) ctx, cancel := context.WithCancelCause(c) t.collectStackTrace(ctx, pos, refs) t.cancel = cancel event.ThreadId = t.id c.C() <- &dap.StoppedEvent{ Event: dap.Event{Event: "stopped"}, Body: event, } return t.paused } func (t *thread) prepareResultHandle(c Context, k string, refs map[string]gateway.Reference, err error) { var ref gateway.Reference if err == nil { var ok bool if ref, ok = refs[k]; !ok { return } } // Create a context for cancellations and make the cancel function // block on the wait group. var wg sync.WaitGroup ctx, cancel := context.WithCancelCause(c) t.cancel = func(cause error) { defer wg.Wait() cancel(cause) } t.rCtx = build.NewResultHandle(ctx, t.c, ref, t.meta, err) if err != nil { gwcaps := t.c.BuildOpts().Caps var solveErr *errdefs.SolveError // If we had a solve error and the exec filesystem capability, we can // get the filesystem mounts used in the actual build rather than only the input // mounts. if gwcaps.Supports(gwpb.CapGatewayExecFilesystem) == nil && errors.As(err, &solveErr) { if exec, ok := solveErr.Op.Op.(*pb.Op_Exec); ok { rCtx := t.rCtx getContainer := sync.OnceValues(func() (*build.Container, error) { return build.NewContainer(c, rCtx, &build.InvokeConfig{}) }) for i, m := range exec.Exec.Mounts { refs[m.Dest] = &mountReference{ getContainer: getContainer, index: i, } } } } } // Start the attach. Use the context we created and perform it in // a goroutine. We aren't necessarily assuming this will actually work. wg.Go(func() { t.sh.Attach(ctx, t) }) } func (t *thread) Continue() { t.resume(stepContinue) } func (t *thread) Next() { t.resume(stepNext) } func (t *thread) StepIn() { t.resume(stepIn) } func (t *thread) StepOut() { t.resume(stepOut) } func (t *thread) resume(step stepType) { t.mu.Lock() defer t.mu.Unlock() if t.paused == nil { return } t.releaseState() t.paused <- step close(t.paused) t.paused = nil } func (t *thread) StackTrace() []dap.StackFrame { t.mu.Lock() defer t.mu.Unlock() if t.paused == nil { // Cannot compute stack trace when not paused. // This should never happen, but protect ourself in // case it does. return []dap.StackFrame{} } frames := make([]dap.StackFrame, len(t.stackTrace)) for i, id := range t.stackTrace { frames[i] = t.frames[id].StackFrame } return frames } func (t *thread) Scopes(frameID int) []dap.Scope { t.mu.Lock() defer t.mu.Unlock() frame := t.frames[int32(frameID)] return frame.Scopes() } func (t *thread) Variables(id int) []dap.Variable { return t.variables.Get(id) } func (t *thread) getLLBState(ctx Context) error { st, err := t.ref.ToState() if err != nil { return err } t.def, err = st.Marshal(ctx) if err != nil { return err } if t.sourceInfoMap != nil { t.def.Source = t.sourceInfoMap(t.def.Source) } for _, src := range t.def.Source.Infos { t.sourceMap.Put(ctx, src.Filename, src.Data) } t.ops = make(map[digest.Digest]*pb.Op, len(t.def.Def)) for _, dt := range t.def.Def { dgst := digest.FromBytes(dt) var op pb.Op if err := op.Unmarshal(dt); err != nil { return err } t.ops[dgst] = &op } t.head, err = t.def.Head() return err } func (t *thread) setBreakpoints(ctx Context) { t.bps = t.breakpointMap.Intersect(ctx, t.def.Source) } func (t *thread) seekNext(ctx Context, from *step, action stepType) (string, *step, map[string]gateway.Reference, error) { // Determine how we are going to limit the scan for the next step. var limit func(s *step) *step switch action { case stepNext: limit = func(s *step) *step { return s.next } case stepIn: limit = func(s *step) *step { return s.in } case stepOut: limit = func(s *step) *step { return s.out } } target := t.continueDigest(from, limit) return t.seek(ctx, target) } func (t *thread) seek(ctx Context, target *step) (k string, result *step, mounts map[string]gateway.Reference, err error) { k = "/" var refs map[string]gateway.Reference if target != nil { k, refs, err = t.solveInputs(ctx, target) if err != nil { return "", nil, nil, err } result = target } else { refs = map[string]gateway.Reference{"/": t.ref} } if len(refs) > 0 { if err := t.evaluateRefs(ctx, refs); err != nil { return t.rewind(ctx, err) } } return k, result, refs, nil } func (t *thread) continueDigest(from *step, limit func(*step) *step) *step { // First chance to exit early. If there's no function for limiting // the until step and no breakpoints then just go directly to the end step. if len(t.bps) == 0 && limit == nil { return nil } isBreakpoint := func(dgst digest.Digest) bool { if dgst == "" { return false } _, ok := t.bps[dgst] return ok } // Special case. When we aren't coming from any step we consider // whether the entrypoint itself is a breakpoint. If it is, we stop // there. Otherwise, we treat the entrypoint as the from location. if from == nil { if isBreakpoint(t.entrypoint.dgst) { return t.entrypoint } from = t.entrypoint } var until *step if limit != nil { until = limit(from) } // Second chance to exit early. If we've fully resolved from and the // limit function doesn't return an end step, just go directly to the end. if len(t.bps) == 0 && until == nil { return nil } next := func(s *step) *step { cur := s.in for cur != nil && cur != until { if isBreakpoint(cur.dgst) { return cur } cur = cur.in } return until } return next(from) } func (t *thread) solveInputs(ctx context.Context, target *step) (string, map[string]gateway.Reference, error) { if target == nil || target.frame.op == nil { return "", nil, nil } op := target.frame.op var root string refs := make(map[string]gateway.Reference) for i, input := range op.Inputs { k := t.determineInputName(op, i, input) if _, ok := refs[k]; ok || k == "" { continue } if i == target.parent { root = k } ref, err := t.solve(ctx, input) if err != nil { return "", nil, err } refs[k] = ref } return root, refs, nil } func (t *thread) determineInputName(op *pb.Op, index int, input *pb.Input) string { // Attempt to match the input to one of the mount destinations if we have // an exec operation. if exec, ok := op.Op.(*pb.Op_Exec); ok { for _, m := range exec.Exec.Mounts { if m.Input >= 0 && m.Input == int64(index) { return m.Dest } } } // Is our input digest a source? Use the identifier if it is. // That should give us something that is at least more user-friendly. if op := t.ops[digest.Digest(input.Digest)]; op != nil && input.Index == 0 { if source, ok := op.Op.(*pb.Op_Source); ok { return source.Source.Identifier } } // Use a default name that matches the input digest. return input.Digest } func (t *thread) evaluateRefs(ctx context.Context, refs map[string]gateway.Reference) error { eg, _ := errgroup.WithContext(ctx) for _, ref := range refs { eg.Go(func() error { return ref.Evaluate(ctx) }) } return eg.Wait() } func (t *thread) solve(ctx context.Context, input *pb.Input) (gateway.Reference, error) { if input.Digest == string(t.head) { return t.ref, nil } head := &pb.Op{ Inputs: []*pb.Input{input}, } dt, err := head.Marshal() if err != nil { return nil, err } def := t.def.ToPB() def.Def[len(def.Def)-1] = dt res, err := t.c.Solve(ctx, gateway.SolveRequest{ Definition: def, }) if err != nil { return nil, err } return res.SingleRef() } func (t *thread) releaseState() { if t.rCtx != nil { t.rCtx.Done() t.rCtx = nil } for _, f := range t.frames { f.ResetVars() } if t.cancel != nil { t.cancel(context.Canceled) t.cancel = nil } t.stackTrace = t.stackTrace[:0] t.variables.Reset() } func (t *thread) collectStackTrace(ctx context.Context, pos *step, mounts map[string]gateway.Reference) { for pos != nil { frame := pos.frame frame.ExportVars(ctx, mounts, t.variables) t.stackTrace = append(t.stackTrace, int32(frame.Id)) pos, mounts = pos.out, nil } } func (t *thread) hasFrame(id int) bool { t.mu.Lock() defer t.mu.Unlock() if t.paused == nil { return false } _, ok := t.frames[int32(id)] return ok } func (t *thread) rewind(ctx Context, inErr error) (k string, result *step, mounts map[string]gateway.Reference, retErr error) { var solveErr *errdefs.SolveError if !errors.As(inErr, &solveErr) { // If this is not a solve error, do not return the // reference and target step. return "", nil, nil, inErr } // Find the error digests we might have failed on. var digests []digest.Digest if dt, err := solveErr.Op.Marshal(); err == nil { digests = append(digests, digest.FromBytes(dt)) } // Include a version of the digest without the platform // if this is a file op. if _, ok := solveErr.Op.Op.(*pb.Op_File); ok && solveErr.Op.Platform != nil { op := solveErr.Op.CloneVT() op.Platform = nil if dt, err := op.Marshal(); err == nil { digests = append(digests, digest.FromBytes(dt)) } } if len(digests) == 0 { return "", nil, nil, inErr } // Iterate from the first step to find the one we failed on. result = t.entrypoint for result != nil && !slices.Contains(digests, result.dgst) { result = result.in } // Seek to this step. This should succeed because otherwise // we wouldn't have been able to even fail on it to begin with. k, result, mounts, retErr = t.seek(ctx, result) if retErr != nil { return k, result, mounts, retErr } return k, result, mounts, inErr } type mountReference struct { getContainer func() (*build.Container, error) index int } func (r *mountReference) ToState() (llb.State, error) { return llb.State{}, errors.New("unimplemented, cannot use ToState with mount reference") } func (r *mountReference) Evaluate(ctx context.Context) error { return nil } func (r *mountReference) ReadFile(ctx context.Context, req gateway.ReadRequest) ([]byte, error) { ctr, err := r.getContainer() if err != nil { return nil, err } return ctr.ReadFile(ctx, gateway.ReadContainerRequest{ ReadRequest: req, MountIndex: r.index, }) } func (r *mountReference) StatFile(ctx context.Context, req gateway.StatRequest) (*types.Stat, error) { ctr, err := r.getContainer() if err != nil { return nil, err } return ctr.StatFile(ctx, gateway.StatContainerRequest{ StatRequest: req, MountIndex: r.index, }) } func (r *mountReference) ReadDir(ctx context.Context, req gateway.ReadDirRequest) ([]*types.Stat, error) { ctr, err := r.getContainer() if err != nil { return nil, err } return ctr.ReadDir(ctx, gateway.ReadDirContainerRequest{ ReadDirRequest: req, MountIndex: r.index, }) }