homelab/hivemind
1
0
Fork 0
Code Activity

fix: don't spam nodes with updates and instead pull the registry on keepalive

Browse source
This commit is contained in:
Arthur K. 2026-01-22 16:02:44 +03:00
parent 7fb90dd1da
commit 476c4b056f
Signed by: wzray
GPG key ID: B97F30FDC4636357
5 changed files with 108 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

46
internal/roles/master/master.go
View file

@ -9,6 +9,7 @@ import (
"git.wzray.com/homelab/hivemind/internal/state"
"git.wzray.com/homelab/hivemind/internal/types"
"git.wzray.com/homelab/hivemind/internal/web/client"
"github.com/rs/zerolog/log"
)
type Role struct {
@ -36,7 +37,7 @@ func New(state *state.RuntimeState, config config.MasterConfig) *Role {
func (r *Role) OnStartup(ctx context.Context) error {
r.tasksGroup.Go(func() {
r.observer.Start(ctx, func(n types.Node) error {
_, err := r.onLeave(n)
_, err := r.onLeave(n, true)
return err
})
})
@ -50,7 +51,7 @@ func (r *Role) OnShutdown() error {
}
func (r *Role) notify(path types.Path, v any) {
for _, n := range r.state.Registry.Nodes() {
for _, n := range r.state.Registry.ByRole(types.MasterRole) {
addr := n.Endpoint
r.tasksGroup.Go(func() {
client.Post[any](addr, path, v)
@ -58,39 +59,58 @@ func (r *Role) notify(path types.Path, v any) {
}
}
func (r *Role) onJoin(node types.Node) (map[string]types.Node, error) {
func (r *Role) onJoin(node types.Node, notify bool) (map[string]types.Node, error) {
if err := r.state.Registry.AddNode(node); err != nil {
return nil, err
}
r.notify(types.PathNodeJoin, node)
if notify {
r.notify(types.PathMasterEventJoin, node)
}
return r.state.Registry.AllNodes(), nil
}
func (r *Role) onLeave(node types.Node) (bool, error) {
func (r *Role) onLeave(node types.Node, notify bool) (bool, error) {
if err := r.state.Registry.RemoveNode(node); err != nil {
return false, err
}
r.notify(types.PathNodeLeave, node)
if notify {
r.notify(types.PathMasterEventLeave, node)
}
return true, nil
}
func (r *Role) onKeepAlive(node types.Node) (bool, error) {
func (r *Role) onKeepAlive(node types.Node, notify bool) (map[string]types.Node, error) {
r.observer.onKeepAlive(node)
if ok := r.state.Registry.Exists(node.Hostname); !ok {
_, err := r.onJoin(node)
return true, err
// TODO: i don't like this side effect
if _, err := r.onJoin(node, true); err != nil {
log.Warn().Err(err).Msg("unable to add node to the registry from keepalive")
}
}
return false, nil
if notify {
r.notify(types.PathMasterEventKeepalive, node)
}
return r.state.Registry.AllNodes(), nil
}
func eventFunc[R any](fn func(types.Node, bool) (R, error), notify bool) func(types.Node) (R, error) {
return func(n types.Node) (R, error) {
return fn(n, notify)
}
}
func (c *Role) RegisterHandlers(r types.Registrator) {
r.Register(types.PostEndpoint(types.PathMasterJoin, c.onJoin))
r.Register(types.PostEndpoint(types.PathMasterLeave, c.onLeave))
r.Register(types.PostEndpoint(types.PathMasterKeepalive, c.onKeepAlive))
r.Register(types.PostEndpoint(types.PathMasterKeepalive, eventFunc(c.onKeepAlive, true)))
r.Register(types.PostEndpoint(types.PathMasterEventKeepalive, eventFunc(c.onKeepAlive, false)))
r.Register(types.PostEndpoint(types.PathMasterJoin, eventFunc(c.onJoin, true)))
r.Register(types.PostEndpoint(types.PathMasterLeave, eventFunc(c.onLeave, true)))
r.Register(types.PostEndpoint(types.PathMasterEventJoin, eventFunc(c.onJoin, false)))
r.Register(types.PostEndpoint(types.PathMasterEventLeave, eventFunc(c.onLeave, false)))
}

32
internal/roles/node/node.go
View file

@ -92,7 +92,6 @@ func (r *Role) Leave() error {
func (r *Role) OnStartup(ctx context.Context) error {
r.keepaliveGroup.Go(r.keepaliveFunc(ctx))
return nil
}
@ -107,11 +106,20 @@ func (r *Role) keepaliveFunc(ctx context.Context) func() {
logger := log.With().Str("name", m.Hostname).Logger()
logger.Debug().Msg("sending keepalive packet")
if _, err := client.Post[any](m.Endpoint, types.PathMasterKeepalive, r.state.Self); err != nil {
nodes, err := client.Post[map[string]types.Node](m.Endpoint, types.PathMasterKeepalive, r.state.Self)
if err != nil {
logger.Info().Err(err).Msg("unable to send keepalive packet")
} else {
logger.Debug().Msg("keepalive packet sent")
continue
}
logger.Debug().Msg("keepalive packet sent")
if err := r.state.Registry.Set(*nodes); err != nil {
logger.Warn().Err(err).Msg("unable to set masters nodes")
continue
}
break
}
}
@ -127,26 +135,10 @@ func (r *Role) keepaliveFunc(ctx context.Context) func() {
}
}
func (r *Role) onJoin(node types.Node) (bool, error) {
if err := r.state.Registry.AddNode(node); err != nil {
return false, err
}
return true, nil
}
func (r *Role) onLeave(node types.Node) (bool, error) {
if err := r.state.Registry.RemoveNode(node); err != nil {
return false, err
}
return true, nil
}
func healthcheck() (string, error) {
return "OK", nil
}
func (n *Role) RegisterHandlers(r types.Registrator) {
r.Register(types.GetEndpoint(types.PathNodeHealthcheck, healthcheck))
r.Register(types.PostEndpoint(types.PathNodeJoin, n.onJoin))
r.Register(types.PostEndpoint(types.PathNodeLeave, n.onLeave))
}

11
internal/types/web.go
View file

@ -15,13 +15,14 @@ func (p Path) String() string {
}
const (
PathMasterJoin Path = "/master/join"
PathMasterLeave Path = "/master/leave"
PathMasterKeepalive Path = "/master/keepalive"
PathMasterJoin Path = "/master/join"
PathMasterLeave Path = "/master/leave"
PathMasterKeepalive Path = "/master/keepalive"
PathMasterEventJoin Path = "/master/event_join"
PathMasterEventLeave Path = "/master/event_leave"
PathMasterEventKeepalive Path = "/master/event_keepalive"
PathNodeHealthcheck Path = "/node/healthcheck"
PathNodeJoin Path = "/node/join"
PathNodeLeave Path = "/node/leave"
PathDnsCallback Path = "/dns/callback"