diff --git a/cmd/hivemind/main.go b/cmd/hivemind/main.go index e6cff3c..617dad7 100644 --- a/cmd/hivemind/main.go +++ b/cmd/hivemind/main.go @@ -76,8 +76,9 @@ func main() { zerolog.SetGlobalLevel(levelToZerolog(config.Node.LogLevel)) self := types.NewNode( - fmt.Sprintf("%v:%v", config.Node.Endpoint, config.Node.Port), config.Node.Hostname, + config.Node.Address, + config.Node.Port, config.Roles, ) diff --git a/internal/config/node.go b/internal/config/node.go index 2ffa6db..4049bfe 100644 --- a/internal/config/node.go +++ b/internal/config/node.go @@ -3,6 +3,7 @@ package config import ( "errors" "fmt" + "net" "strings" ) @@ -28,29 +29,34 @@ func (l *LogLevel) UnmarshalText(data []byte) error { } type NodeConfig struct { - Hostname string `toml:"hostname"` - Endpoint string `toml:"endpoint"` + Hostname string `toml:"hostname"` + Address string `toml:"address"` + Port int `toml:"port"` + KeepaliveInterval int `toml:"keepalive_interval"` LogLevel LogLevel `toml:"log_level"` BootstrapMaster string `toml:"bootstrap_master"` ListenOn string `toml:"listen_on"` - Port int `toml:"port"` } func (c NodeConfig) Validate() error { + if c.Address == "" { + return errors.New("missing address") + } + if c.Hostname == "" { return errors.New("missing hostname") } - if c.Endpoint == "" { - return errors.New("missing endpoint") - } - if c.KeepaliveInterval < 1 && c.KeepaliveInterval != -1 { return errors.New("invalid keepalive_interval") } + if net.ParseIP(c.ListenOn) == nil { + return errors.New("invalid listen_on") + } + return nil } @@ -59,8 +65,8 @@ func (c *NodeConfig) Merge(other NodeConfig) { c.Hostname = other.Hostname } - if other.Endpoint != "" { - c.Endpoint = other.Endpoint + if other.Address != "" { + c.Address = other.Address } if other.BootstrapMaster != "" { diff --git a/internal/registry/registry.go b/internal/registry/registry.go index d3a5a4d..8269c47 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -36,7 +36,7 @@ func New(storage Storage, self types.Node) *Registry { r.nodes = storedData.Nodes ret: - r.nodes[self.Name] = self + r.nodes[self.Hostname] = self return r } @@ -68,7 +68,7 @@ func (r *Registry) AllNodes() []types.Node { func (r *Registry) Nodes() []types.Node { nodes := r.AllNodes() nodes = slices.DeleteFunc(nodes, func(n types.Node) bool { - return n.Name == r.self.Name + return n.Hostname == r.self.Hostname }) return nodes } @@ -79,7 +79,7 @@ func (r *Registry) ByRole(role types.Role) []types.Node { o := make([]types.Node, 0, len(r.nodes)) for _, node := range r.nodes { - if slices.Contains(node.Roles, role) && node.Name != r.self.Name { + if slices.Contains(node.Roles, role) && node.Hostname != r.self.Hostname { o = append(o, node) } } @@ -88,7 +88,7 @@ func (r *Registry) ByRole(role types.Role) []types.Node { func (r *Registry) AddNode(node types.Node) error { r.lock.Lock() - r.nodes[node.Name] = node + r.nodes[node.Hostname] = node r.LastUpdate = time.Now() snapshot := r.snapshot() r.lock.Unlock() @@ -119,7 +119,7 @@ func (r *Registry) Set(nodes []types.Node) error { r.lock.Lock() r.nodes = make(map[string]types.Node) for _, n := range nodes { - r.nodes[n.Name] = n + r.nodes[n.Hostname] = n } snapshot := r.snapshot() r.lock.Unlock() diff --git a/internal/roles/dns/dns.go b/internal/roles/dns/dns.go index 66efb45..2149e5e 100644 --- a/internal/roles/dns/dns.go +++ b/internal/roles/dns/dns.go @@ -48,10 +48,10 @@ func parseState(state types.HostState) (string, []byte) { var builder strings.Builder for _, d := range state.Domains { - builder.WriteString(fmt.Sprintf("%s %s\n", state.Endpoint, d)) + builder.WriteString(fmt.Sprintf("%s %s\n", state.Address, d)) } - return hostsDir + state.Name, []byte(builder.String()) + return hostsDir + state.Hostname, []byte(builder.String()) } func (r *Role) OnStartup(ctx context.Context) error { @@ -76,15 +76,15 @@ func (r *Role) OnStartup(ctx context.Context) error { func (r *Role) syncFromRegistry() { for _, n := range r.state.Registry.ByRole(types.HostRole) { - state, err := client.Get[types.HostState](n.Address, types.PathHostDns) + state, err := client.Get[types.HostState](n.Endpoint, types.PathHostDns) if err != nil { - log.Warn().Str("name", n.Name).Err(err).Msg("unable to get host config") + log.Warn().Str("name", n.Hostname).Err(err).Msg("unable to get host config") continue } filename, data := parseState(*state) if err := r.updateDnsmasq(filename, data); err != nil { - log.Warn().Str("name", n.Name).Err(err).Msg("unable to update dnsmasq") + log.Warn().Str("name", n.Hostname).Err(err).Msg("unable to update dnsmasq") continue } } diff --git a/internal/roles/host/host.go b/internal/roles/host/host.go index 6bdea06..ca91d8b 100644 --- a/internal/roles/host/host.go +++ b/internal/roles/host/host.go @@ -16,11 +16,11 @@ import ( ) type Role struct { - state *state.RuntimeState - config config.HostConfig + state *state.RuntimeState + config config.HostConfig - client *traefikClient - tasksGroup sync.WaitGroup + client *traefikClient + tasksGroup sync.WaitGroup externalDomains []string // TODO: i don't like hardcoding external/internal logic here internalDomains []string @@ -37,15 +37,15 @@ func New(state *state.RuntimeState, config config.HostConfig) *Role { func (r *Role) sendUpdate(domains []string, role types.Role) { state := types.HostState{ Domains: domains, - Name: r.state.Self.Name, - Endpoint: r.state.Self.Address, + Hostname: r.state.Self.Hostname, + Address: r.state.Self.Endpoint, } for _, node := range r.state.Registry.ByRole(role) { r.tasksGroup.Go(func() { - logger := log.With().Str("name", node.Name).Logger() + logger := log.With().Str("name", node.Hostname).Logger() logger.Debug().Msg("sending update") - if _, err := client.Post[any](node.Address, types.PathDnsCallback, state); err != nil { + if _, err := client.Post[any](node.Endpoint, types.PathDnsCallback, state); err != nil { logger.Warn().Err(err).Msg("unable to send dns info") } else { logger.Debug().Msg("update sent") @@ -87,8 +87,8 @@ func (r *Role) onCallback(w http.ResponseWriter, req *http.Request) { func (r *Role) getInternal() (types.HostState, error) { return types.HostState{ Domains: r.internalDomains, - Endpoint: r.config.IpAddress, - Name: r.state.Self.Name, + Address: r.config.IpAddress, + Hostname: r.state.Self.Hostname, }, nil } diff --git a/internal/roles/master/master.go b/internal/roles/master/master.go index 823b385..91ffac8 100644 --- a/internal/roles/master/master.go +++ b/internal/roles/master/master.go @@ -50,7 +50,7 @@ func (r *Role) OnShutdown() error { func (r *Role) notify(path types.Path, v any) { for _, n := range r.state.Registry.Nodes() { - addr := n.Address + addr := n.Endpoint r.tasksGroup.Go(func() { client.Post[any](addr, path, v) }) @@ -68,17 +68,17 @@ func (r *Role) onJoin(node types.Node) ([]types.Node, error) { } func (r *Role) onLeave(node types.Node) (bool, error) { - if err := r.state.Registry.RemoveNode(node.Name); err != nil { + if err := r.state.Registry.RemoveNode(node.Hostname); err != nil { return false, err } - r.notify(types.PathNodeLeave, node.Name) + r.notify(types.PathNodeLeave, node.Hostname) return true, nil } func (r *Role) onKeepAlive(node types.Node) (bool, error) { - if ok := r.state.Registry.Exists(node.Name); !ok { + if ok := r.state.Registry.Exists(node.Hostname); !ok { _, err := r.onJoin(node) return true, err } diff --git a/internal/roles/master/observer.go b/internal/roles/master/observer.go index f293133..8c4c61d 100644 --- a/internal/roles/master/observer.go +++ b/internal/roles/master/observer.go @@ -33,14 +33,14 @@ func newObserver( func (o *observer) pollNodes(ctx context.Context, onLeave func(types.Node) error) { for _, n := range o.state.Registry.Nodes() { - name := n.Name + name := n.Hostname logger := log.With().Str("name", name).Logger() logger.Debug().Msg("checking node") delay := time.Duration(o.backoff) alive := false for i := o.backoffCount; i > 0; i-- { - _, err := client.Get[any](n.Address, types.PathNodeHealthcheck) + _, err := client.Get[any](n.Endpoint, types.PathNodeHealthcheck) if err == nil { logger.Debug().Msg("node is alive") diff --git a/internal/roles/node/node.go b/internal/roles/node/node.go index b795835..9a0c1b9 100644 --- a/internal/roles/node/node.go +++ b/internal/roles/node/node.go @@ -29,10 +29,10 @@ func New(state *state.RuntimeState, config config.NodeConfig) *Role { func (r *Role) Join(bootstrap string) error { masters := make(map[string]struct{}) for _, node := range r.state.Registry.ByRole(types.MasterRole) { - if node.Name == r.state.Self.Name { + if node.Hostname == r.state.Self.Hostname { continue } - masters[node.Address] = struct{}{} + masters[node.Endpoint] = struct{}{} } if bootstrap != "" { masters[bootstrap] = struct{}{} @@ -69,10 +69,10 @@ func (r *Role) Leave() error { sent := false for _, m := range masters { - logger := log.With().Str("name", m.Name).Logger() + logger := log.With().Str("name", m.Hostname).Logger() logger.Debug().Msg("sending leave message") - _, err := client.Post[any](m.Address, types.PathMasterLeave, r.state.Self) + _, err := client.Post[any](m.Endpoint, types.PathMasterLeave, r.state.Self) if err != nil { logger.Debug().Err(err).Msg("unable to send leave message") continue @@ -114,10 +114,10 @@ func (r *Role) keepaliveFunc(ctx context.Context) func() { sent := false for _, m := range masters { - logger := log.With().Str("name", m.Name).Logger() + logger := log.With().Str("name", m.Hostname).Logger() logger.Debug().Msg("sending keepalive packet") - if _, err := client.Post[any](m.Address, types.PathMasterKeepalive, r.state.Self); err != nil { + if _, err := client.Post[any](m.Endpoint, types.PathMasterKeepalive, r.state.Self); err != nil { continue } else { logger.Debug().Msg("keepalive packet sent") @@ -151,7 +151,7 @@ func (r *Role) onJoin(node types.Node) (bool, error) { } func (r *Role) onLeave(node types.Node) (bool, error) { - if err := r.state.Registry.RemoveNode(node.Name); err != nil { + if err := r.state.Registry.RemoveNode(node.Hostname); err != nil { return false, err } return true, nil diff --git a/internal/types/host.go b/internal/types/host.go index c483e75..c9e2219 100644 --- a/internal/types/host.go +++ b/internal/types/host.go @@ -2,6 +2,6 @@ package types type HostState struct { Domains []string - Endpoint string - Name string + Address string + Hostname string } diff --git a/internal/types/node.go b/internal/types/node.go index a01b651..d1ed308 100644 --- a/internal/types/node.go +++ b/internal/types/node.go @@ -1,16 +1,27 @@ package types +import "fmt" + // TODO: consider moving this type back to registry type Node struct { - Address string `json:"address"` - Name string `json:"name"` - Roles []Role `json:"roles"` + Hostname string `json:"hostname"` + Address string `json:"address"` + Port int `json:"port"` + Roles []Role `json:"roles"` + Endpoint string `json:"endpoint"` } -func NewNode(address string, name string, roles []Role) Node { +func NewNode( + hostname string, + address string, + port int, + roles []Role, +) Node { return Node{ - Address: address, - Name: name, - Roles: roles, + Hostname: hostname, + Address: address, + Port: port, + Roles: roles, + Endpoint: fmt.Sprintf("%s:%d", address, port), } } diff --git a/internal/types/rpc.go b/internal/types/rpc.go deleted file mode 100644 index ab1254f..0000000 --- a/internal/types/rpc.go +++ /dev/null @@ -1 +0,0 @@ -package types diff --git a/internal/types/web.go b/internal/types/web.go index 17ab7f9..1416d6a 100644 --- a/internal/types/web.go +++ b/internal/types/web.go @@ -6,6 +6,8 @@ import ( "net/http" ) +// TODO: split this up + type Path string func (p Path) String() string {