From e78aa98c92957b945ce9a80267f087ef34029a22 Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Fri, 30 May 2025 13:41:58 -0500 Subject: [PATCH 1/3] build: refactor some of the build functions into smaller utility functions Signed-off-by: Jonathan A. Sternberg --- build/build.go | 256 ++++++++++++++++++++++++++++++------------------- 1 file changed, 158 insertions(+), 98 deletions(-) diff --git a/build/build.go b/build/build.go index 0f8fca6d8..dbd42e907 100644 --- a/build/build.go +++ b/build/build.go @@ -138,75 +138,68 @@ func filterAvailableNodes(nodes []builder.Node) ([]builder.Node, error) { return nil, err } -func toRepoOnly(in string) (string, error) { - m := map[string]struct{}{} - p := strings.Split(in, ",") - for _, pp := range p { - n, err := reference.ParseNormalizedNamed(pp) - if err != nil { - return "", err - } - m[n.Name()] = struct{}{} - } - out := make([]string, 0, len(m)) - for k := range m { - out = append(out, k) - } - return strings.Join(out, ","), nil -} - -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(driverIndex int, rCtx *ResultHandle)) (resp map[string]*client.SolveResponse, err error) { - if len(nodes) == 0 { - return nil, errors.Errorf("driver required for build") - } - - nodes, err = filterAvailableNodes(nodes) - if err != nil { - return nil, errors.Wrapf(err, "no valid drivers found") - } - - var noMobyDriver *driver.DriverHandle +// findNonMobyDriver returns the first non-moby based driver. +func findNonMobyDriver(nodes []builder.Node) *driver.DriverHandle { for _, n := range nodes { if !n.Driver.IsMobyDriver() { - noMobyDriver = n.Driver - break + return n.Driver + } + } + return nil +} + +// warnOnNoOutput will check if the given nodes and options would result in an output +// and prints a warning if it would not. +func warnOnNoOutput(ctx context.Context, nodes []builder.Node, opts map[string]Options) { + // Return immediately if default load is explicitly disabled or a call + // function is used. + if noDefaultLoad() || !noCallFunc(opts) { + return + } + + // Find the first non-moby driver and return if it either doesn't exist + // or if the driver has default load enabled. + noMobyDriver := findNonMobyDriver(nodes) + if noMobyDriver == nil || noMobyDriver.Features(ctx)[driver.DefaultLoad] { + return + } + + // Produce a warning describing the targets affected. + var noOutputTargets []string + for name, opt := range opts { + if !opt.Linked && len(opt.Exports) == 0 { + noOutputTargets = append(noOutputTargets, name) } } - if noMobyDriver != nil && !noDefaultLoad() && noCallFunc(opts) { - var noOutputTargets []string - for name, opt := range opts { - if noMobyDriver.Features(ctx)[driver.DefaultLoad] { - continue - } - - if !opt.Linked && len(opt.Exports) == 0 { - noOutputTargets = append(noOutputTargets, name) - } - } - if len(noOutputTargets) > 0 { - var warnNoOutputBuf bytes.Buffer - warnNoOutputBuf.WriteString("No output specified ") - if len(noOutputTargets) == 1 && noOutputTargets[0] == "default" { - warnNoOutputBuf.WriteString(fmt.Sprintf("with %s driver", noMobyDriver.Factory().Name())) - } else { - warnNoOutputBuf.WriteString(fmt.Sprintf("for %s target(s) with %s driver", strings.Join(noOutputTargets, ", "), noMobyDriver.Factory().Name())) - } - logrus.Warnf("%s. Build result will only remain in the build cache. To push result image into registry use --push or to load image into docker use --load", warnNoOutputBuf.String()) - } + if len(noOutputTargets) == 0 { + return } - drivers, err := resolveDrivers(ctx, nodes, opts, w) - if err != nil { - return nil, err + var warnNoOutputBuf bytes.Buffer + warnNoOutputBuf.WriteString("No output specified ") + if len(noOutputTargets) == 1 && noOutputTargets[0] == "default" { + warnNoOutputBuf.WriteString(fmt.Sprintf("with %s driver", noMobyDriver.Factory().Name())) + } else { + warnNoOutputBuf.WriteString(fmt.Sprintf("for %s target(s) with %s driver", strings.Join(noOutputTargets, ", "), noMobyDriver.Factory().Name())) } + logrus.Warnf("%s. Build result will only remain in the build cache. To push result image into registry use --push or to load image into docker use --load", warnNoOutputBuf.String()) +} +func newBuildRequests(ctx context.Context, docker *dockerutil.Client, cfg *confutil.Config, drivers map[string][]*resolvedNode, w progress.Writer, opts map[string]Options) (_ map[string][]*reqForNode, _ func(), retErr error) { reqForNodes := make(map[string][]*reqForNode) - eg, ctx := errgroup.WithContext(ctx) + + var releasers []func() + releaseAll := func() { + for _, fn := range releasers { + fn() + } + } + defer func() { + if retErr != nil { + releaseAll() + } + }() for k, opt := range opts { multiDriver := len(drivers[k]) > 1 @@ -226,17 +219,17 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ opt.Platforms = np.platforms gatewayOpts, err := np.BuildOpts(ctx) if err != nil { - return nil, err + return nil, nil, err } localOpt := opt so, release, err := toSolveOpt(ctx, np.Node(), multiDriver, &localOpt, gatewayOpts, cfg, w, docker) opts[k] = localOpt if err != nil { - return nil, err + return nil, nil, err } - defer release() + releasers = append(releasers, release) if err := saveLocalState(so, k, opt, np.Node(), cfg); err != nil { - return nil, err + return nil, nil, err } addGitAttrs(so) reqn = append(reqn, &reqForNode{ @@ -261,15 +254,17 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ for _, e := range np.so.Exports { if e.Type == "moby" { if ok, _ := strconv.ParseBool(e.Attrs["push"]); ok { - return nil, errors.Errorf("multi-node push can't currently be performed with the docker driver, please switch to a different driver") + return nil, nil, errors.Errorf("multi-node push can't currently be performed with the docker driver, please switch to a different driver") } } } } } } + return reqForNodes, releaseAll, nil +} - // validate that all links between targets use same drivers +func validateTargetLinks(reqForNodes map[string][]*reqForNode, drivers map[string][]*resolvedNode, opts map[string]Options) error { for name := range opts { dps := reqForNodes[name] for i, dp := range dps { @@ -279,8 +274,9 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ k2 := strings.TrimPrefix(v, "target:") dps2, ok := drivers[k2] if !ok { - return nil, errors.Errorf("failed to find target %s for context %s", k2, strings.TrimPrefix(k, "context:")) // should be validated before already + return errors.Errorf("failed to find target %s for context %s", k2, strings.TrimPrefix(k, "context:")) // should be validated before already } + var found bool for _, dp2 := range dps2 { if dp2.driverIndex == dp.driverIndex { @@ -289,12 +285,63 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ } } if !found { - return nil, errors.Errorf("failed to use %s as context %s for %s because targets build with different drivers", k2, strings.TrimPrefix(k, "context:"), name) + return errors.Errorf("failed to use %s as context %s for %s because targets build with different drivers", k2, strings.TrimPrefix(k, "context:"), name) } } } } } + return nil +} + +func toRepoOnly(in string) (string, error) { + m := map[string]struct{}{} + p := strings.Split(in, ",") + for _, pp := range p { + n, err := reference.ParseNormalizedNamed(pp) + if err != nil { + return "", err + } + m[n.Name()] = struct{}{} + } + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return strings.Join(out, ","), nil +} + +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) { + if len(nodes) == 0 { + return nil, errors.Errorf("driver required for build") + } + + nodes, err = filterAvailableNodes(nodes) + if err != nil { + return nil, errors.Wrapf(err, "no valid drivers found") + } + warnOnNoOutput(ctx, nodes, opts) + + drivers, err := resolveDrivers(ctx, nodes, opts, w) + if err != nil { + return nil, err + } + + eg, ctx := errgroup.WithContext(ctx) + reqForNodes, release, err := newBuildRequests(ctx, docker, cfg, drivers, w, opts) + if err != nil { + return nil, err + } + defer release() + + // validate that all links between targets use same drivers + if err := validateTargetLinks(reqForNodes, drivers, opts); err != nil { + return nil, err + } sharedSessions, err := detectSharedMounts(ctx, reqForNodes) if err != nil { @@ -311,7 +358,6 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ for k, opt := range opts { err := func(k string) (err error) { - opt := opt dps := drivers[k] multiDriver := len(drivers[k]) > 1 @@ -441,19 +487,11 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ req.FrontendOpt["requestid"] = "frontend." + opt.CallFunc.Name } - res, err := c.Solve(ctx, req) + res, err := solve(ctx, c, req) if err != nil { - req, ok := fallbackPrintError(err, req) - if ok { - res2, err2 := c.Solve(ctx, req) - if err2 != nil { - return nil, err - } - res = res2 - } else { - return nil, err - } + return nil, err } + if opt.CallFunc != nil { callRes = res.Metadata } @@ -461,31 +499,15 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ rKey := resultKey(dp.driverIndex, k) results.Set(rKey, res) - if children, ok := childTargets[rKey]; ok && len(children) > 0 { - // wait for the child targets to register their LLB before evaluating - _, err := results.Get(ctx, children...) - if err != nil { - return nil, err - } - // we need to wait until the child targets have completed before we can release - eg, ctx := errgroup.WithContext(ctx) - eg.Go(func() error { - return res.EachRef(func(ref gateway.Reference) error { - return ref.Evaluate(ctx) - }) - }) - eg.Go(func() error { - _, err := results.Get(ctx, children...) - return err - }) - if err := eg.Wait(); err != nil { + if children := childTargets[rKey]; len(children) > 0 { + if err := waitForChildren(ctx, res, results, children); 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 resultHandleFunc != nil { var resultHandle *ResultHandle @@ -496,6 +518,7 @@ func BuildWithResultHandler(ctx context.Context, nodes []builder.Node, opts map[ rr, err = c.Build(ctx, *so, "buildx", buildFunc, ch) tracing.FinishWithError(span, err) } + if !so.Internal && desktop.BuildBackendEnabled() && node.Driver.HistoryAPISupported(ctx) { if err != nil { return &desktop.ErrorWithBuildRef{ @@ -1146,3 +1169,40 @@ func ReadSourcePolicy() (*spb.Policy, error) { return &pol, nil } + +func solve(ctx context.Context, c gateway.Client, req gateway.SolveRequest) (*gateway.Result, error) { + res, err := c.Solve(ctx, req) + if err != nil { + req, ok := fallbackPrintError(err, req) + if ok { + res2, err2 := c.Solve(ctx, req) + if err2 != nil { + return nil, err + } + res = res2 + } else { + return nil, err + } + } + return res, nil +} + +func waitForChildren(ctx context.Context, 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 { + return err + } + // we need to wait until the child targets have completed before we can release + eg, ctx := errgroup.WithContext(ctx) + eg.Go(func() error { + return res.EachRef(func(ref gateway.Reference) error { + return ref.Evaluate(ctx) + }) + }) + eg.Go(func() error { + _, err := results.Get(ctx, children...) + return err + }) + return eg.Wait() +} From 6a0f5610e331ce846d865dbf4a9cab52e27e9cbe Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Fri, 30 May 2025 14:15:56 -0500 Subject: [PATCH 2/3] controller: remove the controller interface The controller interface is removed and the local controller is used for only the initial build, invoke, and rebuilds. Process control has been moved to the monitor. Signed-off-by: Jonathan A. Sternberg --- commands/build.go | 44 ++++++++---------------- commands/debug/root.go | 11 ++---- controller/control/controller.go | 26 --------------- controller/controller.go | 13 -------- controller/local/controller.go | 41 +++++++---------------- monitor/monitor.go | 57 +++++++++++++++++++++++++------- monitor/types/types.go | 23 +++++++++++-- 7 files changed, 94 insertions(+), 121 deletions(-) delete mode 100644 controller/control/controller.go delete mode 100644 controller/controller.go diff --git a/commands/build.go b/commands/build.go index 7c9027584..60fbb4272 100644 --- a/commands/build.go +++ b/commands/build.go @@ -21,10 +21,9 @@ import ( "github.com/docker/buildx/build" "github.com/docker/buildx/builder" "github.com/docker/buildx/commands/debug" - "github.com/docker/buildx/controller" cbuild "github.com/docker/buildx/controller/build" - "github.com/docker/buildx/controller/control" 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" @@ -431,28 +430,19 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild // stdin must be usable for monitor return nil, nil, errors.Errorf("Dockerfile or context from stdin is not supported with invoke") } - c := controller.NewController(ctx, dockerCli) - defer func() { - if err := c.Close(); err != nil { - logrus.Warnf("failed to close server connection %v", err) - } - }() - // NOTE: buildx server has the current working directory different from the client - // so we need to resolve paths to abosolute ones in the client. - opts, err := cbuild.ResolveOptionPaths(opts) - if err != nil { - return nil, nil, err - } + c := local.NewController(ctx, dockerCli) + defer c.Close() - var ref string - var retErr error - var resp *client.SolveResponse - var inputs *build.Inputs + var ( + ref string + retErr error + + f *ioset.SingleForwarder + pr io.ReadCloser + pw io.WriteCloser + ) - var f *ioset.SingleForwarder - var pr io.ReadCloser - var pw io.WriteCloser if options.invokeConfig == nil { pr = dockerCli.In() } else { @@ -466,7 +456,7 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild }) } - resp, inputs, err = c.Build(ctx, opts, pr, printer) + resp, inputs, err := c.Build(ctx, opts, pr, printer) if err != nil { var be *controllererrors.BuildError if errors.As(err, &be) { @@ -508,10 +498,6 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild // Update return values with the last build result from monitor resp, retErr = monitorBuildResult.Resp, monitorBuildResult.Err } - } else { - if err := c.Close(); err != nil { - logrus.Warnf("close error: %v", err) - } } return resp, inputs, retErr @@ -1003,13 +989,9 @@ func (cfg *invokeConfig) needsDebug(retErr error) bool { } } -func (cfg *invokeConfig) runDebug(ctx context.Context, ref string, options *cbuild.Options, c control.BuildxController, stdin io.ReadCloser, stdout io.WriteCloser, stderr console.File, progress *progress.Printer) (*monitor.MonitorBuildResult, error) { +func (cfg *invokeConfig) runDebug(ctx context.Context, ref string, options *cbuild.Options, c *local.Controller, stdin io.ReadCloser, stdout io.WriteCloser, stderr console.File, progress *progress.Printer) (*monitor.MonitorBuildResult, error) { con := console.Current() if err := con.SetRaw(); err != nil { - // TODO: run disconnect in build command (on error case) - if err := c.Close(); err != nil { - logrus.Warnf("close error: %v", err) - } return nil, errors.Errorf("failed to configure terminal: %v", err) } defer con.Reset() diff --git a/commands/debug/root.go b/commands/debug/root.go index f63bde3ac..9d0d0bb4f 100644 --- a/commands/debug/root.go +++ b/commands/debug/root.go @@ -5,7 +5,7 @@ import ( "os" "github.com/containerd/console" - "github.com/docker/buildx/controller" + "github.com/docker/buildx/controller/local" controllerapi "github.com/docker/buildx/controller/pb" "github.com/docker/buildx/monitor" "github.com/docker/buildx/util/cobrautil" @@ -13,7 +13,6 @@ import ( "github.com/docker/cli/cli/command" "github.com/moby/buildkit/util/progress/progressui" "github.com/pkg/errors" - "github.com/sirupsen/logrus" "github.com/spf13/cobra" ) @@ -47,12 +46,8 @@ func RootCmd(dockerCli command.Cli, children ...DebuggableCmd) *cobra.Command { } ctx := context.TODO() - c := controller.NewController(ctx, dockerCli) - defer func() { - if err := c.Close(); err != nil { - logrus.Warnf("failed to close server connection %v", err) - } - }() + c := local.NewController(ctx, dockerCli) + con := console.Current() if err := con.SetRaw(); err != nil { return errors.Errorf("failed to configure terminal: %v", err) diff --git a/controller/control/controller.go b/controller/control/controller.go deleted file mode 100644 index cc501e9bb..000000000 --- a/controller/control/controller.go +++ /dev/null @@ -1,26 +0,0 @@ -package control - -import ( - "context" - "io" - - "github.com/docker/buildx/build" - cbuild "github.com/docker/buildx/controller/build" - controllerapi "github.com/docker/buildx/controller/pb" - "github.com/docker/buildx/controller/processes" - "github.com/docker/buildx/util/progress" - "github.com/moby/buildkit/client" -) - -type BuildxController interface { - Build(ctx context.Context, options *cbuild.Options, in io.ReadCloser, progress progress.Writer) (resp *client.SolveResponse, inputs *build.Inputs, err error) - // Invoke starts an IO session into the specified process. - // If pid doesn't match to any running processes, it starts a new process with the specified config. - // If there is no container running or InvokeConfig.Rollback is specified, the process will start in a newly created container. - // NOTE: If needed, in the future, we can split this API into three APIs (NewContainer, NewProcess and Attach). - Invoke(ctx context.Context, pid string, options *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error - Close() error - ListProcesses(ctx context.Context) (infos []*processes.ProcessInfo, retErr error) - DisconnectProcess(ctx context.Context, pid string) error - Inspect(ctx context.Context) *cbuild.Options -} diff --git a/controller/controller.go b/controller/controller.go deleted file mode 100644 index dfa0f47d5..000000000 --- a/controller/controller.go +++ /dev/null @@ -1,13 +0,0 @@ -package controller - -import ( - "context" - - "github.com/docker/buildx/controller/control" - "github.com/docker/buildx/controller/local" - "github.com/docker/cli/cli/command" -) - -func NewController(ctx context.Context, dockerCli command.Cli) control.BuildxController { - return local.NewLocalBuildxController(ctx, dockerCli) -} diff --git a/controller/local/controller.go b/controller/local/controller.go index 174a002c7..671c4ba0e 100644 --- a/controller/local/controller.go +++ b/controller/local/controller.go @@ -7,7 +7,6 @@ import ( "github.com/docker/buildx/build" cbuild "github.com/docker/buildx/controller/build" - "github.com/docker/buildx/controller/control" controllererrors "github.com/docker/buildx/controller/errdefs" controllerapi "github.com/docker/buildx/controller/pb" "github.com/docker/buildx/controller/processes" @@ -18,10 +17,9 @@ import ( "github.com/pkg/errors" ) -func NewLocalBuildxController(ctx context.Context, dockerCli command.Cli) control.BuildxController { - return &localController{ +func NewController(ctx context.Context, dockerCli command.Cli) *Controller { + return &Controller{ dockerCli: dockerCli, - processes: processes.NewManager(), } } @@ -32,15 +30,14 @@ type buildConfig struct { buildOptions *cbuild.Options } -type localController struct { +type Controller struct { dockerCli command.Cli buildConfig buildConfig - processes *processes.Manager buildOnGoing atomic.Bool } -func (b *localController) Build(ctx context.Context, options *cbuild.Options, in io.ReadCloser, progress progress.Writer) (*client.SolveResponse, *build.Inputs, error) { +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") } @@ -63,27 +60,15 @@ func (b *localController) Build(ctx context.Context, options *cbuild.Options, in return resp, dockerfileMappings, nil } -func (b *localController) ListProcesses(ctx context.Context) (infos []*processes.ProcessInfo, retErr error) { - return b.processes.ListProcesses(), nil -} - -func (b *localController) DisconnectProcess(ctx context.Context, pid string) error { - return b.processes.DeleteProcess(pid) -} - -func (b *localController) cancelRunningProcesses() { - b.processes.CancelRunningProcesses() -} - -func (b *localController) Invoke(ctx context.Context, pid string, cfg *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error { - proc, ok := b.processes.Get(pid) +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 = b.processes.StartProcess(pid, b.buildConfig.resultCtx, cfg) + proc, err = processes.StartProcess(pid, b.buildConfig.resultCtx, cfg) if err != nil { return err } @@ -103,15 +88,13 @@ func (b *localController) Invoke(ctx context.Context, pid string, cfg *controlle } } -func (b *localController) Close() error { - b.cancelRunningProcesses() +func (b *Controller) Inspect(ctx context.Context) *cbuild.Options { + return b.buildConfig.buildOptions +} + +func (b *Controller) Close() error { if b.buildConfig.resultCtx != nil { b.buildConfig.resultCtx.Done() } - // TODO: cancel ongoing builds? return nil } - -func (b *localController) Inspect(ctx context.Context) *cbuild.Options { - return b.buildConfig.buildOptions -} diff --git a/monitor/monitor.go b/monitor/monitor.go index 750f44393..6156dc834 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -12,8 +12,9 @@ import ( "github.com/containerd/console" "github.com/docker/buildx/build" cbuild "github.com/docker/buildx/controller/build" - "github.com/docker/buildx/controller/control" + "github.com/docker/buildx/controller/local" controllerapi "github.com/docker/buildx/controller/pb" + "github.com/docker/buildx/controller/processes" "github.com/docker/buildx/monitor/commands" "github.com/docker/buildx/monitor/types" "github.com/docker/buildx/util/ioset" @@ -32,13 +33,7 @@ type MonitorBuildResult struct { } // RunMonitor provides an interactive session for running and managing containers via specified IO. -func RunMonitor(ctx context.Context, curRef string, options *cbuild.Options, invokeConfig *controllerapi.InvokeConfig, c control.BuildxController, stdin io.ReadCloser, stdout io.WriteCloser, stderr console.File, progress *progress.Printer) (*MonitorBuildResult, error) { - defer func() { - if err := c.Close(); err != nil { - logrus.Warnf("close error: %v", err) - } - }() - +func RunMonitor(ctx context.Context, curRef string, options *cbuild.Options, invokeConfig *controllerapi.InvokeConfig, c *local.Controller, stdin io.ReadCloser, stdout io.WriteCloser, stderr console.File, progress *progress.Printer) (*MonitorBuildResult, error) { if err := progress.Pause(); err != nil { return nil, err } @@ -70,8 +65,9 @@ func RunMonitor(ctx context.Context, curRef string, options *cbuild.Options, inv invokeForwarder := ioset.NewForwarder() invokeForwarder.SetIn(&containerIn) m := &monitor{ - BuildxController: c, - invokeIO: invokeForwarder, + c: c, + processes: processes.NewManager(), + invokeIO: invokeForwarder, muxIO: ioset.NewMuxIO(ioset.In{ Stdin: io.NopCloser(stdin), Stdout: nopCloser{stdout}, @@ -84,6 +80,12 @@ func RunMonitor(ctx context.Context, curRef string, options *cbuild.Options, inv return "Switched IO\n" }), } + + defer func() { + if err := m.Close(); err != nil { + logrus.Warnf("close error: %v", err) + } + }() m.ref.Store(curRef) // Start container automatically @@ -231,7 +233,7 @@ type readWriter struct { } type monitor struct { - control.BuildxController + c *local.Controller ref atomic.Value muxIO *ioset.MuxIO @@ -240,14 +242,24 @@ type monitor struct { attachedPid atomic.Value lastBuildResult *MonitorBuildResult + + processes *processes.Manager } func (m *monitor) Build(ctx context.Context, options *cbuild.Options, in io.ReadCloser, progress progress.Writer) (resp *client.SolveResponse, input *build.Inputs, err error) { - resp, _, err = m.BuildxController.Build(ctx, options, in, progress) + resp, _, err = m.c.Build(ctx, options, in, progress) m.lastBuildResult = &MonitorBuildResult{Resp: resp, Err: err} // Record build result return } +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) +} + +func (m *monitor) Inspect(ctx context.Context) *cbuild.Options { + return m.c.Inspect(ctx) +} + func (m *monitor) Rollback(ctx context.Context, cfg *controllerapi.InvokeConfig) string { pid := identity.NewID() cfg1 := cfg @@ -332,6 +344,27 @@ func (m *monitor) invoke(ctx context.Context, pid string, cfg *controllerapi.Inv return err } +func (m *monitor) Close() error { + m.cancelRunningProcesses() + // if m.buildConfig.resultCtx != nil { + // b.buildConfig.resultCtx.Done() + // } + // TODO: cancel ongoing builds? + return nil +} + +func (m *monitor) ListProcesses(ctx context.Context) (infos []*processes.ProcessInfo, retErr error) { + return m.processes.ListProcesses(), nil +} + +func (m *monitor) DisconnectProcess(ctx context.Context, pid string) error { + return m.processes.DeleteProcess(pid) +} + +func (m *monitor) cancelRunningProcesses() { + m.processes.CancelRunningProcesses() +} + type nopCloser struct { io.Writer } diff --git a/monitor/types/types.go b/monitor/types/types.go index 201f273ac..da837f32c 100644 --- a/monitor/types/types.go +++ b/monitor/types/types.go @@ -2,14 +2,31 @@ package types import ( "context" + "io" - "github.com/docker/buildx/controller/control" + "github.com/docker/buildx/build" + cbuild "github.com/docker/buildx/controller/build" controllerapi "github.com/docker/buildx/controller/pb" + "github.com/docker/buildx/controller/processes" + "github.com/docker/buildx/util/progress" + "github.com/moby/buildkit/client" ) // Monitor provides APIs for attaching and controlling the buildx server. type Monitor interface { - control.BuildxController + Build(ctx context.Context, options *cbuild.Options, in io.ReadCloser, progress progress.Writer) (resp *client.SolveResponse, inputs *build.Inputs, err error) + + Inspect(ctx context.Context) *cbuild.Options + + // Invoke starts an IO session into the specified process. + // If pid doesn't match to any running processes, it starts a new process with the specified config. + // If there is no container running or InvokeConfig.Rollback is specified, the process will start in a newly created container. + // NOTE: If needed, in the future, we can split this API into three APIs (NewContainer, NewProcess and Attach). + Invoke(ctx context.Context, pid string, options *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error + + ListProcesses(ctx context.Context) (infos []*processes.ProcessInfo, retErr error) + + DisconnectProcess(ctx context.Context, pid string) error // Rollback re-runs the interactive container with initial rootfs contents. Rollback(ctx context.Context, cfg *controllerapi.InvokeConfig) string @@ -25,6 +42,8 @@ type Monitor interface { // Detach detaches IO from the container. Detach() + + io.Closer } // CommandInfo is information about a command. From 65e46cc6af2b3a7f844a353e5548125d89c3bb83 Mon Sep 17 00:00:00 2001 From: "Jonathan A. Sternberg" Date: Mon, 2 Jun 2025 16:31:32 -0500 Subject: [PATCH 3/3] commands: simplify passing stdin to the build when the monitor is configured The monitor needs stdin to run and isn't compatible with loading a context or dockerfile from stdin. We already disallow this combination and, with the removal of the remote controller, there's no way to use stdin during the build when invoke is configured. This just removes the extra code to allow forwarding stdin to the build when the monitor is configured to simplify that section of code. Signed-off-by: Jonathan A. Sternberg --- commands/build.go | 51 +++++++++++++++------------------------------- monitor/monitor.go | 4 ---- 2 files changed, 16 insertions(+), 39 deletions(-) diff --git a/commands/build.go b/commands/build.go index 60fbb4272..93791a4ac 100644 --- a/commands/build.go +++ b/commands/build.go @@ -425,7 +425,7 @@ func runBasicBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild.Opti return resp, dfmap, err } -func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild.Options, options buildOptions, printer *progress.Printer) (*client.SolveResponse, *build.Inputs, error) { +func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild.Options, options buildOptions, printer *progress.Printer) (_ *client.SolveResponse, _ *build.Inputs, retErr error) { if options.invokeConfig != nil && (options.dockerfileName == "-" || options.contextPath == "-") { // stdin must be usable for monitor return nil, nil, errors.Errorf("Dockerfile or context from stdin is not supported with invoke") @@ -434,29 +434,12 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild c := local.NewController(ctx, dockerCli) defer c.Close() - var ( - ref string - retErr error - - f *ioset.SingleForwarder - pr io.ReadCloser - pw io.WriteCloser - ) - + var in io.ReadCloser if options.invokeConfig == nil { - pr = dockerCli.In() - } else { - f = ioset.NewSingleForwarder() - f.SetReader(dockerCli.In()) - pr, pw = io.Pipe() - f.SetWriter(pw, func() io.WriteCloser { - pw.Close() // propagate EOF - logrus.Debug("propagating stdin close") - return nil - }) + in = dockerCli.In() } - resp, inputs, err := c.Build(ctx, opts, pr, printer) + resp, inputs, err := c.Build(ctx, opts, in, printer) if err != nil { var be *controllererrors.BuildError if errors.As(err, &be) { @@ -467,28 +450,26 @@ func runControllerBuild(ctx context.Context, dockerCli command.Cli, opts *cbuild } } - if options.invokeConfig != nil { - if err := pw.Close(); err != nil { - logrus.Debug("failed to close stdin pipe writer") - } - if err := pr.Close(); err != nil { - logrus.Debug("failed to close stdin pipe reader") - } - } - if options.invokeConfig != nil && options.invokeConfig.needsDebug(retErr) { // Print errors before launching monitor if err := printError(retErr, printer); err != nil { logrus.Warnf("failed to print error information: %v", err) } - pr2, pw2 := io.Pipe() - f.SetWriter(pw2, func() io.WriteCloser { - pw2.Close() // propagate EOF + pr, pw := io.Pipe() + + f := ioset.NewSingleForwarder() + f.SetReader(dockerCli.In()) + f.SetWriter(pw, func() io.WriteCloser { + pw.Close() // propagate EOF return nil }) - monitorBuildResult, err := options.invokeConfig.runDebug(ctx, ref, opts, c, pr2, os.Stdout, os.Stderr, printer) - if err := pw2.Close(); err != nil { + + // TODO: ref was never set to a value in the original code. Removed the variable to + // reduce confusion but it also probably means this call is wrong in some way. + // This area should be removed during the refactor anyway so it doesn't matter that much. + monitorBuildResult, err := options.invokeConfig.runDebug(ctx, "", opts, c, pr, os.Stdout, os.Stderr, printer) + if err := pw.Close(); err != nil { logrus.Debug("failed to close monitor stdin pipe reader") } if err != nil { diff --git a/monitor/monitor.go b/monitor/monitor.go index 6156dc834..50cabd672 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -346,10 +346,6 @@ func (m *monitor) invoke(ctx context.Context, pid string, cfg *controllerapi.Inv func (m *monitor) Close() error { m.cancelRunningProcesses() - // if m.buildConfig.resultCtx != nil { - // b.buildConfig.resultCtx.Done() - // } - // TODO: cancel ongoing builds? return nil }