alextab93/buildx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge pull request #3341 from jsternberg/dap-persistent-exec

Browse Source 
dap: make exec shell persistent across the build
This commit is contained in:
Tõnis Tiigi
2025-08-13 15:34:54 +03:00
committed by GitHub
parent da81bc15b3 dbda218489
commit 4f9f47deec
5 changed files with 414 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/result.go
+40
View File
@@ -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")
dap/adapter.go
+16 -23
View File
@@ -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 {
dap/debug_shell.go
+308
View File
@@ -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...)
}
}
dap/eval.go
+15 -20
View File
@@ -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) {
dap/thread.go
+35 -19
View File
@@ -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()
}