dap: support evaluate request to invoke a container

Supports using the `evaluate` request in REPL mode to start a container
with the `exec` command. Presently doesn't support any arguments.

This improves the dap server so it is capable of sending reverse
requests and receiving the response. It also adds a hidden command
`dap attach` that attaches to the socket created by `evaluate`.

This requires the client to support `runInTerminal`.

Likely needs some additional work to make sure resources are cleaned up
cleanly especially when the build is unpaused or terminated, but it
should work as a decent base.

Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com>

Author: jsternberg

commands/dap.go

@@ -3,12 +3,15 @@ package commands
 import (
 	"context"
 	"io"
+	"net"
+	"os"
 
 	"github.com/containerd/console"
 	"github.com/docker/buildx/dap"
 	"github.com/docker/buildx/util/cobrautil"
 	"github.com/docker/buildx/util/ioset"
 	"github.com/docker/buildx/util/progress"
+	"github.com/docker/cli/cli"
 	"github.com/docker/cli/cli/command"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
@@ -28,6 +31,7 @@ func dapCmd(dockerCli command.Cli, rootOpts *rootOptions) *cobra.Command {
 	cobrautil.MarkFlagsExperimental(flags, "on")
 
 	cmd.AddCommand(buildCmd(dockerCli, rootOpts, &options))
+	cmd.AddCommand(dapAttachCmd())
 	return cmd
 }
 
@@ -92,3 +96,39 @@ func (fakeConsole) Fd() uintptr {
 func (fakeConsole) Name() string {
 	return ""
 }
+
+func dapAttachCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:    "attach PATH",
+		Short:  "Attach to a container created by the dap evaluate request",
+		Args:   cli.ExactArgs(1),
+		Hidden: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			c, err := console.ConsoleFromFile(os.Stdout)
+			if err != nil {
+				return err
+			}
+
+			if err := c.SetRaw(); err != nil {
+				return err
+			}
+
+			conn, err := net.Dial("unix", args[0])
+			if err != nil {
+				return err
+			}
+
+			fwd := ioset.NewSingleForwarder()
+			fwd.SetReader(os.Stdin)
+			fwd.SetWriter(conn, func() io.WriteCloser {
+				return conn
+			})
+
+			if _, err := io.Copy(os.Stdout, conn); err != nil && !errors.Is(err, io.EOF) {
+				return err
+			}
+			return nil
+		},
+	}
+	return cmd
+}

dap/adapter.go

@@ -27,6 +27,8 @@ type Adapter struct {
 	threads      map[int]*thread
 	threadsMu    sync.RWMutex
 	nextThreadID int
+
+	supportsExec bool
 }
 
 func New(cfg *build.InvokeConfig) *Adapter {
@@ -86,6 +88,9 @@ func (d *Adapter) Stop() error {
 func (d *Adapter) Initialize(c Context, req *dap.InitializeRequest, resp *dap.InitializeResponse) error {
 	close(d.initialized)
 
+	// Set parameters based on passed client capabilities.
+	d.supportsExec = req.Arguments.SupportsRunInTerminalRequest
+
 	// Set capabilities.
 	resp.Body.SupportsConfigurationDoneRequest = true
 	return nil
@@ -202,6 +207,20 @@ func (d *Adapter) getThread(id int) (t *thread) {
 	return t
 }
 
+func (d *Adapter) getCurrentThread() (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) deleteThread(ctx Context, t *thread) {
 	d.threadsMu.Lock()
 	delete(d.threads, t.id)
@@ -325,6 +344,7 @@ func (d *Adapter) dapHandler() Handler {
 		ConfigurationDone: d.ConfigurationDone,
 		Threads:           d.Threads,
 		StackTrace:        d.StackTrace,
+		Evaluate:          d.Evaluate,
 	}
 }
 
@@ -379,6 +399,13 @@ func (t *thread) pause(c Context, rCtx *build.ResultHandle, reason, desc string)
 	return t.paused
 }
 
+func (t *thread) isPaused() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	return t.paused != nil
+}
+
 func (t *thread) Resume(c Context) {
 	t.mu.Lock()
 	defer t.mu.Unlock()

dap/eval.go

@@ -0,0 +1,137 @@
+package dap
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+
+	"github.com/docker/buildx/build"
+	"github.com/docker/cli/cli-plugins/metadata"
+	"github.com/google/go-dap"
+	"github.com/google/shlex"
+	"github.com/pkg/errors"
+)
+
+func (d *Adapter) Evaluate(ctx Context, req *dap.EvaluateRequest, resp *dap.EvaluateResponse) error {
+	if req.Arguments.Context != "repl" {
+		return errors.Errorf("unsupported evaluate context: %s", req.Arguments.Context)
+	}
+
+	args, err := shlex.Split(req.Arguments.Expression)
+	if err != nil {
+		return errors.Wrapf(err, "cannot parse expression")
+	}
+
+	if len(args) == 0 {
+		return nil
+	}
+
+	switch arg0 := args[0]; arg0 {
+	case "exec":
+		if !d.supportsExec {
+			return errors.New("dap client does not support runInTerminalRequest")
+		}
+
+		t := d.getCurrentThread()
+		if t == nil {
+			return errors.New("no paused thread for exec command")
+		}
+
+		argv := args[1:]
+		if err := t.Exec(ctx, argv, resp); err != nil {
+			return err
+		}
+		return nil
+	default:
+		return errors.Errorf("unknown evalute command: %q", arg0)
+	}
+}
+
+func (t *thread) Exec(ctx Context, args []string, eresp *dap.EvaluateResponse) (retErr error) {
+	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
+	}
+
+	ctr, err := build.NewContainer(ctx, t.rCtx, cfg)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if retErr != nil {
+			ctr.Cancel()
+		}
+	}()
+
+	dir, err := os.MkdirTemp("", "buildx-dap-exec")
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if retErr != nil {
+			os.RemoveAll(dir)
+		}
+	}()
+
+	socketPath := filepath.Join(dir, "s.sock")
+	l, err := net.Listen("unix", socketPath)
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		defer os.RemoveAll(dir)
+		t.runExec(l, ctr, cfg)
+	}()
+
+	// 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", socketPath},
+			Env: map[string]any{
+				"BUILDX_EXPERIMENTAL": "1",
+			},
+		},
+	}
+
+	resp := ctx.Request(req)
+	if !resp.GetResponse().Success {
+		return errors.New(resp.GetResponse().Message)
+	}
+
+	eresp.Body.Result = fmt.Sprintf("Started process attached to %s.", socketPath)
+	return nil
+}
+
+func (t *thread) runExec(l net.Listener, ctr *build.Container, cfg *build.InvokeConfig) {
+	defer l.Close()
+	defer ctr.Cancel()
+
+	conn, err := l.Accept()
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+
+	// start a background goroutine to politely refuse any subsequent connections.
+	go func() {
+		for {
+			conn, err := l.Accept()
+			if err != nil {
+				return
+			}
+			fmt.Fprint(conn, "Error: Already connected to exec instance.")
+			conn.Close()
+		}
+	}()
+	ctr.Exec(context.Background(), cfg, conn, conn, conn)
+}

dap/handler.go

@@ -12,6 +12,7 @@ type Context interface {
 	context.Context
 	C() chan<- dap.Message
 	Go(f func(c Context)) bool
+	Request(req dap.RequestMessage) dap.ResponseMessage
 }
 
 type dispatchContext struct {
@@ -28,6 +29,14 @@ func (c *dispatchContext) Go(f func(c Context)) bool {
 	return c.srv.Go(f)
 }
 
+func (c *dispatchContext) Request(req dap.RequestMessage) dap.ResponseMessage {
+	respCh := make(chan dap.ResponseMessage, 1)
+	c.srv.doRequest(c, req, func(c Context, resp dap.ResponseMessage) {
+		respCh <- resp
+	})
+	return <-respCh
+}
+
 type HandlerFunc[Req dap.RequestMessage, Resp dap.ResponseMessage] func(c Context, req Req, resp Resp) error
 
 func (h HandlerFunc[Req, Resp]) Do(c Context, req Req) (resp Resp, err error) {

dap/server.go

@@ -3,13 +3,16 @@ package dap
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 
 	"github.com/docker/buildx/build"
 	"github.com/google/go-dap"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
+type RequestCallback func(c Context, resp dap.ResponseMessage)
+
 type Server struct {
 	h Handler
 
@@ -20,6 +23,8 @@ type Server struct {
 	ctx    context.Context
 	cancel context.CancelCauseFunc
 
+	seq         atomic.Int64
+	requests    sync.Map
 	initialized bool
 }
 
@@ -72,14 +77,18 @@ func (s *Server) readLoop(conn Conn) error {
 
 		switch m := m.(type) {
 		case dap.RequestMessage:
-			if ok := s.dispatch(m); !ok {
+			if ok := s.dispatchRequest(m); !ok {
+				return nil
+			}
+		case dap.ResponseMessage:
+			if ok := s.dispatchResponse(m); !ok {
 				return nil
 			}
 		}
 	}
 }
 
-func (s *Server) dispatch(m dap.RequestMessage) bool {
+func (s *Server) dispatchRequest(m dap.RequestMessage) bool {
 	fn := func(c Context) {
 		rmsg, err := s.handleMessage(c, m)
 		if err != nil {
@@ -94,6 +103,19 @@ func (s *Server) dispatch(m dap.RequestMessage) bool {
 	return s.Go(fn)
 }
 
+func (s *Server) dispatchResponse(m dap.ResponseMessage) bool {
+	fn := func(c Context) {
+		reqID := m.GetResponse().RequestSeq
+		if v, loaded := s.requests.LoadAndDelete(reqID); loaded {
+			callback := v.(RequestCallback)
+			s.Go(func(c Context) {
+				callback(c, m)
+			})
+		}
+	}
+	return s.Go(fn)
+}
+
 func (s *Server) handleMessage(c Context, m dap.Message) (dap.ResponseMessage, error) {
 	switch req := m.(type) {
 	case *dap.InitializeRequest:
@@ -143,20 +165,24 @@ func (s *Server) handleInitialize(c Context, req *dap.InitializeRequest) (*dap.I
 }
 
 func (s *Server) writeLoop(conn Conn, respCh <-chan dap.Message) error {
-	var seq int
 	for m := range respCh {
 		switch m := m.(type) {
 		case dap.RequestMessage:
-			m.GetRequest().Seq = seq
+			if req := m.GetRequest(); req.Seq == 0 {
+				req.Seq = int(s.seq.Add(1))
+			}
 			m.GetRequest().Type = "request"
 		case dap.EventMessage:
-			m.GetEvent().Seq = seq
+			if event := m.GetEvent(); event.Seq == 0 {
+				event.Seq = int(s.seq.Add(1))
+			}
 			m.GetEvent().Type = "event"
 		case dap.ResponseMessage:
-			m.GetResponse().Seq = seq
+			if resp := m.GetResponse(); resp.Seq == 0 {
+				resp.Seq = int(s.seq.Add(1))
+			}
 			m.GetResponse().Type = "response"
 		}
-		seq++
 
 		if err := conn.SendMsg(m); err != nil {
 			return err
@@ -196,6 +222,12 @@ func (s *Server) Go(fn func(c Context)) bool {
 	return <-started
 }
 
+func (s *Server) doRequest(c Context, req dap.RequestMessage, callback RequestCallback) {
+	req.GetRequest().Seq = int(s.seq.Add(1))
+	s.requests.Store(req.GetRequest().Seq, callback)
+	c.C() <- req
+}
+
 func (s *Server) Stop() {
 	s.mu.Lock()
 	s.ch = nil