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

Merge pull request #3220 from jsternberg/monitor-driven-build

Browse Source 
monitor: move remaining controller functionality into monitor
This commit is contained in:
Tõnis Tiigi
2025-06-04 13:50:24 -07:00
committed by GitHub
parent bb5b5e37e8 8f2604b6b4
commit 60b1eda2df
7 changed files with 193 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/build.go
+7 -3
View File
@@ -311,11 +311,15 @@ func toRepoOnly(in string) (string, error) {
return strings.Join(out, ","), nil
}
type Handler struct {
OnResult func(driverIdx int, rCtx *ResultHandle)
}
func Build(ctx context.Context, nodes []builder.Node, opts map[string]Options, docker *dockerutil.Client, cfg *confutil.Config, w progress.Writer) (resp map[string]*client.SolveResponse, err error) {
return BuildWithResultHandler(ctx, nodes, opts, docker, cfg, w, nil)
}
func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[string]Options, docker *dockerutil.Client, cfg *confutil.Config, w progress.Writer, resultHandleFunc func(driverIdx int, rCtx *ResultHandle)) (resp map[string]*client.SolveResponse, err error) {
func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[string]Options, docker *dockerutil.Client, cfg *confutil.Config, w progress.Writer, bh *Handler) (resp map[string]*client.SolveResponse, err error) {
if len(nodes) == 0 {
return nil, errors.Errorf("driver required for build")
}
@@ -509,10 +513,10 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[
buildRef := fmt.Sprintf("%s/%s/%s", node.Builder, node.Name, so.Ref)
var rr *client.SolveResponse
if resultHandleFunc != nil {
if bh != nil && bh.OnResult != nil {
var resultHandle *ResultHandle
resultHandle, rr, err = NewResultHandle(ctx, cc, *so, "buildx", buildFunc, ch)
resultHandleFunc(dp.driverIndex, resultHandle)
bh.OnResult(dp.driverIndex, resultHandle)
} else {
span, ctx := tracing.StartSpan(ctx, "build")
rr, err = c.Build(ctx, *so, "buildx", buildFunc, ch)
commands/build.go
+16 -70
View File
@@ -23,7 +23,6 @@ import (
"github.com/docker/buildx/commands/debug"
cbuild "github.com/docker/buildx/controller/build"
controllererrors "github.com/docker/buildx/controller/errdefs"
"github.com/docker/buildx/controller/local"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/monitor"
"github.com/docker/buildx/store"
@@ -32,7 +31,6 @@ import (
"github.com/docker/buildx/util/cobrautil"
"github.com/docker/buildx/util/confutil"
"github.com/docker/buildx/util/desktop"
"github.com/docker/buildx/util/ioset"
"github.com/docker/buildx/util/metricutil"
"github.com/docker/buildx/util/osutil"
"github.com/docker/buildx/util/progress"
@@ -418,11 +416,7 @@ func getImageID(resp map[string]string) string {
}
func runBasicBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild.Options, printer *progress.Printer) (*client.SolveResponse, *build.Inputs, error) {
resp, res, dfmap, err := cbuild.RunBuild(ctx, dockerCli, opts, dockerCli.In(), printer, false)
if res != nil {
res.Done()
}
return resp, dfmap, err
return cbuild.RunBuild(ctx, dockerCli, opts, dockerCli.In(), printer, nil)
}
func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild.Options, options buildOptions, printer *progress.Printer) (_ *client.SolveResponse, _ *build.Inputs, retErr error) {
@@ -433,33 +427,32 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild
var (
in io.ReadCloser
f *ioset.SingleForwarder
m *monitor.Monitor
bh build.Handler
)
if options.invokeConfig == nil {
in = dockerCli.In()
} else {
f = ioset.NewSingleForwarder()
f.SetReader(dockerCli.In())
m = monitor.New(&options.invokeConfig.InvokeConfig, dockerCli.In(), os.Stdout, os.Stderr, printer)
defer m.Close()
bh = m.Handler()
}
for {
c := local.NewController(ctx, dockerCli)
resp, inputs, err := c.Build(ctx, opts, in, printer)
resp, inputs, err := cbuild.RunBuild(ctx, dockerCli, opts, in, printer, &bh)
if err != nil {
var be *controllererrors.BuildError
if errors.As(err, &be) {
retErr = err
// We can proceed to monitor
} else {
c.Close()
return nil, nil, errors.Wrapf(err, "failed to build")
}
}
if options.invokeConfig != nil {
if err := runMonitorIfNeeded(ctx, options.invokeConfig, retErr, c, f, os.Stdout, os.Stderr, printer); err != nil {
c.Close()
if m != nil {
if err := m.Run(ctx, err); err != nil {
if errors.Is(err, monitor.ErrReload) {
retErr = nil
continue
@@ -468,55 +461,10 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild
}
}
c.Close()
return resp, inputs, err
}
}
func runMonitorIfNeeded(ctx context.Context, cfg *invokeConfig, retErr error, c *local.Controller, stdin *ioset.SingleForwarder, stdout io.WriteCloser, stderr console.File, printer *progress.Printer) error {
if !cfg.needsDebug(retErr) {
return nil
}
// Print errors before launching monitor
if err := printError(retErr, printer); err != nil {
logrus.Warnf("failed to print error information: %v", err)
}
pr, pw := io.Pipe()
stdin.SetWriter(pw, func() io.WriteCloser {
pw.Close() // propagate EOF
return nil
})
con := console.Current()
if err := con.SetRaw(); err != nil {
return errors.Errorf("failed to configure terminal: %v", err)
}
defer con.Reset()
monitorErr := monitor.RunMonitor(ctx, &cfg.InvokeConfig, c, pr, stdout, stderr, printer)
if err := pw.Close(); err != nil {
logrus.Debug("failed to close monitor stdin pipe reader")
}
return monitorErr
}
func printError(err error, printer *progress.Printer) error {
if err == nil {
return nil
}
if err := printer.Pause(); err != nil {
return err
}
defer printer.Unpause()
for _, s := range errdefs.Sources(err) {
s.Print(os.Stderr)
}
fmt.Fprintf(os.Stderr, "ERROR: %v\n", err)
return nil
}
func newDebuggableBuild(dockerCli command.Cli, rootOpts *rootOptions) debug.DebuggableCmd {
return &debuggableBuild{dockerCli: dockerCli, rootOpts: rootOpts}
}
@@ -973,23 +921,21 @@ func printValue(w io.Writer, printer callFunc, version string, format string, re
type invokeConfig struct {
controllerapi.InvokeConfig
onFlag string
invokeFlag string
}
func (cfg *invokeConfig) needsDebug(retErr error) bool {
switch cfg.onFlag {
func (cfg *invokeConfig) parseInvokeConfig(invoke, on string) error {
switch on {
case "always":
return true
cfg.SuspendOn = controllerapi.SuspendAlways
case "error":
return retErr != nil
cfg.SuspendOn = controllerapi.SuspendError
default:
return cfg.invokeFlag != ""
if invoke != "" {
cfg.SuspendOn = controllerapi.SuspendAlways
}
}
func (cfg *invokeConfig) parseInvokeConfig(invoke, on string) error {
cfg.onFlag = on
cfg.invokeFlag = invoke
cfg.Tty = true
cfg.NoCmd = true
commands/debug/root.go
-31
View File
@@ -1,18 +1,8 @@
package debug
import (
"context"
"os"
"github.com/containerd/console"
"github.com/docker/buildx/controller/local"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/monitor"
"github.com/docker/buildx/util/cobrautil"
"github.com/docker/buildx/util/progress"
"github.com/docker/cli/cli/command"
"github.com/moby/buildkit/util/progress/progressui"
"github.com/pkg/errors"
"github.com/spf13/cobra"
)
@@ -38,27 +28,6 @@ func RootCmd(dockerCli command.Cli, children ...DebuggableCmd) *cobra.Command {
cmd := &cobra.Command{
Use: "debug",
Short: "Start debugger",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, args []string) error {
printer, err := progress.NewPrinter(context.TODO(), os.Stderr, progressui.DisplayMode(progressMode))
if err != nil {
return err
}
ctx := context.TODO()
c := local.NewController(ctx, dockerCli)
con := console.Current()
if err := con.SetRaw(); err != nil {
return errors.Errorf("failed to configure terminal: %v", err)
}
err = monitor.RunMonitor(ctx, &controllerapi.InvokeConfig{
Tty: true,
}, c, dockerCli.In(), os.Stdout, os.Stderr, printer)
con.Reset()
return err
},
}
cobrautil.MarkCommandExperimental(cmd)
controller/build/build.go
+19 -35
View File
@@ -5,10 +5,10 @@ import (
"io"
"path/filepath"
"strings"
"sync"
"github.com/docker/buildx/build"
"github.com/docker/buildx/builder"
"github.com/docker/buildx/controller/errdefs"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/store"
"github.com/docker/buildx/store/storeutil"
@@ -34,9 +34,9 @@ const defaultTargetName = "default"
// NOTE: When an error happens during the build and this function acquires the debuggable *build.ResultHandle,
// this function returns it in addition to the error (i.e. it does "return nil, res, err"). The caller can
// inspect the result and debug the cause of that error.
func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream io.Reader, progress progress.Writer, generateResult bool) (*client.SolveResponse, *build.ResultHandle, *build.Inputs, error) {
func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream io.Reader, progress progress.Writer, bh *build.Handler) (*client.SolveResponse, *build.Inputs, error) {
if in.NoCache && len(in.NoCacheFilter) > 0 {
return nil, nil, nil, errors.Errorf("--no-cache and --no-cache-filter cannot currently be used together")
return nil, nil, errors.Errorf("--no-cache and --no-cache-filter cannot currently be used together")
}
contexts := map[string]build.NamedContext{}
@@ -70,7 +70,7 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
platforms, err := platformutil.Parse(in.Platforms)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
opts.Platforms = platforms
@@ -81,7 +81,7 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
secrets, err := controllerapi.CreateSecrets(in.Secrets)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
opts.Session = append(opts.Session, secrets)
@@ -91,13 +91,13 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
}
ssh, err := controllerapi.CreateSSH(sshSpecs)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
opts.Session = append(opts.Session, ssh)
outputs, _, err := controllerapi.CreateExports(in.Exports)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
if in.ExportPush {
var pushUsed bool
@@ -136,7 +136,7 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
annotations, err := buildflags.ParseAnnotations(in.Annotations)
if err != nil {
return nil, nil, nil, errors.Wrap(err, "parse annotations")
return nil, nil, errors.Wrap(err, "parse annotations")
}
for _, o := range outputs {
@@ -156,7 +156,7 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
allow, err := buildflags.ParseEntitlements(in.Allow)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
opts.Allow = allow
@@ -180,28 +180,27 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
builder.WithContextPathHash(contextPathHash),
)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
if err = updateLastActivity(dockerCli, b.NodeGroup); err != nil {
return nil, nil, nil, errors.Wrapf(err, "failed to update builder last activity time")
return nil, nil, errors.Wrapf(err, "failed to update builder last activity time")
}
nodes, err := b.LoadNodes(ctx)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
var inputs *build.Inputs
buildOptions := map[string]build.Options{defaultTargetName: opts}
resp, res, err := buildTargets(ctx, dockerCli, nodes, buildOptions, progress, generateResult)
resp, err := buildTargets(ctx, dockerCli, nodes, buildOptions, progress, bh)
err = wrapBuildError(err, false)
if err != nil {
// NOTE: buildTargets can return *build.ResultHandle even on error.
return nil, res, nil, err
return nil, nil, errdefs.WrapBuild(err)
}
if i, ok := buildOptions[defaultTargetName]; ok {
inputs = &i.Inputs
}
return resp, res, inputs, nil
return resp, inputs, nil
}
// buildTargets runs the specified build and returns the result.
@@ -209,27 +208,12 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *Options, inStream
// NOTE: When an error happens during the build and this function acquires the debuggable *build.ResultHandle,
// this function returns it in addition to the error (i.e. it does "return nil, res, err"). The caller can
// inspect the result and debug the cause of that error.
func buildTargets(ctx context.Context, dockerCli command.Cli, nodes []builder.Node, opts map[string]build.Options, progress progress.Writer, generateResult bool) (*client.SolveResponse, *build.ResultHandle, error) {
var res *build.ResultHandle
var resp map[string]*client.SolveResponse
var err error
if generateResult {
var mu sync.Mutex
var idx int
resp, err = build.BuildWithResultHandler(ctx, nodes, opts, dockerutil.NewClient(dockerCli), confutil.NewConfig(dockerCli), progress, func(driverIndex int, gotRes *build.ResultHandle) {
mu.Lock()
defer mu.Unlock()
if res == nil || driverIndex < idx {
idx, res = driverIndex, gotRes
}
})
} else {
resp, err = build.Build(ctx, nodes, opts, dockerutil.NewClient(dockerCli), confutil.NewConfig(dockerCli), progress)
}
func buildTargets(ctx context.Context, dockerCli command.Cli, nodes []builder.Node, opts map[string]build.Options, progress progress.Writer, bh *build.Handler) (*client.SolveResponse, error) {
resp, err := build.BuildWithResultHandler(ctx, nodes, opts, dockerutil.NewClient(dockerCli), confutil.NewConfig(dockerCli), progress, bh)
if err != nil {
return nil, res, err
return nil, err
}
return resp[defaultTargetName], res, err
return resp[defaultTargetName], err
}
func wrapBuildError(err error, bake bool) error {
controller/local/controller.go
-96
View File
@@ -1,96 +0,0 @@
package local
import (
"context"
"io"
"sync/atomic"
"github.com/docker/buildx/build"
cbuild "github.com/docker/buildx/controller/build"
controllererrors "github.com/docker/buildx/controller/errdefs"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/controller/processes"
"github.com/docker/buildx/util/ioset"
"github.com/docker/buildx/util/progress"
"github.com/docker/cli/cli/command"
"github.com/moby/buildkit/client"
"github.com/pkg/errors"
)
func NewController(ctx context.Context, dockerCli command.Cli) *Controller {
return &Controller{
dockerCli: dockerCli,
}
}
type buildConfig struct {
// TODO: these two structs should be merged
// Discussion: https://github.com/docker/buildx/pull/1640#discussion_r1113279719
resultCtx *build.ResultHandle
buildOptions *cbuild.Options
}
type Controller struct {
dockerCli command.Cli
buildConfig buildConfig
buildOnGoing atomic.Bool
}
func (b *Controller) Build(ctx context.Context, options *cbuild.Options, in io.ReadCloser, progress progress.Writer) (*client.SolveResponse, *build.Inputs, error) {
if !b.buildOnGoing.CompareAndSwap(false, true) {
return nil, nil, errors.New("build ongoing")
}
defer b.buildOnGoing.Store(false)
resp, res, dockerfileMappings, buildErr := cbuild.RunBuild(ctx, b.dockerCli, options, in, progress, true)
// NOTE: RunBuild can return *build.ResultHandle even on error.
if res != nil {
b.buildConfig = buildConfig{
resultCtx: res,
buildOptions: options,
}
if buildErr != nil {
buildErr = controllererrors.WrapBuild(buildErr)
}
}
if buildErr != nil {
return nil, nil, buildErr
}
return resp, dockerfileMappings, nil
}
func (b *Controller) Invoke(ctx context.Context, processes *processes.Manager, pid string, cfg *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error {
proc, ok := processes.Get(pid)
if !ok {
// Start a new process.
if b.buildConfig.resultCtx == nil {
return errors.New("no build result is registered")
}
var err error
proc, err = processes.StartProcess(pid, b.buildConfig.resultCtx, cfg)
if err != nil {
return err
}
}
// Attach containerIn to this process
ioCancelledCh := make(chan struct{})
proc.ForwardIO(&ioset.In{Stdin: ioIn, Stdout: ioOut, Stderr: ioErr}, func(error) { close(ioCancelledCh) })
select {
case <-ioCancelledCh:
return errors.Errorf("io cancelled")
case err := <-proc.Done():
return err
case <-ctx.Done():
return context.Cause(ctx)
}
}
func (b *Controller) Close() error {
if b.buildConfig.resultCtx != nil {
b.buildConfig.resultCtx.Done()
}
return nil
}
controller/pb/invoke.go
+16
View File
@@ -37,4 +37,20 @@ type InvokeConfig struct {
Tty bool
Rollback bool
Initial bool
SuspendOn SuspendOn
}
func (cfg *InvokeConfig) NeedsDebug(err error) bool {
return cfg.SuspendOn.DebugEnabled(err)
}
type SuspendOn int
const (
SuspendError SuspendOn = iota
SuspendAlways
)
func (s SuspendOn) DebugEnabled(err error) bool {
return err != nil || s == SuspendAlways
}
monitor/monitor.go
+130 -5
View File
@@ -4,13 +4,14 @@ import (
"context"
"fmt"
"io"
"os"
"sort"
"sync"
"sync/atomic"
"text/tabwriter"
"github.com/containerd/console"
"github.com/docker/buildx/controller/local"
"github.com/docker/buildx/build"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/controller/processes"
"github.com/docker/buildx/monitor/commands"
@@ -19,6 +20,7 @@ import (
"github.com/docker/buildx/util/progress"
"github.com/google/shlex"
"github.com/moby/buildkit/identity"
"github.com/moby/buildkit/solver/errdefs"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/term"
@@ -26,8 +28,92 @@ import (
var ErrReload = errors.New("monitor: reload")
type Monitor struct {
invokeConfig *controllerapi.InvokeConfig
printer *progress.Printer
stdin *ioset.SingleForwarder
stdout io.WriteCloser
stderr io.WriteCloser
res *build.ResultHandle
idx int
mu sync.Mutex
}
func New(cfg *controllerapi.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCloser, printer *progress.Printer) *Monitor {
m := &Monitor{
invokeConfig: cfg,
printer: printer,
stdin: ioset.NewSingleForwarder(),
stdout: stdout,
stderr: stderr,
}
m.stdin.SetReader(stdin)
return m
}
func (m *Monitor) Handler() build.Handler {
return build.Handler{
OnResult: func(driverIndex int, gotRes *build.ResultHandle) {
m.mu.Lock()
defer m.mu.Unlock()
if m.res == nil || driverIndex < m.idx {
m.idx, m.res = driverIndex, gotRes
}
},
}
}
func (m *Monitor) Run(ctx context.Context, buildErr error) error {
defer m.reset()
if !m.invokeConfig.NeedsDebug(buildErr) {
return nil
}
// Print errors before launching monitor
if err := printError(buildErr, m.printer); err != nil {
logrus.Warnf("failed to print error information: %v", err)
}
pr, pw := io.Pipe()
m.stdin.SetWriter(pw, func() io.WriteCloser {
pw.Close() // propagate EOF
return nil
})
con := console.Current()
if err := con.SetRaw(); err != nil {
return errors.Errorf("failed to configure terminal: %v", err)
}
defer con.Reset()
monitorErr := RunMonitor(ctx, m.invokeConfig, m.res, pr, m.stdout, m.stderr, m.printer)
if err := pw.Close(); err != nil {
logrus.Debug("failed to close monitor stdin pipe reader")
}
return monitorErr
}
func (m *Monitor) reset() {
m.mu.Lock()
defer m.mu.Unlock()
m.idx = 0
if m.res != nil {
m.res.Done()
m.res = nil
}
}
func (m *Monitor) Close() error {
return m.stdin.Close()
}
// RunMonitor provides an interactive session for running and managing containers via specified IO.
func RunMonitor(ctx context.Context, invokeConfig *controllerapi.InvokeConfig, c *local.Controller, stdin io.ReadCloser, stdout io.WriteCloser, stderr console.File, progress *progress.Printer) error {
func RunMonitor(ctx context.Context, invokeConfig *controllerapi.InvokeConfig, rCtx *build.ResultHandle, stdin io.ReadCloser, stdout, stderr io.WriteCloser, progress *progress.Printer) error {
if err := progress.Pause(); err != nil {
return err
}
@@ -61,7 +147,7 @@ func RunMonitor(ctx context.Context, invokeConfig *controllerapi.InvokeConfig, c
invokeForwarder := ioset.NewForwarder()
invokeForwarder.SetIn(&containerIn)
m := &monitor{
c: c,
rCtx: rCtx,
processes: processes.NewManager(),
invokeIO: invokeForwarder,
muxIO: ioset.NewMuxIO(ioset.In{
@@ -237,7 +323,7 @@ type monitor struct {
ctx context.Context
cancel context.CancelCauseFunc
c *local.Controller
rCtx *build.ResultHandle
muxIO *ioset.MuxIO
invokeIO *ioset.Forwarder
@@ -248,7 +334,31 @@ type monitor struct {
}
func (m *monitor) Invoke(ctx context.Context, pid string, cfg *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error {
return m.c.Invoke(ctx, m.processes, pid, cfg, ioIn, ioOut, ioErr)
proc, ok := m.processes.Get(pid)
if !ok {
// Start a new process.
if m.rCtx == nil {
return errors.New("no build result is registered")
}
var err error
proc, err = m.processes.StartProcess(pid, m.rCtx, cfg)
if err != nil {
return err
}
}
// Attach containerIn to this process
ioCancelledCh := make(chan struct{})
proc.ForwardIO(&ioset.In{Stdin: ioIn, Stdout: ioOut, Stderr: ioErr}, func(error) { close(ioCancelledCh) })
select {
case <-ioCancelledCh:
return errors.Errorf("io cancelled")
case err := <-proc.Done():
return err
case <-ctx.Done():
return context.Cause(ctx)
}
}
func (m *monitor) Rollback(ctx context.Context, cfg *controllerapi.InvokeConfig) string {
@@ -361,3 +471,18 @@ type nopCloser struct {
}
func (c nopCloser) Close() error { return nil }
func printError(err error, printer *progress.Printer) error {
if err == nil {
return nil
}
if err := printer.Pause(); err != nil {
return err
}
defer printer.Unpause()
for _, s := range errdefs.Sources(err) {
s.Print(os.Stderr)
}
fmt.Fprintf(os.Stderr, "ERROR: %v\n", err)
return nil
}