From 34e59ca1bdc2e4552bc441e25da938aed83219ae Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Mon, 9 Jun 2025 12:36:11 -0500 Subject: [PATCH 1/2] progress: fix progress writer pause and unpause to prevent panics This changes the progress printer's pause and unpause implementation to be reentrant to prevent race conditions and it also allows the status updates to be buffered when the display is paused. The previous implementation mixed the pause implementation with the finish implementation and could cause a send on closed channel panic because it could close the status channel before it had finished being used. Now, the status channel is not closed. When the display is enabled, the status channel will be forwarded to an internal channel that is used to display the updates. When the display is paused, the status channel will have the statuses buffered in memory to be sent when the progress display is resumed. The `Unpause` method has also been renamed to `Resume`. Signed-off-by: Jonathan A. Sternberg --- monitor/monitor.go | 14 ++- util/progress/printer.go | 181 +++++++++++++++++++++++++++++++-------- 2 files changed, 149 insertions(+), 46 deletions(-) diff --git a/monitor/monitor.go b/monitor/monitor.go index 09901291c..c30d5e66d 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -113,10 +113,8 @@ func (m *Monitor) Close() error { // RunMonitor provides an interactive session for running and managing containers via specified IO. func RunMonitor(ctx context.Context, invokeConfig *build.InvokeConfig, rCtx *build.ResultHandle, stdin io.ReadCloser, stdout, stderr io.WriteCloser, progress *progress.Printer) error { - if err := progress.Pause(); err != nil { - return err - } - defer progress.Unpause() + progress.Pause() + defer progress.Resume() defer stdin.Close() @@ -475,10 +473,10 @@ func printError(err error, printer *progress.Printer) error { if err == nil { return nil } - if err := printer.Pause(); err != nil { - return err - } - defer printer.Unpause() + + printer.Pause() + defer printer.Resume() + for _, s := range errdefs.Sources(err) { s.Print(os.Stderr) } diff --git a/util/progress/printer.go b/util/progress/printer.go index 482641a18..85208b7e2 100644 --- a/util/progress/printer.go +++ b/util/progress/printer.go @@ -16,12 +16,24 @@ import ( "go.opentelemetry.io/otel/metric" ) +type printerState int + +const ( + printerStateDone printerState = iota + printerStateRunning + printerStatePaused +) + type Printer struct { - status chan *client.SolveStatus + out console.File + mode progressui.DisplayMode + opt *printerOpts + + status chan *client.SolveStatus + interrupt chan interruptRequest + state printerState - ready chan struct{} done chan struct{} - paused chan struct{} closeOnce sync.Once err error @@ -54,13 +66,23 @@ func (p *Printer) IsDone() bool { } func (p *Printer) Pause() error { - p.paused = make(chan struct{}) - return p.Wait() + done := make(chan struct{}) + p.interrupt <- interruptRequest{ + desiredState: printerStatePaused, + done: done, + } + + // Need to wait for a response to confirm we have control + // of the console output. + <-done + return nil } -func (p *Printer) Unpause() { - close(p.paused) - <-p.ready +func (p *Printer) Resume() { + p.interrupt <- interruptRequest{ + desiredState: printerStateRunning, + } + // Do not care about waiting for a response. } func (p *Printer) Write(s *client.SolveStatus) { @@ -115,42 +137,114 @@ func NewPrinter(ctx context.Context, out console.File, mode progressui.DisplayMo } pw := &Printer{ - ready: make(chan struct{}), - metrics: opt.mw, + out: out, + mode: mode, + opt: opt, + status: make(chan *client.SolveStatus), + interrupt: make(chan interruptRequest), + state: printerStateRunning, + done: make(chan struct{}), } + go pw.run(ctx, d) + + return pw, nil +} + +func (p *Printer) run(ctx context.Context, d progressui.Display) { + defer close(p.done) + defer close(p.interrupt) + + var ss []*client.SolveStatus + for p.state != printerStateDone { + switch p.state { + case printerStatePaused: + ss, p.err = p.bufferDisplay(ctx, ss) + case printerStateRunning: + var warnings []client.VertexWarning + warnings, ss, p.err = p.updateDisplay(ctx, d, ss) + p.warnings = append(p.warnings, warnings...) + + d, _ = p.newDisplay() + } + } + + if p.opt.onclose != nil { + p.opt.onclose() + } +} + +func (p *Printer) newDisplay() (progressui.Display, error) { + return progressui.NewDisplay(p.out, p.mode, p.opt.displayOpts...) +} + +func (p *Printer) updateDisplay(ctx context.Context, d progressui.Display, ss []*client.SolveStatus) ([]client.VertexWarning, []*client.SolveStatus, error) { + p.logMu.Lock() + p.logSourceMap = map[digest.Digest]any{} + p.logMu.Unlock() + + resumeLogs := logutil.Pause(logrus.StandardLogger()) + defer resumeLogs() + + interruptCh := make(chan interruptRequest, 1) + ingress := make(chan *client.SolveStatus) + go func() { + defer close(ingress) + defer close(interruptCh) + + for _, s := range ss { + ingress <- s + } + for { - pw.status = make(chan *client.SolveStatus) - pw.done = make(chan struct{}) - pw.closeOnce = sync.Once{} - - pw.logMu.Lock() - pw.logSourceMap = map[digest.Digest]any{} - pw.logMu.Unlock() - - resumeLogs := logutil.Pause(logrus.StandardLogger()) - close(pw.ready) - // not using shared context to not disrupt display but let is finish reporting errors - pw.warnings, pw.err = d.UpdateFrom(ctx, pw.status) - resumeLogs() - close(pw.done) - - if opt.onclose != nil { - opt.onclose() + select { + case s, ok := <-p.status: + if !ok { + return + } + ingress <- s + case req := <-p.interrupt: + interruptCh <- req + return + case <-ctx.Done(): + return } - if pw.paused == nil { - break - } - - pw.ready = make(chan struct{}) - <-pw.paused - pw.paused = nil - - d, _ = progressui.NewDisplay(out, mode, opt.displayOpts...) } }() - <-pw.ready - return pw, nil + + warnings, err := d.UpdateFrom(context.Background(), ingress) + if err == nil { + err = context.Cause(ctx) + } + + interrupt := <-interruptCh + p.state = interrupt.desiredState + interrupt.close() + return warnings, nil, err +} + +// bufferDisplay will buffer display updates from the status channel into a +// slice. +// +// This method returns if either status gets closed or if an interrupt is received. +func (p *Printer) bufferDisplay(ctx context.Context, ss []*client.SolveStatus) ([]*client.SolveStatus, error) { + for { + select { + case s, ok := <-p.status: + if !ok { + p.state = printerStateDone + return ss, nil + } + ss = append(ss, s) + case req := <-p.interrupt: + p.state = req.desiredState + req.close() + return ss, nil + case <-ctx.Done(): + p.state = printerStateDone + return nil, context.Cause(ctx) + } + } } func (p *Printer) WriteBuildRef(target string, ref string) { @@ -221,3 +315,14 @@ func dedupWarnings(inp []client.VertexWarning) []client.VertexWarning { } return res } + +type interruptRequest struct { + desiredState printerState + done chan<- struct{} +} + +func (req *interruptRequest) close() { + if req.done != nil { + close(req.done) + } +} From 38cf84346c18f837cabfa663395e92b20682f3d0 Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Tue, 10 Jun 2025 11:48:41 -0500 Subject: [PATCH 2/2] build: change build handler to evaluate instead of onresult This changes the build handler to customize the behavior of evaluate rather than onresult and also simplifies the `ResultHandle`. The `ResultHandle` is now only valid within the gateway callback and can be used to start containers from the handler. `Evaluate` now executes inside of the gateway callback rather than having a separate implementation that executes or re-invokes the build. This keeps the gateway callback session open until the debugger has returned. The `ErrReload` for monitor has now been moved into the `build` package and been renamed to `ErrRestart`. This is because it restarts the build so the name makes a bit more sense. The actual use of this functionality is still tied to the monitor reload. Signed-off-by: Jonathan A. Sternberg --- build/build.go | 50 ++++++--- build/invoke.go | 26 ++--- build/result.go | 250 +++------------------------------------------ commands/build.go | 41 +------- monitor/monitor.go | 62 ++++++----- 5 files changed, 95 insertions(+), 334 deletions(-) diff --git a/build/build.go b/build/build.go index 6c20d0eb3..12006a9da 100644 --- a/build/build.go +++ b/build/build.go @@ -59,6 +59,8 @@ const ( printLintFallbackImage = "docker/dockerfile:1.8.1@sha256:e87caa74dcb7d46cd820352bfea12591f3dba3ddc4285e19c7dcd13359f7cefd" ) +var ErrRestart = errors.New("build: restart") + type Options struct { Inputs Inputs @@ -312,7 +314,7 @@ func toRepoOnly(in string) (string, error) { } type Handler struct { - OnResult func(driverIdx int, rCtx *ResultHandle) + Evaluate func(ctx context.Context, c gateway.Client, res *gateway.Result) error } 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) { @@ -479,9 +481,14 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ ch, done := progress.NewChannel(pw) defer func() { <-done }() - cc := c - var callRes map[string][]byte - buildFunc := func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { + var ( + callRes map[string][]byte + frontendErr error + ) + buildFunc := func(ctx context.Context, c gateway.Client) (_ *gateway.Result, retErr error) { + // Capture the error from this build function. + defer catchFrontendError(&retErr, &frontendErr) + if opt.CallFunc != nil { if _, ok := req.FrontendOpt["frontend.caps"]; !ok { req.FrontendOpt["frontend.caps"] = "moby.buildkit.frontend.subrequests+forward" @@ -504,24 +511,25 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ results.Set(rKey, res) if children := childTargets[rKey]; len(children) > 0 { - if err := waitForChildren(ctx, res, results, children); err != nil { + if err := waitForChildren(ctx, bh, c, res, results, children); err != nil { + return nil, err + } + } else if bh != nil && bh.Evaluate != nil { + if err := bh.Evaluate(ctx, c, res); err != nil { return nil, err } } return res, nil } + buildRef := fmt.Sprintf("%s/%s/%s", node.Builder, node.Name, so.Ref) - var rr *client.SolveResponse - if bh != nil && bh.OnResult != nil { - var resultHandle *ResultHandle - resultHandle, rr, err = NewResultHandle(ctx, cc, *so, "buildx", buildFunc, ch) - bh.OnResult(dp.driverIndex, resultHandle) - } else { - span, ctx := tracing.StartSpan(ctx, "build") - rr, err = c.Build(ctx, *so, "buildx", buildFunc, ch) - tracing.FinishWithError(span, err) + span, ctx := tracing.StartSpan(ctx, "build") + rr, err := c.Build(ctx, *so, "buildx", buildFunc, ch) + if errors.Is(frontendErr, ErrRestart) { + err = ErrRestart } + tracing.FinishWithError(span, err) if !so.Internal && desktop.BuildBackendEnabled() && node.Driver.HistoryAPISupported(ctx) { if err != nil { @@ -1191,7 +1199,7 @@ func solve(ctx context.Context, c gateway.Client, req gateway.SolveRequest) (*ga return res, nil } -func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap.Map, children []string) error { +func waitForChildren(ctx context.Context, bh *Handler, c gateway.Client, res *gateway.Result, results *waitmap.Map, children []string) error { // wait for the child targets to register their LLB before evaluating _, err := results.Get(ctx, children...) if err != nil { @@ -1200,6 +1208,9 @@ func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap. // we need to wait until the child targets have completed before we can release eg, ctx := errgroup.WithContext(ctx) eg.Go(func() error { + if bh != nil && bh.Evaluate != nil { + return bh.Evaluate(ctx, c, res) + } return res.EachRef(func(ref gateway.Reference) error { return ref.Evaluate(ctx) }) @@ -1210,3 +1221,12 @@ func waitForChildren(ctx context.Context, res *gateway.Result, results *waitmap. }) return eg.Wait() } + +func catchFrontendError(retErr, frontendErr *error) { + *frontendErr = *retErr + if errors.Is(*retErr, ErrRestart) { + // Overwrite the sentinel error with a more user friendly message. + // This gets stored only in the return error. + *retErr = errors.New("build restarted by client") + } +} diff --git a/build/invoke.go b/build/invoke.go index 5075530b3..1b79a987c 100644 --- a/build/invoke.go +++ b/build/invoke.go @@ -56,26 +56,18 @@ type Container struct { func NewContainer(ctx context.Context, resultCtx *ResultHandle, cfg *InvokeConfig) (*Container, error) { mainCtx := ctx - ctrCh := make(chan *Container) - errCh := make(chan error) + ctrCh := make(chan *Container, 1) + errCh := make(chan error, 1) go func() { - err := resultCtx.build(func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { - ctx, cancel := context.WithCancelCause(ctx) - go func() { - <-mainCtx.Done() - cancel(errors.WithStack(context.Canceled)) - }() - - containerCfg, err := resultCtx.getContainerConfig(cfg) - if err != nil { - return nil, err - } + err := func() error { containerCtx, containerCancel := context.WithCancelCause(ctx) defer containerCancel(errors.WithStack(context.Canceled)) - bkContainer, err := c.NewContainer(containerCtx, containerCfg) + + bkContainer, err := resultCtx.NewContainer(containerCtx, cfg) if err != nil { - return nil, err + return err } + releaseCh := make(chan struct{}) container := &Container{ containerCancel: containerCancel, @@ -92,8 +84,8 @@ func NewContainer(ctx context.Context, resultCtx *ResultHandle, cfg *InvokeConfi ctrCh <- container <-container.releaseCh - return nil, bkContainer.Release(ctx) - }) + return bkContainer.Release(ctx) + }() if err != nil { errCh <- err } diff --git a/build/result.go b/build/result.go index d18f4a0b3..8e999944a 100644 --- a/build/result.go +++ b/build/result.go @@ -7,259 +7,41 @@ import ( "io" "sync" - "github.com/moby/buildkit/client" "github.com/moby/buildkit/exporter/containerimage/exptypes" gateway "github.com/moby/buildkit/frontend/gateway/client" "github.com/moby/buildkit/solver/errdefs" "github.com/moby/buildkit/solver/pb" - "github.com/moby/buildkit/solver/result" ocispecs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/pkg/errors" "github.com/sirupsen/logrus" - "golang.org/x/sync/errgroup" ) -// NewResultHandle makes a call to client.Build, additionally returning a -// opaque ResultHandle alongside the standard response and error. +// NewResultHandle stores a gateway client, gateway result, and the error from +// an evaluate call if it is present. // // This ResultHandle can be used to execute additional build steps in the same // context as the build occurred, which can allow easy debugging of build // failures and successes. // // If the returned ResultHandle is not nil, the caller must call Done() on it. -func NewResultHandle(ctx context.Context, cc *client.Client, opt client.SolveOpt, product string, buildFunc gateway.BuildFunc, ch chan *client.SolveStatus) (*ResultHandle, *client.SolveResponse, error) { - // Create a new context to wrap the original, and cancel it when the - // caller-provided context is cancelled. - // - // We derive the context from the background context so that we can forbid - // cancellation of the build request after <-done is closed (which we do - // before returning the ResultHandle). - baseCtx := ctx - ctx, cancel := context.WithCancelCause(context.Background()) - done := make(chan struct{}) - go func() { - select { - case <-baseCtx.Done(): - cancel(baseCtx.Err()) - case <-done: - // Once done is closed, we've recorded a ResultHandle, so we - // shouldn't allow cancelling the underlying build request anymore. - } - }() - - // Create a new channel to forward status messages to the original. - // - // We do this so that we can discard status messages after the main portion - // of the build is complete. This is necessary for the solve error case, - // where the original gateway is kept open until the ResultHandle is - // closed - we don't want progress messages from operations in that - // ResultHandle to display after this function exits. - // - // Additionally, callers should wait for the progress channel to be closed. - // If we keep the session open and never close the progress channel, the - // caller will likely hang. - baseCh := ch - ch = make(chan *client.SolveStatus) - go func() { - for { - s, ok := <-ch - if !ok { - return - } - select { - case <-baseCh: - // base channel is closed, discard status messages - default: - baseCh <- s - } - } - }() - defer close(baseCh) - - var resp *client.SolveResponse - var respErr error - var respHandle *ResultHandle - - go func() { - defer func() { cancel(errors.WithStack(context.Canceled)) }() // ensure no dangling processes - - var res *gateway.Result - var err error - resp, err = cc.Build(ctx, opt, product, func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { - var err error - res, err = buildFunc(ctx, c) - - if res != nil && err == nil { - // Force evaluation of the build result (otherwise, we likely - // won't get a solve error) - def, err2 := getDefinition(ctx, res) - if err2 != nil { - return nil, err2 - } - res, err = evalDefinition(ctx, c, def) - } - - if err != nil { - // Scenario 1: we failed to evaluate a node somewhere in the - // build graph. - // - // In this case, we construct a ResultHandle from this - // original Build session, and return it alongside the original - // build error. We then need to keep the gateway session open - // until the caller explicitly closes the ResultHandle. - - var se *errdefs.SolveError - if errors.As(err, &se) { - respHandle = &ResultHandle{ - done: make(chan struct{}), - solveErr: se, - gwClient: c, - gwCtx: ctx, - } - respErr = err // return original error to preserve stacktrace - close(done) - - // Block until the caller closes the ResultHandle. - select { - case <-respHandle.done: - case <-ctx.Done(): - } - } - } - return res, err - }, ch) - if respHandle != nil { - return - } - if err != nil { - // Something unexpected failed during the build, we didn't succeed, - // but we also didn't make it far enough to create a ResultHandle. - respErr = err - close(done) - return - } - - // Scenario 2: we successfully built the image with no errors. - // - // In this case, the original gateway session has now been closed - // since the Build has been completed. So, we need to create a new - // gateway session to populate the ResultHandle. To do this, we - // need to re-evaluate the target result, in this new session. This - // should be instantaneous since the result should be cached. - - def, err := getDefinition(ctx, res) - if err != nil { - respErr = err - close(done) - return - } - - // NOTE: ideally this second connection should be lazily opened - opt := opt - opt.Ref = "" - opt.Exports = nil - opt.CacheExports = nil - opt.Internal = true - _, respErr = cc.Build(ctx, opt, "buildx", func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { - res, err := evalDefinition(ctx, c, def) - if err != nil { - // This should probably not happen, since we've previously - // successfully evaluated the same result with no issues. - return nil, errors.Wrap(err, "inconsistent solve result") - } - respHandle = &ResultHandle{ - done: make(chan struct{}), - res: res, - gwClient: c, - gwCtx: ctx, - } - close(done) - - // Block until the caller closes the ResultHandle. - select { - case <-respHandle.done: - case <-ctx.Done(): - } - return nil, context.Cause(ctx) - }, nil) - if respHandle != nil { - return - } - close(done) - }() - - // Block until the other thread signals that it's completed the build. - select { - case <-done: - case <-baseCtx.Done(): - if respErr == nil { - respErr = baseCtx.Err() - } +func NewResultHandle(ctx context.Context, c gateway.Client, res *gateway.Result, err error) *ResultHandle { + rCtx := &ResultHandle{ + res: res, + gwClient: c, } - return respHandle, resp, respErr -} - -// getDefinition converts a gateway result into a collection of definitions for -// each ref in the result. -func getDefinition(ctx context.Context, res *gateway.Result) (*result.Result[*pb.Definition], error) { - return result.ConvertResult(res, func(ref gateway.Reference) (*pb.Definition, error) { - st, err := ref.ToState() - if err != nil { - return nil, err - } - def, err := st.Marshal(ctx) - if err != nil { - return nil, err - } - return def.ToPB(), nil - }) -} - -// evalDefinition performs the reverse of getDefinition, converting a -// collection of definitions into a gateway result. -func evalDefinition(ctx context.Context, c gateway.Client, defs *result.Result[*pb.Definition]) (*gateway.Result, error) { - // force evaluation of all targets in parallel - results := make(map[*pb.Definition]*gateway.Result) - resultsMu := sync.Mutex{} - eg, egCtx := errgroup.WithContext(ctx) - defs.EachRef(func(def *pb.Definition) error { - eg.Go(func() error { - res, err := c.Solve(egCtx, gateway.SolveRequest{ - Evaluate: true, - Definition: def, - }) - if err != nil { - return err - } - resultsMu.Lock() - results[def] = res - resultsMu.Unlock() - return nil - }) + if err != nil && !errors.As(err, &rCtx.solveErr) { return nil - }) - if err := eg.Wait(); err != nil { - return nil, err } - res, _ := result.ConvertResult(defs, func(def *pb.Definition) (gateway.Reference, error) { - if res, ok := results[def]; ok { - return res.Ref, nil - } - return nil, nil - }) - return res, nil + return rCtx } // ResultHandle is a build result with the client that built it. type ResultHandle struct { res *gateway.Result solveErr *errdefs.SolveError - - done chan struct{} - doneOnce sync.Once - gwClient gateway.Client - gwCtx context.Context + + doneOnce sync.Once cleanups []func() cleanupsMu sync.Mutex @@ -274,9 +56,6 @@ func (r *ResultHandle) Done() { for _, f := range cleanups { f() } - - close(r.done) - <-r.gwCtx.Done() }) } @@ -286,9 +65,12 @@ func (r *ResultHandle) registerCleanup(f func()) { r.cleanupsMu.Unlock() } -func (r *ResultHandle) build(buildFunc gateway.BuildFunc) (err error) { - _, err = buildFunc(r.gwCtx, r.gwClient) - return err +func (r *ResultHandle) NewContainer(ctx context.Context, cfg *InvokeConfig) (gateway.Container, error) { + req, err := r.getContainerConfig(cfg) + if err != nil { + return nil, err + } + return r.gwClient.NewContainer(ctx, req) } func (r *ResultHandle) getContainerConfig(cfg *InvokeConfig) (containerCfg gateway.NewContainerRequest, _ error) { diff --git a/commands/build.go b/commands/build.go index a4c4e2588..e1b8f86b6 100644 --- a/commands/build.go +++ b/commands/build.go @@ -430,23 +430,11 @@ func runBuildWithOptions(ctx context.Context, dockerCli command.Cli, opts *Build for { resp, inputs, err := RunBuild(ctx, dockerCli, opts, in, printer, &bh) if err != nil { - var be *BuildError - if errors.As(err, &be) { - retErr = err - // We can proceed to monitor - } else { - return nil, nil, errors.Wrapf(err, "failed to build") - } - } - - if m != nil { - if err := m.Run(ctx, err); err != nil { - if errors.Is(err, monitor.ErrReload) { - retErr = nil - continue - } - logrus.Warnf("failed to run monitor: %v", err) + if errors.Is(err, build.ErrRestart) { + retErr = nil + continue } + return nil, nil, errors.Wrapf(err, "failed to build") } return resp, inputs, err @@ -1229,29 +1217,10 @@ func RunBuild(ctx context.Context, dockerCli command.Cli, in *BuildOptions, inSt resp, err := build.BuildWithResultHandler(ctx, nodes, buildOptions, dockerutil.NewClient(dockerCli), confutil.NewConfig(dockerCli), progress, bh) err = wrapBuildError(err, false) if err != nil { - return nil, nil, WrapBuild(err) + return nil, nil, err } if i, ok := buildOptions[defaultTargetName]; ok { inputs = &i.Inputs } return resp[defaultTargetName], inputs, nil } - -type BuildError struct { - err error -} - -func (e *BuildError) Unwrap() error { - return e.err -} - -func (e *BuildError) Error() string { - return e.err.Error() -} - -func WrapBuild(err error) error { - if err == nil { - return nil - } - return &BuildError{err: err} -} diff --git a/monitor/monitor.go b/monitor/monitor.go index c30d5e66d..6981c5ccb 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -9,6 +9,7 @@ import ( "sync" "sync/atomic" "text/tabwriter" + "time" "github.com/containerd/console" "github.com/docker/buildx/build" @@ -18,6 +19,7 @@ import ( "github.com/docker/buildx/util/ioset" "github.com/docker/buildx/util/progress" "github.com/google/shlex" + gateway "github.com/moby/buildkit/frontend/gateway/client" "github.com/moby/buildkit/identity" "github.com/moby/buildkit/solver/errdefs" "github.com/pkg/errors" @@ -34,10 +36,6 @@ type Monitor struct { stdin *ioset.SingleForwarder stdout io.WriteCloser stderr io.WriteCloser - - res *build.ResultHandle - idx int - mu sync.Mutex } func New(cfg *build.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCloser, printer *progress.Printer) *Monitor { @@ -54,27 +52,38 @@ func New(cfg *build.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCl 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 - } - }, + Evaluate: m.Evaluate, } } -func (m *Monitor) Run(ctx context.Context, buildErr error) error { - defer m.reset() +func (m *Monitor) Evaluate(ctx context.Context, c gateway.Client, res *gateway.Result) error { + buildErr := res.EachRef(func(ref gateway.Reference) error { + return ref.Evaluate(ctx) + }) - if !m.invokeConfig.NeedsDebug(buildErr) { - return nil + if m.invokeConfig.NeedsDebug(buildErr) { + // Allow some time to ensure status updates are sent. + time.Sleep(200 * time.Millisecond) + + // Print errors before launching monitor + if err := printError(buildErr, m.printer); err != nil { + logrus.Warnf("failed to print error information: %v", err) + } + + rCtx := build.NewResultHandle(ctx, c, res, buildErr) + if monitorErr := m.Run(ctx, rCtx); monitorErr != nil { + if errors.Is(monitorErr, build.ErrRestart) { + return build.ErrRestart + } + logrus.Warnf("failed to run monitor: %v", monitorErr) + } } + return buildErr +} - // Print errors before launching monitor - if err := printError(buildErr, m.printer); err != nil { - logrus.Warnf("failed to print error information: %v", err) +func (m *Monitor) Run(ctx context.Context, rCtx *build.ResultHandle) error { + if rCtx != nil { + defer rCtx.Done() } pr, pw := io.Pipe() @@ -89,24 +98,13 @@ func (m *Monitor) Run(ctx context.Context, buildErr error) error { } defer con.Reset() - monitorErr := RunMonitor(ctx, m.invokeConfig, m.res, pr, m.stdout, m.stderr, m.printer) + monitorErr := RunMonitor(ctx, m.invokeConfig, rCtx, 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() } @@ -380,7 +378,7 @@ func (m *monitor) Detach() { } func (m *monitor) Reload() { - m.cancel(ErrReload) + m.cancel(build.ErrRestart) } func (m *monitor) AttachedPID() string {