From 8f2604b6b4ec314f7c141cc246f9024b7ef97fa5 Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Wed, 4 Jun 2025 11:33:29 -0500 Subject: [PATCH] monitor: move remaining controller functionality into monitor This creates a `Monitor` type that keeps the global state between monitor invocations and allows the monitor to exist during the build so it can be utilized for callbacks. The result handler is now registered with the monitor during the build and `Run` will use the result if it is present and the configuration intends the monitor to be invoked with the given result. Signed-off-by: Jonathan A. Sternberg --- build/build.go | 10 ++- commands/build.go | 96 +++++------------------ commands/debug/root.go | 31 -------- controller/build/build.go | 54 +++++-------- controller/local/controller.go | 96 ----------------------- controller/pb/invoke.go | 16 ++++ monitor/monitor.go | 135 +++++++++++++++++++++++++++++++-- 7 files changed, 193 insertions(+), 245 deletions(-) delete mode 100644 controller/local/controller.go diff --git a/build/build.go b/build/build.go index dbd42e907..a8f691286 100644 --- a/build/build.go +++ b/build/build.go @@ -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) diff --git a/commands/build.go b/commands/build.go index 6e8052ba2..60062a9af 100644 --- a/commands/build.go +++ b/commands/build.go @@ -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 { - case "always": - return true - case "error": - return retErr != nil - default: - return cfg.invokeFlag != "" - } -} - func (cfg *invokeConfig) parseInvokeConfig(invoke, on string) error { - cfg.onFlag = on + switch on { + case "always": + cfg.SuspendOn = controllerapi.SuspendAlways + case "error": + cfg.SuspendOn = controllerapi.SuspendError + default: + if invoke != "" { + cfg.SuspendOn = controllerapi.SuspendAlways + } + } + cfg.invokeFlag = invoke cfg.Tty = true cfg.NoCmd = true diff --git a/commands/debug/root.go b/commands/debug/root.go index e9294f022..8ddc0df1d 100644 --- a/commands/debug/root.go +++ b/commands/debug/root.go @@ -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) diff --git a/controller/build/build.go b/controller/build/build.go index 4eebaabf3..207936097 100644 --- a/controller/build/build.go +++ b/controller/build/build.go @@ -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 { diff --git a/controller/local/controller.go b/controller/local/controller.go deleted file mode 100644 index aa741e280..000000000 --- a/controller/local/controller.go +++ /dev/null @@ -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 -} diff --git a/controller/pb/invoke.go b/controller/pb/invoke.go index 76839cc96..505d2c561 100644 --- a/controller/pb/invoke.go +++ b/controller/pb/invoke.go @@ -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 } diff --git a/monitor/monitor.go b/monitor/monitor.go index f7710778a..35ef37bf5 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -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 +}