286 lines
9.5 KiB
Go
286 lines
9.5 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"github.com/novox/mesh-host/internal/apply"
|
|
"github.com/novox/mesh-host/internal/link"
|
|
"github.com/novox/mesh-host/internal/store"
|
|
)
|
|
|
|
// A library is judged on every look (novox/hq ADR 0328 §8, §9).
|
|
//
|
|
// Each library this machine is reached by is said as a resource `library:<name>` of kind library, with no module:
|
|
// healthy while it is kept (or, off its home, while its mount is there), waiting with an access wait of reason keep
|
|
// or disk, unknown while its filesystem could not be named, unhealthy otherwise. The root is the target, never in a
|
|
// reason nor in a wait's words: the controller says them on the operator's channel, which withholds a path (ADR
|
|
// 0234 §6). An access is judged by nothing, as before ADR 0325.
|
|
|
|
// libraryJudge reads every library on every look. Safe for the apply and the looking loop at once.
|
|
type libraryJudge struct {
|
|
mu sync.Mutex
|
|
libraries []declaration.Library
|
|
declared *declaration.Declaration
|
|
// was is each library's last verdict, with since when and how many looks in a row found it unhealthy.
|
|
was map[string]link.ResourceHealth
|
|
|
|
// reading guards inFlight: every library whose look has not returned yet.
|
|
reading sync.Mutex
|
|
inFlight map[string]bool
|
|
}
|
|
|
|
// accesses is the one library judge on this machine (its name from when it judged accesses).
|
|
var accesses = newLibraryJudge()
|
|
|
|
func newLibraryJudge() *libraryJudge {
|
|
return &libraryJudge{was: map[string]link.ResourceHealth{}, inFlight: map[string]bool{}}
|
|
}
|
|
|
|
// set is what the declaration just applied declares: every library. One no longer declared is forgotten.
|
|
func (j *libraryJudge) set(d *declaration.Declaration, _ map[string]bool) {
|
|
var ls []declaration.Library
|
|
if d != nil {
|
|
ls = append(ls, d.Libraries...)
|
|
}
|
|
sort.Slice(ls, func(a, b int) bool { return ls[a].Library < ls[b].Library })
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
j.libraries, j.declared = ls, d
|
|
declared := map[string]bool{}
|
|
for _, l := range ls {
|
|
declared[apply.LibraryAccess(l.Library)] = true
|
|
}
|
|
for id := range j.was {
|
|
if !declared[id] {
|
|
delete(j.was, id)
|
|
}
|
|
}
|
|
}
|
|
|
|
// declaration is the declaration last set.
|
|
func (j *libraryJudge) declaration() *declaration.Declaration {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
return j.declared
|
|
}
|
|
|
|
// lookBound is how long a look waits for one library to answer. A hard network mount whose server is down blocks
|
|
// a stat for ever, and one library must never stop every health statement this machine makes.
|
|
var lookBound = 5 * time.Second
|
|
|
|
// verdict is one library's judgement in a look.
|
|
type verdict struct {
|
|
state, reason string
|
|
waits []link.AccessWait
|
|
act *link.AccessActRecord
|
|
}
|
|
|
|
// judgeBounded judges one library within lookBound, outside any lock. A look that has not answered is said so;
|
|
// while it still hangs, later looks start no other on the same library.
|
|
func (j *libraryJudge) judgeBounded(l declaration.Library) (verdict, bool) {
|
|
j.reading.Lock()
|
|
if j.inFlight[l.Library] {
|
|
j.reading.Unlock()
|
|
return verdict{}, false
|
|
}
|
|
j.inFlight[l.Library] = true
|
|
j.reading.Unlock()
|
|
done := make(chan verdict, 1)
|
|
go func() {
|
|
v := judgeLibrary(l)
|
|
j.reading.Lock()
|
|
delete(j.inFlight, l.Library)
|
|
j.reading.Unlock()
|
|
done <- v
|
|
}()
|
|
select {
|
|
case v := <-done:
|
|
return v, true
|
|
case <-time.After(lookBound):
|
|
return verdict{}, false
|
|
}
|
|
}
|
|
|
|
// judgeLibrary is one library's judgement now (ADR 0328 §7, §8, §9).
|
|
func judgeLibrary(l declaration.Library) verdict {
|
|
v := apply.JudgeLibrary(context.Background(), l)
|
|
access := apply.LibraryAccess(l.Library)
|
|
var out verdict
|
|
if a := v.Act; a != nil {
|
|
out.act = &link.AccessActRecord{Act: a.Act, Access: access, Warrant: a.Warrant, By: a.By, State: a.State,
|
|
Started: a.Started, Ended: a.Ended, Folders: a.Folders, Files: a.Files, Kept: a.Kept, Links: a.Links,
|
|
Other: a.Other, Mounts: a.Mounts, Shared: a.Shared, Linked: a.Linked, Stripped: a.Stripped, Why: a.Why}
|
|
}
|
|
switch v.Kind {
|
|
case apply.VerdictWaitKeep:
|
|
out.state, out.reason = link.StateWaiting, "its root holds data the mesh has not taken on the filesystem "+
|
|
"it is on now, and waits for the operator's word (novox/hq ADR 0328 §7, §8)"
|
|
out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
|
|
What: apply.KeepWords(l.Library), Placed: v.Placed, Reason: link.WaitKeep}}
|
|
case apply.VerdictWaitDisk:
|
|
out.state, out.reason = link.StateWaiting, v.Why+" (novox/hq ADR 0328 §9)"
|
|
out.waits = []link.AccessWait{{Kind: link.WaitKindAccess, Access: access, Part: l.Library,
|
|
What: apply.DiskWords(l.Library), Reason: link.WaitDisk}}
|
|
case apply.VerdictUnknown:
|
|
// Not a fault and not the operator's: said as not known, and judged again on the next look.
|
|
out.state, out.reason = link.StateUnknown, v.Why+"; judged again on the next look"
|
|
case apply.VerdictBroken:
|
|
out.state, out.reason = link.StateUnhealthy, v.Why
|
|
default:
|
|
out.state = link.StateHealthy
|
|
switch {
|
|
case v.Act != nil && v.Act.State == link.ActRunning:
|
|
out.state, out.reason = link.StateStarting, "the take act is setting its group under its root"
|
|
case v.Act != nil && (v.Act.State == link.ActStopped || v.Act.State == link.ActFailed):
|
|
out.state, out.reason = link.StateStarting, "the take act goes on from where it stopped: "+v.Act.Why
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// look reads every library now. Whether any verdict changed since the last look, for the loop to say it.
|
|
func (j *libraryJudge) look(now time.Time) ([]link.ResourceHealth, bool) {
|
|
j.mu.Lock()
|
|
ls := append([]declaration.Library(nil), j.libraries...)
|
|
j.mu.Unlock()
|
|
|
|
read := make([]verdict, len(ls))
|
|
for i, l := range ls {
|
|
v, answered := j.judgeBounded(l)
|
|
if !answered {
|
|
v = verdict{state: link.StateUnknown, reason: "its root does not answer: judging it did not return " +
|
|
"within " + lookBound.String()}
|
|
}
|
|
read[i] = v
|
|
}
|
|
|
|
// And every tool runner's account the last apply could not put in a group, as a resource of its own (ADR 0334
|
|
// §2): unhealthy, never failing the apply.
|
|
warnings := apply.RunnerWarnings()
|
|
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
changed := false
|
|
out := make([]link.ResourceHealth, 0, len(ls)+len(warnings))
|
|
for _, w := range warnings {
|
|
v := link.ResourceHealth{Module: w.Module, Resource: w.Resource, Kind: link.KindRunner, State: link.StateUnhealthy,
|
|
Reason: "the tool runner's account joins no group: " + w.Why, Since: now.UTC(), Streak: 1}
|
|
if was, had := j.was[w.Resource]; had && was.Reason == v.Reason {
|
|
v.Since, v.Streak = was.Since, was.Streak+1
|
|
} else {
|
|
changed = true
|
|
}
|
|
j.was[w.Resource] = v
|
|
out = append(out, v)
|
|
}
|
|
for i, l := range ls {
|
|
r := read[i]
|
|
id := apply.LibraryAccess(l.Library)
|
|
v := link.ResourceHealth{Resource: id, Kind: link.KindLibrary, Target: l.Root, State: r.state,
|
|
Reason: r.reason, Since: now.UTC(), AccessWaits: r.waits, Act: r.act}
|
|
if was, had := j.was[id]; had && was.State == r.state && was.Reason == r.reason {
|
|
v.Since = was.Since
|
|
v.Streak = was.Streak
|
|
} else {
|
|
changed = true
|
|
}
|
|
if r.state == link.StateUnhealthy {
|
|
v.Streak++
|
|
}
|
|
j.was[id] = v
|
|
out = append(out, v)
|
|
}
|
|
return out, changed
|
|
}
|
|
|
|
// withAccesses adds to a statement every library, as a resource of kind library.
|
|
func withAccesses(h *link.Health, vs []link.ResourceHealth) *link.Health {
|
|
h.Resources = append(h.Resources, vs...)
|
|
return h
|
|
}
|
|
|
|
// accessesHealthy says no library in vs is anything but healthy.
|
|
func accessesHealthy(vs []link.ResourceHealth) bool {
|
|
for _, v := range vs {
|
|
if v.State != link.StateHealthy {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// keeper goes on with take acts stopped at their bound, beside the applies, one at a time (ADR 0328 §7).
|
|
type keeper struct {
|
|
running atomic.Bool
|
|
say func(string)
|
|
}
|
|
|
|
// keeping is the one keeper on this machine.
|
|
var keeping = &keeper{say: func(string) {}}
|
|
|
|
// poke goes on with every stopped take act of the declaration last applied, unless that is running already.
|
|
func (k *keeper) poke(ctx context.Context) {
|
|
d := accesses.declaration()
|
|
if d == nil || len(d.Libraries) == 0 || !k.running.CompareAndSwap(false, true) {
|
|
return
|
|
}
|
|
go func() {
|
|
defer k.running.Store(false)
|
|
for _, line := range apply.ReconcileLibraries(ctx, d) {
|
|
k.say(line)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// startKeeping has the keeper say what it does.
|
|
func startKeeping(ctx context.Context, say func(string)) {
|
|
keeping.say = say
|
|
keeping.poke(ctx)
|
|
}
|
|
|
|
// accessActFrom is the take act as the controller asks it over the link (ADR 0328 §7): judged against the
|
|
// declaration this node keeps, and run beside the applies.
|
|
func accessActFrom(statePath string, timeout time.Duration) link.AccessActFunc {
|
|
return func(act link.AccessAct) (string, error) {
|
|
d, err := declaredNow(statePath, nil)
|
|
if errors.Is(err, store.ErrNothingDeclared) {
|
|
return "", errors.New("this node keeps no declaration, so it is declared no library")
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("what runs on this node does not read: %w", err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
_, said, err := apply.TakeLibrary(ctx, d, act.Access, filepath.Clean(act.Path), act.Warrant, act.By)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "mesh-host: the take act on %s was refused: %v\n", act.Access, err)
|
|
return "", err
|
|
}
|
|
return said, nil
|
|
}
|
|
}
|
|
|
|
// withWaiting is held with every id of waiting beside it: what liveness does not judge.
|
|
func withWaiting(held, waiting map[string]bool) map[string]bool {
|
|
if len(waiting) == 0 {
|
|
return held
|
|
}
|
|
out := map[string]bool{}
|
|
for id := range held {
|
|
out[id] = true
|
|
}
|
|
for id := range waiting {
|
|
out[id] = true
|
|
}
|
|
return out
|
|
}
|