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

Merge pull request #3224 from jsternberg/evaluate-handler

Browse Source 
build: change build handler to evaluate instead of onresult
This commit is contained in:
Tõnis Tiigi
2025-06-10 11:07:31 -07:00
committed by GitHub
parent d1e22e5fc3 38cf84346c
commit 52b5d0862f
6 changed files with 244 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/build.go
+35 -15
View File
@@ -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")
}
}
build/invoke.go
+9 -17
View File
@@ -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
}
build/result.go
+16 -234
View File
@@ -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) {
commands/build.go
+5 -36
View File
@@ -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}
}
monitor/monitor.go
+36 -40
View File
@@ -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,34 +98,21 @@ 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()
}
// 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()
@@ -382,7 +378,7 @@ func (m *monitor) Detach() {
}
func (m *monitor) Reload() {
m.cancel(ErrReload)
m.cancel(build.ErrRestart)
}
func (m *monitor) AttachedPID() string {
@@ -475,10 +471,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)
}
util/progress/printer.go
+143 -38
View File
@@ -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)
}
}