Manage child processes individually, not via Pmux. Also route child logs through mlog

This commit is contained in:
Brian Picciano 2024-10-29 15:11:13 +01:00
parent b7c097ef63
commit 7274815cfd
13 changed files with 243 additions and 206 deletions

View File

@ -32,7 +32,7 @@ func main() {
logger := mlog.NewLogger(&mlog.LoggerOpts{
MessageHandler: newLogMsgHandler(),
MaxLevel: mlog.LevelInfo.Int(),
MaxLevel: mlog.LevelInfo.Int(), // TODO make this configurable
})
defer logger.Close()

View File

@ -6,8 +6,6 @@ import (
"context"
"errors"
"fmt"
"io"
"os"
"code.betamike.com/micropelago/pmux/pmuxlib"
"dev.mediocregopher.com/mediocre-go-lib.git/mctx"
@ -21,25 +19,13 @@ import (
// Opts are optional parameters which can be passed in when initializing a new
// Children instance. A nil Opts is equivalent to a zero value.
type Opts struct {
// Stdout and Stderr are what the associated outputs from child processes
// will be directed to.
Stdout, Stderr io.Writer
}
type Opts struct{}
func (o *Opts) withDefaults() *Opts {
if o == nil {
o = new(Opts)
}
if o.Stdout == nil {
o.Stdout = os.Stdout
}
if o.Stderr == nil {
o.Stderr = os.Stderr
}
return o
}
@ -50,13 +36,16 @@ func (o *Opts) withDefaults() *Opts {
// - garage (0 or more, depending on configured storage allocations)
type Children struct {
logger *mlog.Logger
binDirPath string
runtimeDir toolkit.Dir
garageAdminToken string
opts Opts
garageRPCSecret string
pmux *pmuxlib.Pmux
nebulaProc *pmuxlib.Process
dnsmasqProc *pmuxlib.Process
garageProcs map[string]*pmuxlib.Process
}
// New initializes and returns a Children instance. If initialization fails an
@ -84,33 +73,66 @@ func New(
c := &Children{
logger: logger,
binDirPath: binDirPath,
runtimeDir: runtimeDir,
garageAdminToken: garageAdminToken,
opts: *opts,
garageRPCSecret: garageRPCSecret,
}
pmuxConfig, err := c.newPmuxConfig(
if c.nebulaProc, err = nebulaPmuxProc(
ctx,
c.logger,
c.runtimeDir.Path,
c.binDirPath,
networkConfig,
hostBootstrap,
); err != nil {
return nil, fmt.Errorf("starting nebula: %w", err)
}
if err := waitForNebula(ctx, c.logger, hostBootstrap); err != nil {
logger.Warn(ctx, "Failed waiting for nebula to initialize, shutting down child processes", err)
c.Shutdown()
return nil, fmt.Errorf("waiting for nebula to start: %w", err)
}
if c.dnsmasqProc, err = dnsmasqPmuxProc(
ctx,
c.logger,
c.runtimeDir.Path,
c.binDirPath,
networkConfig,
hostBootstrap,
); err != nil {
logger.Warn(ctx, "Failed to start dnsmasq, shutting down child processes", err)
c.Shutdown()
return nil, fmt.Errorf("starting dnsmasq: %w", err)
}
// TODO wait for dnsmasq to come up
if c.garageProcs, err = garagePmuxProcs(
ctx,
c.logger,
garageRPCSecret,
binDirPath,
c.runtimeDir.Path,
c.binDirPath,
networkConfig,
garageAdminToken,
hostBootstrap,
)
if err != nil {
return nil, fmt.Errorf("generating pmux config: %w", err)
); err != nil {
logger.Warn(ctx, "Failed to start garage processes, shutting down child processes", err)
c.Shutdown()
return nil, fmt.Errorf("starting garage processes: %w", err)
}
c.pmux = pmuxlib.NewPmux(pmuxConfig, c.opts.Stdout, c.opts.Stderr)
initErr := c.postPmuxInit(
ctx, networkConfig, garageAdminToken, hostBootstrap,
)
if initErr != nil {
logger.Warn(ctx, "failed to initialize Children, shutting down child processes", err)
if err := waitForGarage(
ctx, c.logger, networkConfig, garageAdminToken, hostBootstrap,
); err != nil {
logger.Warn(ctx, "Failed waiting for garage processes to initialize, shutting down child processes", err)
c.Shutdown()
return nil, initErr
return nil, fmt.Errorf("waiting for garage processes to initialize: %w", err)
}
return c, nil
@ -133,7 +155,7 @@ func (c *Children) reloadDNSMasq(
}
c.logger.Info(ctx, "dnsmasq config file has changed, restarting process")
c.pmux.Restart("dnsmasq")
c.dnsmasqProc.Restart()
return nil
}
@ -153,7 +175,7 @@ func (c *Children) reloadNebula(
}
c.logger.Info(ctx, "nebula config file has changed, restarting process")
c.pmux.Restart("nebula")
c.nebulaProc.Restart()
if err := waitForNebula(ctx, c.logger, hostBootstrap); err != nil {
return fmt.Errorf("waiting for nebula to start: %w", err)
@ -184,7 +206,7 @@ func (c *Children) reloadGarage(
)
)
_, changed, err := garageWriteChildConfig(
childConfigPath, changed, err := garageWriteChildConfig(
ctx,
c.logger,
c.garageRPCSecret,
@ -202,8 +224,16 @@ func (c *Children) reloadGarage(
anyChanged = true
c.logger.Info(ctx, "garage config has changed, restarting process")
c.pmux.Restart(procName)
if proc, ok := c.garageProcs[procName]; ok {
c.logger.Info(ctx, "garage config has changed, restarting process")
proc.Restart()
continue
}
c.logger.Info(ctx, "garage config has been added, creating process")
c.garageProcs[procName] = garagePmuxProc(
ctx, c.logger, c.binDirPath, procName, childConfigPath,
)
}
if anyChanged {
@ -244,5 +274,15 @@ func (c *Children) Reload(
// Shutdown blocks until all child processes have gracefully shut themselves
// down.
func (c *Children) Shutdown() {
c.pmux.Stop()
for _, proc := range c.garageProcs {
proc.Stop()
}
if c.dnsmasqProc != nil {
c.dnsmasqProc.Stop()
}
if c.nebulaProc != nil {
c.nebulaProc.Stop()
}
}

View File

@ -49,37 +49,36 @@ func dnsmasqWriteConfig(
return confPath, changed, nil
}
func dnsmasqPmuxProcConfig(
// TODO consider a shared dnsmasq across all the daemon's networks.
// This would have a few benefits:
// - Less processes, less problems
// - Less configuration for the user in the case of more than one network.
// - Can listen on 127.0.0.x:53, rather than on the nebula address. This
// allows DNS to come up before nebula, which is helpful when nebula depends
// on DNS.
func dnsmasqPmuxProc(
ctx context.Context,
logger *mlog.Logger,
runtimeDirPath, binDirPath string,
networkConfig daecommon.NetworkConfig,
hostBootstrap bootstrap.Bootstrap,
) (
pmuxlib.ProcessConfig, error,
*pmuxlib.Process, error,
) {
confPath, _, err := dnsmasqWriteConfig(
ctx, logger, runtimeDirPath, networkConfig, hostBootstrap,
)
if err != nil {
return pmuxlib.ProcessConfig{}, fmt.Errorf(
return nil, fmt.Errorf(
"writing dnsmasq config: %w", err,
)
}
return pmuxlib.ProcessConfig{
cfg := pmuxlib.ProcessConfig{
Cmd: filepath.Join(binDirPath, "dnsmasq"),
Args: []string{"-d", "-C", confPath},
StartAfterFunc: func(ctx context.Context) error {
// TODO consider a shared dnsmasq across all the daemon's networks.
// This would have a few benefits:
// - Less processes, less problems
// - Less configuration for the user in the case of more than one
// network.
// - Can listen on 127.0.0.x:53, rather than on the nebula address.
// This allows DNS to come up before nebula, which is helpful when
// nebula depends on DNS.
return waitForNebula(ctx, logger, hostBootstrap)
},
}, nil
}
cfg = withPmuxLoggers(ctx, logger, "dnsmasq", cfg)
return pmuxlib.NewProcess(cfg), nil
}

View File

@ -120,7 +120,24 @@ func garagePmuxProcName(alloc daecommon.ConfigStorageAllocation) string {
return fmt.Sprintf("garage-%d", alloc.RPCPort)
}
func garagePmuxProcConfigs(
func garagePmuxProc(
ctx context.Context,
logger *mlog.Logger,
binDirPath string,
procName string,
childConfigPath string,
) *pmuxlib.Process {
cfg := pmuxlib.ProcessConfig{
Cmd: filepath.Join(binDirPath, "garage"),
Args: []string{"-c", childConfigPath, "server"},
}
cfg = withPmuxLoggers(ctx, logger, procName, cfg)
return pmuxlib.NewProcess(cfg)
}
func garagePmuxProcs(
ctx context.Context,
logger *mlog.Logger,
rpcSecret, runtimeDirPath, binDirPath string,
@ -128,11 +145,11 @@ func garagePmuxProcConfigs(
adminToken string,
hostBootstrap bootstrap.Bootstrap,
) (
map[string]pmuxlib.ProcessConfig, error,
map[string]*pmuxlib.Process, error,
) {
var (
pmuxProcConfigs = map[string]pmuxlib.ProcessConfig{}
allocs = networkConfig.Storage.Allocations
pmuxProcs = map[string]*pmuxlib.Process{}
allocs = networkConfig.Storage.Allocations
)
if len(allocs) > 0 && rpcSecret == "" {
@ -152,14 +169,10 @@ func garagePmuxProcConfigs(
}
procName := garagePmuxProcName(alloc)
pmuxProcConfigs[procName] = pmuxlib.ProcessConfig{
Cmd: filepath.Join(binDirPath, "garage"),
Args: []string{"-c", childConfigPath, "server"},
StartAfterFunc: func(ctx context.Context) error {
return waitForNebula(ctx, logger, hostBootstrap)
},
}
pmuxProcs[procName] = garagePmuxProc(
ctx, logger, binDirPath, procName, childConfigPath,
)
}
return pmuxProcConfigs, nil
return pmuxProcs, nil
}

View File

@ -0,0 +1,59 @@
package children
import (
"context"
"fmt"
"isle/toolkit"
"code.betamike.com/micropelago/pmux/pmuxlib"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
type pmuxLogger struct {
ctx context.Context
logger *mlog.Logger
}
func newPmuxStdoutLogger(
ctx context.Context, logger *mlog.Logger, name string,
) pmuxlib.Logger {
return &pmuxLogger{ctx, logger.WithNamespace(name).WithNamespace("out")}
}
func newPmuxStderrLogger(
ctx context.Context, logger *mlog.Logger, name string,
) pmuxlib.Logger {
return &pmuxLogger{ctx, logger.WithNamespace(name).WithNamespace("err")}
}
func newPmuxSysLogger(
ctx context.Context, logger *mlog.Logger, name string,
) pmuxlib.Logger {
return &pmuxLogger{ctx, logger.WithNamespace(name).WithNamespace("sys")}
}
func (l *pmuxLogger) Println(line string) {
l.logger.Log(mlog.Message{
Context: l.ctx,
Level: toolkit.LogLevelChild,
Description: line,
})
}
func (l *pmuxLogger) Printf(format string, args ...any) {
l.Println(fmt.Sprintf(format, args...))
}
////////////////////////////////////////////////////////////////////////////////
func withPmuxLoggers(
ctx context.Context,
logger *mlog.Logger,
name string,
cfg pmuxlib.ProcessConfig,
) pmuxlib.ProcessConfig {
cfg.StdoutLogger = newPmuxStdoutLogger(ctx, logger, name)
cfg.StderrLogger = newPmuxStderrLogger(ctx, logger, name)
cfg.SysLogger = newPmuxSysLogger(ctx, logger, name)
return cfg
}

View File

@ -174,27 +174,27 @@ func nebulaWriteConfig(
return nebulaYmlPath, changed, nil
}
func nebulaPmuxProcConfig(
func nebulaPmuxProc(
ctx context.Context,
logger *mlog.Logger,
runtimeDirPath, binDirPath string,
networkConfig daecommon.NetworkConfig,
hostBootstrap bootstrap.Bootstrap,
) (
pmuxlib.ProcessConfig, error,
*pmuxlib.Process, error,
) {
nebulaYmlPath, _, err := nebulaWriteConfig(
ctx, logger, runtimeDirPath, networkConfig, hostBootstrap,
)
if err != nil {
return pmuxlib.ProcessConfig{}, fmt.Errorf(
"writing nebula config: %w", err,
)
return nil, fmt.Errorf("writing nebula config: %w", err)
}
return pmuxlib.ProcessConfig{
Cmd: filepath.Join(binDirPath, "nebula"),
Args: []string{"-config", nebulaYmlPath},
Group: -1, // Make sure nebula is shut down last.
}, nil
cfg := pmuxlib.ProcessConfig{
Cmd: filepath.Join(binDirPath, "nebula"),
Args: []string{"-config", nebulaYmlPath},
}
cfg = withPmuxLoggers(ctx, logger, "nebula", cfg)
return pmuxlib.NewProcess(cfg), nil
}

View File

@ -1,90 +0,0 @@
package children
import (
"context"
"fmt"
"isle/bootstrap"
"isle/daemon/daecommon"
"code.betamike.com/micropelago/pmux/pmuxlib"
)
func (c *Children) newPmuxConfig(
ctx context.Context,
garageRPCSecret, binDirPath string,
networkConfig daecommon.NetworkConfig,
garageAdminToken string,
hostBootstrap bootstrap.Bootstrap,
) (
pmuxlib.Config, error,
) {
nebulaPmuxProcConfig, err := nebulaPmuxProcConfig(
ctx,
c.logger,
c.runtimeDir.Path,
binDirPath,
networkConfig,
hostBootstrap,
)
if err != nil {
return pmuxlib.Config{}, fmt.Errorf("generating nebula config: %w", err)
}
dnsmasqPmuxProcConfig, err := dnsmasqPmuxProcConfig(
ctx,
c.logger,
c.runtimeDir.Path,
binDirPath,
networkConfig,
hostBootstrap,
)
if err != nil {
return pmuxlib.Config{}, fmt.Errorf(
"generating dnsmasq config: %w", err,
)
}
garagePmuxProcConfigs, err := garagePmuxProcConfigs(
ctx,
c.logger,
garageRPCSecret,
c.runtimeDir.Path,
binDirPath,
networkConfig,
garageAdminToken,
hostBootstrap,
)
if err != nil {
return pmuxlib.Config{}, fmt.Errorf(
"generating garage children configs: %w", err,
)
}
pmuxProcConfigs := garagePmuxProcConfigs
pmuxProcConfigs["nebula"] = nebulaPmuxProcConfig
pmuxProcConfigs["dnsmasq"] = dnsmasqPmuxProcConfig
return pmuxlib.Config{
Processes: pmuxProcConfigs,
}, nil
}
func (c *Children) postPmuxInit(
ctx context.Context,
networkConfig daecommon.NetworkConfig,
garageAdminToken string,
hostBootstrap bootstrap.Bootstrap,
) error {
if err := waitForNebula(ctx, c.logger, hostBootstrap); err != nil {
return fmt.Errorf("waiting for nebula to start: %w", err)
}
err := waitForGarage(
ctx, c.logger, networkConfig, garageAdminToken, hostBootstrap,
)
if err != nil {
return fmt.Errorf("waiting for garage to start: %w", err)
}
return nil
}

View File

@ -4,6 +4,7 @@ import (
"context"
"errors"
"io"
"isle/toolkit"
"net"
"net/http/httptest"
"path/filepath"
@ -11,8 +12,6 @@ import (
"sync"
"testing"
"time"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
type DivideParams struct {
@ -72,7 +71,7 @@ type divider interface {
func testHandler(t *testing.T) Handler {
var (
logger = mlog.NewTestLogger(t)
logger = toolkit.NewTestLogger(t)
d = divider(dividerImpl{})
)

View File

@ -4,9 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"isle/bootstrap"
"isle/daemon/children"
"isle/daemon/daecommon"
"isle/nebula"
"isle/toolkit"
@ -100,10 +98,8 @@ func newIntegrationHarness(t *testing.T) *integrationHarness {
})
return &integrationHarness{
ctx: context.Background(),
logger: mlog.NewLogger(&mlog.LoggerOpts{
MessageHandler: mlog.NewTestMessageHandler(t),
}),
ctx: context.Background(),
logger: toolkit.NewTestLogger(t),
rootDir: toolkit.Dir{Path: rootDir},
}
}
@ -170,36 +166,6 @@ func (h *integrationHarness) mkNetworkConfig(
)
}
func (h *integrationHarness) mkChildrenOpts(
t *testing.T, runtimeDir toolkit.Dir,
) *children.Opts {
var (
childrenLogFilePath = filepath.Join(runtimeDir.Path, "children.log")
childrenOpts children.Opts
)
childrenLogFile, err := os.Create(childrenLogFilePath)
require.NoError(t, err)
t.Cleanup(func() {
assert.NoError(t, err)
})
if os.Getenv("ISLE_INTEGRATION_TEST_CHILDREN_LOG_STDOUT") == "" {
childrenOpts = children.Opts{
Stdout: childrenLogFile,
Stderr: childrenLogFile,
}
} else {
childrenOpts = children.Opts{
Stdout: io.MultiWriter(os.Stdout, childrenLogFile),
Stderr: io.MultiWriter(os.Stdout, childrenLogFile),
}
}
return &childrenOpts
}
type createNetworkOpts struct {
creationParams bootstrap.CreationParams
manualShutdown bool
@ -248,7 +214,6 @@ func (h *integrationHarness) createNetwork(
hostName = nebula.HostName(hostNameStr)
networkOpts = &Opts{
ChildrenOpts: h.mkChildrenOpts(t, runtimeDir),
GarageAdminToken: "admin_token",
}
)
@ -327,7 +292,6 @@ func (h *integrationHarness) joinNetwork(
stateDir = h.mkDir(t, "state")
runtimeDir = h.mkDir(t, "runtime")
networkOpts = &Opts{
ChildrenOpts: h.mkChildrenOpts(t, runtimeDir),
GarageAdminToken: "admin_token",
}
)

View File

@ -3,7 +3,7 @@ module isle
go 1.22
require (
code.betamike.com/micropelago/pmux v0.0.0-20240719134913-f5fce902e8c4
code.betamike.com/micropelago/pmux v0.0.0-20241029124408-55bc7097f7b8
dev.mediocregopher.com/mediocre-go-lib.git v0.0.0-20241023182613-55984cdf5233
github.com/adrg/xdg v0.4.0
github.com/jxskiss/base62 v1.1.0

View File

@ -1,5 +1,7 @@
code.betamike.com/micropelago/pmux v0.0.0-20240719134913-f5fce902e8c4 h1:n4pGP12kgdH5kCTF4TZYpOjWuAR/zZLpM9j3neeVMEk=
code.betamike.com/micropelago/pmux v0.0.0-20240719134913-f5fce902e8c4/go.mod h1:WlEWacLREVfIQl1IlBjKzuDgL56DFRvyl7YiL5gGP4w=
code.betamike.com/micropelago/pmux v0.0.0-20241029124408-55bc7097f7b8 h1:DTAMr60/y9ZtpMORg50LYGpxyvA9+3UFKVW4mb4CwAE=
code.betamike.com/micropelago/pmux v0.0.0-20241029124408-55bc7097f7b8/go.mod h1:WlEWacLREVfIQl1IlBjKzuDgL56DFRvyl7YiL5gGP4w=
dev.mediocregopher.com/mediocre-go-lib.git v0.0.0-20241023182613-55984cdf5233 h1:Ea4HixNfDNDPh7zMngPpEeDf8gpociSPEROBFBedqIY=
dev.mediocregopher.com/mediocre-go-lib.git v0.0.0-20241023182613-55984cdf5233/go.mod h1:nP+AtQWrc3k5qq5y3ABiBLkOfUPlk/FO9fpTFpF+jgs=
github.com/adrg/xdg v0.4.0 h1:RzRqFcjH4nE5C6oTAxhBtoE2IRyjBSa62SCbyPidvls=

32
go/toolkit/logging.go Normal file
View File

@ -0,0 +1,32 @@
package toolkit
import (
"strings"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
type logLevel struct {
level int
name string
}
func (l logLevel) Int() int { return l.level }
func (l logLevel) String() string { return l.name }
var (
// LogLevelChild is used for logging out the stdout, stderr, and system logs
// (from pmux) related to child processes.
LogLevelChild mlog.Level = logLevel{mlog.LevelInfo.Int() + 1, "CHILD"}
)
// LogLevelFromString parses a string as a log level, taking into account custom
// log levels introduced in Isle.
func LogLevelFromString(str string) mlog.Level {
switch strings.TrimSpace(strings.ToUpper(str)) {
case LogLevelChild.String():
return LogLevelChild
default:
return mlog.LevelFromString(str)
}
}

View File

@ -1,8 +1,11 @@
package toolkit
import (
"fmt"
"os"
"testing"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
// MarkIntegrationTest marks a test as being an integration test. It will be
@ -12,3 +15,19 @@ func MarkIntegrationTest(t *testing.T) {
t.Skip("Skipped because ISLE_INTEGRATION_TEST isn't set")
}
}
// NewTestLogger returns a Logger which should be used for testing purposes. The
// log level of the Logger can be adjusted using the ISLE_LOG_LEVEL envvar.
func NewTestLogger(t *testing.T) *mlog.Logger {
level := mlog.LevelInfo
if levelStr := os.Getenv("ISLE_LOG_LEVEL"); levelStr != "" {
if level = LogLevelFromString(levelStr); level == nil {
panic(fmt.Sprintf("invalid log level: %q", levelStr))
}
}
return mlog.NewLogger(&mlog.LoggerOpts{
MessageHandler: mlog.NewTestMessageHandler(t),
MaxLevel: level.Int(),
})
}