Merge pull request #1620 from jedevc/remote-controller-fixes

Improvements for remote controller code
pull/1644/head
Justin Chadwell 2 years ago committed by GitHub
commit ed4fd965ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -392,10 +392,6 @@ func launchControllerAndRunBuild(dockerCli command.Cli, options buildOptions) er
if err := c.Disconnect(ctx, ref); err != nil { if err := c.Disconnect(ctx, ref); err != nil {
logrus.Warnf("disconnect error: %v", err) logrus.Warnf("disconnect error: %v", err)
} }
// If "invoke" isn't specified, further inspection ins't provided. Finish the buildx server.
if err := c.Kill(ctx); err != nil {
return err
}
} }
return nil return nil
} }

@ -20,20 +20,21 @@ import (
"google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/credentials/insecure"
) )
func NewClient(addr string) (*Client, error) { func NewClient(ctx context.Context, addr string) (*Client, error) {
backoffConfig := backoff.DefaultConfig backoffConfig := backoff.DefaultConfig
backoffConfig.MaxDelay = 3 * time.Second backoffConfig.MaxDelay = 3 * time.Second
connParams := grpc.ConnectParams{ connParams := grpc.ConnectParams{
Backoff: backoffConfig, Backoff: backoffConfig,
} }
gopts := []grpc.DialOption{ gopts := []grpc.DialOption{
grpc.WithBlock(),
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithConnectParams(connParams), grpc.WithConnectParams(connParams),
grpc.WithContextDialer(dialer.ContextDialer), grpc.WithContextDialer(dialer.ContextDialer),
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize)), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(defaults.DefaultMaxRecvMsgSize)),
grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)), grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(defaults.DefaultMaxSendMsgSize)),
} }
conn, err := grpc.Dial(dialer.DialAddress(addr), gopts...) conn, err := grpc.DialContext(ctx, dialer.DialAddress(addr), gopts...)
if err != nil { if err != nil {
return nil, err return nil, err
} }

@ -35,6 +35,12 @@ const (
serveCommandName = "_INTERNAL_SERVE" serveCommandName = "_INTERNAL_SERVE"
) )
var (
defaultLogFilename = fmt.Sprintf("buildx.%s.log", version.Revision)
defaultSocketFilename = fmt.Sprintf("buildx.%s.sock", version.Revision)
defaultPIDFilename = fmt.Sprintf("buildx.%s.pid", version.Revision)
)
type serverConfig struct { type serverConfig struct {
// Specify buildx server root // Specify buildx server root
Root string `toml:"root"` Root string `toml:"root"`
@ -52,10 +58,21 @@ func NewRemoteBuildxController(ctx context.Context, dockerCli command.Cli, opts
rootDir = rootDataDir(dockerCli) rootDir = rootDataDir(dockerCli)
} }
serverRoot := filepath.Join(rootDir, "shared") serverRoot := filepath.Join(rootDir, "shared")
c, err := newBuildxClientAndCheck(filepath.Join(serverRoot, "buildx.sock"), 1, 0)
// connect to buildx server if it is already running
ctx2, cancel := context.WithTimeout(ctx, 1*time.Second)
c, err := newBuildxClientAndCheck(ctx2, filepath.Join(serverRoot, defaultSocketFilename))
cancel()
if err != nil { if err != nil {
logrus.Info("no buildx server found; launching...") if !errors.Is(err, context.DeadlineExceeded) {
return nil, errors.Wrap(err, "cannot connect to the buildx server")
}
} else {
return &buildxController{c, serverRoot}, nil
}
// start buildx server via subcommand // start buildx server via subcommand
logrus.Info("no buildx server found; launching...")
launchFlags := []string{} launchFlags := []string{}
if opts.ServerConfig != "" { if opts.ServerConfig != "" {
launchFlags = append(launchFlags, "--config", opts.ServerConfig) launchFlags = append(launchFlags, "--config", opts.ServerConfig)
@ -69,11 +86,14 @@ func NewRemoteBuildxController(ctx context.Context, dockerCli command.Cli, opts
return nil, err return nil, err
} }
go wait() go wait()
c, err = newBuildxClientAndCheck(filepath.Join(serverRoot, "buildx.sock"), 10, time.Second)
// wait for buildx server to be ready
ctx2, cancel = context.WithTimeout(ctx, 10*time.Second)
c, err = newBuildxClientAndCheck(ctx2, filepath.Join(serverRoot, defaultSocketFilename))
cancel()
if err != nil { if err != nil {
return nil, errors.Wrap(err, "cannot connect to the buildx server") return nil, errors.Wrap(err, "cannot connect to the buildx server")
} }
}
return &buildxController{c, serverRoot}, nil return &buildxController{c, serverRoot}, nil
} }
@ -110,7 +130,7 @@ func serveCmd(dockerCli command.Cli) *cobra.Command {
if err != nil { if err != nil {
return err return err
} }
pidF := filepath.Join(root, "pid") pidF := filepath.Join(root, defaultPIDFilename)
if err := os.WriteFile(pidF, []byte(fmt.Sprintf("%d", os.Getpid())), 0600); err != nil { if err := os.WriteFile(pidF, []byte(fmt.Sprintf("%d", os.Getpid())), 0600); err != nil {
return err return err
} }
@ -127,7 +147,7 @@ func serveCmd(dockerCli command.Cli) *cobra.Command {
defer b.Close() defer b.Close()
// serve server // serve server
addr := filepath.Join(root, "buildx.sock") addr := filepath.Join(root, defaultSocketFilename)
if err := os.Remove(addr); err != nil && !os.IsNotExist(err) { // avoid EADDRINUSE if err := os.Remove(addr); err != nil && !os.IsNotExist(err) { // avoid EADDRINUSE
return err return err
} }
@ -151,18 +171,22 @@ func serveCmd(dockerCli command.Cli) *cobra.Command {
errCh <- errors.Wrapf(err, "error on serving via socket %q", addr) errCh <- errors.Wrapf(err, "error on serving via socket %q", addr)
} }
}() }()
var s os.Signal var s os.Signal
sigCh := make(chan os.Signal, 1) sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, os.Interrupt) signal.Notify(sigCh, syscall.SIGINT)
signal.Notify(sigCh, syscall.SIGTERM)
select { select {
case s = <-sigCh:
logrus.Debugf("got signal %v", s)
case err := <-errCh: case err := <-errCh:
logrus.Errorf("got error %s, exiting", err)
return err return err
case s = <-sigCh:
logrus.Infof("got signal %s, exiting", s)
return nil
case <-doneCh: case <-doneCh:
} logrus.Infof("rpc server done, exiting")
return nil return nil
}
}, },
} }
@ -176,15 +200,14 @@ func getLogFilePath(dockerCli command.Cli, configPath string) (string, error) {
if err != nil { if err != nil {
return "", err return "", err
} }
logFile := config.LogFile if config.LogFile == "" {
if logFile == "" {
root, err := prepareRootDir(dockerCli, config) root, err := prepareRootDir(dockerCli, config)
if err != nil { if err != nil {
return "", err return "", err
} }
logFile = filepath.Join(root, "log") return filepath.Join(root, defaultLogFilename), nil
} }
return logFile, nil return config.LogFile, nil
} }
func getConfig(dockerCli command.Cli, configPath string) (*serverConfig, error) { func getConfig(dockerCli command.Cli, configPath string) (*serverConfig, error) {
@ -228,34 +251,18 @@ func rootDataDir(dockerCli command.Cli) string {
return filepath.Join(confutil.ConfigDir(dockerCli), "controller") return filepath.Join(confutil.ConfigDir(dockerCli), "controller")
} }
func newBuildxClientAndCheck(addr string, checkNum int, duration time.Duration) (*Client, error) { func newBuildxClientAndCheck(ctx context.Context, addr string) (*Client, error) {
c, err := NewClient(addr) c, err := NewClient(ctx, addr)
if err != nil { if err != nil {
return nil, err return nil, err
} }
var lastErr error p, v, r, err := c.Version(ctx)
for i := 0; i < checkNum; i++ {
_, err := c.List(context.TODO())
if err == nil {
lastErr = nil
break
}
err = errors.Wrapf(err, "failed to access server (tried %d times)", i)
logrus.Debugf("connection failure: %v", err)
lastErr = err
time.Sleep(duration)
}
if lastErr != nil {
return nil, lastErr
}
p, v, r, err := c.Version(context.TODO())
if err != nil { if err != nil {
return nil, err return nil, err
} }
logrus.Debugf("connected to server (\"%v %v %v\")", p, v, r) logrus.Debugf("connected to server (\"%v %v %v\")", p, v, r)
if !(p == version.Package && v == version.Version && r == version.Revision) { if !(p == version.Package && v == version.Version && r == version.Revision) {
logrus.Warnf("version mismatch (server: \"%v %v %v\", client: \"%v %v %v\"); please kill and restart buildx server", return nil, errors.Errorf("version mismatch (client: \"%v %v %v\", server: \"%v %v %v\")", version.Package, version.Version, version.Revision, p, v, r)
p, v, r, version.Package, version.Version, version.Revision)
} }
return c, nil return c, nil
} }
@ -266,7 +273,7 @@ type buildxController struct {
} }
func (c *buildxController) Kill(ctx context.Context) error { func (c *buildxController) Kill(ctx context.Context) error {
pidB, err := os.ReadFile(filepath.Join(c.serverRoot, "pid")) pidB, err := os.ReadFile(filepath.Join(c.serverRoot, defaultPIDFilename))
if err != nil { if err != nil {
return err return err
} }
@ -289,7 +296,12 @@ func (c *buildxController) Kill(ctx context.Context) error {
} }
func launch(ctx context.Context, logFile string, args ...string) (func() error, error) { func launch(ctx context.Context, logFile string, args ...string) (func() error, error) {
bCmd := exec.CommandContext(ctx, os.Args[0], args...) // set absolute path of binary, since we set the working directory to the root
pathname, err := filepath.Abs(os.Args[0])
if err != nil {
return nil, err
}
bCmd := exec.CommandContext(ctx, pathname, args...)
if logFile != "" { if logFile != "" {
f, err := os.OpenFile(logFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) f, err := os.OpenFile(logFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil { if err != nil {

Loading…
Cancel
Save