micropelago/isle
4
1
Fork 0
Code Issues 1 Pull Requests Projects Releases 5 Activity

Implement JoinNetwork RPC method, and accompanying sub-command

Browse Source
This commit is contained in:
Brian Picciano 2024-07-07 12:44:49 +02:00
parent 81368821b7
commit 7d8b274445
12 changed files with 273 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
go/bootstrap/bootstrap.go
View File

@ -103,16 +103,29 @@ func FromFile(path string) (Bootstrap, error) {
defer f.Close()
var b Bootstrap
if err := json.NewDecoder(f).Decode(&b); err != nil {
return Bootstrap{}, fmt.Errorf("decoding json: %w", err)
}
if b.HostAssigned, err = b.SignedHostAssigned.UnwrapUnsafe(); err != nil {
return Bootstrap{}, fmt.Errorf("unwrapping host assigned: %w", err)
return b, nil
}
func (b *Bootstrap) UnmarshalJSON(data []byte) error {
type inner Bootstrap
err := json.Unmarshal(data, (*inner)(b))
if err != nil {
return err
}
return b, nil
b.HostAssigned, err = b.SignedHostAssigned.Unwrap(
b.CAPublicCredentials.SigningKey,
)
if err != nil {
return fmt.Errorf("unwrapping HostAssigned: %w", err)
}
return nil
}
// WriteTo writes the Bootstrap as a new bootstrap to the given io.Writer.
@ -123,7 +136,6 @@ func (b Bootstrap) WriteTo(into io.Writer) error {
// ThisHost is a shortcut for b.Hosts[b.HostName], but will panic if the
// HostName isn't found in the Hosts map.
func (b Bootstrap) ThisHost() Host {
host, ok := b.Hosts[b.Name]
if !ok {
panic(fmt.Sprintf("hostname %q not defined in bootstrap's hosts", b.Name))

6
go/bootstrap/hosts.go
View File

@ -78,5 +78,9 @@ type Host struct {
// This assumes that the Host and its data has already been verified against the
// CA signing key.
func (h Host) IP() net.IP {
return h.PublicCredentials.Cert.Unwrap().Details.Ips[0].IP
cert := h.PublicCredentials.Cert.Unwrap()
if len(cert.Details.Ips) == 0 {
panic(fmt.Sprintf("host %q not configured with any ips: %+v", h.Name, h))
}
return cert.Details.Ips[0].IP
}

74
go/cmd/entrypoint/daemon.go
View File

@ -2,15 +2,11 @@ package main
import (
"context"
"errors"
"fmt"
"io/fs"
"os"
"isle/bootstrap"
"isle/daemon"
"dev.mediocregopher.com/mediocre-go-lib.git/mctx"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
@ -31,11 +27,6 @@ var subCmdDaemon = subCmd{
"Write the default configuration file to stdout and exit.",
)
bootstrapPath := flags.StringP(
"bootstrap-path", "b", "",
`Path to a bootstrap.json file. This only needs to be provided the first time the daemon is started, after that it is ignored. If the isle binary has a bootstrap built into it then this argument is always optional.`,
)
logLevelStr := flags.StringP(
"log-level", "l", "info",
`Maximum log level which should be output. Values can be "debug", "info", "warn", "error", "fatal". Does not apply to sub-processes`,
@ -64,72 +55,17 @@ var subCmdDaemon = subCmd{
}
defer runtimeDirCleanup()
var (
bootstrapStateDirPath = bootstrap.StateDirPath(daemonEnvVars.StateDirPath)
bootstrapAppDirPath = bootstrap.AppDirPath(envAppDirPath)
hostBootstrapPath string
hostBootstrap bootstrap.Bootstrap
)
tryLoadBootstrap := func(path string) bool {
ctx := mctx.Annotate(ctx, "bootstrapFilePath", path)
if err != nil {
return false
} else if hostBootstrap, err = bootstrap.FromFile(path); errors.Is(err, fs.ErrNotExist) {
logger.WarnString(ctx, "bootstrap file not found")
err = nil
return false
} else if err != nil {
err = fmt.Errorf("parsing bootstrap.json at %q: %w", path, err)
return false
}
logger.Info(ctx, "bootstrap file found")
hostBootstrapPath = path
return true
}
switch {
case tryLoadBootstrap(bootstrapStateDirPath):
case *bootstrapPath != "" && tryLoadBootstrap(*bootstrapPath):
case tryLoadBootstrap(bootstrapAppDirPath):
case err != nil:
return fmt.Errorf("attempting to load bootstrap.json file: %w", err)
default:
return errors.New("No bootstrap.json file could be found, and one is not provided with --bootstrap-path")
}
if hostBootstrapPath != bootstrapStateDirPath {
// If the bootstrap file is not being stored in the data dir, copy
// it there, so it can be loaded from there next time.
if err := writeBootstrapToStateDir(hostBootstrap); err != nil {
return fmt.Errorf("writing bootstrap.json to data dir: %w", err)
}
}
daemonConfig, err := daemon.LoadConfig(envAppDirPath, *daemonConfigPath)
if err != nil {
return fmt.Errorf("loading daemon config: %w", err)
}
// we update this Host's data using whatever configuration has been
// provided by the daemon config. This way the daemon has the most
// up-to-date possible bootstrap. This updated bootstrap will later get
// updated in garage as a background daemon task, so other hosts will
// see it as well.
if hostBootstrap, err = coalesceDaemonConfigAndBootstrap(hostBootstrap, daemonConfig); err != nil {
return fmt.Errorf("merging daemon config into bootstrap data: %w", err)
}
daemonInst := daemon.NewDaemon(
logger, daemonConfig, envBinDirPath, hostBootstrap, nil,
daemonInst, err := daemon.NewDaemon(
logger, daemonConfig, envBinDirPath, nil,
)
if err != nil {
return fmt.Errorf("starting daemon: %w", err)
}
defer func() {
logger.Info(ctx, "Stopping child processes")
if err := daemonInst.Shutdown(); err != nil {

10
go/cmd/entrypoint/hosts.go
View File

@ -1,7 +1,6 @@
package main
import (
"encoding/json"
"errors"
"fmt"
"isle/bootstrap"
@ -30,17 +29,12 @@ var subCmdHostsList = subCmd{
ctx := subCmdCtx.ctx
var resRaw json.RawMessage
err := subCmdCtx.daemonRCPClient.Call(ctx, &resRaw, "GetHosts", nil)
var res daemon.GetHostsResult
err := subCmdCtx.daemonRCPClient.Call(ctx, &res, "GetHosts", nil)
if err != nil {
return fmt.Errorf("calling GetHosts: %w", err)
}
var res daemon.GetHostsResult
if err := json.Unmarshal(resRaw, &res); err != nil {
return fmt.Errorf("unmarshaling %s into %T: %w", string(resRaw), res, err)
}
type host struct {
Name string
VPN struct {

1
go/cmd/entrypoint/main.go
View File

@ -66,6 +66,7 @@ func main() {
subCmdGarage,
subCmdHosts,
subCmdNebula,
subCmdNetwork,
subCmdVersion,
)

51
go/cmd/entrypoint/network.go Normal file
View File

@ -0,0 +1,51 @@
package main
import (
"errors"
"fmt"
"isle/bootstrap"
)
var subCmdNetworkJoin = subCmd{
name: "join",
descr: "Joins this host to an existing network",
do: func(subCmdCtx subCmdCtx) error {
var (
ctx = subCmdCtx.ctx
flags = subCmdCtx.flagSet(false)
)
bootstrapPath := flags.StringP(
"bootstrap-path", "b", "", "Path to a bootstrap.json file.",
)
if err := flags.Parse(subCmdCtx.args); err != nil {
return fmt.Errorf("parsing flags: %w", err)
}
if *bootstrapPath == "" {
return errors.New("--bootstrap-path is required")
}
newBootstrap, err := bootstrap.FromFile(*bootstrapPath)
if err != nil {
return fmt.Errorf(
"loading bootstrap from %q: %w", *bootstrapPath, err,
)
}
return subCmdCtx.daemonRCPClient.Call(
ctx, nil, "JoinNetwork", newBootstrap,
)
},
}
var subCmdNetwork = subCmd{
name: "network",
descr: "Sub-commands related to network membership",
do: func(subCmdCtx subCmdCtx) error {
return subCmdCtx.doSubCmd(
subCmdNetworkJoin,
)
},
}

60
go/daemon/bootstrap.go
View File

@ -1,32 +1,14 @@
package daemon
import (
"errors"
"fmt"
"io/fs"
"os"
"path/filepath"
"isle/bootstrap"
"isle/garage/garagesrv"
)
func loadHostBootstrap(stateDirPath string) (bootstrap.Bootstrap, error) {
path := bootstrap.StateDirPath(stateDirPath)
hostBootstrap, err := bootstrap.FromFile(path)
if errors.Is(err, fs.ErrNotExist) {
return bootstrap.Bootstrap{}, fmt.Errorf(
"%q not found, has the daemon ever been run?",
stateDirPath,
)
} else if err != nil {
return bootstrap.Bootstrap{}, fmt.Errorf("loading %q: %w", stateDirPath, err)
}
return hostBootstrap, nil
}
func writeBootstrapToStateDir(
stateDirPath string, hostBootstrap bootstrap.Bootstrap,
) error {
@ -48,3 +30,43 @@ func writeBootstrapToStateDir(
return hostBootstrap.WriteTo(f)
}
func coalesceDaemonConfigAndBootstrap(
daemonConfig Config, hostBootstrap bootstrap.Bootstrap,
) (
bootstrap.Bootstrap, error,
) {
host := bootstrap.Host{
HostAssigned: hostBootstrap.HostAssigned,
HostConfigured: bootstrap.HostConfigured{
Nebula: bootstrap.NebulaHost{
PublicAddr: daemonConfig.VPN.PublicAddr,
},
},
}
if allocs := daemonConfig.Storage.Allocations; len(allocs) > 0 {
for i, alloc := range allocs {
id, rpcPort, err := garagesrv.InitAlloc(alloc.MetaPath, alloc.RPCPort)
if err != nil {
return bootstrap.Bootstrap{}, fmt.Errorf(
"initializing alloc at %q: %w", alloc.MetaPath, err,
)
}
host.Garage.Instances = append(host.Garage.Instances, bootstrap.GarageHostInstance{
ID: id,
RPCPort: rpcPort,
S3APIPort: alloc.S3APIPort,
})
allocs[i].RPCPort = rpcPort
}
}
hostBootstrap.Hosts[host.Name] = host
return hostBootstrap, nil
}

162
go/daemon/daemon.go
View File

@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"io"
"io/fs"
"isle/bootstrap"
"os"
"sync"
@ -20,6 +21,13 @@ import (
// with isle, typically via the unix socket.
type Daemon interface {
// JoinNetwork joins the Daemon to an existing network using the given
// Bootstrap.
//
// Errors:
// - ErrAlreadyJoined
JoinNetwork(context.Context, bootstrap.Bootstrap) error
// GetGarageBootstrapHosts loads (and verifies) the <hostname>.json.signed
// file for all hosts stored in garage.
GetGarageBootstrapHosts(
@ -67,7 +75,8 @@ func (o *Opts) withDefaults() *Opts {
}
const (
daemonStateInitializing = iota
daemonStateNoNetwork = iota
daemonStateInitializing
daemonStateOk
daemonStateRestarting
daemonStateShutdown
@ -79,13 +88,13 @@ type daemon struct {
envBinDirPath string
opts *Opts
l sync.Mutex
l sync.RWMutex
state int
children *Children
currBootstrap bootstrap.Bootstrap
cancelFn context.CancelFunc
stoppedCh chan struct{}
shutdownCh chan struct{}
wg sync.WaitGroup
}
// NewDaemon initializes and returns a Daemon instance which will manage all
@ -110,43 +119,89 @@ func NewDaemon(
logger *mlog.Logger,
daemonConfig Config,
envBinDirPath string,
currBootstrap bootstrap.Bootstrap,
opts *Opts,
) Daemon {
ctx, cancelFn := context.WithCancel(context.Background())
) (
Daemon, error,
) {
var (
d = &daemon{
logger: logger,
daemonConfig: daemonConfig,
envBinDirPath: envBinDirPath,
opts: opts.withDefaults(),
shutdownCh: make(chan struct{}),
}
bootstrapFilePath = bootstrap.StateDirPath(d.opts.EnvVars.StateDirPath)
)
d := &daemon{
logger: logger,
daemonConfig: daemonConfig,
envBinDirPath: envBinDirPath,
opts: opts.withDefaults(),
currBootstrap: currBootstrap,
cancelFn: cancelFn,
stoppedCh: make(chan struct{}),
currBootstrap, err := bootstrap.FromFile(bootstrapFilePath)
if errors.Is(err, fs.ErrNotExist) {
// daemon has never had a network created or joined
} else if err != nil {
return nil, fmt.Errorf(
"loading bootstrap from %q: %w", bootstrapFilePath, err,
)
} else if err := d.initialize(currBootstrap); err != nil {
return nil, fmt.Errorf("initializing with bootstrap: %w", err)
}
go func() {
d.restartLoop(ctx)
d.logger.Debug(ctx, "DaemonRestarter stopped")
close(d.stoppedCh)
}()
return d
return d, nil
}
func withInnerChildren[Res any](
d *daemon, fn func(*Children) (Res, error),
func (d *daemon) initialize(currBootstrap bootstrap.Bootstrap) error {
// we update this Host's data using whatever configuration has been provided
// by the daemon config. This way the daemon has the most up-to-date
// possible bootstrap. This updated bootstrap will later get updated in
// garage as a background daemon task, so other hosts will see it as well.
currBootstrap, err := coalesceDaemonConfigAndBootstrap(
d.daemonConfig, currBootstrap,
)
if err != nil {
return fmt.Errorf("combining daemon configuration into bootstrap: %w", err)
}
err = writeBootstrapToStateDir(d.opts.EnvVars.StateDirPath, currBootstrap)
if err != nil {
return fmt.Errorf("writing bootstrap to state dir: %w", err)
}
d.currBootstrap = currBootstrap
d.state = daemonStateInitializing
ctx, cancel := context.WithCancel(context.Background())
d.wg.Add(1)
go func() {
defer d.wg.Done()
<-d.shutdownCh
cancel()
}()
d.wg.Add(1)
go func() {
defer d.wg.Done()
d.restartLoop(ctx)
d.logger.Debug(ctx, "Daemon restart loop stopped")
}()
return nil
}
func withCurrBootstrap[Res any](
d *daemon, fn func(bootstrap.Bootstrap) (Res, error),
) (Res, error) {
var zero Res
d.l.Lock()
children, state := d.children, d.state
d.l.Unlock()
d.l.RLock()
defer d.l.RUnlock()
currBootstrap, state := d.currBootstrap, d.state
switch state {
case daemonStateNoNetwork:
return zero, ErrNoNetwork
case daemonStateInitializing:
return zero, ErrInitializing
case daemonStateOk:
return fn(children)
return fn(currBootstrap)
case daemonStateRestarting:
return zero, ErrRestarting
case daemonStateShutdown:
@ -167,7 +222,7 @@ func (d *daemon) checkBootstrap(
thisHost := hostBootstrap.ThisHost()
newHosts, err := d.getGarageBootstrapHosts(ctx)
newHosts, err := getGarageBootstrapHosts(ctx, d.logger, hostBootstrap)
if err != nil {
return bootstrap.Bootstrap{}, false, fmt.Errorf("getting hosts from garage: %w", err)
}
@ -233,19 +288,6 @@ func (d *daemon) watchForChanges(ctx context.Context) bootstrap.Bootstrap {
}
func (d *daemon) restartLoop(ctx context.Context) {
defer func() {
d.l.Lock()
d.state = daemonStateShutdown
children := d.children
d.l.Unlock()
if children != nil {
if err := children.Shutdown(); err != nil {
d.logger.Fatal(ctx, "Failed to cleanly shutdown daemon children, there may be orphaned child processes", err)
}
}
}()
wait := func(d time.Duration) bool {
select {
case <-ctx.Done():
@ -328,18 +370,46 @@ func (d *daemon) restartLoop(ctx context.Context) {
}
}
func (d *daemon) JoinNetwork(
ctx context.Context, newBootstrap bootstrap.Bootstrap,
) error {
d.l.Lock()
defer d.l.Unlock()
if d.state != daemonStateNoNetwork {
return ErrAlreadyJoined
}
return d.initialize(newBootstrap)
}
func (d *daemon) GetGarageBootstrapHosts(
ctx context.Context,
) (
map[string]bootstrap.Host, error,
) {
return withInnerChildren(d, func(*Children) (map[string]bootstrap.Host, error) {
return d.getGarageBootstrapHosts(ctx)
return withCurrBootstrap(d, func(
currBootstrap bootstrap.Bootstrap,
) (
map[string]bootstrap.Host, error,
) {
return getGarageBootstrapHosts(ctx, d.logger, currBootstrap)
})
}
func (d *daemon) Shutdown() error {
d.cancelFn()
<-d.stoppedCh
d.l.Lock()
defer d.l.Unlock()
close(d.shutdownCh)
d.wg.Wait()
d.state = daemonStateShutdown
if d.children != nil {
if err := d.children.Shutdown(); err != nil {
return fmt.Errorf("shutting down children: %w", err)
}
}
return nil
}

12
go/daemon/errors.go
View File

@ -3,11 +3,19 @@ package daemon
import "isle/daemon/jsonrpc2"
var (
// ErrNoNetwork is returned when the daemon has never been configured with a
// network.
ErrNoNetwork = jsonrpc2.NewError(1, "No network configured")
// ErrInitializing is returned when a network is unavailable due to still
// being initialized.
ErrInitializing = jsonrpc2.NewError(1, "Network is being initialized")
ErrInitializing = jsonrpc2.NewError(2, "Network is being initialized")
// ErrRestarting is returned when a network is unavailable due to being
// restarted.
ErrRestarting = jsonrpc2.NewError(2, "Network is being restarted")
ErrRestarting = jsonrpc2.NewError(3, "Network is being restarted")
// ErrAlreadyJoined is returned when the daemon is instructed to create or
// join a new network, but it is already joined to a network.
ErrAlreadyJoined = jsonrpc2.NewError(4, "Already joined to a network")
)

11
go/daemon/global_bucket.go
View File

@ -11,6 +11,7 @@ import (
"path/filepath"
"dev.mediocregopher.com/mediocre-go-lib.git/mctx"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
"github.com/minio/minio-go/v7"
)
@ -65,13 +66,13 @@ func (d *daemon) putGarageBoostrapHost(ctx context.Context) error {
return nil
}
func (d *daemon) getGarageBootstrapHosts(
ctx context.Context,
func getGarageBootstrapHosts(
ctx context.Context, logger *mlog.Logger, currBootstrap bootstrap.Bootstrap,
) (
map[string]bootstrap.Host, error,
) {
var (
b = d.currBootstrap
b = currBootstrap
client = b.GlobalBucketS3APIClient()
hosts = map[string]bootstrap.Host{}
@ -106,13 +107,13 @@ func (d *daemon) getGarageBootstrapHosts(
obj.Close()
if err != nil {
d.logger.Warn(ctx, "Object contains invalid json", err)
logger.Warn(ctx, "Object contains invalid json", err)
continue
}
host, err := authedHost.Unwrap(b.CAPublicCredentials)
if err != nil {
d.logger.Warn(ctx, "Host could not be authenticated", err)
logger.Warn(ctx, "Host could not be authenticated", err)
}
hosts[host.Name] = host

9
go/daemon/rpc.go
View File

@ -26,6 +26,15 @@ func NewRPC(daemon Daemon) *RPC {
return &RPC{daemon}
}
// JoinNetwork passes through to the Daemon method of the same name.
func (r *RPC) JoinNetwork(
ctx context.Context, req bootstrap.Bootstrap,
) (
struct{}, error,
) {
return struct{}{}, r.daemon.JoinNetwork(ctx, req)
}
// GetHosts returns all hosts known to the network, sorted by their name.
func (r *RPC) GetHosts(
ctx context.Context, req struct{},

15
tests/utils/with-1-data-1-empty-node-network.sh
View File

@ -60,10 +60,9 @@ EOF
isle daemon -l debug --config-path daemon.yml >daemon.log 2>&1 &
pid="$!"
echo "Waiting for primus daemon (process $pid) to initialize"
$SHELL "$UTILS/register-cleanup.sh" "$pid" "1-data-1-empty-node-network/primus"
echo "Waiting for primus daemon (process $pid) to initialize"
while ! isle hosts list >/dev/null; do sleep 1; done
echo "Creating secondus bootstrap"
@ -82,11 +81,17 @@ EOF
device: isle-secondus
EOF
isle daemon -l debug -c daemon.yml -b "$secondus_bootstrap" >daemon.log 2>&1 &
isle daemon -l debug -c daemon.yml >daemon.log 2>&1 &
pid="$!"
echo "Waiting for secondus daemon (process $!) to initialize"
$SHELL "$UTILS/register-cleanup.sh" "$pid" "1-data-1-empty-node-network/secondus"
echo "Waiting for secondus daemon (process $!) to start"
while ! [ -e "$ISLE_DAEMON_HTTP_SOCKET_PATH" ]; do sleep 1; done
echo "Joining secondus to the network"
isle network join -b "$secondus_bootstrap"
echo "Waiting for secondus daemon to join"
while ! isle hosts list >/dev/null; do sleep 1; done
)
fi