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

controller: remove remaining parts of the controller

Browse Source 
Removes all references to the controller and moves the remaining
sections of code to other packages.

Processes has been moved to monitor where it is used and the data
structs have been removed so buildflags is used directly. The controller
build function has been moved to the commands package.

Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com>
This commit is contained in:
Jonathan A. Sternberg
2025-06-05 11:57:03 -05:00
parent 60b1eda2df
commit 1d7cda1232
36 changed files with 659 additions and 1502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
monitor/commands/exec.go
+4 -4
View File
@@ -5,7 +5,7 @@ import (
"fmt"
"io"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/build"
"github.com/docker/buildx/monitor/types"
"github.com/pkg/errors"
)
@@ -13,11 +13,11 @@ import (
type ExecCmd struct {
m types.Monitor
invokeConfig *controllerapi.InvokeConfig
invokeConfig *build.InvokeConfig
stdout io.WriteCloser
}
func NewExecCmd(m types.Monitor, invokeConfig *controllerapi.InvokeConfig, stdout io.WriteCloser) types.Command {
func NewExecCmd(m types.Monitor, invokeConfig *build.InvokeConfig, stdout io.WriteCloser) types.Command {
return &ExecCmd{m, invokeConfig, stdout}
}
@@ -38,7 +38,7 @@ func (cm *ExecCmd) Exec(ctx context.Context, args []string) error {
if len(args) < 2 {
return errors.Errorf("command must be passed")
}
cfg := &controllerapi.InvokeConfig{
cfg := &build.InvokeConfig{
Entrypoint: []string{args[1]},
Cmd: args[2:],
NoCmd: false,
monitor/commands/rollback.go
+3 -3
View File
@@ -5,18 +5,18 @@ import (
"fmt"
"io"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/build"
"github.com/docker/buildx/monitor/types"
)
type RollbackCmd struct {
m types.Monitor
invokeConfig *controllerapi.InvokeConfig
invokeConfig *build.InvokeConfig
stdout io.WriteCloser
}
func NewRollbackCmd(m types.Monitor, invokeConfig *controllerapi.InvokeConfig, stdout io.WriteCloser) types.Command {
func NewRollbackCmd(m types.Monitor, invokeConfig *build.InvokeConfig, stdout io.WriteCloser) types.Command {
return &RollbackCmd{m, invokeConfig, stdout}
}
monitor/monitor.go
+10 -11
View File
@@ -12,9 +12,8 @@ import (
"github.com/containerd/console"
"github.com/docker/buildx/build"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/controller/processes"
"github.com/docker/buildx/monitor/commands"
"github.com/docker/buildx/monitor/processes"
"github.com/docker/buildx/monitor/types"
"github.com/docker/buildx/util/ioset"
"github.com/docker/buildx/util/progress"
@@ -29,7 +28,7 @@ import (
var ErrReload = errors.New("monitor: reload")
type Monitor struct {
invokeConfig *controllerapi.InvokeConfig
invokeConfig *build.InvokeConfig
printer *progress.Printer
stdin *ioset.SingleForwarder
@@ -41,7 +40,7 @@ type Monitor struct {
mu sync.Mutex
}
func New(cfg *controllerapi.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCloser, printer *progress.Printer) *Monitor {
func New(cfg *build.InvokeConfig, stdin io.ReadCloser, stdout, stderr io.WriteCloser, printer *progress.Printer) *Monitor {
m := &Monitor{
invokeConfig: cfg,
printer: printer,
@@ -113,7 +112,7 @@ func (m *Monitor) Close() error {
}
// RunMonitor provides an interactive session for running and managing containers via specified IO.
func RunMonitor(ctx context.Context, invokeConfig *controllerapi.InvokeConfig, rCtx *build.ResultHandle, stdin io.ReadCloser, stdout, stderr io.WriteCloser, progress *progress.Printer) error {
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
}
@@ -333,7 +332,7 @@ type monitor struct {
processes *processes.Manager
}
func (m *monitor) Invoke(ctx context.Context, pid string, cfg *controllerapi.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error {
func (m *monitor) Invoke(ctx context.Context, pid string, cfg *build.InvokeConfig, ioIn io.ReadCloser, ioOut io.WriteCloser, ioErr io.WriteCloser) error {
proc, ok := m.processes.Get(pid)
if !ok {
// Start a new process.
@@ -361,19 +360,19 @@ func (m *monitor) Invoke(ctx context.Context, pid string, cfg *controllerapi.Inv
}
}
func (m *monitor) Rollback(ctx context.Context, cfg *controllerapi.InvokeConfig) string {
func (m *monitor) Rollback(ctx context.Context, cfg *build.InvokeConfig) string {
pid := identity.NewID()
cfg1 := cfg
cfg1.Rollback = true
return m.startInvoke(ctx, pid, cfg1)
}
func (m *monitor) Exec(ctx context.Context, cfg *controllerapi.InvokeConfig) string {
func (m *monitor) Exec(ctx context.Context, cfg *build.InvokeConfig) string {
return m.startInvoke(ctx, identity.NewID(), cfg)
}
func (m *monitor) Attach(ctx context.Context, pid string) {
m.startInvoke(ctx, pid, &controllerapi.InvokeConfig{})
m.startInvoke(ctx, pid, &build.InvokeConfig{})
}
func (m *monitor) Detach() {
@@ -394,7 +393,7 @@ func (m *monitor) close() {
m.Detach()
}
func (m *monitor) startInvoke(ctx context.Context, pid string, cfg *controllerapi.InvokeConfig) string {
func (m *monitor) startInvoke(ctx context.Context, pid string, cfg *build.InvokeConfig) string {
if m.invokeCancel != nil {
m.invokeCancel() // Finish existing attach
}
@@ -420,7 +419,7 @@ func (m *monitor) startInvoke(ctx context.Context, pid string, cfg *controllerap
return pid
}
func (m *monitor) invoke(ctx context.Context, pid string, cfg *controllerapi.InvokeConfig) error {
func (m *monitor) invoke(ctx context.Context, pid string, cfg *build.InvokeConfig) error {
m.muxIO.Enable(1)
defer m.muxIO.Disable(1)
if err := m.muxIO.SwitchTo(1); err != nil {
monitor/processes/processes.go
+160
View File
@@ -0,0 +1,160 @@
package processes
import (
"context"
"sync"
"sync/atomic"
"github.com/docker/buildx/build"
"github.com/docker/buildx/util/ioset"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
// Process provides methods to control a process.
type Process struct {
inEnd *ioset.Forwarder
invokeConfig *build.InvokeConfig
errCh chan error
processCancel func()
serveIOCancel func(error)
}
// ForwardIO forwards process's io to the specified reader/writer.
// Optionally specify ioCancelCallback which will be called when
// the process closes the specified IO. This will be useful for additional cleanup.
func (p *Process) ForwardIO(in *ioset.In, ioCancelCallback func(error)) {
p.inEnd.SetIn(in)
if f := p.serveIOCancel; f != nil {
f(errors.WithStack(context.Canceled))
}
p.serveIOCancel = ioCancelCallback
}
// Done returns a channel where error or nil will be sent
// when the process exits.
// TODO: change this to Wait()
func (p *Process) Done() <-chan error {
return p.errCh
}
// Manager manages a set of processes.
type Manager struct {
container atomic.Value
processes sync.Map
}
// NewManager creates and returns a Manager.
func NewManager() *Manager {
return &Manager{}
}
// Get returns the specified process.
func (m *Manager) Get(id string) (*Process, bool) {
v, ok := m.processes.Load(id)
if !ok {
return nil, false
}
return v.(*Process), true
}
// CancelRunningProcesses cancels execution of all running processes.
func (m *Manager) CancelRunningProcesses() {
var funcs []func()
m.processes.Range(func(key, value any) bool {
funcs = append(funcs, value.(*Process).processCancel)
m.processes.Delete(key)
return true
})
for _, f := range funcs {
f()
}
}
// ListProcesses lists all running processes.
func (m *Manager) ListProcesses() (res []*ProcessInfo) {
m.processes.Range(func(key, value any) bool {
res = append(res, &ProcessInfo{
ProcessID: key.(string),
InvokeConfig: value.(*Process).invokeConfig,
})
return true
})
return res
}
// DeleteProcess deletes the specified process.
func (m *Manager) DeleteProcess(id string) error {
p, ok := m.processes.LoadAndDelete(id)
if !ok {
return errors.Errorf("unknown process %q", id)
}
p.(*Process).processCancel()
return nil
}
// StartProcess starts a process in the container.
// When a container isn't available (i.e. first time invoking or the container has exited) or cfg.Rollback is set,
// this method will start a new container and run the process in it. Otherwise, this method starts a new process in the
// existing container.
func (m *Manager) StartProcess(pid string, resultCtx *build.ResultHandle, cfg *build.InvokeConfig) (*Process, error) {
// Get the target result to invoke a container from
var ctr *build.Container
if a := m.container.Load(); a != nil {
ctr = a.(*build.Container)
}
if cfg.Rollback || ctr == nil || ctr.IsUnavailable() {
go m.CancelRunningProcesses()
// (Re)create a new container if this is rollback or first time to invoke a process.
if ctr != nil {
go ctr.Cancel() // Finish the existing container
}
var err error
ctr, err = build.NewContainer(context.TODO(), resultCtx, cfg)
if err != nil {
return nil, errors.Errorf("failed to create container %v", err)
}
m.container.Store(ctr)
}
// [client(ForwardIO)] <-forwarder(switchable)-> [out] <-pipe-> [in] <- [process]
in, out := ioset.Pipe()
f := ioset.NewForwarder()
f.PropagateStdinClose = false
f.SetOut(&out)
// Register process
ctx, cancel := context.WithCancelCause(context.TODO())
var cancelOnce sync.Once
processCancelFunc := func() {
cancelOnce.Do(func() {
cancel(errors.WithStack(context.Canceled))
f.Close()
in.Close()
out.Close()
})
}
p := &Process{
inEnd: f,
invokeConfig: cfg,
processCancel: processCancelFunc,
errCh: make(chan error),
}
m.processes.Store(pid, p)
go func() {
var err error
if err = ctr.Exec(ctx, cfg, in.Stdin, in.Stdout, in.Stderr); err != nil {
logrus.Debugf("process error: %v", err)
}
logrus.Debugf("finished process %s %v", pid, cfg.Entrypoint)
m.processes.Delete(pid)
processCancelFunc()
p.errCh <- err
}()
return p, nil
}
type ProcessInfo struct {
ProcessID string
InvokeConfig *build.InvokeConfig
}
monitor/types/types.go
+5 -5
View File
@@ -4,8 +4,8 @@ import (
"context"
"io"
controllerapi "github.com/docker/buildx/controller/pb"
"github.com/docker/buildx/controller/processes"
"github.com/docker/buildx/build"
"github.com/docker/buildx/monitor/processes"
)
// Monitor provides APIs for attaching and controlling the buildx server.
@@ -14,17 +14,17 @@ type Monitor interface {
// 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
Invoke(ctx context.Context, pid string, options *build.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
Rollback(ctx context.Context, cfg *build.InvokeConfig) string
// Rollback executes a process in the interactive container.
Exec(ctx context.Context, cfg *controllerapi.InvokeConfig) string
Exec(ctx context.Context, cfg *build.InvokeConfig) string
// Attach attaches IO to a process in the container.
Attach(ctx context.Context, pid string)