monitor: support step-by-step breakpoint debugger

This commit adds a set of commands to monitor for enabling breakpoint debugger.
This is implemented based on the walker utility for step-by-step LLB inspection.
For each vertex and breakpoint, monitor calls Solve API so the user can enter to
the debugger container on each vertex for inspection.

User can enter to the breakpoint debugger mode by --invoke=debug-step flag.

Signed-off-by: Kohei Tokunaga <ktokunaga.mail@gmail.com>

monitor/commands/attach.go

@@ -5,18 +5,25 @@ import (
 	"fmt"
 	"io"
 
+	controllererrors "github.com/docker/buildx/controller/errdefs"
+	controllerapi "github.com/docker/buildx/controller/pb"
 	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/monitor/utils"
+	"github.com/docker/buildx/util/progress"
+	solverpb "github.com/moby/buildkit/solver/pb"
 	"github.com/pkg/errors"
 )
 
 type AttachCmd struct {
 	m types.Monitor
 
-	stdout io.WriteCloser
+	stdout       io.WriteCloser
+	progress     *progress.Printer
+	invokeConfig controllerapi.InvokeConfig
 }
 
-func NewAttachCmd(m types.Monitor, stdout io.WriteCloser) types.Command {
-	return &AttachCmd{m, stdout}
+func NewAttachCmd(m types.Monitor, stdout io.WriteCloser, progress *progress.Printer, invokeConfig controllerapi.InvokeConfig) types.Command {
+	return &AttachCmd{m, stdout, progress, invokeConfig}
 }
 
 func (cm *AttachCmd) Info() types.CommandInfo {
@@ -40,7 +47,7 @@ func (cm *AttachCmd) Exec(ctx context.Context, args []string) error {
 	ref := args[1]
 	var id string
 
-	isProcess, err := isProcessID(ctx, cm.m, ref)
+	isProcess, err := utils.IsProcessID(ctx, cm.m, cm.m.AttachedSessionID(), ref)
 	if err == nil && isProcess {
 		cm.m.Attach(ctx, ref)
 		id = ref
@@ -63,23 +70,34 @@ func (cm *AttachCmd) Exec(ctx context.Context, args []string) error {
 		cm.m.Detach() // Finish existing attach
 		cm.m.AttachSession(ref)
 	}
+	if !isProcess && id != "" {
+		var walkerDef *solverpb.Definition
+		if res, err := cm.m.Inspect(ctx, id); err == nil {
+			walkerDef = res.Definition
+			if !utils.IsSameDefinition(res.Definition, res.CurrentDefinition) && res.Options != nil {
+				// Reload the current build if breakpoint debugger was ongoing on this session
+				ref, _, err := cm.m.Build(ctx, *res.Options, nil, cm.progress)
+				if err != nil {
+					var be *controllererrors.BuildError
+					if errors.As(err, &be) {
+						ref = be.Ref
+					} else {
+						return errors.Errorf("failed to reload after attach: %v", err)
+					}
+				}
+				st, err := cm.m.Inspect(ctx, ref)
+				if err != nil {
+					return err
+				}
+				walkerDef = st.Definition
+				cm.m.AttachSession(ref)
+				// rollback the running container with the new result
+				id := cm.m.Rollback(ctx, cm.invokeConfig)
+				fmt.Fprintf(cm.stdout, "Interactive container was restarted with process %q. Press Ctrl-a-c to switch to the new container", id)
+			}
+		}
+		cm.m.RegisterWalkerController(utils.NewWalkerController(cm.m, cm.stdout, cm.invokeConfig, cm.progress, walkerDef))
+	}
 	fmt.Fprintf(cm.stdout, "Attached to process %q. Press Ctrl-a-c to switch to the new container\n", id)
 	return nil
 }
-
-func isProcessID(ctx context.Context, c types.Monitor, ref string) (bool, error) {
-	sid := c.AttachedSessionID()
-	if sid == "" {
-		return false, errors.Errorf("no attaching session")
-	}
-	infos, err := c.ListProcesses(ctx, sid)
-	if err != nil {
-		return false, err
-	}
-	for _, p := range infos {
-		if p.ProcessID == ref {
-			return true, nil
-		}
-	}
-	return false, nil
-}

monitor/commands/break.go

@@ -0,0 +1,43 @@
+package commands
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/util/walker"
+	"github.com/pkg/errors"
+)
+
+type BreakCmd struct {
+	m types.Monitor
+}
+
+func NewBreakCmd(m types.Monitor) types.Command {
+	return &BreakCmd{m}
+}
+
+func (cm *BreakCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "break",
+		HelpMessage: "sets a breakpoint",
+		HelpMessageLong: `
+Usage:
+  break LINE
+
+LINE is a line number to set a breakpoint.
+`,
+	}
+}
+
+func (cm *BreakCmd) Exec(ctx context.Context, args []string) error {
+	if len(args) < 2 {
+		return errors.Errorf("break: specify line")
+	}
+	line, err := strconv.ParseInt(args[1], 10, 64)
+	if err != nil {
+		return errors.Errorf("break: invalid line number: %q: %v", args[1], err)
+	}
+	cm.m.GetWalkerController().Breakpoints().Add("", walker.NewLineBreakpoint(line))
+	return nil
+}

monitor/commands/breakpoints.go

@@ -0,0 +1,36 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/util/walker"
+)
+
+type BreakpointsCmd struct {
+	m types.Monitor
+}
+
+func NewBreakpointsCmd(m types.Monitor) types.Command {
+	return &BreakpointsCmd{m}
+}
+
+func (cm *BreakpointsCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "breakpoints",
+		HelpMessage: "lists registered breakpoints",
+		HelpMessageLong: `
+Usage:
+  breakpoints
+`,
+	}
+}
+
+func (cm *BreakpointsCmd) Exec(ctx context.Context, args []string) error {
+	cm.m.GetWalkerController().Breakpoints().ForEach(func(key string, bp walker.Breakpoint) bool {
+		fmt.Printf("%s %s\n", key, bp.String())
+		return true
+	})
+	return nil
+}

monitor/commands/clear.go

@@ -0,0 +1,38 @@
+package commands
+
+import (
+	"context"
+
+	"github.com/docker/buildx/monitor/types"
+	"github.com/pkg/errors"
+)
+
+type ClearCmd struct {
+	m types.Monitor
+}
+
+func NewClearCmd(m types.Monitor) types.Command {
+	return &ClearCmd{m}
+}
+
+func (cm *ClearCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "clear",
+		HelpMessage: "clears a breakpoint",
+		HelpMessageLong: `
+Usage:
+  clear KEY
+
+KEY is the name of the breakpoint.
+Use "breakpoints" command to list keys of the breakpoints.
+`,
+	}
+}
+
+func (cm *ClearCmd) Exec(ctx context.Context, args []string) error {
+	if len(args) < 2 {
+		return errors.Errorf("clear: specify breakpoint key")
+	}
+	cm.m.GetWalkerController().Breakpoints().Clear(args[1])
+	return nil
+}

monitor/commands/clearall.go

@@ -0,0 +1,32 @@
+package commands
+
+import (
+	"context"
+
+	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/monitor/utils"
+)
+
+type ClearallCmd struct {
+	m types.Monitor
+}
+
+func NewClearallCmd(m types.Monitor) types.Command {
+	return &ClearallCmd{m}
+}
+
+func (cm *ClearallCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "clearall",
+		HelpMessage: "clears all breakpoints",
+		HelpMessageLong: `
+Usage:
+  clearall
+`,
+	}
+}
+
+func (cm *ClearallCmd) Exec(ctx context.Context, args []string) error {
+	utils.SetDefaultBreakpoints(cm.m.GetWalkerController().Breakpoints())
+	return nil
+}

monitor/commands/continue.go

@@ -0,0 +1,39 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/buildx/monitor/types"
+)
+
+type ContinueCmd struct {
+	m types.Monitor
+}
+
+func NewContinueCmd(m types.Monitor) types.Command {
+	return &ContinueCmd{m}
+}
+
+func (cm *ContinueCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "continue",
+		HelpMessage: "resumes the build until the next breakpoint",
+		HelpMessageLong: `
+Usage:
+  continue
+`,
+	}
+}
+
+func (cm *ContinueCmd) Exec(ctx context.Context, args []string) error {
+	wc := cm.m.GetWalkerController()
+	wc.Continue()
+	if (len(args) >= 2 && args[1] == "init") || !wc.IsStarted() {
+		wc.WalkCancel() // Cancel current walking (needed especially for "init" option)
+		if err := wc.StartWalk(); err != nil {
+			fmt.Printf("failed to walk LLB: %v\n", err)
+		}
+	}
+	return nil
+}

monitor/commands/disconnect.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/monitor/utils"
 	"github.com/pkg/errors"
 )
 
@@ -36,7 +37,7 @@ func (cm *DisconnectCmd) Exec(ctx context.Context, args []string) error {
 	} else if target == "" {
 		return errors.Errorf("no attaching session")
 	}
-	isProcess, err := isProcessID(ctx, cm.m, target)
+	isProcess, err := utils.IsProcessID(ctx, cm.m, cm.m.AttachedSessionID(), target)
 	if err == nil && isProcess {
 		sid := cm.m.AttachedSessionID()
 		if sid == "" {

monitor/commands/next.go

@@ -0,0 +1,34 @@
+package commands
+
+import (
+	"context"
+
+	"github.com/docker/buildx/monitor/types"
+	"github.com/pkg/errors"
+)
+
+type NextCmd struct {
+	m types.Monitor
+}
+
+func NewNextCmd(m types.Monitor) types.Command {
+	return &NextCmd{m}
+}
+
+func (cm *NextCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "next",
+		HelpMessage: "resumes the build until the next vertex",
+		HelpMessageLong: `
+Usage:
+  next
+`,
+	}
+}
+
+func (cm *NextCmd) Exec(ctx context.Context, args []string) error {
+	if err := cm.m.GetWalkerController().Next(); err != nil {
+		return errors.Errorf("next: %s : If walker isn't runnig, might need to run \"continue\" command first", err)
+	}
+	return nil
+}

monitor/commands/reload.go

@@ -8,6 +8,7 @@ import (
 	controllererrors "github.com/docker/buildx/controller/errdefs"
 	controllerapi "github.com/docker/buildx/controller/pb"
 	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/monitor/utils"
 	"github.com/docker/buildx/util/progress"
 	"github.com/pkg/errors"
 )
@@ -58,6 +59,7 @@ func (cm *ReloadCmd) Exec(ctx context.Context, args []string) error {
 			fmt.Println("disconnect error", err)
 		}
 	}
+
 	var resultUpdated bool
 	cm.progress.Unpause()
 	ref, _, err := cm.m.Build(ctx, *bo, nil, cm.progress) // TODO: support stdin, hold build ref
@@ -74,6 +76,13 @@ func (cm *ReloadCmd) Exec(ctx context.Context, args []string) error {
 		resultUpdated = true
 	}
 	cm.m.AttachSession(ref)
+
+	st, err := cm.m.Inspect(ctx, ref)
+	if err != nil {
+		return err
+	}
+	cm.m.RegisterWalkerController(utils.NewWalkerController(cm.m, cm.stdout, cm.invokeConfig, cm.progress, st.Definition))
+
 	if resultUpdated {
 		// rollback the running container with the new result
 		id := cm.m.Rollback(ctx, cm.invokeConfig)

monitor/commands/show.go

@@ -0,0 +1,39 @@
+package commands
+
+import (
+	"context"
+	"io"
+
+	"github.com/docker/buildx/monitor/types"
+	monitorutils "github.com/docker/buildx/monitor/utils"
+	"github.com/pkg/errors"
+)
+
+type ShowCmd struct {
+	m      types.Monitor
+	stdout io.WriteCloser
+}
+
+func NewShowCmd(m types.Monitor, stdout io.WriteCloser) types.Command {
+	return &ShowCmd{m, stdout}
+}
+
+func (cm *ShowCmd) Info() types.CommandInfo {
+	return types.CommandInfo{
+		Name:        "show",
+		HelpMessage: "shows the debugging Dockerfile with breakpoint information",
+		HelpMessageLong: `
+Usage:
+  show
+`,
+	}
+}
+
+func (cm *ShowCmd) Exec(ctx context.Context, args []string) error {
+	st := cm.m.GetWalkerController().Inspect()
+	if len(st.Definition.Source.Infos) != 1 {
+		return errors.Errorf("list: multiple sources isn't supported")
+	}
+	monitorutils.PrintLines(cm.stdout, st.Definition.Source.Infos[0], st.Cursors, cm.m.GetWalkerController().Breakpoints(), 0, 0, true)
+	return nil
+}

monitor/monitor.go

@@ -14,8 +14,10 @@ import (
 	controllerapi "github.com/docker/buildx/controller/pb"
 	"github.com/docker/buildx/monitor/commands"
 	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/monitor/utils"
 	"github.com/docker/buildx/util/ioset"
 	"github.com/docker/buildx/util/progress"
+	"github.com/docker/buildx/util/walker"
 	"github.com/google/shlex"
 	"github.com/moby/buildkit/identity"
 	"github.com/pkg/errors"
@@ -85,15 +87,30 @@ func RunMonitor(ctx context.Context, curRef string, options *controllerapi.Build
 	id := m.Rollback(ctx, invokeConfig)
 	fmt.Fprintf(stdout, "Interactive container was restarted with process %q. Press Ctrl-a-c to switch to the new container\n", id)
 
+	st, err := c.Inspect(ctx, curRef)
+	if err != nil {
+		return err
+	}
+	wc := utils.NewWalkerController(m, stdout, invokeConfig, progress, st.Definition)
+	m.RegisterWalkerController(wc)
 	availableCommands := []types.Command{
 		commands.NewReloadCmd(m, stdout, progress, options, invokeConfig),
 		commands.NewRollbackCmd(m, invokeConfig, stdout),
 		commands.NewListCmd(m, stdout),
 		commands.NewDisconnectCmd(m),
 		commands.NewKillCmd(m),
-		commands.NewAttachCmd(m, stdout),
+		commands.NewAttachCmd(m, stdout, progress, invokeConfig),
 		commands.NewExecCmd(m, invokeConfig, stdout),
 		commands.NewPsCmd(m, stdout),
+
+		// breakpoint debugger
+		commands.NewBreakCmd(m),
+		commands.NewBreakpointsCmd(m),
+		commands.NewClearCmd(m),
+		commands.NewClearallCmd(m),
+		commands.NewContinueCmd(m),
+		commands.NewNextCmd(m),
+		commands.NewShowCmd(m, stdout),
 	}
 	registeredCommands := make(map[string]types.Command)
 	for _, c := range availableCommands {
@@ -233,6 +250,19 @@ type monitor struct {
 	invokeIO     *ioset.Forwarder
 	invokeCancel func()
 	attachedPid  atomic.Value
+
+	walkerController *walker.Controller
+}
+
+func (m *monitor) RegisterWalkerController(wc *walker.Controller) {
+	if m.walkerController != nil {
+		m.walkerController.Close()
+	}
+	m.walkerController = wc
+}
+
+func (m *monitor) GetWalkerController() *walker.Controller {
+	return m.walkerController
 }
 
 func (m *monitor) DisconnectSession(ctx context.Context, targetID string) error {

monitor/types/types.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/docker/buildx/controller/control"
 	controllerapi "github.com/docker/buildx/controller/pb"
+	"github.com/docker/buildx/util/walker"
 )
 
 // Monitor provides APIs for attaching and controlling the buildx server.
@@ -34,6 +35,12 @@ type Monitor interface {
 
 	// AttachedSessionID returns the ID of the attached session.
 	AttachedSessionID() string
+
+	// RegisterWalkerController registers the specified walker to the monitor.
+	RegisterWalkerController(wc *walker.Controller)
+
+	// GetWalkerController returns the currently registered walker.
+	GetWalkerController() *walker.Controller
 }
 
 // CommandInfo is information about a command.

monitor/utils/utils.go

@@ -0,0 +1,139 @@
+package utils
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/docker/buildx/controller/control"
+	controllerapi "github.com/docker/buildx/controller/pb"
+	"github.com/docker/buildx/monitor/types"
+	"github.com/docker/buildx/util/progress"
+	"github.com/docker/buildx/util/walker"
+	"github.com/moby/buildkit/client/llb"
+	solverpb "github.com/moby/buildkit/solver/pb"
+	"github.com/pkg/errors"
+)
+
+func IsProcessID(ctx context.Context, c control.BuildxController, curRef, ref string) (bool, error) {
+	infos, err := c.ListProcesses(ctx, curRef)
+	if err != nil {
+		return false, err
+	}
+	for _, p := range infos {
+		if p.ProcessID == ref {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func PrintLines(w io.Writer, source *solverpb.SourceInfo, positions []solverpb.Range, bps *walker.Breakpoints, before, after int, all bool) {
+	fmt.Fprintf(w, "Filename: %q\n", source.Filename)
+	scanner := bufio.NewScanner(bytes.NewReader(source.Data))
+	lastLinePrinted := false
+	firstPrint := true
+	for i := 1; scanner.Scan(); i++ {
+		print := false
+		target := false
+		if len(positions) == 0 {
+			print = true
+		} else {
+			for _, r := range positions {
+				if all || int(r.Start.Line)-before <= i && i <= int(r.End.Line)+after {
+					print = true
+					if int(r.Start.Line) <= i && i <= int(r.End.Line) {
+						target = true
+						break
+					}
+				}
+			}
+		}
+
+		if !print {
+			lastLinePrinted = false
+			continue
+		}
+		if !lastLinePrinted && !firstPrint {
+			fmt.Fprintln(w, "----------------")
+		}
+
+		prefix := " "
+		bps.ForEach(func(key string, b walker.Breakpoint) bool {
+			if b.IsMarked(int64(i)) {
+				prefix = "*"
+				return false
+			}
+			return true
+		})
+		prefix2 := "  "
+		if target {
+			prefix2 = "=>"
+		}
+		fmt.Fprintln(w, prefix+prefix2+fmt.Sprintf("%4d| ", i)+scanner.Text())
+		lastLinePrinted = true
+		firstPrint = false
+	}
+}
+
+func IsSameDefinition(a *solverpb.Definition, b *solverpb.Definition) bool {
+	ctx := context.TODO()
+	opA, err := llb.NewDefinitionOp(a)
+	if err != nil {
+		return false
+	}
+	dgstA, _, _, _, err := llb.NewState(opA).Output().Vertex(ctx, nil).Marshal(ctx, nil)
+	if err != nil {
+		return false
+	}
+	opB, err := llb.NewDefinitionOp(b)
+	if err != nil {
+		return false
+	}
+	dgstB, _, _, _, err := llb.NewState(opB).Output().Vertex(ctx, nil).Marshal(ctx, nil)
+	if err != nil {
+		return false
+	}
+	return dgstA.String() == dgstB.String()
+}
+
+func SetDefaultBreakpoints(bps *walker.Breakpoints) {
+	bps.ClearAll()
+	bps.Add("stopOnEntry", walker.NewStopOnEntryBreakpoint()) // always enabled
+	bps.Add("stopOnErr", walker.NewOnErrorBreakpoint())
+}
+
+func NewWalkerController(m types.Monitor, stdout io.WriteCloser, invokeConfig controllerapi.InvokeConfig, progress *progress.Printer, def *solverpb.Definition) *walker.Controller {
+	bps := walker.NewBreakpoints()
+	SetDefaultBreakpoints(bps)
+	return walker.NewController(def, bps, func(ctx context.Context, bCtx *walker.BreakContext) error {
+		var keys []string
+		for k := range bCtx.Hits {
+			keys = append(keys, k)
+		}
+		fmt.Fprintf(stdout, "Break at %+v\n", keys)
+		PrintLines(stdout, bCtx.Definition.Source.Infos[0], bCtx.Cursors, bCtx.Breakpoints, 0, 0, true)
+		m.Rollback(ctx, invokeConfig)
+		return nil
+	}, func(ctx context.Context, st llb.State) error {
+		d, err := st.Marshal(ctx)
+		if err != nil {
+			return errors.Errorf("solve: failed to marshal definition: %v", err)
+		}
+		progress.Unpause()
+		err = m.Solve(ctx, m.AttachedSessionID(), d.ToPB(), progress)
+		progress.Pause()
+		if err != nil {
+			fmt.Fprintf(stdout, "failed during walk: %v\n", err)
+		}
+		return err
+	}, func(err error) {
+		if err == nil {
+			fmt.Fprintf(stdout, "walker finished\n")
+		} else {
+			fmt.Fprintf(stdout, "walker finished with error %v\n", err)
+		}
+	})
+}