From dbda218489e735de434808cc1086dacfc70dcbce Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Mon, 11 Aug 2025 12:40:09 -0500 Subject: [PATCH] dap: make exec shell persistent across the build Invoking the shell will cause it to persist across the entire build and to re-execute whenever the builder pauses at another location again. This still requires using `exec` to launch the shell. Launching by frame id is also removed since it no longer applies to this version. Signed-off-by: Jonathan A. Sternberg --- build/result.go | 40 ++++++ dap/adapter.go | 39 +++--- dap/debug_shell.go | 308 +++++++++++++++++++++++++++++++++++++++++++++ dap/eval.go | 35 +++--- dap/thread.go | 54 +++++--- 5 files changed, 414 insertions(+), 62 deletions(-) create mode 100644 dap/debug_shell.go diff --git a/build/result.go b/build/result.go index 13fefba47..b940fce0d 100644 --- a/build/result.go +++ b/build/result.go @@ -1,10 +1,15 @@ package build import ( + "cmp" "context" _ "crypto/sha256" // ensure digests can be computed "encoding/json" "io" + iofs "io/fs" + "path/filepath" + "slices" + "strings" "sync" "github.com/moby/buildkit/exporter/containerimage/exptypes" @@ -14,6 +19,7 @@ import ( ocispecs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/pkg/errors" "github.com/sirupsen/logrus" + "github.com/tonistiigi/fsutil/types" ) // NewResultHandle stores a gateway client, gateway reference, and the error from @@ -75,6 +81,40 @@ func (r *ResultHandle) NewContainer(ctx context.Context, cfg *InvokeConfig) (gat return r.gwClient.NewContainer(ctx, req) } +func (r *ResultHandle) StatFile(ctx context.Context, fpath string, cfg *InvokeConfig) (*types.Stat, error) { + containerCfg, err := r.getContainerConfig(cfg) + if err != nil { + return nil, err + } + + candidateMounts := make([]gateway.Mount, 0, len(containerCfg.Mounts)) + for _, m := range containerCfg.Mounts { + if strings.HasPrefix(fpath, m.Dest) { + candidateMounts = append(candidateMounts, m) + } + } + if len(candidateMounts) == 0 { + return nil, iofs.ErrNotExist + } + + slices.SortFunc(candidateMounts, func(a, b gateway.Mount) int { + return cmp.Compare(len(a.Dest), len(b.Dest)) + }) + + m := candidateMounts[len(candidateMounts)-1] + relpath, err := filepath.Rel(m.Dest, fpath) + if err != nil { + return nil, err + } + + if m.Ref == nil { + return nil, iofs.ErrNotExist + } + + req := gateway.StatRequest{Path: filepath.ToSlash(relpath)} + return m.Ref.StatFile(ctx, req) +} + func (r *ResultHandle) getContainerConfig(cfg *InvokeConfig) (containerCfg gateway.NewContainerRequest, _ error) { if r.ref != nil && r.solveErr == nil { logrus.Debugf("creating container from successful build") diff --git a/dap/adapter.go b/dap/adapter.go index ec07df4bb..ade3139cc 100644 --- a/dap/adapter.go +++ b/dap/adapter.go @@ -38,9 +38,14 @@ type Adapter[C LaunchConfig] struct { threadsMu sync.RWMutex nextThreadID int + sharedState +} + +type sharedState struct { breakpointMap *breakpointMap - sourceMap sourceMap + sourceMap *sourceMap idPool *idPool + sh *shell } func New[C LaunchConfig]() *Adapter[C] { @@ -51,8 +56,12 @@ func New[C LaunchConfig]() *Adapter[C] { evaluateReqCh: make(chan *evaluateRequest), threads: make(map[int]*thread), nextThreadID: 1, - breakpointMap: newBreakpointMap(), - idPool: new(idPool), + sharedState: sharedState{ + breakpointMap: newBreakpointMap(), + sourceMap: new(sourceMap), + idPool: new(idPool), + sh: newShell(), + }, } d.srv = NewServer(d.dapHandler()) return d @@ -233,12 +242,10 @@ func (d *Adapter[C]) newThread(ctx Context, name string) (t *thread) { d.threadsMu.Lock() id := d.nextThreadID t = &thread{ - id: id, - name: name, - sourceMap: &d.sourceMap, - breakpointMap: d.breakpointMap, - idPool: d.idPool, - variables: newVariableReferences(), + id: id, + name: name, + sharedState: d.sharedState, + variables: newVariableReferences(), } d.threads[t.id] = t d.nextThreadID++ @@ -261,20 +268,6 @@ func (d *Adapter[C]) getThread(id int) (t *thread) { return t } -func (d *Adapter[C]) getFirstThread() (t *thread) { - d.threadsMu.Lock() - defer d.threadsMu.Unlock() - - for _, thread := range d.threads { - if thread.isPaused() { - if t == nil || thread.id < t.id { - t = thread - } - } - } - return t -} - func (d *Adapter[C]) deleteThread(ctx Context, t *thread) { d.threadsMu.Lock() if t := d.threads[t.id]; t != nil { diff --git a/dap/debug_shell.go b/dap/debug_shell.go new file mode 100644 index 000000000..cf7a51cd5 --- /dev/null +++ b/dap/debug_shell.go @@ -0,0 +1,308 @@ +package dap + +import ( + "context" + "fmt" + "io" + "io/fs" + "net" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/docker/buildx/build" + "github.com/docker/buildx/util/ioset" + "github.com/docker/cli/cli-plugins/metadata" + "github.com/google/go-dap" + "github.com/pkg/errors" + "golang.org/x/sync/errgroup" + "golang.org/x/sync/semaphore" +) + +type shell struct { + // SocketPath is set on the first time Init is invoked + // and stays that way. + SocketPath string + + // Locks access to the session from the debug adapter. + // Only one debug thread can access the shell at a time. + sem *semaphore.Weighted + + // Initialized once per shell and reused. + once sync.Once + err error + l net.Listener + eg *errgroup.Group + + // For the specific session. + fwd *ioset.Forwarder + connected chan struct{} + mu sync.RWMutex +} + +func newShell() *shell { + sh := &shell{ + sem: semaphore.NewWeighted(1), + } + sh.resetSession() + return sh +} + +func (s *shell) resetSession() { + s.mu.Lock() + defer s.mu.Unlock() + + s.fwd = nil + s.connected = make(chan struct{}) +} + +// Init initializes the shell for connections on the client side. +// Attach will block until the terminal has been initialized. +func (s *shell) Init() error { + return s.listen() +} + +func (s *shell) listen() error { + s.once.Do(func() { + var dir string + dir, s.err = os.MkdirTemp("", "buildx-dap-exec") + if s.err != nil { + return + } + defer func() { + if s.err != nil { + os.RemoveAll(dir) + } + }() + s.SocketPath = filepath.Join(dir, "s.sock") + + s.l, s.err = net.Listen("unix", s.SocketPath) + if s.err != nil { + return + } + + s.eg, _ = errgroup.WithContext(context.Background()) + s.eg.Go(s.acceptLoop) + }) + return s.err +} + +func (s *shell) acceptLoop() error { + for { + if err := s.accept(); err != nil { + if errors.Is(err, net.ErrClosed) { + return nil + } + return err + } + } +} + +func (s *shell) accept() error { + conn, err := s.l.Accept() + if err != nil { + return err + } + + s.mu.Lock() + defer s.mu.Unlock() + + if s.fwd != nil { + writeLine(conn, "Error: Already connected to exec instance.") + conn.Close() + return nil + } + + // Set the input of the forwarder to the connection. + s.fwd = ioset.NewForwarder() + s.fwd.SetIn(&ioset.In{ + Stdin: io.NopCloser(conn), + Stdout: conn, + Stderr: nopCloser{conn}, + }) + + close(s.connected) + writeLine(conn, "Attached to build process.") + return nil +} + +// Attach will attach the given thread to the shell. +// Only one container can attach to a shell at any given time. +// Other attaches will block until the context is canceled or it is +// able to reserve the shell for its own use. +// +// This method is intended to be called by paused threads. +func (s *shell) Attach(ctx context.Context, t *thread) { + rCtx := t.rCtx + if rCtx == nil { + return + } + + var f dap.StackFrame + if len(t.stackTrace) > 0 { + f = t.frames[t.stackTrace[0]].StackFrame + } + + cfg := &build.InvokeConfig{Tty: true} + if len(cfg.Entrypoint) == 0 && len(cfg.Cmd) == 0 { + cfg.Entrypoint = []string{"/bin/sh"} // launch shell by default + cfg.Cmd = []string{} + cfg.NoCmd = false + } + + for { + if err := s.attach(ctx, f, rCtx, cfg); err != nil { + return + } + } +} + +func (s *shell) wait(ctx context.Context) error { + s.mu.RLock() + connected := s.connected + s.mu.RUnlock() + + select { + case <-connected: + return nil + case <-ctx.Done(): + return context.Cause(ctx) + } +} + +func (s *shell) attach(ctx context.Context, f dap.StackFrame, rCtx *build.ResultHandle, cfg *build.InvokeConfig) (retErr error) { + if err := s.wait(ctx); err != nil { + return err + } + + in, out := ioset.Pipe() + defer in.Close() + defer out.Close() + + s.mu.RLock() + fwd := s.fwd + s.mu.RUnlock() + + fwd.SetOut(&out) + defer func() { + if retErr != nil { + fwd.SetOut(nil) + } + }() + + // Check if the entrypoint is executable. If it isn't, don't bother + // trying to invoke. + if reason, ok := s.canInvoke(ctx, rCtx, cfg); !ok { + writeLineF(in.Stdout, "Build container is not executable. (reason: %s)", reason) + <-ctx.Done() + return context.Cause(ctx) + } + + if err := s.sem.Acquire(ctx, 1); err != nil { + return err + } + defer s.sem.Release(1) + + ctr, err := build.NewContainer(ctx, rCtx, cfg) + if err != nil { + return err + } + defer ctr.Cancel() + + writeLineF(in.Stdout, "Running %s in build container from line %d.", + strings.Join(append(cfg.Entrypoint, cfg.Cmd...), " "), + f.Line, + ) + + writeLine(in.Stdout, "Changes to the container will be reset after the next step is executed.") + err = ctr.Exec(ctx, cfg, in.Stdin, in.Stdout, in.Stderr) + + // Send newline to properly terminate the output. + writeLine(in.Stdout, "") + + if err != nil { + return err + } + + fwd.Close() + s.resetSession() + return nil +} + +func (s *shell) canInvoke(ctx context.Context, rCtx *build.ResultHandle, cfg *build.InvokeConfig) (reason string, ok bool) { + var cmd string + if len(cfg.Entrypoint) > 0 { + cmd = cfg.Entrypoint[0] + } else if len(cfg.Cmd) > 0 { + cmd = cfg.Cmd[0] + } + + if cmd == "" { + return "no command specified", false + } + + st, err := rCtx.StatFile(ctx, cmd, cfg) + if err != nil { + return fmt.Sprintf("stat error: %s", err), false + } + + mode := fs.FileMode(st.Mode) + if !mode.IsRegular() { + return fmt.Sprintf("%s: not a file", cmd), false + } + if mode&0111 == 0 { + return fmt.Sprintf("%s: not an executable", cmd), false + } + return "", true +} + +// SendRunInTerminalRequest will send the request to the client to attach to +// the socket path that was created by Init. This is intended to be run +// from the adapter and interact directly with the client. +func (s *shell) SendRunInTerminalRequest(ctx Context) error { + // TODO: this should work in standalone mode too. + docker := os.Getenv(metadata.ReexecEnvvar) + req := &dap.RunInTerminalRequest{ + Request: dap.Request{ + Command: "runInTerminal", + }, + Arguments: dap.RunInTerminalRequestArguments{ + Kind: "integrated", + Args: []string{docker, "buildx", "dap", "attach", s.SocketPath}, + Env: map[string]any{ + "BUILDX_EXPERIMENTAL": "1", + }, + }, + } + + resp := ctx.Request(req) + if !resp.GetResponse().Success { + return errors.New(resp.GetResponse().Message) + } + return nil +} + +type nopCloser struct { + io.Writer +} + +func (nopCloser) Close() error { + return nil +} + +func writeLine(w io.Writer, msg string) { + if os.PathSeparator == '\\' { + fmt.Fprint(w, msg+"\r\n") + } else { + fmt.Fprintln(w, msg) + } +} + +func writeLineF(w io.Writer, format string, a ...any) { + if os.PathSeparator == '\\' { + fmt.Fprintf(w, format+"\r\n", a...) + } else { + fmt.Fprintf(w, format+"\n", a...) + } +} diff --git a/dap/eval.go b/dap/eval.go index a7618a59a..0072535b1 100644 --- a/dap/eval.go +++ b/dap/eval.go @@ -30,7 +30,7 @@ func (d *Adapter[C]) Evaluate(ctx Context, req *dap.EvaluateRequest, resp *dap.E } var retErr error - cmd := d.replCommands(ctx, req, resp, &retErr) + cmd := d.replCommands(ctx, resp, &retErr) cmd.SetArgs(args) cmd.SetErr(d.Out()) if err := cmd.Execute(); err != nil { @@ -42,39 +42,34 @@ func (d *Adapter[C]) Evaluate(ctx Context, req *dap.EvaluateRequest, resp *dap.E return retErr } -func (d *Adapter[C]) replCommands(ctx Context, req *dap.EvaluateRequest, resp *dap.EvaluateResponse, retErr *error) *cobra.Command { +func (d *Adapter[C]) replCommands(ctx Context, resp *dap.EvaluateResponse, retErr *error) *cobra.Command { rootCmd := &cobra.Command{ SilenceErrors: true, } - execCmd, execOpts := replCmd(ctx, "exec", resp, retErr, d.execCmd) - execCmd.PreRun = func(cmd *cobra.Command, args []string) { - execOpts.FrameID = req.Arguments.FrameId - } + execCmd, _ := replCmd(ctx, "exec", resp, retErr, d.execCmd) rootCmd.AddCommand(execCmd) return rootCmd } -type execOptions struct { - FrameID int -} +type execOptions struct{} -func (d *Adapter[C]) execCmd(ctx Context, args []string, flags execOptions) (string, error) { +func (d *Adapter[C]) execCmd(ctx Context, _ []string, _ execOptions) (string, error) { if !d.supportsExec { return "", errors.New("cannot exec without runInTerminal client capability") } - var t *thread - if flags.FrameID > 0 { - if t = d.getThreadByFrameID(flags.FrameID); t == nil { - return "", errors.Errorf("no thread with frame id %d", flags.FrameID) - } - } else { - if t = d.getFirstThread(); t == nil { - return "", errors.New("no paused thread") - } + // Initialize the shell if it hasn't been done before. This will allow any + // containers that are attempting to attach to actually attach. + if err := d.sh.Init(); err != nil { + return "", err } - return t.Exec(ctx, args) + + // Send the request to attach to the terminal. + if err := d.sh.SendRunInTerminalRequest(ctx); err != nil { + return "", err + } + return fmt.Sprintf("Started process attached to %s.", d.sh.SocketPath), nil } func replCmd[Flags any, RetVal any](ctx Context, name string, resp *dap.EvaluateResponse, retErr *error, fn func(ctx Context, args []string, flags Flags) (RetVal, error)) (*cobra.Command, *Flags) { diff --git a/dap/thread.go b/dap/thread.go index 5f87b024c..52e6bd6a6 100644 --- a/dap/thread.go +++ b/dap/thread.go @@ -22,10 +22,8 @@ type thread struct { name string // Persistent state from the adapter. - idPool *idPool - sourceMap *sourceMap - breakpointMap *breakpointMap - variables *variableReferences + sharedState + variables *variableReferences // Inputs to the evaluate call. c gateway.Client @@ -50,7 +48,7 @@ type thread struct { mu sync.Mutex // Attributes set when a thread is paused. - cancel context.CancelCauseFunc + cancel context.CancelCauseFunc // invoked when the thread is resumed rCtx *build.ResultHandle curPos digest.Digest stackTrace []int32 @@ -254,9 +252,6 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e } t.paused = make(chan stepType, 1) - if ref != nil || err != nil { - t.rCtx = build.NewResultHandle(c, t.c, ref, t.meta, err) - } if err != nil { var solveErr *errdefs.SolveError if errors.As(err, &solveErr) { @@ -270,6 +265,10 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e t.collectStackTrace(ctx, pos, ref) t.cancel = cancel + if ref != nil || err != nil { + t.prepareResultHandle(c, ref, err) + } + event.ThreadId = t.id c.C() <- &dap.StoppedEvent{ Event: dap.Event{Event: "stopped"}, @@ -278,6 +277,27 @@ func (t *thread) pause(c Context, ref gateway.Reference, err error, pos *step, e return t.paused } +func (t *thread) prepareResultHandle(c Context, ref gateway.Reference, err error) { + // 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) + + // Start the attach. Use the context we created and perform it in + // a goroutine. We aren't necessarily assuming this will actually work. + wg.Add(1) + go func() { + defer wg.Done() + t.sh.Attach(ctx, t) + }() +} + func (t *thread) Continue() { t.resume(stepContinue) } @@ -308,13 +328,6 @@ func (t *thread) resume(step stepType) { t.paused = nil } -func (t *thread) isPaused() bool { - t.mu.Lock() - defer t.mu.Unlock() - - return t.paused != nil -} - func (t *thread) StackTrace() []dap.StackFrame { t.mu.Lock() defer t.mu.Unlock() @@ -494,17 +507,20 @@ func (t *thread) solve(ctx context.Context, target digest.Digest) (gateway.Refer } func (t *thread) releaseState() { - if t.cancel != nil { - t.cancel(context.Canceled) - t.cancel = nil - } 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() }