From 38cf84346c18f837cabfa663395e92b20682f3d0 Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Tue, 10 Jun 2025 11:48:41 -0500 Subject: [PATCH] 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 {