mesh-tools
6a2fb63bfc
A module that left a node kept its have key at running for ever, so nothing could tell it stopped and the gate would never take the seat. The key is written once with the phase removed over the revision read, never deleted. Pins mesh-sdk to the commit that adds the phase.
319 lines
10 KiB
Go
319 lines
10 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"sync"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold.
|
|
var ErrHaveAbsent = errors.New("link: no such key")
|
|
|
|
// HaveBus is a link that can bind the have bucket (novox/hq ADR 0429 §4). Separate from Bus, as HealthBus is: a link that
|
|
// cannot is still a link, and the write waits for the next apply or look.
|
|
type HaveBus interface {
|
|
// HaveBind binds the bucket the controller made, once for a pass of writes, within ctx. A node whose grant is not there
|
|
// yet gets an error here (or the context's end), and writes nothing.
|
|
HaveBind(ctx context.Context) (HaveBucket, error)
|
|
}
|
|
|
|
// HaveBucket is the bound bucket.
|
|
type HaveBucket interface {
|
|
// Get is a key's value and last revision, or ErrHaveAbsent.
|
|
Get(ctx context.Context, key string) (value []byte, revision uint64, err error)
|
|
// Set writes a key over the revision read: zero creates it, and a key that moved since is refused.
|
|
Set(ctx context.Context, key string, value []byte, revision uint64) error
|
|
}
|
|
|
|
// haveState is what the node-engine last learned of each module of the declaration it applied, and of each module's health
|
|
// as the last look found it — kept so a look that changes only a health can write only that module's have (issue 1132).
|
|
type haveState struct {
|
|
mu sync.Mutex
|
|
modules map[string]ModuleFact
|
|
health map[string]delivery.Health
|
|
node *delivery.HaveNode
|
|
// written is the build of each module this node-engine wrote a have key for and that the declaration it applied
|
|
// still holds; removed is each such module the latest declaration no longer holds, with the build its key last
|
|
// said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted).
|
|
written map[string]string
|
|
removed map[string]string
|
|
}
|
|
|
|
// moduleHealths is the health of each module a look names, from the states of its resources: failed when any is
|
|
// unhealthy, waits when any waits (waiting, starting, held), ok when all are healthy, unknown otherwise.
|
|
func moduleHealths(h *Health) map[string]delivery.Health {
|
|
out := map[string]delivery.Health{}
|
|
if h == nil {
|
|
return out
|
|
}
|
|
rank := map[string]int{}
|
|
for _, r := range h.Resources {
|
|
if r.Module == "" {
|
|
continue
|
|
}
|
|
var this int
|
|
switch r.State {
|
|
case StateUnhealthy:
|
|
this = 3
|
|
case StateWaiting, StateStarting, StateHeld:
|
|
this = 2
|
|
case StateHealthy:
|
|
this = 0
|
|
default:
|
|
this = 1
|
|
}
|
|
if was, seen := rank[r.Module]; !seen || this > was {
|
|
rank[r.Module] = this
|
|
}
|
|
}
|
|
for module, r := range rank {
|
|
out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r]
|
|
}
|
|
return out
|
|
}
|
|
|
|
// nodeFacts is the node's own facts of a report, each as the JSON the report carries it in.
|
|
func nodeFacts(r Report) delivery.HaveNode {
|
|
n := delivery.HaveNode{V: delivery.Version}
|
|
carry := func(into *json.RawMessage, empty bool, v any) {
|
|
if empty {
|
|
return
|
|
}
|
|
if raw, err := json.Marshal(v); err == nil {
|
|
*into = raw
|
|
}
|
|
}
|
|
carry(&n.Windows, len(r.Windows) == 0, r.Windows)
|
|
carry(&n.Busy, len(r.Busy) == 0, r.Busy)
|
|
carry(&n.Strays, len(r.Strays) == 0, r.Strays)
|
|
carry(&n.Filters, len(r.Filters) == 0, r.Filters)
|
|
carry(&n.Profile, len(r.Profile) == 0, r.Profile)
|
|
return n
|
|
}
|
|
|
|
// noteApply takes what an apply's report says of the modules and the node.
|
|
func (q *Queue) noteApply(r Report) bool {
|
|
if r.Modules == nil {
|
|
return false
|
|
}
|
|
s := &q.haves
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
next := map[string]ModuleFact{}
|
|
for _, m := range r.Modules {
|
|
next[m.Module] = m
|
|
}
|
|
for module, build := range s.written {
|
|
if _, held := next[module]; !held {
|
|
if s.removed == nil {
|
|
s.removed = map[string]string{}
|
|
}
|
|
s.removed[module] = build
|
|
delete(s.written, module)
|
|
}
|
|
}
|
|
for module := range next {
|
|
delete(s.removed, module) // added back: its have is written with its new phase
|
|
}
|
|
s.modules = next
|
|
if r.Health != nil {
|
|
s.health = moduleHealths(r.Health)
|
|
}
|
|
n := nodeFacts(r)
|
|
s.node = &n
|
|
return true
|
|
}
|
|
|
|
// noteHealth takes a look's health, and says whether any module's changed.
|
|
func (q *Queue) noteHealth(h Health) bool {
|
|
s := &q.haves
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
next := moduleHealths(&h)
|
|
changed := false
|
|
for module := range s.modules {
|
|
want, ok := next[module]
|
|
if !ok {
|
|
want = delivery.HealthUnknown
|
|
}
|
|
have, known := s.health[module]
|
|
if !known {
|
|
have = delivery.HealthUnknown
|
|
}
|
|
if have != want {
|
|
changed = true
|
|
}
|
|
}
|
|
s.health = next
|
|
return changed
|
|
}
|
|
|
|
// sayHave is a line in the node-engine's log, if it has one.
|
|
func (q *Queue) sayHave(s string) {
|
|
if q.Say != nil {
|
|
q.Say(s)
|
|
}
|
|
}
|
|
|
|
// writeHaves writes this node's keys in the have bucket where they differ from what the bucket holds (novox/hq ADR 0429
|
|
// §4, issue 1132): each module's have, and the node's own facts. It reads the key first and writes nothing when the value
|
|
// is the same, otherwise writes over the revision it read, and creates the key when there is none. A write the bucket
|
|
// refuses is said and left: the next apply or look writes again, and the report was sent before this ran.
|
|
func (q *Queue) writeHaves(ctx context.Context, bus Bus) {
|
|
hb, ok := bus.(HaveBus)
|
|
if !ok || bus == nil {
|
|
return
|
|
}
|
|
s := &q.haves
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.modules == nil && s.node == nil && len(s.removed) == 0 {
|
|
return
|
|
}
|
|
node := q.Membership.Node
|
|
// The whole pass is bounded by the queue's timeout, and it stops at the first error: a node whose grant is not there
|
|
// yet would otherwise make the worker wait the bound again for every key (issue 1132).
|
|
write, cancel := context.WithTimeout(context.WithoutCancel(ctx), q.timeout())
|
|
defer cancel()
|
|
bucket, err := hb.HaveBind(write)
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("could not bind the have bucket, and wrote nothing there; the next apply or look writes again: %v", err))
|
|
return
|
|
}
|
|
modules := make([]string, 0, len(s.modules))
|
|
for module := range s.modules {
|
|
modules = append(modules, module)
|
|
}
|
|
sort.Strings(modules)
|
|
for _, module := range modules {
|
|
f := s.modules[module]
|
|
h := delivery.Have{V: delivery.Version, Build: f.Build, Phase: delivery.PhaseRunning,
|
|
Health: delivery.HealthUnknown, Detail: f.Detail}
|
|
if f.Failed {
|
|
h.Phase = delivery.PhaseFailed
|
|
}
|
|
if got, known := s.health[module]; known {
|
|
h.Health = got
|
|
}
|
|
key, err := delivery.HaveKey(node, module)
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
|
|
continue
|
|
}
|
|
raw, err := delivery.EncodeHave(h)
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
|
|
continue
|
|
}
|
|
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
|
c, err := delivery.DecodeHave(current)
|
|
return err == nil && delivery.SameHave(c, h)
|
|
}) {
|
|
return
|
|
}
|
|
if s.written == nil {
|
|
s.written = map[string]string{}
|
|
}
|
|
s.written[module] = f.Build
|
|
}
|
|
if !q.writeRemoved(write, bucket, node) {
|
|
return
|
|
}
|
|
if s.node != nil {
|
|
key, err := delivery.HaveNodeKey(node)
|
|
if err == nil {
|
|
var raw []byte
|
|
if raw, err = delivery.EncodeHaveNode(*s.node); err == nil {
|
|
want := *s.node
|
|
q.putHave(write, bucket, key, raw, func(current []byte) bool {
|
|
c, err := delivery.DecodeHaveNode(current)
|
|
return err == nil && delivery.SameHaveNode(c, want)
|
|
})
|
|
}
|
|
}
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("cannot write this node's own facts: %v", err))
|
|
}
|
|
}
|
|
}
|
|
|
|
// writeRemoved writes, once, each module the applied declaration no longer holds and this node-engine wrote a have key
|
|
// for, with the phase removed over the revision it reads (issue 1369). The key is never deleted, and a key that is not
|
|
// in the bucket is not created. False when the pass should stop. The caller holds the lock of the haves.
|
|
func (q *Queue) writeRemoved(ctx context.Context, b HaveBucket, node string) bool {
|
|
s := &q.haves
|
|
gone := make([]string, 0, len(s.removed))
|
|
for module := range s.removed {
|
|
gone = append(gone, module)
|
|
}
|
|
sort.Strings(gone)
|
|
for _, module := range gone {
|
|
key, err := delivery.HaveKey(node, module)
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
|
|
delete(s.removed, module)
|
|
continue
|
|
}
|
|
h := delivery.Have{V: delivery.Version, Build: s.removed[module], Phase: delivery.PhaseRemoved,
|
|
Health: delivery.HealthUnknown, Detail: "the applied declaration no longer holds the module"}
|
|
raw, err := delivery.EncodeHave(h)
|
|
if err != nil {
|
|
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
|
|
delete(s.removed, module)
|
|
continue
|
|
}
|
|
if ctx.Err() != nil {
|
|
q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
|
|
return false
|
|
}
|
|
current, revision, err := b.Get(ctx, key)
|
|
switch {
|
|
case errors.Is(err, ErrHaveAbsent):
|
|
delete(s.removed, module) // nothing to say: the key is not there
|
|
continue
|
|
case err != nil:
|
|
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
|
|
return false
|
|
}
|
|
if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved {
|
|
delete(s.removed, module)
|
|
continue
|
|
}
|
|
if err := b.Set(ctx, key, raw, revision); err != nil {
|
|
q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
|
|
return false
|
|
}
|
|
delete(s.removed, module)
|
|
}
|
|
return true
|
|
}
|
|
|
|
// putHave reads a key and writes it unless same says it already holds the value. False when the pass should stop: the
|
|
// bucket failed, the error is said once, and the rest wait for the next pass.
|
|
func (q *Queue) putHave(ctx context.Context, b HaveBucket, key string, value []byte, same func([]byte) bool) bool {
|
|
if ctx.Err() != nil {
|
|
q.sayHave(fmt.Sprintf("the time for writing the have bucket ran out before %s; the next apply or look writes it", key))
|
|
return false
|
|
}
|
|
current, revision, err := b.Get(ctx, key)
|
|
switch {
|
|
case errors.Is(err, ErrHaveAbsent):
|
|
revision = 0
|
|
case err != nil:
|
|
q.sayHave(fmt.Sprintf("could not read %s in the have bucket, and wrote nothing more there: %v", key, err))
|
|
return false
|
|
case same(current):
|
|
return true
|
|
}
|
|
if err := b.Set(ctx, key, value, revision); err != nil {
|
|
q.sayHave(fmt.Sprintf("could not write %s in the have bucket; the next apply or look writes it again: %v", key, err))
|
|
return false
|
|
}
|
|
return true
|
|
}
|