isle/go/daemon/daemon.go

477 lines
11 KiB
Go

// Package daemon implements the isle daemon, which is a long-running service
// managing all isle background tasks and sub-processes for a single network.
package daemon
import (
"context"
"errors"
"fmt"
"isle/bootstrap"
"isle/daemon/children"
"isle/daemon/daecommon"
"isle/daemon/network"
"isle/nebula"
"isle/toolkit"
"sort"
"sync"
"dev.mediocregopher.com/mediocre-go-lib.git/mctx"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
)
// Opts are optional parameters which can be passed in when initializing a new
// Daemon instance. A nil Opts is equivalent to a zero value.
type Opts struct {
ChildrenOpts *children.Opts
// Defaults to that returned by daecommon.GetEnvVars.
EnvVars daecommon.EnvVars
}
func (o *Opts) withDefaults() *Opts {
if o == nil {
o = new(Opts)
}
if o.EnvVars == (daecommon.EnvVars{}) {
o.EnvVars = daecommon.GetEnvVars()
}
return o
}
var _ RPC = (*Daemon)(nil)
// Daemon implements all methods of the Daemon interface, plus others used
// to manage this particular implementation.
//
// Daemon manages all child processes and state required by the isle
// service, as well as an HTTP socket over which RPC calls will be served.
//
// Inner Children instance(s) will be wrapped such that they will be
// automatically shutdown and re-created whenever there's changes in the network
// which require the configuration to be changed (e.g. a new nebula lighthouse).
// During such an inner restart all methods will return ErrRestarting, except
// Shutdown which will block until the currently executing restart is finished
// and then shutdown cleanly from there.
//
// While still starting up the Daemon for the first time all methods will return
// ErrInitializing, except Shutdown which will block until initialization is
// canceled.
type Daemon struct {
logger *mlog.Logger
daemonConfig daecommon.Config
envBinDirPath string
opts *Opts
networksStateDir toolkit.Dir
networksRuntimeDir toolkit.Dir
l sync.RWMutex
networks map[string]network.Network
}
// New initializes and returns a Daemon.
func New(
ctx context.Context,
logger *mlog.Logger,
daemonConfig daecommon.Config,
envBinDirPath string,
opts *Opts,
) (
*Daemon, error,
) {
opts = opts.withDefaults()
if err := migrateToMultiNetworkStateDirectory(
ctx,
logger.WithNamespace("migration-multi-network-state-dir"),
opts.EnvVars,
); err != nil {
return nil, fmt.Errorf(
"migrating to multi-network state directory: %w", err,
)
}
d := &Daemon{
logger: logger,
daemonConfig: daemonConfig,
envBinDirPath: envBinDirPath,
opts: opts,
networks: map[string]network.Network{},
}
{
h := new(toolkit.MkDirHelper)
d.networksStateDir, _ = h.Maybe(
d.opts.EnvVars.StateDir.MkChildDir("networks", true),
)
d.networksRuntimeDir, _ = h.Maybe(
d.opts.EnvVars.RuntimeDir.MkChildDir("networks", true),
)
if err := h.Err(); err != nil {
return nil, fmt.Errorf("creating networks sub-directories: %w", err)
}
}
loadableNetworks, err := loadableNetworks(d.networksStateDir)
if err != nil {
return nil, fmt.Errorf("listing loadable networks: %w", err)
}
for _, creationParams := range loadableNetworks {
id := creationParams.ID
ctx = mctx.WithAnnotator(ctx, creationParams)
networkStateDir, networkRuntimeDir, err := networkDirs(
d.networksStateDir, d.networksRuntimeDir, id, true,
)
if err != nil {
return nil, fmt.Errorf(
"creating sub-directories for network %q: %w", id, err,
)
}
networkConfig, _ := pickNetworkConfig(daemonConfig, creationParams)
d.networks[id], err = network.Load(
ctx,
logger.WithNamespace("network"),
id,
networkConfig,
d.envBinDirPath,
networkStateDir,
networkRuntimeDir,
&network.Opts{
ChildrenOpts: d.opts.ChildrenOpts,
},
)
if err != nil {
return nil, fmt.Errorf("loading network %q: %w", id, err)
}
}
return d, nil
}
// CreateNetwork will initialize a new network using the given parameters.
// - name: Human-readable name of the network.
// - domain: Primary domain name that network services are served under.
// - ipNet: An IP subnet, in CIDR form, which will be the overall range of
// possible IPs in the network. The first IP in this network range will become
// this first host's IP.
// - hostName: The name of this first host in the network.
//
// The daemon on which this is called will become the first host in the network,
// and will have full administrative privileges.
//
// Errors:
// - network.ErrInvalidConfig
func (d *Daemon) CreateNetwork(
ctx context.Context,
name, domain string, ipNet nebula.IPNet, hostName nebula.HostName,
) error {
creationParams := bootstrap.NewCreationParams(name, domain)
ctx = mctx.WithAnnotator(ctx, creationParams)
networkConfig, ok := pickNetworkConfig(
d.daemonConfig, creationParams,
)
if !ok {
return errors.New("couldn't find network config for network being created")
}
d.l.Lock()
defer d.l.Unlock()
if joined, err := alreadyJoined(ctx, d.networks, creationParams); err != nil {
return fmt.Errorf("checking if already joined to network: %w", err)
} else if joined {
return ErrAlreadyJoined
}
networkStateDir, networkRuntimeDir, err := networkDirs(
d.networksStateDir, d.networksRuntimeDir, creationParams.ID, false,
)
if err != nil {
return fmt.Errorf(
"creating sub-directories for network %q: %w",
creationParams.ID,
err,
)
}
d.logger.Info(ctx, "Creating network")
n, err := network.Create(
ctx,
d.logger.WithNamespace("network"),
networkConfig,
d.envBinDirPath,
networkStateDir,
networkRuntimeDir,
creationParams,
ipNet,
hostName,
&network.Opts{
ChildrenOpts: d.opts.ChildrenOpts,
},
)
if err != nil {
return fmt.Errorf("creating network: %w", err)
}
d.logger.Info(ctx, "Network created successfully")
d.networks[creationParams.ID] = n
return nil
}
// JoinNetwork joins the Daemon to an existing network using the given
// Bootstrap.
//
// Errors:
// - ErrAlreadyJoined
func (d *Daemon) JoinNetwork(
ctx context.Context, newBootstrap network.JoiningBootstrap,
) error {
var (
creationParams = newBootstrap.Bootstrap.NetworkCreationParams
networkConfig, _ = pickNetworkConfig(d.daemonConfig, creationParams)
networkID = creationParams.ID
)
ctx = mctx.WithAnnotator(ctx, newBootstrap.Bootstrap.NetworkCreationParams)
d.l.Lock()
defer d.l.Unlock()
if joined, err := alreadyJoined(ctx, d.networks, creationParams); err != nil {
return fmt.Errorf("checking if already joined to network: %w", err)
} else if joined {
return ErrAlreadyJoined
}
networkStateDir, networkRuntimeDir, err := networkDirs(
d.networksStateDir, d.networksRuntimeDir, networkID, false,
)
if err != nil {
return fmt.Errorf(
"creating sub-directories for network %q: %w", networkID, err,
)
}
d.logger.Info(ctx, "Joining network")
n, err := network.Join(
ctx,
d.logger.WithNamespace("network"),
networkConfig,
newBootstrap,
d.envBinDirPath,
networkStateDir,
networkRuntimeDir,
&network.Opts{
ChildrenOpts: d.opts.ChildrenOpts,
},
)
if err != nil {
return fmt.Errorf(
"joining network %q: %w", networkID, err,
)
}
d.logger.Info(ctx, "Network joined successfully")
d.networks[networkID] = n
return nil
}
func withNetwork[Res any](
ctx context.Context,
d *Daemon,
fn func(context.Context, network.Network) (Res, error),
) (
Res, error,
) {
d.l.RLock()
defer d.l.RUnlock()
network, err := pickNetwork(ctx, d.networks, d.networksStateDir)
if err != nil {
var zero Res
return zero, nil
}
return fn(ctx, network)
}
// GetNetworks returns all networks which have been joined by the Daemon,
// ordered by their name.
func (d *Daemon) GetNetworks(
ctx context.Context,
) (
[]bootstrap.CreationParams, error,
) {
d.l.RLock()
defer d.l.RUnlock()
res := make([]bootstrap.CreationParams, 0, len(d.networks))
for id, network := range d.networks {
creationParams, err := network.GetNetworkCreationParams(ctx)
if err != nil {
return nil, fmt.Errorf(
"getting network creation params of network %q: %w", id, err,
)
}
res = append(res, creationParams)
}
sort.Slice(res, func(i, j int) bool {
return res[i].Name < res[j].Name
})
return res, nil
}
// GetHost implements the method for the network.RPC interface.
func (d *Daemon) GetHosts(ctx context.Context) ([]bootstrap.Host, error) {
return withNetwork(
ctx,
d,
func(ctx context.Context, n network.Network) ([]bootstrap.Host, error) {
return n.GetHosts(ctx)
},
)
}
// GetGarageClientParams implements the method for the network.RPC interface.
func (d *Daemon) GetGarageClientParams(
ctx context.Context,
) (
network.GarageClientParams, error,
) {
return withNetwork(
ctx,
d,
func(
ctx context.Context, n network.Network,
) (
network.GarageClientParams, error,
) {
return n.GetGarageClientParams(ctx)
},
)
}
// GetNebulaCAPublicCredentials implements the method for the network.RPC
// interface.
func (d *Daemon) GetNebulaCAPublicCredentials(
ctx context.Context,
) (
nebula.CAPublicCredentials, error,
) {
return withNetwork(
ctx,
d,
func(
ctx context.Context, n network.Network,
) (
nebula.CAPublicCredentials, error,
) {
return n.GetNebulaCAPublicCredentials(ctx)
},
)
}
// RemoveHost implements the method for the network.RPC interface.
func (d *Daemon) RemoveHost(ctx context.Context, hostName nebula.HostName) error {
_, err := withNetwork(
ctx,
d,
func(
ctx context.Context, n network.Network,
) (
struct{}, error,
) {
return struct{}{}, n.RemoveHost(ctx, hostName)
},
)
return err
}
// CreateHost implements the method for the network.RPC interface.
func (d *Daemon) CreateHost(
ctx context.Context,
hostName nebula.HostName,
opts network.CreateHostOpts,
) (
network.JoiningBootstrap, error,
) {
return withNetwork(
ctx,
d,
func(
ctx context.Context, n network.Network,
) (
network.JoiningBootstrap, error,
) {
return n.CreateHost(ctx, hostName, opts)
},
)
}
// CreateNebulaCertificate implements the method for the network.RPC interface.
func (d *Daemon) CreateNebulaCertificate(
ctx context.Context,
hostName nebula.HostName,
hostPubKey nebula.EncryptingPublicKey,
) (
nebula.Certificate, error,
) {
return withNetwork(
ctx,
d,
func(
ctx context.Context, n network.Network,
) (
nebula.Certificate, error,
) {
return n.CreateNebulaCertificate(ctx, hostName, hostPubKey)
},
)
}
// Shutdown blocks until all resources held or created by the daemon,
// including child processes it has started, have been cleaned up.
//
// If this returns an error then it's possible that child processes are
// still running and are no longer managed.
func (d *Daemon) Shutdown() error {
d.l.Lock()
defer d.l.Unlock()
var (
errCh = make(chan error, len(d.networks))
errs []error
)
for id := range d.networks {
var (
id = id
n = d.networks[id]
)
go func() {
if err := n.Shutdown(); err != nil {
errCh <- fmt.Errorf("shutting down network %q: %w", id, err)
}
errCh <- nil
}()
}
for range cap(errCh) {
errs = append(errs, <-errCh)
}
return errors.Join(errs...)
}