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

@@ -2,11 +2,19 @@ package commands
 
 import (
 	"context"
+<<<<<<< HEAD
+=======
+	"io"
+	"net"
+	"os"
+>>>>>>> cc003efc (dap: support evaluate request to invoke a container)
 
+	"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 +36,8 @@ func dapCmd(dockerCli command.Cli, rootOpts *rootOptions) *cobra.Command {
 	dapBuildCmd := buildCmd(dockerCli, rootOpts, &options)
 	dapBuildCmd.Args = cobra.RangeArgs(0, 1)
 	cmd.AddCommand(dapBuildCmd)
+
+	cmd.AddCommand(dapAttachCmd())
 	return cmd
 }
 
@@ -79,3 +89,39 @@ func (d *adapterProtocolDebugger) Stop() error {
 	defer d.conn.Close()
 	return d.Adapter.Stop()
 }
+
+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

@@ -28,6 +28,8 @@ type Adapter[T any] struct {
 	threads      map[int]*thread
 	threadsMu    sync.RWMutex
 	nextThreadID int
+
+	supportsExec bool
 }
 
 func New[T any](cfg *build.InvokeConfig) *Adapter[T] {
@@ -92,6 +94,9 @@ func (d *Adapter[T]) Stop() error {
 func (d *Adapter[T]) 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
@@ -221,6 +226,20 @@ func (d *Adapter[T]) getThread(id int) (t *thread) {
 	return t
 }
 
+func (d *Adapter[T]) 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[T]) deleteThread(ctx Context, t *thread) {
 	d.threadsMu.Lock()
 	delete(d.threads, t.id)
@@ -346,6 +365,7 @@ func (d *Adapter[T]) dapHandler() Handler {
 		Disconnect:        d.Disconnect,
 		Threads:           d.Threads,
 		StackTrace:        d.StackTrace,
+		Evaluate:          d.Evaluate,
 	}
 }
 
@@ -422,6 +442,13 @@ func (t *thread) Resume(c Context) {
 	t.paused = nil
 }
 
+func (t *thread[T]) isPaused() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	return t.paused != nil
+}
+
 // TODO: return a suitable stack frame for the thread.
 // For now, just returns nothing.
 func (t *thread) StackFrames() []dap.StackFrame {

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[T]) 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[T]) 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[T]) 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) {