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 <[email protected]>

Author: jsternberg

commands/dap.go

@@ -2,12 +2,17 @@ package commands
 
 import (
 	"context"
+	"io"
+	"net"
+	"os"
 
+	"github.com/containerd/console"
 	"github.com/docker/buildx/dap"
 	"github.com/docker/buildx/dap/common"
 	"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"
@@ -24,6 +29,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
 }
 
@@ -71,3 +78,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

@@ -30,6 +30,7 @@ type Adapter[C LaunchConfig] struct {
 	initialized   chan struct{}
 	started       chan launchResponse[C]
 	configuration chan struct{}
+	supportsExec  bool
 
 	evaluateReqCh chan *evaluateRequest
 
@@ -104,6 +105,9 @@ func (d *Adapter[C]) Stop() error {
 func (d *Adapter[C]) 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
@@ -262,6 +266,20 @@ 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 {
@@ -454,6 +472,7 @@ func (d *Adapter[C]) dapHandler() Handler {
 		StackTrace:        d.StackTrace,
 		Scopes:            d.Scopes,
 		Variables:         d.Variables,
+		Evaluate:          d.Evaluate,
 		Source:            d.Source,
 	}
 }

dap/eval.go

@@ -0,0 +1,153 @@
+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"
+	"github.com/spf13/cobra"
+)
+
+func (d *Adapter[C]) 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
+	}
+
+	var t *thread
+	if req.Arguments.FrameId > 0 {
+		if t = d.getThreadByFrameID(req.Arguments.FrameId); t == nil {
+			return errors.Errorf("no thread with frame id %d", req.Arguments.FrameId)
+		}
+	} else {
+		if t = d.getFirstThread(); t == nil {
+			return errors.New("no paused thread")
+		}
+	}
+
+	cmd := d.replCommands(ctx, t, resp)
+	cmd.SetArgs(args)
+	cmd.SetErr(d.Out())
+	if err := cmd.Execute(); err != nil {
+		fmt.Fprintf(d.Out(), "ERROR: %+v\n", err)
+	}
+	return nil
+}
+
+func (d *Adapter[C]) replCommands(ctx Context, t *thread, resp *dap.EvaluateResponse) *cobra.Command {
+	rootCmd := &cobra.Command{}
+
+	execCmd := &cobra.Command{
+		Use: "exec",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if !d.supportsExec {
+				return errors.New("cannot exec without runInTerminal client capability")
+			}
+			return t.Exec(ctx, args, resp)
+		},
+	}
+	rootCmd.AddCommand(execCmd)
+	return rootCmd
+}
+
+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) {