Watchman: hold mesh-watchdogs and write its watchdogs as findings, so health can leave the controller (issue 1952 WM1-WM5)
The signals rows S1, S8, S9, S10, S11, S21 and the haves' unhealthy and needs-operator are pulses whose expiry is a finding; stuck keys and presence are its own findings; it keeps the dead letters from the lease write on its own consumer, lists and shows them, and refuses deliver and drop until the hand-act log's home (issue 1953). Tools read every state it keeps and start resync. Pins mesh-sdk 2a04bf0.
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
// Command mesh-delivery-watchman-tools serves the seat mesh-watchdogs' verbs and the watchman's read tools (novox/hq ADR 0564
|
||||
// §4, §5, §12, issue 1952 WM1): the runtime launches it for a call. It reads every state the watchman keeps (presence, pulse,
|
||||
// stuck, findings: its keys, one key, how many it holds) and asks the watchman's program, on its socket, for what only the
|
||||
// program reads or does: DEAD_LETTERS, a stream, and resync. A process never serves (ADR 0436).
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
)
|
||||
|
||||
// Seat is the seat the watchman holds.
|
||||
const Seat = "mesh-watchdogs"
|
||||
|
||||
// The seat's verbs and the module's own tools, as the manifest names them.
|
||||
const (
|
||||
VerbSignals = "signals"
|
||||
VerbStuck = "stuck"
|
||||
VerbPresence = "presence"
|
||||
VerbDeadLetters = "dead-letters"
|
||||
VerbResync = "resync"
|
||||
ToolFindings = "watchman_findings"
|
||||
)
|
||||
|
||||
// Verbs are the seat's verbs, in the manifest's order.
|
||||
var Verbs = []string{VerbSignals, VerbStuck, VerbPresence, VerbDeadLetters, VerbResync}
|
||||
|
||||
// ModuleTools are the module's own tools, in the manifest's order.
|
||||
var ModuleTools = []string{ToolFindings}
|
||||
|
||||
// Reads names, per tool, the states it reads whole; Acts, per tool, the program's act it starts. The clause-12 test holds
|
||||
// that every state the manifest keeps is read by one, and every act the program takes by hand is started by one.
|
||||
var (
|
||||
Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}}
|
||||
Acts = map[string][]string{VerbDeadLetters: {socket.AskDeadLetters}, VerbResync: {socket.AskResync}}
|
||||
)
|
||||
|
||||
// state is what the tools need of a state: the runtime's, or a test's.
|
||||
type state interface {
|
||||
Keys() ([]string, error)
|
||||
Get(key string) (*stdio.StateEntry, error)
|
||||
}
|
||||
|
||||
// keyed is one key as a read answers it.
|
||||
type keyed struct {
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
Revision uint64 `json:"revision"`
|
||||
}
|
||||
|
||||
// readAll answers a state whole, or one key, with how many keys it holds; prefix narrows the keys.
|
||||
func readAll(st state, name, key, prefix string) (any, error) {
|
||||
if key != "" {
|
||||
e, err := st.Get(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s %s cannot be read: %w", name, key, err)
|
||||
}
|
||||
if e == nil {
|
||||
return nil, fmt.Errorf("%s holds no key %s", name, key)
|
||||
}
|
||||
return keyed{Key: key, Value: e.Value, Revision: e.Revision}, nil
|
||||
}
|
||||
keys, err := st.Keys()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s cannot be listed: %w", name, err)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
out := []keyed{}
|
||||
for _, k := range keys {
|
||||
if prefix != "" && !strings.HasPrefix(k, prefix) {
|
||||
continue
|
||||
}
|
||||
e, err := st.Get(k)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s %s cannot be read: %w", name, k, err)
|
||||
}
|
||||
if e != nil {
|
||||
out = append(out, keyed{Key: k, Value: e.Value, Revision: e.Revision})
|
||||
}
|
||||
}
|
||||
return map[string]any{"state": name, "holds": len(keys), "keys": out}, nil
|
||||
}
|
||||
|
||||
func str(args map[string]any, name string) string {
|
||||
v, _ := args[name].(string)
|
||||
return strings.TrimSpace(v)
|
||||
}
|
||||
|
||||
var keyArgs = map[string]any{"type": "object", "properties": map[string]any{
|
||||
"key": map[string]any{"type": "string", "description": "one key only"},
|
||||
"prefix": map[string]any{"type": "string", "description": "only the keys that begin with it"}}}
|
||||
|
||||
// Tools are the seat's verbs and the module's tools, over the states states gives and the program asked through ask.
|
||||
func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, error)) []stdio.Tool {
|
||||
read := func(name string) func(map[string]any) (any, error) {
|
||||
return func(args map[string]any) (any, error) {
|
||||
return readAll(states(name), name, str(args, "key"), str(args, "prefix"))
|
||||
}
|
||||
}
|
||||
signals := func(args map[string]any) (any, error) {
|
||||
pulses, err := readAll(states("pulse"), "pulse", str(args, "key"), str(args, "prefix"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"rows": watch.SignalRows, "pulses": pulses}, nil
|
||||
}
|
||||
dead := func(args map[string]any) (any, error) {
|
||||
a := socket.Ask{Ask: socket.AskDeadLetters, Consumer: str(args, "consumer"), Act: str(args, "act"), Why: str(args, "why")}
|
||||
if v, ok := args["limit"].(float64); ok {
|
||||
a.Limit = int(v)
|
||||
}
|
||||
if v, ok := args["id"].(float64); ok && v > 0 {
|
||||
a.ID = uint64(v)
|
||||
}
|
||||
return ask(a)
|
||||
}
|
||||
resync := func(map[string]any) (any, error) { return ask(socket.Ask{Ask: socket.AskResync}) }
|
||||
verb := func(v string) string { return Seat + "." + v }
|
||||
return []stdio.Tool{
|
||||
{Name: verb(VerbSignals), Input: keyArgs, Run: signals,
|
||||
Description: "The signals table's rows the watchman holds (S1, S8, S9, S10, S11, S21 and the haves' health), each a pulse key " +
|
||||
"whose time to live is its bound and whose expiry is its finding (novox/hq ADR 0564 §4), and every pulse the watchman " +
|
||||
"keeps: the moves it mirrors (a build step, a want, a presence) and the signals (signal.<row>.<id>, with what their " +
|
||||
"expiry does). Give key for one, prefix to narrow. Read-only."},
|
||||
{Name: verb(VerbStuck), Input: keyArgs, Run: read("stuck"),
|
||||
Description: "What stopped moving: every key of the watchman's stuck bucket, <bucket>.<key> with what it waited for since " +
|
||||
"when, and whether it moved since (never deleted). A key not moved is also the watchman's finding " +
|
||||
"delivery.<key>.stuck, a node's presence machine.<node>.silent. Give key for one, prefix to narrow. Read-only."},
|
||||
{Name: verb(VerbPresence), Input: keyArgs, Run: read("presence"),
|
||||
Description: "Who is online on the bus as the watchman hears it: each node's agent (<node>) and each delivery part's instance " +
|
||||
"(part.<part>.<node>), online or not since the entry's time. Give key for one, prefix to narrow. Read-only."},
|
||||
{Name: verb(VerbDeadLetters), Run: dead,
|
||||
Description: "What a consumer gave up on, kept in DEAD_LETTERS (novox/hq ADR 0564 §5, ADR 0264, ADR 0376): newest first, " +
|
||||
"for one consumer (its name or <stream>.<consumer>), at most limit; one whole by id (a restricted one's body " +
|
||||
"withheld). Who keeps them now: the controller until the lease names no controller, the watchman from then. " +
|
||||
"Delivering again and dropping (act deliver or drop, with why) are refused here until the hand-act log's home " +
|
||||
"(issue 1953): before the lease write use mesh-controller.dead-letters.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{
|
||||
"consumer": map[string]any{"type": "string", "description": "only this consumer's, its name or <stream>.<consumer>"},
|
||||
"limit": map[string]any{"type": "integer", "description": "at most this many, newest first"},
|
||||
"id": map[string]any{"type": "integer", "description": "one dead letter whole, by its id"},
|
||||
"act": map[string]any{"type": "string", "enum": []string{"deliver", "drop"}, "description": "deliver again or drop (refused until issue 1953)"},
|
||||
"why": map[string]any{"type": "string", "description": "why, with act"}}}},
|
||||
{Name: verb(VerbResync), Run: resync,
|
||||
Description: "Rebuild what the watchman holds by hand (novox/hq ADR 0564 §12): presence from the bus's open connections, " +
|
||||
"the dead letters' count from DEAD_LETTERS and whether to take the notices, and every finding it observes written " +
|
||||
"again to its bucket. Answers how many findings it wrote."},
|
||||
{Name: ToolFindings, Input: keyArgs, Run: read("findings"),
|
||||
Description: "The watchman's findings (novox/hq ADR 0564 §2): each thing it observes wrong now, <scope>.<id>.<kind>, with its " +
|
||||
"severity, subject, evidence and the row that observed it; the condition keeper raises and clears the conditions from " +
|
||||
"them. Give key for one, prefix to narrow. Read-only."},
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
path := socket.PathOf(os.Getenv)
|
||||
tools := Tools(func(name string) state { return stdio.State(name) }, func(a socket.Ask) (json.RawMessage, error) { return socket.On(path, a) })
|
||||
if err := stdio.Serve("", tools); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "mesh-delivery-watchman-tools:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
)
|
||||
|
||||
type fakeState map[string]string
|
||||
|
||||
func (f fakeState) Keys() ([]string, error) {
|
||||
out := []string{}
|
||||
for k := range f {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (f fakeState) Get(key string) (*stdio.StateEntry, error) {
|
||||
v, ok := f[key]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return &stdio.StateEntry{Key: key, Value: json.RawMessage(v), Revision: 3}, nil
|
||||
}
|
||||
|
||||
// 1952/WM1, ADR 0564 §12: every state the manifest keeps is read by a tool, every act the program starts by hand is
|
||||
// started by one, and every verb the claim serves and every tool the manifest names is served; and each answers.
|
||||
func TestEveryStateAndActHasATool(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m struct {
|
||||
Claims []struct {
|
||||
Name string `json:"name"`
|
||||
Serves []string `json:"serves"`
|
||||
} `json:"claims"`
|
||||
Tools []string `json:"tools"`
|
||||
State []struct {
|
||||
Name string `json:"name"`
|
||||
} `json:"state"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
asked := []socket.Ask{}
|
||||
states := map[string]fakeState{"findings": {"machine.ace.silent": `{"v":1}`, "core.doctor.self-check-silent": `{"v":1}`}}
|
||||
tools := Tools(func(name string) state { return states[name] }, func(a socket.Ask) (json.RawMessage, error) {
|
||||
asked = append(asked, a)
|
||||
if a.Act != "" {
|
||||
return nil, errors.New("refused")
|
||||
}
|
||||
return json.RawMessage(`{}`), nil
|
||||
})
|
||||
served := map[string]stdio.Tool{}
|
||||
for _, tool := range tools {
|
||||
served[tool.Name] = tool
|
||||
}
|
||||
var names []string
|
||||
for _, c := range m.Claims {
|
||||
if c.Name != Seat || !slices.Equal(c.Serves, Verbs) {
|
||||
t.Errorf("claim %+v, want %s serving %v", c, Seat, Verbs)
|
||||
}
|
||||
for _, v := range c.Serves {
|
||||
names = append(names, Seat+"."+v)
|
||||
}
|
||||
}
|
||||
if !slices.Equal(m.Tools, ModuleTools) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, ModuleTools)
|
||||
}
|
||||
names = append(names, m.Tools...)
|
||||
if len(served) != len(names) {
|
||||
t.Errorf("served %d tools, the manifest names %v", len(served), names)
|
||||
}
|
||||
for _, n := range names {
|
||||
if _, ok := served[n]; !ok {
|
||||
t.Errorf("%s is named and not served", n)
|
||||
}
|
||||
}
|
||||
read := map[string]bool{}
|
||||
for _, r := range Reads {
|
||||
for _, s := range r {
|
||||
read[s] = true
|
||||
}
|
||||
}
|
||||
for _, s := range m.State {
|
||||
if !read[s.Name] {
|
||||
t.Errorf("the state %s is read by no tool", s.Name)
|
||||
}
|
||||
}
|
||||
acts := map[string]bool{}
|
||||
for _, a := range Acts {
|
||||
for _, s := range a {
|
||||
acts[s] = true
|
||||
}
|
||||
}
|
||||
for _, a := range []string{socket.AskDeadLetters, socket.AskResync} {
|
||||
if !acts[a] {
|
||||
t.Errorf("the program's act %s is started by no tool", a)
|
||||
}
|
||||
}
|
||||
t.Run("a read answers every key and how many", func(t *testing.T) {
|
||||
out, err := served[ToolFindings].Run(map[string]any{"prefix": "machine."})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, _ := json.Marshal(out)
|
||||
var got struct {
|
||||
Holds int `json:"holds"`
|
||||
Keys []struct {
|
||||
Key string `json:"key"`
|
||||
} `json:"keys"`
|
||||
}
|
||||
_ = json.Unmarshal(b, &got)
|
||||
if got.Holds != 2 || len(got.Keys) != 1 || got.Keys[0].Key != "machine.ace.silent" {
|
||||
t.Fatalf("answer = %s", b)
|
||||
}
|
||||
if _, err := served[ToolFindings].Run(map[string]any{"key": "no.such.key"}); err == nil {
|
||||
t.Fatal("a key not held is answered")
|
||||
}
|
||||
})
|
||||
t.Run("dead-letters and resync ask the program", func(t *testing.T) {
|
||||
if _, err := served[Seat+"."+VerbDeadLetters].Run(map[string]any{"consumer": "ace_web", "limit": float64(3), "id": float64(7)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := served[Seat+"."+VerbDeadLetters].Run(map[string]any{"act": "drop", "why": "x"}); err == nil {
|
||||
t.Fatal("a refused act answered")
|
||||
}
|
||||
if _, err := served[Seat+"."+VerbResync].Run(nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []socket.Ask{{Ask: socket.AskDeadLetters, Consumer: "ace_web", Limit: 3, ID: 7}, {Ask: socket.AskDeadLetters, Act: "drop", Why: "x"}, {Ask: socket.AskResync}}
|
||||
if !slices.Equal(asked, want) {
|
||||
t.Fatalf("asked %+v", asked)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -17,8 +17,13 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-watchman/internal/deadletters"
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
)
|
||||
|
||||
// busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system
|
||||
@@ -103,6 +108,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: delivery.AssignerWantsBucket, History: 20},
|
||||
{Bucket: delivery.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
} {
|
||||
kv, err := js.CreateKeyValue(ctx, c)
|
||||
if err != nil {
|
||||
@@ -111,6 +117,31 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
kvs[c.Bucket] = kv
|
||||
}
|
||||
|
||||
// The controller's lease, its consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one
|
||||
// message a consumer will give up on (1952/WM5).
|
||||
lease, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buscompose.LeaseBucket})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, c := range []jetstream.StreamConfig{
|
||||
{Name: buscompose.DeadLetterNoticesStream, Subjects: []string{buscompose.MaxDeliveriesAdvisories}, Retention: jetstream.WorkQueuePolicy},
|
||||
{Name: buscompose.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}, Duplicates: 24 * time.Hour, AllowMsgTTL: true},
|
||||
{Name: "SRC", Subjects: []string{"src.>"}},
|
||||
} {
|
||||
if _, err := js.CreateStream(ctx, c); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := js.CreateConsumer(ctx, buscompose.DeadLetterNoticesStream, jetstream.ConsumerConfig{Durable: buscompose.ControllerName, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
given := nats.NewMsg("src.ask")
|
||||
given.Header.Set("X-Asker", "n1")
|
||||
given.Data = []byte("do it")
|
||||
if _, err := js.PublishMsg(ctx, given); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A node already connected when the watchman starts: its presence comes from the one connections request.
|
||||
connect("node.n0")
|
||||
|
||||
@@ -119,7 +150,13 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_PULSE_BOUND": "3s"}
|
||||
// A unix socket's path is at most 108 bytes, which a test's TMPDIR may pass.
|
||||
sockDir, err := os.MkdirTemp("/tmp", "ws")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(sockDir) })
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_PULSE_BOUND": "3s", "MESH_WATCHMAN_SOCKET": filepath.Join(sockDir, "tools.sock")}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
@@ -224,6 +261,63 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
|
||||
t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
|
||||
}
|
||||
}
|
||||
|
||||
// 1952/WM4: what stopped moving is the watchman's own finding: the ask no node ended, and node n1 silent.
|
||||
finding := func(key string) bool { _, ok := has(watch.FindingsBucket, key); return ok }
|
||||
until("the stuck ask's finding", func() bool {
|
||||
return finding(watch.FindingKey(watch.ScopeDelivery, stuckOf(askStream, ask1), watch.KindStuck))
|
||||
})
|
||||
until("node n1's silence", func() bool { return finding("machine.n1.silent") })
|
||||
|
||||
// 1952/WM5: while the lease names a controller the watchman takes no notice; handed away while the controller's consumer
|
||||
// stands, it says the notices wait; the consumer removed (the lease write's step), it makes its own and keeps a notice
|
||||
// with its headers.
|
||||
if _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer); err == nil {
|
||||
t.Fatal("the watchman made its consumer while the lease names a controller")
|
||||
}
|
||||
if _, err := lease.Put(ctx, buscompose.ControllerLeaseKey, []byte(`{"handed-to":"mesh-vault","handed-by":"operator","why":"the lease write"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waits := watch.FindingKey(watch.ScopeBus, "DEAD_LETTER_NOTICES.controller", watch.KindNoticesWaiting)
|
||||
until("the notices said waiting behind the controller's consumer", func() bool { return finding(waits) })
|
||||
if _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer); err == nil {
|
||||
t.Fatal("the watchman made its consumer while the controller's stands")
|
||||
}
|
||||
if err := js.DeleteConsumer(ctx, buscompose.DeadLetterNoticesStream, buscompose.ControllerName); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
until("the watchman's own consumer on the notices", func() bool {
|
||||
_, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer)
|
||||
return err == nil
|
||||
})
|
||||
until("the wait no longer said", func() bool { return !finding(waits) })
|
||||
notice, _ := json.Marshal(map[string]any{"stream": "SRC", "consumer": "c1", "stream_seq": 1, "deliveries": 5, "timestamp": time.Now().UTC()})
|
||||
if err := ctl.Publish(buscompose.MaxDeliveriesAdvisories[:len(buscompose.MaxDeliveriesAdvisories)-1]+"SRC.c1", notice); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dl, err := js.Stream(ctx, buscompose.DeadLettersStream)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var kept *jetstream.RawStreamMsg
|
||||
until("the notice kept in DEAD_LETTERS", func() bool {
|
||||
kept, err = dl.GetLastMsgForSubject(ctx, buscompose.DeadLetterSubject("SRC", "c1"))
|
||||
return err == nil
|
||||
})
|
||||
if kept.Header.Get(deadletters.HeaderStream) != "SRC" || kept.Header.Get(deadletters.HeaderSequence) != "1" ||
|
||||
kept.Header.Get(deadletters.HeaderSubject) != "src.ask" || kept.Header.Get("X-Asker") != "n1" || string(kept.Data) != "do it" {
|
||||
t.Fatalf("kept = %v %q", kept.Header, kept.Data)
|
||||
}
|
||||
until("the consumer's dead letters said", func() bool { return finding("bus.SRC.c1.max-deliveries") })
|
||||
// The tools' socket lists it, and refuses to drop it.
|
||||
answer, err := socket.On(env["MESH_WATCHMAN_SOCKET"], socket.Ask{Ask: socket.AskDeadLetters, Consumer: "c1"})
|
||||
if err != nil || !strings.Contains(string(answer), `"total":1`) || !strings.Contains(string(answer), "the watchman") {
|
||||
t.Fatalf("dead-letters = %s, %v", answer, err)
|
||||
}
|
||||
if _, err := socket.On(env["MESH_WATCHMAN_SOCKET"], socket.Ask{Ask: socket.AskDeadLetters, ID: kept.Sequence, Act: "drop", Why: "test"}); err == nil ||
|
||||
!strings.Contains(err.Error(), "issue 1953") {
|
||||
t.Fatalf("a drop was not refused naming the hand-act log's home: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnAskRunningBeforeTheStartGetsItsPulse puts a running step in the build module's steps before the watchman starts, and
|
||||
@@ -271,6 +365,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: delivery.AssignerWantsBucket, History: 20},
|
||||
{Bucket: delivery.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
} {
|
||||
kv, err := js.CreateKeyValue(ctx, c)
|
||||
if err != nil {
|
||||
@@ -291,7 +386,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_WATCHMAN_SOCKET": filepath.Join(dir, "tools.sock")}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
@@ -355,6 +450,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
|
||||
{Bucket: delivery.AssignerWantsBucket, History: 20},
|
||||
{Bucket: delivery.HaveBucket, History: 20},
|
||||
{Bucket: watch.FindingsBucket, History: 5},
|
||||
} {
|
||||
kv, err := js.CreateKeyValue(ctx, c)
|
||||
if err != nil {
|
||||
@@ -393,7 +489,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_WATCHMAN_SOCKET": filepath.Join(dir, "tools.sock")}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
|
||||
@@ -22,8 +22,10 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
)
|
||||
|
||||
@@ -44,12 +46,19 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
// kvOut writes the watchman's three buckets.
|
||||
// kvOut writes the watchman's four buckets.
|
||||
type kvOut struct {
|
||||
ctx context.Context
|
||||
presence, pulse, stuk jetstream.KeyValue
|
||||
ctx context.Context
|
||||
presence, pulse, stuk, findings jetstream.KeyValue
|
||||
}
|
||||
|
||||
func (o *kvOut) PutFinding(key string, value []byte) error {
|
||||
_, err := o.findings.Put(o.ctx, key, value)
|
||||
return err
|
||||
}
|
||||
|
||||
func (o *kvOut) DeleteFinding(key string) error { return o.findings.Delete(o.ctx, key) }
|
||||
|
||||
func (o *kvOut) PutPresence(key string, online bool) error {
|
||||
value, err := delivery.EncodePresence(delivery.Presence{Online: online})
|
||||
if err != nil {
|
||||
@@ -161,11 +170,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
|
||||
// and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build
|
||||
// seats' steps, and says that moves are not watched (issue 1945).
|
||||
var kvs struct{ presence, pulse, stuck, wants, haves jetstream.KeyValue }
|
||||
var kvs struct{ presence, pulse, stuck, findings, wants, haves, lease jetstream.KeyValue }
|
||||
for _, b := range []struct {
|
||||
name string
|
||||
into *jetstream.KeyValue
|
||||
}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} {
|
||||
}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}, {watch.FindingsBucket, &kvs.findings}} {
|
||||
if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -176,7 +185,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it
|
||||
}
|
||||
|
||||
w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck}, time.Now)
|
||||
kvs.lease = process.OptionalBucket(ctx, js, buscompose.LeaseBucket, "the lease is read as naming a controller, so the dead letters stay the controller's", logger.Printf)
|
||||
w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now)
|
||||
w.Logf(logger.Printf)
|
||||
// A test shortens how long a move may take; the mesh never sets it.
|
||||
if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" {
|
||||
@@ -197,6 +207,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
w.SeedPresence(e.Key(), p.Online, e.Created())
|
||||
}
|
||||
}},
|
||||
{kvs.findings, func(e jetstream.KeyValueEntry) {
|
||||
if opOf(e) == watch.Put {
|
||||
w.SeedFinding(e.Key(), e.Value())
|
||||
}
|
||||
}},
|
||||
{kvs.stuck, func(e jetstream.KeyValueEntry) {
|
||||
if opOf(e) == watch.Put {
|
||||
w.SeedStuck(e.Key(), e.Value())
|
||||
@@ -212,6 +227,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
pulses, err := settle(ctx, kvs.pulse, func(e jetstream.KeyValueEntry) {
|
||||
if opOf(e) == watch.Put {
|
||||
w.SeedPulse(e.Key())
|
||||
w.SeedSignal(e.Key(), e.Value())
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
@@ -281,9 +297,39 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := w.Rebuild(ask); err != nil {
|
||||
return err
|
||||
}
|
||||
dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
|
||||
dead.count()
|
||||
dead.check()
|
||||
w.Reconcile()
|
||||
if err := hear(conn, w, dead, logger.Printf); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
errs := make(chan error, 8)
|
||||
if kvs.lease != nil {
|
||||
lw, err := kvs.lease.Watch(ctx, buscompose.ControllerLeaseKey)
|
||||
if err != nil {
|
||||
process.NotWatched(logger.Printf, buscompose.LeaseBucket, err, "the lease is read as naming a controller, so the dead letters stay the controller's")
|
||||
} else {
|
||||
go func() {
|
||||
errs <- follow(ctx, lw, func(e jetstream.KeyValueEntry) {
|
||||
var raw []byte
|
||||
if opOf(e) == watch.Put {
|
||||
raw = e.Value()
|
||||
}
|
||||
on, _ := buscompose.LeaseHandedAway(raw, len(raw) > 0, nil)
|
||||
dead.lease(on)
|
||||
})
|
||||
}()
|
||||
}
|
||||
}
|
||||
if l, err := socket.Listen(socket.PathOf(getenv)); err != nil {
|
||||
logger.Printf("the tools' socket cannot be opened (%v): dead-letters and resync answer nothing until the watchman starts again", err)
|
||||
} else {
|
||||
go func() {
|
||||
errs <- socket.Serve(ctx, l, func(a socket.Ask) (any, error) { return answer(a, w, dead, ask) })
|
||||
}()
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
|
||||
}()
|
||||
|
||||
@@ -0,0 +1,185 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
|
||||
"mesh-delivery-watchman/internal/deadletters"
|
||||
"mesh-delivery-watchman/internal/socket"
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
)
|
||||
|
||||
// The subjects the watchman hears for the signals table's rows (novox/hq ADR 0564 §4, Limits 3 and 11), each on the bus,
|
||||
// none in the controller's store. Its grant names them (mesh-sdk buscompose, the watchman's principal).
|
||||
const (
|
||||
PowerSubject = "mesh.mod.power.event.*"
|
||||
FailingSubject = "mesh.mod.*.event.provisioner.failing"
|
||||
RecoveredSubject = "mesh.mod.*.event.provisioner.recovered"
|
||||
ToolsAliveSubject = "mesh.control.*.tools-alive"
|
||||
CheckedSubject = "mesh.seat.mesh-self-check.event.checked"
|
||||
DeadKeptSubject = "mesh.events.dead.>"
|
||||
)
|
||||
|
||||
// HeardSubjects is every subject the watchman subscribes for its rows, beside the bus's connect and disconnect.
|
||||
var HeardSubjects = []string{PowerSubject, FailingSubject, RecoveredSubject, ToolsAliveSubject, CheckedSubject,
|
||||
watch.ConsumerDeletedAdvisories, DeadKeptSubject}
|
||||
|
||||
// heard hands one message of a row's subject to the watchman.
|
||||
func heard(w *watch.Watchman, dead *deadKeeper, m *nats.Msg, logf func(string, ...any)) {
|
||||
switch {
|
||||
case strings.HasPrefix(m.Subject, "mesh.mod.power.event."):
|
||||
w.Power(m.Header.Get("x-node"), strings.TrimPrefix(m.Subject, "mesh.mod.power.event."))
|
||||
case strings.HasSuffix(m.Subject, ".event.provisioner.failing") || strings.HasSuffix(m.Subject, ".event.provisioner.recovered"):
|
||||
st, err := watch.ReadStanding(m.Subject, m.Data)
|
||||
if err != nil {
|
||||
logf("%v; left", err)
|
||||
return
|
||||
}
|
||||
w.Stood(st)
|
||||
case strings.HasSuffix(m.Subject, ".tools-alive"):
|
||||
var body struct {
|
||||
Every float64 `json:"every"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Data, &body)
|
||||
w.ToolsAlive(watch.ToolsNode(m.Subject), time.Duration(body.Every*float64(time.Second)))
|
||||
case m.Subject == CheckedSubject:
|
||||
w.Checked(m.Header.Get("x-node"))
|
||||
case strings.HasPrefix(m.Subject, "$JS.EVENT.ADVISORY."):
|
||||
a, ok := watch.ReadAdvisory(m.Subject, m.Data, deadletters.ConsumerInWords)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if dead.isTheControllersNotices(a.ID) {
|
||||
dead.check() // the lease write's step removed it: the watchman may take the notices now
|
||||
return
|
||||
}
|
||||
w.Advised(a)
|
||||
case strings.HasPrefix(m.Subject, "mesh.events.dead."):
|
||||
dead.count()
|
||||
}
|
||||
}
|
||||
|
||||
// hear subscribes every row's subject; a subject the bus refuses stops the start, naming it.
|
||||
func hear(conn *nats.Conn, w *watch.Watchman, dead *deadKeeper, logf func(string, ...any)) error {
|
||||
for _, subject := range HeardSubjects {
|
||||
if _, err := conn.Subscribe(subject, func(m *nats.Msg) { heard(w, dead, m, logf) }); err != nil {
|
||||
return fmt.Errorf("the watchman cannot hear %s: %w", subject, err)
|
||||
}
|
||||
}
|
||||
return conn.Flush()
|
||||
}
|
||||
|
||||
// deadKeeper keeps what consumers gave up on from the lease write (ADR 0564 §5): it counts DEAD_LETTERS into findings,
|
||||
// and takes DEAD_LETTER_NOTICES on its own consumer only once the lease names no controller and `controller` is gone.
|
||||
type deadKeeper struct {
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
w *watch.Watchman
|
||||
logf func(string, ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
handedAway bool
|
||||
taking jetstream.ConsumeContext
|
||||
}
|
||||
|
||||
// count reads DEAD_LETTERS' count per consumer into the findings.
|
||||
func (d *deadKeeper) count() {
|
||||
held, err := deadletters.Held(d.ctx, d.js)
|
||||
if err != nil {
|
||||
d.logf("the dead letters cannot be counted, their findings are left as they are: %v", err)
|
||||
return
|
||||
}
|
||||
d.w.DeadLettersHeld(held, deadletters.ConsumerInWords)
|
||||
}
|
||||
|
||||
// lease is the lease's change heard: handed away or not.
|
||||
func (d *deadKeeper) lease(handedAway bool) {
|
||||
d.mu.Lock()
|
||||
d.handedAway = handedAway
|
||||
d.mu.Unlock()
|
||||
d.check()
|
||||
}
|
||||
|
||||
func (d *deadKeeper) away() bool {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.handedAway
|
||||
}
|
||||
|
||||
// isTheControllersNotices says an advisory is about the controller's consumer on DEAD_LETTER_NOTICES once the lease is
|
||||
// handed away: its removal is the lease write's step, not a loss.
|
||||
func (d *deadKeeper) isTheControllersNotices(id string) bool {
|
||||
return d.away() && id == buscompose.DeadLetterNoticesStream+"."+buscompose.ControllerName
|
||||
}
|
||||
|
||||
// check takes the notices, or says they wait, as Gate has it.
|
||||
func (d *deadKeeper) check() {
|
||||
away := d.away()
|
||||
stands := true
|
||||
if away {
|
||||
var err error
|
||||
if stands, err = deadletters.ControllerStands(d.ctx, d.js); err != nil {
|
||||
d.logf("whether the controller's consumer on %s stands cannot be read (%v): it is taken to stand", buscompose.DeadLetterNoticesStream, err)
|
||||
stands = true
|
||||
}
|
||||
}
|
||||
take, waiting := deadletters.Gate(away, stands)
|
||||
d.w.NoticesWaiting(waiting)
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if !take || d.taking != nil {
|
||||
return
|
||||
}
|
||||
cc, err := deadletters.TakeNotices(d.ctx, d.js, func(deadletters.DeadLetter) { go d.count() }, d.logf)
|
||||
if err != nil {
|
||||
d.logf("the dead-letter notices cannot be taken yet: %v", err)
|
||||
return
|
||||
}
|
||||
d.taking = cc
|
||||
d.logf("the lease names no controller and its consumer on %s is gone: the watchman keeps the dead letters", buscompose.DeadLetterNoticesStream)
|
||||
}
|
||||
|
||||
// answer is the program's answer to the tools' ask on its socket.
|
||||
func answer(a socket.Ask, w *watch.Watchman, dead *deadKeeper, ask func([]byte) ([]byte, error)) (any, error) {
|
||||
switch a.Ask {
|
||||
case socket.AskResync:
|
||||
if err := w.Rebuild(ask); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dead.count()
|
||||
dead.check()
|
||||
return map[string]any{"findings-written": w.Rewrite(), "presence": "rebuilt from the bus's connections"}, nil
|
||||
case socket.AskDeadLetters:
|
||||
switch {
|
||||
case a.Act == "deliver" || a.Act == "drop":
|
||||
if strings.TrimSpace(a.Why) == "" {
|
||||
return nil, errors.New("a dead letter is delivered again or dropped only with why")
|
||||
}
|
||||
return nil, deadletters.Refusal(dead.away(), a.Act)
|
||||
case a.Act != "":
|
||||
return nil, fmt.Errorf("%q is not deliver or drop", a.Act)
|
||||
case a.ID > 0:
|
||||
return deadletters.Named(dead.ctx, dead.js, a.ID)
|
||||
}
|
||||
list, total, err := deadletters.List(dead.ctx, dead.js, a.Consumer, a.Limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keeper := "the controller (it holds its lease)"
|
||||
if dead.away() {
|
||||
keeper = "the watchman"
|
||||
}
|
||||
return map[string]any{"dead-letters": list, "total": total, "kept-by": keeper}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("refused: the watchman's socket answers %q and %q", socket.AskDeadLetters, socket.AskResync)
|
||||
}
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-watchman
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf h1:Yy51rUWqyk+cZPAcSX02dnK4JVSTUJIfLa9j+741+1Y=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -0,0 +1,358 @@
|
||||
// Package deadletters is what a consumer gave up on, kept by the watchman from the lease write (novox/hq ADR 0564 §5,
|
||||
// ADR 0264, ADR 0376, issue 1952 WM5): ported from mesh-controller internal/link/deadletters.go at e04846e9 with nothing
|
||||
// changed in what is kept, aged or withheld, on the jetstream API.
|
||||
//
|
||||
// DEAD_LETTER_NOTICES is a work queue that holds one consumer. The controller's consumer `controller` never expires, so
|
||||
// the watchman makes its own (Consumer) only when the lease names no controller and `controller` is gone (Gate); while
|
||||
// `controller` still stands after the lease write it says so as a finding. Removing `controller` is a step of the lease
|
||||
// write, with the bus module's tool. Before the lease write the watchman only lists and shows DEAD_LETTERS, which the
|
||||
// controller still fills, and refuses to deliver or drop, naming the controller's verb; from it, delivering and dropping
|
||||
// wait for the hand-act log's home (issue 1953), so they are refused naming that.
|
||||
package deadletters
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
)
|
||||
|
||||
// Consumer is the watchman's own consumer on DEAD_LETTER_NOTICES, made only by Gate's leave.
|
||||
const Consumer = "mesh-delivery-watchman"
|
||||
|
||||
// The headers a kept message carries beside its own (mesh-controller link.DeadHeader*: the same names, so the
|
||||
// controller's copies and the watchman's are read alike).
|
||||
const (
|
||||
HeaderStream = "Mesh-Dead-Stream"
|
||||
HeaderConsumer = "Mesh-Dead-Consumer"
|
||||
HeaderSequence = "Mesh-Dead-Sequence"
|
||||
HeaderSubject = "Mesh-Dead-Subject"
|
||||
HeaderDeliveries = "Mesh-Dead-Deliveries"
|
||||
HeaderGaveUp = "Mesh-Dead-Gave-Up"
|
||||
HeaderPublished = "Mesh-Dead-Published"
|
||||
HeaderLost = "Mesh-Dead-Lost"
|
||||
HeaderMsgID = "Mesh-Dead-Msg-Id"
|
||||
HeaderRestricted = "Mesh-Dead-Restricted"
|
||||
HeaderAgesOut = "Mesh-Dead-Ages-Out"
|
||||
HeaderWorkQueue = "Mesh-Dead-Work-Queue"
|
||||
)
|
||||
|
||||
// DeadLetter is one message a consumer gave up on, as DEAD_LETTERS keeps it (the controller's JSON).
|
||||
type DeadLetter struct {
|
||||
ID uint64 `json:"id"`
|
||||
Stream string `json:"stream"`
|
||||
Consumer string `json:"consumer"`
|
||||
Who string `json:"who"`
|
||||
Subject string `json:"subject"`
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Deliveries uint64 `json:"deliveries"`
|
||||
GaveUp time.Time `json:"gave_up"`
|
||||
Published time.Time `json:"published,omitzero"`
|
||||
Lost string `json:"lost,omitempty"`
|
||||
Size int `json:"size"`
|
||||
Restricted bool `json:"restricted,omitempty"`
|
||||
AgesOut time.Time `json:"ages_out,omitzero"`
|
||||
Withheld string `json:"withheld,omitempty"`
|
||||
WorkQueue bool `json:"work-queue,omitempty"`
|
||||
Body string `json:"body,omitempty"`
|
||||
Headers map[string][]string `json:"headers,omitempty"`
|
||||
}
|
||||
|
||||
// Now is the clock dead letters age by: a variable, so a test can move it.
|
||||
var Now = time.Now
|
||||
|
||||
// WithheldWords is what is said in place of a restricted dead letter's body.
|
||||
const WithheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " +
|
||||
"the modules its grant names may read (novox/hq ADR 0376)"
|
||||
|
||||
// ConsumerInWords is a durable consumer as the mesh says it: whose, and for what.
|
||||
func ConsumerInWords(stream, name string) string {
|
||||
switch {
|
||||
case name == buscompose.ControllerName || name == buscompose.MergesConsumerName:
|
||||
return "the controller's consumer on " + stream
|
||||
case stream == "NODES":
|
||||
return "how " + name + " hears what it should be (its declaration consumer)"
|
||||
case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(name, "_worker"):
|
||||
seat := strings.ToLower(strings.ReplaceAll(strings.TrimPrefix(stream, "SEAT_"), "_", "-"))
|
||||
return "the worker every holder of " + seat + " takes its asks from"
|
||||
case stream == buscompose.EventsStream:
|
||||
if node, module, ok := strings.Cut(name, "_"); ok {
|
||||
return "how " + module + " on " + node + " hears what it consumes"
|
||||
}
|
||||
}
|
||||
return "the consumer " + name + " on " + stream
|
||||
}
|
||||
|
||||
// sourceAge is whether a stream is restricted, a module's declared work queue, and its maximum age; a stream gone is
|
||||
// neither.
|
||||
func sourceAge(ctx context.Context, js jetstream.JetStream, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) {
|
||||
s, err := js.Stream(ctx, stream)
|
||||
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return false, false, 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err)
|
||||
}
|
||||
cfg := s.CachedInfo().Config
|
||||
return cfg.Metadata[buscompose.RestrictedMetadata] == "true", cfg.Metadata[buscompose.WorkQueueMetadata] == "true", cfg.MaxAge, nil
|
||||
}
|
||||
|
||||
// notice is the part of the server's maximum-deliveries advisory a kept message is made from.
|
||||
type notice struct {
|
||||
Stream string `json:"stream"`
|
||||
Consumer string `json:"consumer"`
|
||||
StreamSeq uint64 `json:"stream_seq"`
|
||||
Deliveries uint64 `json:"deliveries"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// Keep keeps the message one maximum-deliveries notice is about. An error keeps nothing and the notice is to be taken
|
||||
// again; a source that no longer holds the message is no error: the record is kept with why.
|
||||
func Keep(ctx context.Context, js jetstream.JetStream, body []byte) (DeadLetter, error) {
|
||||
var a notice
|
||||
if err := json.Unmarshal(body, &a); err != nil || a.Stream == "" || a.Consumer == "" || a.StreamSeq == 0 {
|
||||
return DeadLetter{}, fmt.Errorf("not a maximum-deliveries notice the mesh can read: %.200s", body)
|
||||
}
|
||||
gaveUp := a.Timestamp
|
||||
if gaveUp.IsZero() {
|
||||
gaveUp = Now()
|
||||
}
|
||||
kept := &nats.Msg{Subject: buscompose.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}}
|
||||
restricted, workQueue, maxAge, err := sourceAge(ctx, js, a.Stream)
|
||||
if err != nil {
|
||||
return DeadLetter{}, err
|
||||
}
|
||||
if restricted {
|
||||
kept.Header.Set(HeaderRestricted, "true")
|
||||
}
|
||||
if workQueue {
|
||||
kept.Header.Set(HeaderWorkQueue, "true")
|
||||
}
|
||||
var raw *jetstream.RawStreamMsg
|
||||
src, err := js.Stream(ctx, a.Stream)
|
||||
if err == nil {
|
||||
raw, err = src.GetMsg(ctx, a.StreamSeq)
|
||||
}
|
||||
agesOut := time.Time{}
|
||||
if err == nil && restricted && maxAge > 0 {
|
||||
agesOut = raw.Time.Add(maxAge)
|
||||
kept.Header.Set(HeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano))
|
||||
}
|
||||
switch {
|
||||
case err == nil && !agesOut.IsZero() && agesOut.Sub(Now()) < time.Second:
|
||||
kept.Header.Set(HeaderSubject, raw.Subject)
|
||||
kept.Header.Set(HeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano))
|
||||
kept.Header.Set(HeaderLost, fmt.Sprintf("message %d of %s had passed its stream's maximum age of %s when the notice "+
|
||||
"was taken, so it is kept without its body (novox/hq ADR 0376)", a.StreamSeq, a.Stream, maxAge))
|
||||
case err == nil:
|
||||
for k, v := range raw.Header {
|
||||
if !strings.HasPrefix(k, "Nats-") {
|
||||
kept.Header[k] = v
|
||||
}
|
||||
}
|
||||
if id := raw.Header.Get(nats.MsgIdHdr); id != "" {
|
||||
kept.Header.Set(HeaderMsgID, id)
|
||||
}
|
||||
kept.Header.Set(HeaderSubject, raw.Subject)
|
||||
kept.Header.Set(HeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano))
|
||||
kept.Data = raw.Data
|
||||
if !agesOut.IsZero() {
|
||||
kept.Header.Set(nats.MsgTTLHdr, strconv.FormatInt(int64(agesOut.Sub(Now())/time.Second), 10)+"s")
|
||||
}
|
||||
case errors.Is(err, jetstream.ErrMsgNotFound) || errors.Is(err, jetstream.ErrStreamNotFound):
|
||||
kept.Header.Set(HeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v", a.Stream, a.StreamSeq, err))
|
||||
default:
|
||||
return DeadLetter{}, fmt.Errorf("message %d of %s could not be read: %w", a.StreamSeq, a.Stream, err)
|
||||
}
|
||||
kept.Header.Set(HeaderStream, a.Stream)
|
||||
kept.Header.Set(HeaderConsumer, a.Consumer)
|
||||
kept.Header.Set(HeaderSequence, strconv.FormatUint(a.StreamSeq, 10))
|
||||
kept.Header.Set(HeaderDeliveries, strconv.FormatUint(a.Deliveries, 10))
|
||||
kept.Header.Set(HeaderGaveUp, gaveUp.UTC().Format(time.RFC3339Nano))
|
||||
// One copy per message given up, however often its notice is taken, by whichever keeper takes it.
|
||||
kept.Header.Set(nats.MsgIdHdr, fmt.Sprintf("%s.%s.%d", a.Stream, a.Consumer, a.StreamSeq))
|
||||
ack, err := js.PublishMsg(ctx, kept)
|
||||
if err != nil {
|
||||
return DeadLetter{}, fmt.Errorf("message %d of %s could not be kept in %s: %w", a.StreamSeq, a.Stream, buscompose.DeadLettersStream, err)
|
||||
}
|
||||
return Of(ack.Sequence, kept.Subject, kept.Header, kept.Data, false), nil
|
||||
}
|
||||
|
||||
// Of reads a kept message; whole gives its body and its own headers.
|
||||
func Of(id uint64, subject string, h nats.Header, data []byte, whole bool) DeadLetter {
|
||||
d := DeadLetter{ID: id, Stream: h.Get(HeaderStream), Consumer: h.Get(HeaderConsumer), Subject: h.Get(HeaderSubject),
|
||||
Lost: h.Get(HeaderLost), Size: len(data)}
|
||||
if d.Stream == "" || d.Consumer == "" {
|
||||
d.Stream, d.Consumer, _ = buscompose.DeadLetterOf(subject)
|
||||
}
|
||||
d.Who = ConsumerInWords(d.Stream, d.Consumer)
|
||||
d.Sequence, _ = strconv.ParseUint(h.Get(HeaderSequence), 10, 64)
|
||||
d.Deliveries, _ = strconv.ParseUint(h.Get(HeaderDeliveries), 10, 64)
|
||||
d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderGaveUp))
|
||||
d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderPublished))
|
||||
d.Restricted = h.Get(HeaderRestricted) == "true"
|
||||
d.WorkQueue = h.Get(HeaderWorkQueue) == "true"
|
||||
d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderAgesOut))
|
||||
if whole {
|
||||
d.Body = string(data)
|
||||
d.Headers = map[string][]string{}
|
||||
for k, v := range h {
|
||||
if !strings.HasPrefix(k, "Mesh-Dead-") && !strings.HasPrefix(k, "Nats-") {
|
||||
d.Headers[k] = append([]string(nil), v...)
|
||||
}
|
||||
}
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// Held is how many messages DEAD_LETTERS holds for each consumer, by `<stream>.<consumer>`.
|
||||
func Held(ctx context.Context, js jetstream.JetStream) (map[string]int, error) {
|
||||
s, err := js.Stream(ctx, buscompose.DeadLettersStream)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err)
|
||||
}
|
||||
info, err := s.Info(ctx, jetstream.WithSubjectFilter(">"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err)
|
||||
}
|
||||
held := map[string]int{}
|
||||
for subject, n := range info.State.Subjects {
|
||||
if stream, consumer, ok := buscompose.DeadLetterOf(subject); ok && n > 0 {
|
||||
held[stream+"."+consumer] += int(n)
|
||||
}
|
||||
}
|
||||
return held, nil
|
||||
}
|
||||
|
||||
// List is what DEAD_LETTERS holds, newest first, at most most (all when most is not positive); for one consumer when
|
||||
// consumer names one (its name, or `<stream>.<consumer>`). A restricted one's body is withheld. The total is the stream's
|
||||
// own count.
|
||||
func List(ctx context.Context, js jetstream.JetStream, consumer string, most int) ([]DeadLetter, int, error) {
|
||||
held, err := Held(ctx, js)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
total := 0
|
||||
for key, n := range held {
|
||||
if consumer == "" || key == consumer || strings.HasSuffix(key, "."+consumer) {
|
||||
total += n
|
||||
}
|
||||
}
|
||||
if total == 0 {
|
||||
return nil, 0, nil
|
||||
}
|
||||
s, err := js.Stream(ctx, buscompose.DeadLettersStream)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
info, err := s.Info(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err)
|
||||
}
|
||||
var out []DeadLetter
|
||||
for seq := info.State.LastSeq; seq >= info.State.FirstSeq && seq > 0; seq-- {
|
||||
if most > 0 && len(out) >= most || len(out) >= total {
|
||||
break
|
||||
}
|
||||
raw, err := s.GetMsg(ctx, seq)
|
||||
if errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return out, total, fmt.Errorf("dead letter %d cannot be read: %w", seq, err)
|
||||
}
|
||||
d := Of(seq, raw.Subject, raw.Header, raw.Data, false)
|
||||
if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer {
|
||||
continue
|
||||
}
|
||||
if d.Restricted {
|
||||
d.Withheld = WithheldWords
|
||||
}
|
||||
out = append(out, d)
|
||||
}
|
||||
return out, total, nil
|
||||
}
|
||||
|
||||
// ErrNoDeadLetter is an id DEAD_LETTERS does not hold.
|
||||
var ErrNoDeadLetter = errors.New("no such dead letter")
|
||||
|
||||
// Named is one kept message whole, but for a restricted one, whose body and own headers are never shown.
|
||||
func Named(ctx context.Context, js jetstream.JetStream, id uint64) (DeadLetter, error) {
|
||||
s, err := js.Stream(ctx, buscompose.DeadLettersStream)
|
||||
if err != nil {
|
||||
return DeadLetter{}, err
|
||||
}
|
||||
raw, err := s.GetMsg(ctx, id)
|
||||
if errors.Is(err, jetstream.ErrMsgNotFound) {
|
||||
return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, or never kept; "+
|
||||
"dead-letters lists what it holds", ErrNoDeadLetter, buscompose.DeadLettersStream, id)
|
||||
}
|
||||
if err != nil {
|
||||
return DeadLetter{}, fmt.Errorf("dead letter %d cannot be read: %w", id, err)
|
||||
}
|
||||
d := Of(id, raw.Subject, raw.Header, raw.Data, true)
|
||||
if d.Restricted {
|
||||
d.Body, d.Headers, d.Withheld = "", nil, WithheldWords
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// Refusal is why the watchman delivers nothing again and drops nothing: before the lease write the controller keeps them
|
||||
// and its verb acts; from it the act waits for the hand-act log's home (issue 1953), where a person's act is recorded.
|
||||
// act is deliver or drop.
|
||||
func Refusal(handedAway bool, act string) error {
|
||||
done := map[string]string{"deliver": "delivered again", "drop": "dropped"}[act]
|
||||
if done == "" {
|
||||
return fmt.Errorf("%q is not deliver or drop", act)
|
||||
}
|
||||
if !handedAway {
|
||||
return fmt.Errorf("the controller still holds its lease and keeps the dead letters: %s it with "+
|
||||
"mesh-controller.dead-letters (novox/hq ADR 0564 §5); nothing is %s here", act, done)
|
||||
}
|
||||
return fmt.Errorf("a dead letter is %s only as a person's act recorded in the hand-act log, whose home is not "+
|
||||
"delivered yet (issue 1953): nothing is %s", done, done)
|
||||
}
|
||||
|
||||
// Gate is whether the watchman takes the notices itself: only when the lease names no controller (handedAway) and the
|
||||
// controller's consumer is gone. waiting says the lease is handed away while `controller` still stands, which is the
|
||||
// watchman's finding: each notice waits in the queue until the lease write's step removes it.
|
||||
func Gate(handedAway, controllerStands bool) (take, waiting bool) {
|
||||
return handedAway && !controllerStands, handedAway && controllerStands
|
||||
}
|
||||
|
||||
// ControllerStands asks the bus whether the controller's consumer on DEAD_LETTER_NOTICES stands.
|
||||
func ControllerStands(ctx context.Context, js jetstream.JetStream) (bool, error) {
|
||||
_, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, buscompose.ControllerName)
|
||||
if errors.Is(err, jetstream.ErrConsumerNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
return err == nil, err
|
||||
}
|
||||
|
||||
// TakeNotices makes the watchman's own durable consumer on DEAD_LETTER_NOTICES (never touching `controller`) and keeps
|
||||
// each notice it takes; a notice not kept is handed back to be taken again. kept hears each one kept.
|
||||
func TakeNotices(ctx context.Context, js jetstream.JetStream, kept func(DeadLetter), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
|
||||
c, err := js.CreateOrUpdateConsumer(ctx, buscompose.DeadLetterNoticesStream, jetstream.ConsumerConfig{
|
||||
Durable: Consumer, AckPolicy: jetstream.AckExplicitPolicy, AckWait: time.Minute})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the consumer %s on %s cannot be made: %w", Consumer, buscompose.DeadLetterNoticesStream, err)
|
||||
}
|
||||
return c.Consume(func(m jetstream.Msg) {
|
||||
d, err := Keep(ctx, js, m.Data())
|
||||
if err != nil {
|
||||
logf("a dead-letter notice was not kept, offered again in a minute: %v", err)
|
||||
_ = m.NakWithDelay(time.Minute)
|
||||
return
|
||||
}
|
||||
_ = m.Ack()
|
||||
kept(d)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package deadletters
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// 1952/WM5, ADR 0564 §5: the watchman makes no consumer while `controller` exists or the lease names a controller, makes
|
||||
// one once both are false, and says the wait meanwhile; it delivers and drops nothing, naming who does; and a kept message
|
||||
// reads as the controller's copies do, a restricted one's age and work queue kept.
|
||||
func TestTheWatchmanKeepsTheDeadLettersOnlyFromTheLeaseWrite(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
handedAway, controllerStands, take, waiting bool
|
||||
}{
|
||||
{false, true, false, false},
|
||||
{false, false, false, false},
|
||||
{true, true, false, true},
|
||||
{true, false, true, false},
|
||||
} {
|
||||
take, waiting := Gate(c.handedAway, c.controllerStands)
|
||||
if take != c.take || waiting != c.waiting {
|
||||
t.Errorf("Gate(%v, %v) = %v, %v", c.handedAway, c.controllerStands, take, waiting)
|
||||
}
|
||||
}
|
||||
for _, c := range []struct {
|
||||
away bool
|
||||
act string
|
||||
says string
|
||||
}{
|
||||
{false, "deliver", "mesh-controller.dead-letters"},
|
||||
{false, "drop", "nothing is dropped"},
|
||||
{true, "drop", "issue 1953"},
|
||||
{true, "deliver", "delivered again"},
|
||||
{true, "send", "not deliver or drop"},
|
||||
} {
|
||||
if err := Refusal(c.away, c.act); err == nil || !strings.Contains(err.Error(), c.says) {
|
||||
t.Errorf("Refusal(%v, %s) = %v, want it to say %q", c.away, c.act, err, c.says)
|
||||
}
|
||||
}
|
||||
at := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
|
||||
h := nats.Header{}
|
||||
h.Set(HeaderStream, "EVENTS")
|
||||
h.Set(HeaderConsumer, "ace_web")
|
||||
h.Set(HeaderSequence, "41")
|
||||
h.Set(HeaderDeliveries, "5")
|
||||
h.Set(HeaderGaveUp, at.Format(time.RFC3339Nano))
|
||||
h.Set(HeaderRestricted, "true")
|
||||
h.Set(HeaderWorkQueue, "true")
|
||||
h.Set(HeaderAgesOut, at.Add(time.Hour).Format(time.RFC3339Nano))
|
||||
h.Set("X-Own", "kept")
|
||||
h.Set("Nats-Msg-Id", "EVENTS.ace_web.41")
|
||||
d := Of(9, "mesh.events.dead.EVENTS.ace_web", h, []byte("body"), true)
|
||||
if d.ID != 9 || d.Sequence != 41 || d.Deliveries != 5 || !d.GaveUp.Equal(at) || !d.Restricted || !d.WorkQueue ||
|
||||
!d.AgesOut.Equal(at.Add(time.Hour)) || d.Who != "how web on ace hears what it consumes" ||
|
||||
d.Body != "body" || len(d.Headers) != 1 || d.Headers["X-Own"][0] != "kept" {
|
||||
t.Fatalf("dead letter = %+v", d)
|
||||
}
|
||||
if d := Of(10, "mesh.events.dead.SEAT_X.y_worker", nats.Header{}, nil, false); d.Stream != "SEAT_X" || d.Consumer != "y_worker" || d.Body != "" {
|
||||
t.Fatalf("a copy without headers reads its stream and consumer from its subject: %+v", d)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
// Package socket is how the watchman's tools ask its program what only the program reads (novox/hq ADR 0564 §12, as the
|
||||
// view's tools ask the view, mesh-delivery-view internal/view/serve.go): DEAD_LETTERS is a stream, which the tool runtime
|
||||
// grants no bundle, and resync is an act of the program. One line of JSON each way, on a unix socket on the program's
|
||||
// machine; the socket keeps nothing.
|
||||
package socket
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Path is where the program listens unless MESH_WATCHMAN_SOCKET says otherwise.
|
||||
const Path = "/var/lib/mesh-delivery-watchman-tools/tools.sock"
|
||||
|
||||
// PathOf is the socket's path as getenv says it.
|
||||
func PathOf(getenv func(string) string) string {
|
||||
if p := strings.TrimSpace(getenv("MESH_WATCHMAN_SOCKET")); p != "" {
|
||||
return p
|
||||
}
|
||||
return Path
|
||||
}
|
||||
|
||||
// The asks the program answers.
|
||||
const (
|
||||
AskDeadLetters = "dead-letters"
|
||||
AskResync = "resync"
|
||||
)
|
||||
|
||||
// Ask is one ask of the tools: dead-letters (consumer, limit, id; deliver or drop with why) or resync.
|
||||
type Ask struct {
|
||||
Ask string `json:"ask"`
|
||||
Consumer string `json:"consumer,omitempty"`
|
||||
Limit int `json:"limit,omitempty"`
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Act string `json:"act,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// Listen listens on the unix socket at path, in a directory of mode 0755, the socket 0666.
|
||||
func Listen(path string) (net.Listener, error) {
|
||||
dir := filepath.Dir(path)
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return nil, err
|
||||
}
|
||||
l, err := net.Listen("unix", path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := os.Chmod(path, 0o666); err != nil {
|
||||
l.Close()
|
||||
return nil, err
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// Serve answers each ask on l with answer until ctx ends.
|
||||
func Serve(ctx context.Context, l net.Listener, answer func(Ask) (any, error)) error {
|
||||
go func() { <-ctx.Done(); l.Close() }()
|
||||
for {
|
||||
c, err := l.Accept()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
go func() {
|
||||
defer c.Close()
|
||||
_ = c.SetDeadline(time.Now().Add(30 * time.Second))
|
||||
line, err := bufio.NewReader(io.LimitReader(c, 4096)).ReadBytes('\n')
|
||||
if err != nil && len(line) == 0 {
|
||||
return
|
||||
}
|
||||
var a Ask
|
||||
enc := json.NewEncoder(c)
|
||||
if json.Unmarshal(line, &a) != nil {
|
||||
_ = enc.Encode(map[string]any{"error": "refused: an ask is one line of JSON"})
|
||||
return
|
||||
}
|
||||
out, err := answer(a)
|
||||
if err != nil {
|
||||
_ = enc.Encode(map[string]any{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
_ = enc.Encode(map[string]any{"answer": out})
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// On asks the program at path and returns its answer.
|
||||
func On(path string, a Ask) (json.RawMessage, error) {
|
||||
c, err := net.DialTimeout("unix", path, 5*time.Second)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cannot reach the watchman at %s: %w", path, err)
|
||||
}
|
||||
defer c.Close()
|
||||
_ = c.SetDeadline(time.Now().Add(30 * time.Second))
|
||||
line, _ := json.Marshal(a)
|
||||
if _, err := c.Write(append(line, '\n')); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var reply struct {
|
||||
Answer json.RawMessage `json:"answer"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
if err := json.NewDecoder(c).Decode(&reply); err != nil {
|
||||
return nil, fmt.Errorf("the watchman's answer cannot be read: %w", err)
|
||||
}
|
||||
if reply.Error != "" {
|
||||
return nil, errors.New(reply.Error)
|
||||
}
|
||||
return reply.Answer, nil
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The bus's advisories the watchman hears (S9, mesh-controller internal/link/advisories.go at e04846e9): a durable consumer
|
||||
// the mesh defines deleted is `consumer-lost`. A message given up on is said from DEAD_LETTERS, where it is kept, for as
|
||||
// long as it is kept (DeadLettersHeld), not from its advisory. The controller's own slow consumer and refused subjects end
|
||||
// with it (ADR 0564 §4).
|
||||
|
||||
// Advisory subjects the watchman subscribes.
|
||||
const (
|
||||
ConsumerDeletedAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>"
|
||||
MaxDeliveriesAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>"
|
||||
)
|
||||
|
||||
// KindConsumerLost is a durable consumer of the mesh deleted.
|
||||
const KindConsumerLost = "consumer-lost"
|
||||
|
||||
// ReadAdvisory is one JetStream advisory as the mesh says it; false for one the watchman does not say. who words a
|
||||
// consumer.
|
||||
func ReadAdvisory(subject string, body []byte, who func(stream, consumer string) string) (Advisory, bool) {
|
||||
var a struct {
|
||||
Stream string `json:"stream"`
|
||||
Consumer string `json:"consumer"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &a); err != nil || a.Stream == "" || a.Consumer == "" {
|
||||
return Advisory{}, false
|
||||
}
|
||||
if !strings.HasPrefix(subject, "$JS.EVENT.ADVISORY.CONSUMER.DELETED.") || !MeshNamed(a.Stream, a.Consumer) {
|
||||
return Advisory{}, false
|
||||
}
|
||||
return Advisory{Kind: KindConsumerLost, ID: a.Stream + "." + a.Consumer, Machine: consumerMachine(a.Stream, a.Consumer),
|
||||
Said: fmt.Sprintf("%s was deleted from the bus", who(a.Stream, a.Consumer))}, true
|
||||
}
|
||||
|
||||
// MeshNamed says a consumer is one of the durable consumers the mesh defines, by its name's shape; every other is a
|
||||
// reader's own, named at random and deleted when its read is done.
|
||||
func MeshNamed(stream, name string) bool {
|
||||
switch {
|
||||
case strings.HasPrefix(stream, "KV_") || stream == "ASSIGNMENTS":
|
||||
return false
|
||||
case stream == "NODES":
|
||||
return true
|
||||
case strings.HasPrefix(stream, "SEAT_"):
|
||||
return strings.HasSuffix(name, "_worker")
|
||||
case name == "controller" || name == "controller-merges":
|
||||
return true
|
||||
case stream == "EVENTS":
|
||||
return strings.Contains(name, "_")
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The dead letters' findings (novox/hq ADR 0564 §5, issue 330): a consumer DEAD_LETTERS holds a message for is said for as
|
||||
// long as it holds any, `bus.<stream>.<consumer>.max-deliveries`, and the notices waiting behind the controller's consumer
|
||||
// after the lease write are said as `bus.DEAD_LETTER_NOTICES.controller.notices-waiting`.
|
||||
|
||||
// KindNoticesWaiting is the finding of notices waiting while the controller's consumer stands after the lease write.
|
||||
const KindNoticesWaiting = "notices-waiting"
|
||||
|
||||
// DeadLettersHeld is DEAD_LETTERS' count per consumer, `<stream>.<consumer>`, read whole: each consumer held writes its
|
||||
// finding, each one no longer held has it deleted. who words a consumer.
|
||||
func (w *Watchman) DeadLettersHeld(held map[string]int, who func(stream, consumer string) string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
want := map[string]bool{}
|
||||
keys := make([]string, 0, len(held))
|
||||
for k := range held {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
n := held[key]
|
||||
stream, consumer, ok := strings.Cut(key, ".")
|
||||
if !ok || n <= 0 {
|
||||
continue
|
||||
}
|
||||
messages, them := "a message", "it is"
|
||||
if n > 1 {
|
||||
messages, them = fmt.Sprintf("%d messages", n), "they are"
|
||||
}
|
||||
f := Finding{Scope: ScopeBus, ID: key, Kind: KindDeadLetters, Severity: Warning, Source: "S9",
|
||||
Machine: consumerMachine(stream, consumer),
|
||||
Summary: fmt.Sprintf("%s gave up on %s; %s kept in DEAD_LETTERS until delivered again or dropped, with why",
|
||||
who(stream, consumer), messages, them),
|
||||
Said: fmt.Sprintf("%d held for %s", n, key),
|
||||
Needs: "deliver them again or drop them, from the mesh MCP server."}
|
||||
want[f.Key()] = true
|
||||
w.find(f)
|
||||
}
|
||||
for key := range w.findings {
|
||||
if strings.HasPrefix(key, ScopeBus+".") && strings.HasSuffix(key, "."+KindDeadLetters) && !want[key] {
|
||||
w.unfind(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// consumerMachine is the machine a module's consumer on EVENTS is on; empty for the others.
|
||||
func consumerMachine(stream, consumer string) string {
|
||||
if stream == "EVENTS" {
|
||||
if node, _, ok := strings.Cut(consumer, "_"); ok {
|
||||
return node
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// NoticesWaiting says whether the lease is handed away while the controller's consumer on DEAD_LETTER_NOTICES still
|
||||
// stands: every notice then waits in the queue, kept a week, until the lease write's step removes the consumer.
|
||||
func (w *Watchman) NoticesWaiting(waiting bool) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
f := Finding{Scope: ScopeBus, ID: "DEAD_LETTER_NOTICES.controller", Kind: KindNoticesWaiting, Severity: Warning, Source: "dead-letters",
|
||||
Summary: "the lease names no controller but the controller's consumer on DEAD_LETTER_NOTICES still stands, so every " +
|
||||
"notice of a message given up on waits in the queue, kept a week, and nothing keeps it in DEAD_LETTERS",
|
||||
Said: "the consumer controller on DEAD_LETTER_NOTICES stands after the lease write",
|
||||
Needs: "remove the consumer controller on DEAD_LETTER_NOTICES with the bus module's tool, the lease write's step."}
|
||||
if waiting {
|
||||
w.find(f)
|
||||
return
|
||||
}
|
||||
w.unfind(f.Key())
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// The watchman is an observer of the seat mesh-watchdogs (novox/hq ADR 0564 §2, §4): what it observes wrong it writes as a
|
||||
// finding, one key `<scope>.<id>.<kind>` of its own bucket `findings` in mesh-sdk's layout (health.Finding), only when what
|
||||
// it says changes, and it deletes the key when it no longer observes it. The condition keeper (mesh-conditions) raises and
|
||||
// clears the conditions from them; the watchman writes no condition and reads none (Limits 11).
|
||||
|
||||
// FindingsBucket is the watchman's findings.
|
||||
var FindingsBucket = health.FindingsBucket(health.WatchmanModule)
|
||||
|
||||
// Observer is the name a finding of the watchman carries.
|
||||
const Observer = health.WatchmanModule
|
||||
|
||||
// The scopes and severities a finding of the watchman names, as the controller's conditions name them, so the keeper's
|
||||
// `compare` meets the same keys.
|
||||
const (
|
||||
ScopeMachine = health.ScopeMachine
|
||||
ScopeDelivery = health.ScopeDelivery
|
||||
ScopeModule = health.ScopeModule
|
||||
ScopeProvider = health.ScopeProvider
|
||||
ScopeBus = health.ScopeBus
|
||||
ScopeCore = health.ScopeCore
|
||||
|
||||
Urgent = string(health.Urgent)
|
||||
Warning = string(health.Warning)
|
||||
)
|
||||
|
||||
// Finding is what the watchman builds of one thing it observes wrong: its subject (scope, id, the machine and the others
|
||||
// it concerns), kind and severity, its evidence (Summary, and Said with Since, its one observation), the words the
|
||||
// operator reads (ADR 0253; empty leaves the keeper's wording of the kind), and the row or fact it comes from (Source).
|
||||
// It is written as health.Finding.
|
||||
type Finding struct {
|
||||
Scope, ID, Token, Kind, Machine string
|
||||
Also []string
|
||||
Severity string
|
||||
Summary, Said string
|
||||
Headline, Explanation string
|
||||
Needs, Resolved string
|
||||
Source string
|
||||
Observer string
|
||||
// Since is when what it says began, as the watchman saw it: the finding's time and its observation's, so the same
|
||||
// thing observed again writes the same bytes.
|
||||
Since time.Time
|
||||
}
|
||||
|
||||
// FindingKey composes a finding's key as the controller composes a condition's (health.Key).
|
||||
func FindingKey(scope, id, token string) string { return health.FindingKey(scope, id, token) }
|
||||
|
||||
// Key is the finding's key: its token, or its kind when it has none.
|
||||
func (f Finding) Key() string {
|
||||
token := f.Token
|
||||
if token == "" {
|
||||
token = f.Kind
|
||||
}
|
||||
return FindingKey(f.Scope, f.ID, token)
|
||||
}
|
||||
|
||||
// EncodeFinding is f in mesh-sdk's layout, refusing one the keeper could not make a condition of.
|
||||
func EncodeFinding(f Finding) ([]byte, error) {
|
||||
said := f.Said
|
||||
if said == "" {
|
||||
said = f.Summary
|
||||
}
|
||||
return health.EncodeFinding(health.Finding{Key: f.Key(), Kind: f.Kind, Token: f.Token,
|
||||
Subject: health.Subject{Scope: f.Scope, ID: f.ID, Machine: f.Machine, Also: f.Also},
|
||||
Severity: health.Severity(f.Severity), Summary: f.Summary, Headline: f.Headline, Explanation: f.Explanation,
|
||||
Resolved: f.Resolved, Needs: f.Needs, Evidence: []health.Evidence{{At: f.Since.UTC(), Said: said}},
|
||||
Source: f.Source, Observer: Observer, At: f.Since.UTC()})
|
||||
}
|
||||
|
||||
// DecodeFinding reads a finding the watchman wrote.
|
||||
func DecodeFinding(data []byte) (Finding, error) {
|
||||
h, err := health.DecodeFinding(data)
|
||||
if err != nil {
|
||||
return Finding{}, err
|
||||
}
|
||||
f := Finding{Scope: h.Subject.Scope, ID: h.Subject.ID, Token: h.Token, Kind: h.Kind, Machine: h.Subject.Machine,
|
||||
Also: h.Subject.Also, Severity: string(h.Severity), Summary: h.Summary, Headline: h.Headline, Explanation: h.Explanation,
|
||||
Needs: h.Needs, Resolved: h.Resolved, Source: h.Source, Observer: h.Observer, Since: h.At}
|
||||
if len(h.Evidence) > 0 {
|
||||
f.Said = h.Evidence[0].Said
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// find writes a finding when what it says differs from what the bucket holds (ADR 0289 §1).
|
||||
func (w *Watchman) find(f Finding) {
|
||||
value, err := EncodeFinding(f)
|
||||
if err != nil {
|
||||
w.say("no finding: %v", err)
|
||||
return
|
||||
}
|
||||
key := f.Key()
|
||||
delete(w.seededFindings, key)
|
||||
if string(w.findings[key]) == string(value) {
|
||||
return
|
||||
}
|
||||
if err := w.out.PutFinding(key, value); err != nil {
|
||||
w.say("cannot write the finding %s: %v", key, err)
|
||||
return
|
||||
}
|
||||
w.findings[key] = value
|
||||
}
|
||||
|
||||
// unfind deletes a finding the watchman no longer observes.
|
||||
func (w *Watchman) unfind(key string) {
|
||||
if _, ok := w.findings[key]; !ok {
|
||||
return
|
||||
}
|
||||
if err := w.out.DeleteFinding(key); err != nil {
|
||||
w.say("cannot delete the finding %s: %v", key, err)
|
||||
return
|
||||
}
|
||||
delete(w.findings, key)
|
||||
}
|
||||
|
||||
// SeedFinding tells the watchman a finding that stands in its bucket at its start; ReconcileFindings deletes it unless what
|
||||
// the watchman reads at its start observes it again.
|
||||
func (w *Watchman) SeedFinding(key string, value []byte) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.findings[key] = value
|
||||
w.seededFindings[key] = true
|
||||
}
|
||||
|
||||
// Findings is every finding the watchman holds now, by key.
|
||||
func (w *Watchman) Findings() map[string]Finding {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
out := map[string]Finding{}
|
||||
for k, v := range w.findings {
|
||||
if f, err := DecodeFinding(v); err == nil {
|
||||
out[k] = f
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Rewrite writes every finding the watchman holds again (resync, ADR 0564 §12): a bucket cleared or a key lost by hand is
|
||||
// written back as the watchman observes it. It answers how many it wrote.
|
||||
func (w *Watchman) Rewrite() int {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
n := 0
|
||||
for key, value := range w.findings {
|
||||
if err := w.out.PutFinding(key, value); err != nil {
|
||||
w.say("cannot write the finding %s again: %v", key, err)
|
||||
continue
|
||||
}
|
||||
n++
|
||||
}
|
||||
return n
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func haveOf(t *testing.T, phase delivery.Phase, health delivery.Health, detail string) []byte {
|
||||
t.Helper()
|
||||
b, err := delivery.EncodeHave(delivery.Have{FP: fpNew, Phase: phase, Health: health, Detail: detail})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// expire hears the expiry of the pulse at pk.
|
||||
func expire(w *Watchman, c *clock, pk string, after time.Duration) {
|
||||
c.t = c.t.Add(after)
|
||||
w.Pulse(pk, Expired, c.t)
|
||||
}
|
||||
|
||||
// The watchman's findings (novox/hq ADR 0564 §2, §4, §5, issue 1952 WM2-WM5): per row of the signals table a key not renewed
|
||||
// writes its finding and a renewed one none; the haves' rows; stuck keys and presence as its own findings; the dead letters.
|
||||
func TestTheWatchmansFindings(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
run func(t *testing.T, w *Watchman, o *fakeOut, c *clock)
|
||||
// want are the finding keys standing at the end, with their severity.
|
||||
want map[string]string
|
||||
}{
|
||||
{"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Step("go", "n1.ask1", runningStep("ask1", t0), Put)
|
||||
expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second)
|
||||
k := FindingKey(ScopeDelivery, stuckKeyOf(t, checksKey, "ask1"), KindStuck)
|
||||
if f := o.findings[k]; f.Kind != KindStuck || !strings.Contains(f.Summary, "the ask ask1 to end on n1") {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.Step("go", "n1.ask1", endedStep("ask1", delivery.StepPassed, t0), Put)
|
||||
}, map[string]string{}},
|
||||
{"WM2 S1: a node offline past its bound is silent, and back online it is not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Connect("node.ace")
|
||||
w.Disconnect("node.ace")
|
||||
expire(w, c, pulseKeyOf(t, presenceNS, "ace"), 120*time.Second)
|
||||
if _, ok := o.findings["machine.ace.silent"]; !ok {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.Connect("node.ace")
|
||||
}, map[string]string{}},
|
||||
{"WM2 S1: a node that said it sleeps is not silent", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Connect("node.lap")
|
||||
w.Power("lap", "sleeping")
|
||||
w.Disconnect("node.lap")
|
||||
expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second)
|
||||
}, map[string]string{}},
|
||||
{"WM2 S1: a node back from sleep and still offline is silent after its bound", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Connect("node.lap")
|
||||
w.Power("lap", "sleeping")
|
||||
w.Disconnect("node.lap")
|
||||
expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second)
|
||||
w.Power("lap", "woke")
|
||||
expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second)
|
||||
}, map[string]string{"machine.lap.silent": Warning}},
|
||||
{"WM2 S11: a tool runner not heard for three intervals is silent; a heartbeat ends it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.ToolsAlive("ace", 0)
|
||||
pk := signalKey("S11", "ace")
|
||||
if o.ttls[pk] != 3*HeartbeatEvery {
|
||||
t.Fatalf("ttl = %v", o.ttls[pk])
|
||||
}
|
||||
expire(w, c, pk, 3*HeartbeatEvery)
|
||||
if _, ok := o.findings["machine.ace.tools-silent"]; !ok {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.ToolsAlive("ace", 20*time.Second)
|
||||
if o.ttls[pk] != time.Minute {
|
||||
t.Fatalf("ttl = %v, want three of the 20 s it says", o.ttls[pk])
|
||||
}
|
||||
w.ToolsAlive("g14", 0) // renewed, never expired: no finding
|
||||
}, map[string]string{}},
|
||||
{"WM2 S10: the self-check not heard for three runs is urgent; not before its first run", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
if len(o.pulses) != 0 {
|
||||
t.Fatal("a pulse before the first run")
|
||||
}
|
||||
w.Checked("novox")
|
||||
expire(w, c, signalKey("S10", "doctor"), 3*SelfCheckEvery)
|
||||
}, map[string]string{"core.doctor.self-check-silent": Urgent}},
|
||||
{"WM3 S8: a failing word is its finding; not said again for 30 min it is silent; recovered ends both", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
st, err := ReadStanding("mesh.mod.keycloak.event.provisioner.failing",
|
||||
[]byte(`{"provider":"oidc","provider-node":"ace","consumer":"photos","node":"g14","class":"refused","attempts":4,"since":"2026-10-18T11:00:00Z"}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.Stood(st)
|
||||
if f := o.findings["provider.keycloak.ace.photos.failing"]; f.Kind != KindProviderFailing || len(f.Also) != 1 {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
expire(w, c, signalKey("S8", "keycloak.ace.photos"), ProviderSaysAgainWithin)
|
||||
if _, ok := o.findings["provider.keycloak.ace.photos.silent"]; !ok {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
st.Failing = false
|
||||
w.Stood(st)
|
||||
}, map[string]string{}},
|
||||
{"WM3 S9: an advisory is a finding counted, ended by an hour's quiet", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
a, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.DELETED.EVENTS.ace_web", []byte(`{"stream":"EVENTS","consumer":"ace_web"}`),
|
||||
func(s, c string) string { return c + " on " + s })
|
||||
if !ok {
|
||||
t.Fatal("a mesh consumer's deletion is not read")
|
||||
}
|
||||
if _, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.DELETED.KV_x.abc", []byte(`{"stream":"KV_x","consumer":"abc"}`), nil); ok {
|
||||
t.Fatal("a reader's own consumer is said")
|
||||
}
|
||||
w.Advised(a)
|
||||
w.Advised(a)
|
||||
k := "bus.EVENTS.ace_web.consumer-lost"
|
||||
if f := o.findings[k]; !strings.Contains(f.Summary, "2 times") || f.Machine != "ace" {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
expire(w, c, signalKey("S9", k), AdvisoryQuiet)
|
||||
}, map[string]string{}},
|
||||
{"WM3 S21: a have failed for one apply interval is apply-failing; applied again it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Have("ace.web", haveOf(t, delivery.PhaseFailed, delivery.HealthUnknown, "refused: the package manager (pacman) does not answer"), Put)
|
||||
expire(w, c, signalKey("S21", "ace.web"), ApplyEvery)
|
||||
if f := o.findings["machine.ace.web.apply-failing"]; !strings.Contains(f.Said, "pacman") {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseFailed, delivery.HealthUnknown, "refused"), Put)
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
|
||||
expire(w, c, signalKey("S21", "ace.mail"), ApplyEvery) // applied before its bound: nothing
|
||||
}, map[string]string{}},
|
||||
{"WM3: unhealthy only at the second write in a row, urgent after four hours", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "the unit failed"), Put)
|
||||
if len(o.findings) != 0 {
|
||||
t.Fatalf("one write raised %v", o.findings)
|
||||
}
|
||||
w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "the unit failed"), Put)
|
||||
if o.findings["module.web.ace.unhealthy"].Severity != Warning {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter)
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "a part waits for the secret token"), Put)
|
||||
}, map[string]string{"module.web.ace.unhealthy": Urgent, "module.mail.ace.needs-operator": Warning}},
|
||||
{"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "waits"), Put)
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseRemoved, delivery.HealthUnknown, ""), Put)
|
||||
}, map[string]string{}},
|
||||
{"WM4: resync after a cleared bucket writes the same findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "waits"), Put)
|
||||
o.findings = map[string]Finding{}
|
||||
if n := w.Rewrite(); n != 1 {
|
||||
t.Fatalf("rewrote %d", n)
|
||||
}
|
||||
}, map[string]string{"module.mail.ace.needs-operator": Warning}},
|
||||
{"WM4: a finding read at the start and not observed again is deleted; a signal's stays", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
for _, f := range []Finding{
|
||||
{Scope: ScopeDelivery, ID: "x.y", Kind: KindStuck, Severity: Warning, Summary: "old", Source: "stuck"},
|
||||
{Scope: ScopeMachine, ID: "ace", Kind: KindToolsSilent, Severity: Warning, Summary: "silent", Source: "S11"},
|
||||
} {
|
||||
v, _ := EncodeFinding(f)
|
||||
o.findings[f.Key()], _ = DecodeFinding(v)
|
||||
w.SeedFinding(f.Key(), v)
|
||||
}
|
||||
w.Reconcile()
|
||||
}, map[string]string{"machine.ace.tools-silent": Warning}},
|
||||
{"WM5: a consumer DEAD_LETTERS holds for is its finding until it holds none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
|
||||
who := func(s, c string) string { return c }
|
||||
w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 2, "SEAT_X.y_worker": 1}, who)
|
||||
if f := o.findings["bus.EVENTS.ace_web.max-deliveries"]; f.Machine != "ace" || !strings.Contains(f.Summary, "2 messages") {
|
||||
t.Fatalf("findings = %v", o.findings)
|
||||
}
|
||||
w.DeadLettersHeld(map[string]int{"SEAT_X.y_worker": 1}, who)
|
||||
w.NoticesWaiting(true)
|
||||
}, map[string]string{"bus.SEAT_X.y_worker.max-deliveries": Warning, "bus.DEAD_LETTER_NOTICES.controller.notices-waiting": Warning}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
w, o, c := setup()
|
||||
tc.run(t, w, o, c)
|
||||
got := map[string]string{}
|
||||
for k, f := range o.findings {
|
||||
got[k] = f.Severity
|
||||
if f.Key() != k || f.Observer != Observer {
|
||||
t.Errorf("finding %s = %+v", k, f)
|
||||
}
|
||||
}
|
||||
if len(got) != len(tc.want) {
|
||||
t.Fatalf("findings = %v, want %v", got, tc.want)
|
||||
}
|
||||
for k, sev := range tc.want {
|
||||
if got[k] != sev {
|
||||
t.Fatalf("findings = %v, want %v", got, tc.want)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -116,6 +116,7 @@ func (w *Watchman) Have(key string, value []byte, op Op) {
|
||||
return
|
||||
case Delete:
|
||||
delete(w.haves, key)
|
||||
w.haveFindings(key, node, module, nil)
|
||||
return
|
||||
}
|
||||
have, err := delivery.DecodeHave(value)
|
||||
@@ -124,6 +125,7 @@ func (w *Watchman) Have(key string, value []byte, op Op) {
|
||||
return
|
||||
}
|
||||
w.haves[key] = have.FP
|
||||
w.haveFindings(key, node, module, &have)
|
||||
wk, err := delivery.WantKey(node, module)
|
||||
if err != nil {
|
||||
return
|
||||
|
||||
@@ -0,0 +1,524 @@
|
||||
package watch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The rows of the signals table the watchman holds (novox/hq ADR 0564 §4, Limits 3 and 11). Each is a key of the pulse
|
||||
// bucket whose time to live is the row's bound, renewed by the signal's own event: its expiry is the finding, so no loop
|
||||
// looks every half minute. A signal's pulse is kept under the pulse bucket's pseudo-bucket `signal`, `signal.<row>.<id>`,
|
||||
// and its value says what its expiry does (signalDue), so a pulse that stood across a restart still acts when it expires.
|
||||
//
|
||||
// What each row hears is on the bus, never in the controller's store:
|
||||
// - S1, a machine silent: the watchman's own presence of `node.<node>` (its stuck key), not while the machine said it
|
||||
// sleeps or shuts down (the power seat's events, ADR 0211);
|
||||
// - S8, a provider's failing word repeated: the providers' `provisioner.failing` and `provisioner.recovered` (ADR 0224);
|
||||
// the failing word is itself the finding `provider-failing`, and its pulse of 30 minutes, renewed by each repeat, is S8;
|
||||
// - S9, the bus's advisories: each occurrence a finding, cleared by the expiry of an hour's pulse renewed by the next;
|
||||
// - S10, the self-check's `checked`: a pulse of three runs' length, from the first run heard;
|
||||
// - S11, a machine's tool runner heartbeat on `mesh.control.<node>.tools-alive`: a pulse of three intervals;
|
||||
// - S21, a module's have failed: a pulse of one apply interval from the first failure, whose expiry while it still fails
|
||||
// is the finding `apply-failing`;
|
||||
// - a module's have unhealthy in two writes in a row, `unhealthy`, urgent after four hours; and a have that waits for
|
||||
// the operator, `needs-operator` (ADR 0240 §12, §17).
|
||||
|
||||
// The bounds of the rows.
|
||||
const (
|
||||
// ProviderSaysAgainWithin is how long a provider's failing word stays current without being said again (S8): twice
|
||||
// the quarter of an hour it repeats it at.
|
||||
ProviderSaysAgainWithin = 30 * time.Minute
|
||||
// AdvisoryQuiet is how long the bus must be quiet about a thing before its advisory clears (S9).
|
||||
AdvisoryQuiet = time.Hour
|
||||
// SelfCheckEvery is the self-check's run interval; S10's bound is three of them.
|
||||
SelfCheckEvery = 5 * time.Minute
|
||||
// HeartbeatEvery is a tool runner's interval when it says none; S11's bound is three of them.
|
||||
HeartbeatEvery = 60 * time.Second
|
||||
// ApplyEvery is one apply interval (S21): a failure still there after it is not one the next apply clears.
|
||||
ApplyEvery = 5 * time.Minute
|
||||
// UnhealthyUrgentAfter is how long a module stays unhealthy before it is urgent.
|
||||
UnhealthyUrgentAfter = 4 * time.Hour
|
||||
)
|
||||
|
||||
// signalBucket is the pseudo-bucket of a signal's pulse key.
|
||||
const signalBucket = "signal"
|
||||
|
||||
// signalDue is what a signal's pulse does when it expires: raise the finding (find), or end it (end, an advisory's quiet
|
||||
// hour). Kept as the pulse's Owes, so it is read back at a restart.
|
||||
type signalDue struct {
|
||||
Do string `json:"do"`
|
||||
Finding Finding `json:"finding"`
|
||||
// While names what must still stand at the expiry for a find to be written: a have's key still failing or unhealthy;
|
||||
// empty for a row whose expiry alone is the finding.
|
||||
While string `json:"while,omitempty"`
|
||||
}
|
||||
|
||||
const (
|
||||
dueFind = "find"
|
||||
dueEnd = "end"
|
||||
)
|
||||
|
||||
// signalKey is the pulse key of a row's signal about id.
|
||||
func signalKey(row, id string) string {
|
||||
parts := []string{signalBucket, row}
|
||||
for _, p := range strings.Split(id, ".") {
|
||||
if p = unsafeKey.ReplaceAllString(strings.TrimSpace(p), "_"); p == "" {
|
||||
p = "_"
|
||||
}
|
||||
parts = append(parts, p)
|
||||
}
|
||||
return strings.Join(parts, ".")
|
||||
}
|
||||
|
||||
// unsafeKey is what a key's token may not hold: the bus takes letters, digits and `-_/=`.
|
||||
var unsafeKey = regexp.MustCompile(`[^A-Za-z0-9_=/-]`)
|
||||
|
||||
// signal writes (or renews) the pulse of a row's signal, whose expiry does due.
|
||||
func (w *Watchman) signal(row, id string, ttl time.Duration, due signalDue) {
|
||||
pk := signalKey(row, id)
|
||||
owes, err := json.Marshal(due)
|
||||
if err != nil {
|
||||
w.say("no signal %s: %v", pk, err)
|
||||
return
|
||||
}
|
||||
value, err := delivery.EncodePulse(delivery.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)})
|
||||
if err != nil {
|
||||
w.say("no signal %s: %v", pk, err)
|
||||
return
|
||||
}
|
||||
if err := w.out.PutPulse(pk, value, ttl); err != nil {
|
||||
w.say("cannot write the signal %s: %v", pk, err)
|
||||
return
|
||||
}
|
||||
w.signals[pk] = due
|
||||
}
|
||||
|
||||
// unsignal ends a row's signal about id: its pulse deleted, so it never expires.
|
||||
func (w *Watchman) unsignal(row, id string) {
|
||||
pk := signalKey(row, id)
|
||||
if _, ok := w.signals[pk]; !ok {
|
||||
return
|
||||
}
|
||||
delete(w.signals, pk)
|
||||
if err := w.out.DeletePulse(pk); err != nil {
|
||||
w.say("cannot delete the signal %s: %v", pk, err)
|
||||
}
|
||||
}
|
||||
|
||||
// SeedSignal tells the watchman a signal's pulse that stands at its start, with its value: it is kept, since its event
|
||||
// renews it and its expiry still acts.
|
||||
func (w *Watchman) SeedSignal(key string, value []byte) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if due, ok := dueOf(value); ok {
|
||||
w.signals[key] = due
|
||||
}
|
||||
}
|
||||
|
||||
func dueOf(value []byte) (signalDue, bool) {
|
||||
p, err := delivery.DecodePulse(value)
|
||||
if err != nil || p.Bucket != signalBucket {
|
||||
return signalDue{}, false
|
||||
}
|
||||
var due signalDue
|
||||
if json.Unmarshal([]byte(p.Owes), &due) != nil || (due.Do != dueFind && due.Do != dueEnd) {
|
||||
return signalDue{}, false
|
||||
}
|
||||
return due, true
|
||||
}
|
||||
|
||||
// signalExpired is a signal's pulse expired: its due is done.
|
||||
func (w *Watchman) signalExpired(pk string) {
|
||||
due, ok := w.signals[pk]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(w.signals, pk)
|
||||
switch due.Do {
|
||||
case dueEnd:
|
||||
w.unfind(due.Finding.Key())
|
||||
case dueFind:
|
||||
if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) {
|
||||
return
|
||||
}
|
||||
w.find(due.Finding)
|
||||
}
|
||||
}
|
||||
|
||||
// stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for
|
||||
// unhealthy.
|
||||
func (w *Watchman) stillWrong(haveKey, kind string) bool {
|
||||
h, ok := w.haveState[haveKey]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
switch kind {
|
||||
case KindApplyFailing:
|
||||
return h.Phase == delivery.PhaseFailed
|
||||
case KindUnhealthy:
|
||||
return h.Health == delivery.HealthFailed
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// The kinds the watchman's rows write, as the controller's conditions name them.
|
||||
const (
|
||||
KindSilent = "silent"
|
||||
KindStuck = "stuck"
|
||||
KindProviderFailing = "provider-failing"
|
||||
KindProviderSilent = "provider-silent"
|
||||
KindSelfCheckSilent = "self-check-silent"
|
||||
KindToolsSilent = "tools-silent"
|
||||
KindApplyFailing = "apply-failing"
|
||||
KindUnhealthy = "module-unhealthy"
|
||||
KindNeedsOperator = "needs-operator"
|
||||
KindDeadLetters = "max-deliveries"
|
||||
)
|
||||
|
||||
func stamp(t time.Time) string { return t.UTC().Format("2006-01-02 15:04 MST") }
|
||||
|
||||
// ---- S1 and the power seat ----
|
||||
|
||||
// PowerEvents are the power seat's events that say whether a machine is about to be away or is back (ADR 0211).
|
||||
var PowerEvents = map[string]bool{"sleeping": true, "shutting-down": true, "booted": true, "woke": true}
|
||||
|
||||
// Power is a machine's power event: sleeping and shutting-down say it is away, booted and woke that it is back. A machine
|
||||
// that said it is away is not silent; one back that is still offline is watched again from now.
|
||||
func (w *Watchman) Power(node, state string) {
|
||||
if !PowerEvents[state] || node == "" {
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
away := state == "sleeping" || state == "shutting-down"
|
||||
w.away[node] = away
|
||||
key, err := delivery.PresenceNodeKey(node)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if away {
|
||||
w.unfind(FindingKey(ScopeMachine, node, KindSilent))
|
||||
return
|
||||
}
|
||||
if p, ok := w.presence[key]; ok && !p.online {
|
||||
p.since = w.now()
|
||||
w.presence[key] = p
|
||||
w.ensureOffline(key)
|
||||
}
|
||||
}
|
||||
|
||||
// stuckFinding is the finding of a stuck key as it stands (ADR 0433 §2, §3, now the watchman's own): a node's presence not
|
||||
// moved is S1, the machine silent, unless it said it is away; any other key not moved is `delivery.<key>.stuck`; a key
|
||||
// moved ends its finding.
|
||||
func (w *Watchman) stuckFinding(sk string, s delivery.Stuck) {
|
||||
if s.Bucket == delivery.PresenceBucket {
|
||||
if node, part, _, err := delivery.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" {
|
||||
key := FindingKey(ScopeMachine, node, KindSilent)
|
||||
if s.Moved || w.away[node] {
|
||||
w.unfind(key)
|
||||
return
|
||||
}
|
||||
w.find(Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1",
|
||||
Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), delivery.PulseMoveBound),
|
||||
Said: fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since})
|
||||
return
|
||||
}
|
||||
}
|
||||
key := FindingKey(ScopeDelivery, sk, KindStuck)
|
||||
if s.Moved {
|
||||
w.unfind(key)
|
||||
return
|
||||
}
|
||||
waited := s.Waited
|
||||
if waited == "" {
|
||||
waited = "a move"
|
||||
}
|
||||
w.find(Finding{Scope: ScopeDelivery, ID: sk, Kind: KindStuck, Severity: Warning, Source: "stuck",
|
||||
Summary: fmt.Sprintf("%s has not moved since %s: it waited for %s", sk, stamp(s.Since), waited),
|
||||
Said: fmt.Sprintf("%s: waiting for %s since %s", sk, waited, s.Since.UTC().Format(time.RFC3339)), Since: s.Since})
|
||||
}
|
||||
|
||||
// ---- S8 ----
|
||||
|
||||
// Standing is a provider's word about one consumer (ADR 0224): Module is the emitter, read from the subject.
|
||||
type Standing struct {
|
||||
Module string `json:"-"`
|
||||
Failing bool `json:"-"`
|
||||
Provider string `json:"provider"`
|
||||
ProviderNode string `json:"provider-node"`
|
||||
Consumer string `json:"consumer"`
|
||||
Node string `json:"node"`
|
||||
Class string `json:"class,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
Attempts int `json:"attempts"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
// ReadStanding reads a provider's word from its subject, `mesh.mod.<module>.event.provisioner.failing` or `.recovered`.
|
||||
func ReadStanding(subject string, body []byte) (Standing, error) {
|
||||
rest, ok := strings.CutPrefix(subject, "mesh.mod.")
|
||||
if !ok {
|
||||
return Standing{}, fmt.Errorf("%s is not a provider's word", subject)
|
||||
}
|
||||
module, event, ok := strings.Cut(rest, ".event.")
|
||||
if !ok || module == "" || (event != "provisioner.failing" && event != "provisioner.recovered") {
|
||||
return Standing{}, fmt.Errorf("%s is not a provider's word", subject)
|
||||
}
|
||||
var st Standing
|
||||
if err := json.Unmarshal(body, &st); err != nil {
|
||||
return Standing{}, fmt.Errorf("%s's word could not be read: %w", module, err)
|
||||
}
|
||||
if st.Consumer == "" {
|
||||
return Standing{}, fmt.Errorf("%s's word names no consumer", module)
|
||||
}
|
||||
st.Module, st.Failing = module, event == "provisioner.failing"
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// Stood is a provider's word: failing writes `provider.<module>.<node>.<consumer>.failing` and renews S8's pulse, whose
|
||||
// expiry writes `provider-silent`; recovered deletes both (Limits 11).
|
||||
func (w *Watchman) Stood(st Standing) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
id := st.Module + "." + st.ProviderNode + "." + st.Consumer
|
||||
failing := FindingKey(ScopeProvider, id, "failing")
|
||||
silentKey := FindingKey(ScopeProvider, id, "silent")
|
||||
if !st.Failing {
|
||||
w.unfind(failing)
|
||||
w.unfind(silentKey)
|
||||
w.unsignal("S8", id)
|
||||
return
|
||||
}
|
||||
class := st.Class
|
||||
if class == "" {
|
||||
class = "failing"
|
||||
}
|
||||
whom := st.Consumer
|
||||
if st.Node != "" {
|
||||
whom += " on " + st.Node
|
||||
}
|
||||
said := class
|
||||
if st.Error != "" {
|
||||
said += ": " + firstLine(st.Error)
|
||||
}
|
||||
var also []string
|
||||
if st.Node != "" && st.Node != st.ProviderNode {
|
||||
also = []string{st.Node}
|
||||
}
|
||||
w.find(Finding{Scope: ScopeProvider, ID: id, Token: "failing", Kind: KindProviderFailing, Machine: st.ProviderNode, Also: also,
|
||||
Severity: Warning, Source: "provisioner.failing", Since: st.Since, Said: said,
|
||||
Summary: fmt.Sprintf("%s on %s keeps failing %s: %s, %d attempt(s) since %s", st.Module, st.ProviderNode, whom, class,
|
||||
st.Attempts, stamp(st.Since))})
|
||||
w.unfind(silentKey)
|
||||
now := w.now()
|
||||
w.signal("S8", id, ProviderSaysAgainWithin, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeProvider, ID: id, Token: "silent",
|
||||
Kind: KindProviderSilent, Machine: st.ProviderNode, Severity: Warning, Source: "S8", Since: now,
|
||||
Summary: fmt.Sprintf("%s on %s said it keeps failing %s and has said nothing since %s: it stopped saying anything, so "+
|
||||
"its last word is all the mesh has", st.Module, st.ProviderNode, st.Consumer, stamp(now)),
|
||||
Said: fmt.Sprintf("not said again within %s", ProviderSaysAgainWithin)}})
|
||||
}
|
||||
|
||||
func firstLine(s string) string {
|
||||
s, _, _ = strings.Cut(strings.TrimSpace(s), "\n")
|
||||
return s
|
||||
}
|
||||
|
||||
// ---- S9 ----
|
||||
|
||||
// Advisory is one thing the bus said (S9): its kind (slow-consumer, connection-closed, max-deliveries, refused,
|
||||
// consumer-lost), what it is about, its words, and the machine when it names one.
|
||||
type Advisory struct {
|
||||
Kind, ID, Said, Machine string
|
||||
}
|
||||
|
||||
// Advised is an advisory heard: its finding `bus.<id>.<kind>` is written at once, counting the times since its first, and
|
||||
// an hour's pulse renewed by each, whose expiry ends it.
|
||||
func (w *Watchman) Advised(a Advisory) {
|
||||
if a.Kind == "" || a.ID == "" {
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
f := Finding{Scope: ScopeBus, ID: a.ID, Kind: a.Kind, Machine: a.Machine, Severity: Warning, Source: "S9", Said: a.Said}
|
||||
key := f.Key()
|
||||
n := w.advised[key] + 1
|
||||
if _, open := w.findings[key]; !open {
|
||||
n = 1
|
||||
w.advisedSince[key] = w.now()
|
||||
}
|
||||
w.advised[key] = n
|
||||
since := w.advisedSince[key]
|
||||
f.Since, f.Summary = since, a.Said
|
||||
if n > 1 {
|
||||
f.Summary = fmt.Sprintf("%s (%d times since %s)", a.Said, n, since.UTC().Format("15:04 MST"))
|
||||
}
|
||||
w.find(f)
|
||||
w.signal("S9", key, AdvisoryQuiet, signalDue{Do: dueEnd, Finding: f})
|
||||
}
|
||||
|
||||
// ---- S10 ----
|
||||
|
||||
// Checked is the self-check's `checked` heard: its pulse of three runs' length is renewed, and its silence ended. Before
|
||||
// the first run heard there is no pulse, so a self-check not yet placed is not said silent.
|
||||
func (w *Watchman) Checked(machine string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.unfind(FindingKey(ScopeCore, "doctor", KindSelfCheckSilent))
|
||||
now := w.now()
|
||||
w.signal("S10", "doctor", 3*SelfCheckEvery, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeCore, ID: "doctor",
|
||||
Kind: KindSelfCheckSilent, Machine: machine, Severity: Urgent, Source: "S10", Since: now,
|
||||
Summary: fmt.Sprintf("the self-check has not finished a run since %s (it runs every %s): the mesh's invariants are not "+
|
||||
"being checked", stamp(now), SelfCheckEvery),
|
||||
Said: fmt.Sprintf("no run within %s", 3*SelfCheckEvery)}})
|
||||
}
|
||||
|
||||
// ---- S11 ----
|
||||
|
||||
// ToolsAlive is a machine's tool runner heartbeat, saying its interval (zero: HeartbeatEvery): its pulse of three
|
||||
// intervals is renewed and its silence ended.
|
||||
func (w *Watchman) ToolsAlive(node string, every time.Duration) {
|
||||
if node == "" {
|
||||
return
|
||||
}
|
||||
if every <= 0 {
|
||||
every = HeartbeatEvery
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.unfind(FindingKey(ScopeMachine, node, KindToolsSilent))
|
||||
if w.away[node] {
|
||||
w.unsignal("S11", node)
|
||||
return
|
||||
}
|
||||
now := w.now()
|
||||
w.signal("S11", node, 3*every, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeMachine, ID: node, Kind: KindToolsSilent,
|
||||
Machine: node, Severity: Warning, Source: "S11", Since: now,
|
||||
Summary: fmt.Sprintf("the node tools on %s have not said they are there since %s (bound %s): nothing can ask that "+
|
||||
"machine anything", node, stamp(now), 3*every),
|
||||
Said: fmt.Sprintf("silent for more than %s", 3*every)}})
|
||||
}
|
||||
|
||||
// ToolsNode is the machine a tool runner heartbeat's subject names, `mesh.control.<node>.tools-alive`.
|
||||
func ToolsNode(subject string) string {
|
||||
rest, ok := strings.CutPrefix(subject, "mesh.control.")
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
node, ok := strings.CutSuffix(rest, ".tools-alive")
|
||||
if !ok || strings.Contains(node, ".") {
|
||||
return ""
|
||||
}
|
||||
return node
|
||||
}
|
||||
|
||||
// ---- S21, unhealthy and needs-operator, from the haves ----
|
||||
|
||||
// haveFindings reads a have's change for the rows the haves feed: S21 (failed for one apply interval), unhealthy (two
|
||||
// writes in a row, urgent after four hours) and needs-operator (it waits for the operator). A have removed or deleted ends
|
||||
// them all.
|
||||
func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Have) {
|
||||
applyKey := FindingKey(ScopeMachine, node+"."+module, KindApplyFailing)
|
||||
unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy")
|
||||
waitsKey := FindingKey(ScopeModule, module+"."+node, KindNeedsOperator)
|
||||
if h == nil || h.Phase == delivery.PhaseRemoved {
|
||||
delete(w.haveState, key)
|
||||
delete(w.unhealthy, key)
|
||||
w.unsignal("S21", node+"."+module)
|
||||
w.unsignal("unhealthy", module+"."+node)
|
||||
w.unfind(applyKey)
|
||||
w.unfind(unhealthyKey)
|
||||
w.unfind(waitsKey)
|
||||
return
|
||||
}
|
||||
was, had := w.haveState[key]
|
||||
w.haveState[key] = *h
|
||||
now := w.now()
|
||||
|
||||
// S21: a have refused or applied in part says phase failed, with the node's words in its detail.
|
||||
if h.Phase == delivery.PhaseFailed {
|
||||
if !had || was.Phase != delivery.PhaseFailed {
|
||||
w.signal("S21", node+"."+module, ApplyEvery, signalDue{Do: dueFind, While: key, Finding: Finding{Scope: ScopeMachine,
|
||||
ID: node + "." + module, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: now,
|
||||
Summary: fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, module, stamp(now), ApplyEvery),
|
||||
Said: firstLine(orWords(h.Detail, "no reason given")),
|
||||
Headline: node + " is not applying its update", Resolved: node + " applies what it is sent again"}})
|
||||
}
|
||||
} else {
|
||||
w.unsignal("S21", node+"."+module)
|
||||
w.unfind(applyKey)
|
||||
}
|
||||
|
||||
// unhealthy: two writes in a row that say it failed.
|
||||
if h.Health == delivery.HealthFailed {
|
||||
w.unhealthy[key]++
|
||||
if w.unhealthy[key] == 2 {
|
||||
f := Finding{Scope: ScopeModule, ID: module + "." + node, Token: "unhealthy", Kind: KindUnhealthy, Machine: node,
|
||||
Severity: Warning, Source: "health", Since: now,
|
||||
Summary: fmt.Sprintf("%s on %s is not healthy: %s", module, node, orWords(h.Detail, "its check failed")),
|
||||
Said: firstLine(orWords(h.Detail, "failed")),
|
||||
Headline: fmt.Sprintf("%s not working on %s", module, node)}
|
||||
w.find(f)
|
||||
f.Severity = Urgent
|
||||
w.signal("unhealthy", module+"."+node, UnhealthyUrgentAfter, signalDue{Do: dueFind, While: key, Finding: f})
|
||||
}
|
||||
} else {
|
||||
delete(w.unhealthy, key)
|
||||
w.unsignal("unhealthy", module+"."+node)
|
||||
w.unfind(unhealthyKey)
|
||||
}
|
||||
|
||||
// needs-operator: a part waits for the operator's secret or setting.
|
||||
if h.Health == delivery.HealthWaits {
|
||||
w.find(Finding{Scope: ScopeModule, ID: module + "." + node, Kind: KindNeedsOperator, Machine: node, Severity: Warning,
|
||||
Source: "health", Since: now,
|
||||
Summary: fmt.Sprintf("%s on %s waits for the operator: %s", module, node, orWords(h.Detail, "a part waits for a secret or a setting")),
|
||||
Said: firstLine(orWords(h.Detail, "waits"))})
|
||||
} else {
|
||||
w.unfind(waitsKey)
|
||||
}
|
||||
}
|
||||
|
||||
func orWords(s, otherwise string) string {
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return otherwise
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// SignalRows are the rows the watchman holds, in the table's order, for `signals`.
|
||||
var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
|
||||
{"S1", "a machine's presence (not while it said it sleeps or shuts down)", delivery.PulseMoveBound.String(), KindSilent},
|
||||
{"S8", "a provider's failing word repeated", ProviderSaysAgainWithin.String(), KindProviderSilent},
|
||||
{"S9", "the bus's advisories", "any occurrence; clears after " + AdvisoryQuiet.String() + " without another", "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost"},
|
||||
{"S10", "the self-check's checked", (3 * SelfCheckEvery).String(), KindSelfCheckSilent},
|
||||
{"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent},
|
||||
{"S21", "a have refused or applied in part", ApplyEvery.String(), KindApplyFailing},
|
||||
{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
|
||||
{"health", "a have that waits for the operator", "at once", KindNeedsOperator},
|
||||
}
|
||||
|
||||
// Signal is one signal's pulse as `signals` shows it.
|
||||
type Signal struct {
|
||||
Key string `json:"key"`
|
||||
Do string `json:"do"`
|
||||
Finding Finding `json:"finding"`
|
||||
}
|
||||
|
||||
// Signals is every signal's pulse the watchman holds, sorted by key.
|
||||
func (w *Watchman) Signals() []Signal {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
out := make([]Signal, 0, len(w.signals))
|
||||
for k, d := range w.signals {
|
||||
out = append(out, Signal{Key: k, Do: d.Do, Finding: d.Finding})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
|
||||
return out
|
||||
}
|
||||
@@ -31,6 +31,9 @@ type Out interface {
|
||||
// DeletePulse deletes the pulse key (a delete, not an expiry).
|
||||
DeletePulse(key string) error
|
||||
PutStuck(key string, value []byte) error
|
||||
// PutFinding writes one of the watchman's findings; DeleteFinding deletes one it no longer observes.
|
||||
PutFinding(key string, value []byte) error
|
||||
DeleteFinding(key string) error
|
||||
}
|
||||
|
||||
type presence struct {
|
||||
@@ -60,6 +63,20 @@ type Watchman struct {
|
||||
wants map[string]string // the fingerprint of each want that stands, by want key
|
||||
haves map[string]string // the fingerprint each node runs of a module, by have key
|
||||
bound time.Duration
|
||||
// findings is every finding written, by key, as written; seededFindings those read at the start and not yet observed
|
||||
// again.
|
||||
findings map[string][]byte
|
||||
seededFindings map[string]bool
|
||||
// signals is every signal's pulse standing, by pulse key, with what its expiry does.
|
||||
signals map[string]signalDue
|
||||
// away is each machine that said it sleeps or shuts down, until it says it is back.
|
||||
away map[string]bool
|
||||
// haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each
|
||||
// advisory's count while its finding stands and advisedSince since when.
|
||||
haveState map[string]delivery.Have
|
||||
unhealthy map[string]int
|
||||
advised map[string]int
|
||||
advisedSince map[string]time.Time
|
||||
// stepBound is how long a build step may run once a node took the ask; a test shortens it.
|
||||
stepBound time.Duration
|
||||
}
|
||||
@@ -67,7 +84,9 @@ type Watchman struct {
|
||||
// New is a watchman writing to out, reading the time from now.
|
||||
func New(out Out, now func() time.Time) *Watchman {
|
||||
return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
|
||||
pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound}
|
||||
pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
|
||||
findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{},
|
||||
haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}}
|
||||
}
|
||||
|
||||
// Logf sets where a write that failed is said; it is silent until set.
|
||||
@@ -178,6 +197,7 @@ func (w *Watchman) putStuck(sk string, s delivery.Stuck) {
|
||||
}
|
||||
s.V = delivery.Version
|
||||
w.stuck[sk] = s
|
||||
w.stuckFinding(sk, s)
|
||||
}
|
||||
|
||||
// Pulse is a change heard on the pulse bucket. Only the server's expiry marker of a pulse the watchman made writes a
|
||||
@@ -188,6 +208,10 @@ func (w *Watchman) Pulse(key string, op Op, at time.Time) {
|
||||
if op != Expired {
|
||||
return
|
||||
}
|
||||
if _, ok := w.signals[key]; ok {
|
||||
w.signalExpired(key)
|
||||
return
|
||||
}
|
||||
p, ok := w.pulses[key]
|
||||
if !ok || p.seeded {
|
||||
return
|
||||
@@ -202,6 +226,7 @@ func (w *Watchman) SeedStuck(key string, value []byte) {
|
||||
defer w.mu.Unlock()
|
||||
if s, err := delivery.DecodeStuck(value); err == nil {
|
||||
w.stuck[key] = s
|
||||
w.stuckFinding(key, s)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -211,7 +236,7 @@ func (w *Watchman) SeedPulse(key string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
bucket, k, err := delivery.ParsePulseKey(key)
|
||||
if err != nil {
|
||||
if err != nil || bucket == signalBucket {
|
||||
return
|
||||
}
|
||||
w.pulses[key] = &pulse{bucket: bucket, key: k, seeded: true}
|
||||
@@ -227,6 +252,38 @@ func (w *Watchman) Reconcile() {
|
||||
w.deletePulse(pk)
|
||||
}
|
||||
}
|
||||
w.reconcileFindings()
|
||||
}
|
||||
|
||||
// reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not
|
||||
// observe again is deleted, unless a signal still standing keeps it (a provider's failing word, an advisory's quiet hour, a
|
||||
// machine's silent tools, a silent self-check): those end by their own events or expiry.
|
||||
func (w *Watchman) reconcileFindings() {
|
||||
kept := map[string]bool{}
|
||||
for _, due := range w.signals {
|
||||
kept[due.Finding.Key()] = true
|
||||
}
|
||||
for key := range w.seededFindings {
|
||||
delete(w.seededFindings, key)
|
||||
if kept[key] || signalKind(w.findings[key]) {
|
||||
continue
|
||||
}
|
||||
w.unfind(key)
|
||||
}
|
||||
}
|
||||
|
||||
// signalKind says a finding is one a signal row writes and ends by its own event (S8, S9, S10, S11): it outlives a
|
||||
// restart, since what ends it is heard, not read.
|
||||
func signalKind(value []byte) bool {
|
||||
f, err := DecodeFinding(value)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
switch f.Source {
|
||||
case "provisioner.failing", "S8", "S9", "S10", "S11":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func describe(format string, args ...any) string { return fmt.Sprintf(format, args...) }
|
||||
|
||||
@@ -19,10 +19,31 @@ type fakeOut struct {
|
||||
pulseDeletes []string
|
||||
stuck map[string]delivery.Stuck
|
||||
stuckWrites int
|
||||
findings map[string]Finding
|
||||
findingWrites int
|
||||
}
|
||||
|
||||
func newOut() *fakeOut {
|
||||
return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{}}
|
||||
return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{},
|
||||
findings: map[string]Finding{}}
|
||||
}
|
||||
|
||||
func (o *fakeOut) PutFinding(key string, value []byte) error {
|
||||
f, err := DecodeFinding(value)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if f.Key() != key {
|
||||
return errors.New("a finding written under another key than its own: " + key)
|
||||
}
|
||||
o.findings[key] = f
|
||||
o.findingWrites++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (o *fakeOut) DeleteFinding(key string) error {
|
||||
delete(o.findings, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (o *fakeOut) PutPresence(key string, online bool) error {
|
||||
@@ -116,7 +137,12 @@ func stuckKeyOf(t *testing.T, bucket, key string) string {
|
||||
return k
|
||||
}
|
||||
|
||||
func TestPresenceFromEvents(t *testing.T) { // 1135/AC1
|
||||
func TestPresenceFromEventsWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1
|
||||
t.Run("from events", presenceFromEvents)
|
||||
t.Run("written only when it differs", presenceWrittenOnlyWhenItDiffers)
|
||||
}
|
||||
|
||||
func presenceFromEvents(t *testing.T) { // 1135/AC1
|
||||
w, o, _ := setup()
|
||||
w.Connect("node.ace")
|
||||
w.Connect("ace.mesh-delivery-view")
|
||||
@@ -135,7 +161,7 @@ func TestPresenceFromEvents(t *testing.T) { // 1135/AC1
|
||||
}
|
||||
}
|
||||
|
||||
func TestPresenceWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1
|
||||
func presenceWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1
|
||||
w, o, _ := setup()
|
||||
w.Connect("node.ace")
|
||||
w.Connect("node.ace")
|
||||
@@ -178,7 +204,12 @@ func connzAnswer(users ...string) []byte {
|
||||
return b
|
||||
}
|
||||
|
||||
func TestRebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2
|
||||
func TestRebuildAsksConnectionsAndTakesWhoIsGone(t *testing.T) { // 1135/AC2
|
||||
t.Run("asks the connections with auth", rebuildAsksConnectionsWithAuth)
|
||||
t.Run("takes the presence of who is gone", rebuildTakesThePresenceOfWhoIsGone)
|
||||
}
|
||||
|
||||
func rebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2
|
||||
w, o, _ := setup()
|
||||
var body []byte
|
||||
err := w.Rebuild(func(b []byte) ([]byte, error) {
|
||||
@@ -201,7 +232,7 @@ func TestRebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2
|
||||
}
|
||||
}
|
||||
|
||||
func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
|
||||
func rebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
|
||||
w, o, c := setup()
|
||||
w.SeedPresence("ace", true, t0.Add(-time.Hour))
|
||||
w.SeedPresence("g14", true, t0.Add(-time.Hour))
|
||||
@@ -416,7 +447,12 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
func TestOnlyTheExpiryOfItsOwnPulseWritesStuck(t *testing.T) { // 1135/AC5
|
||||
t.Run("the expiry marker writes stuck", expiryMarkerWritesStuck)
|
||||
t.Run("its own delete of a pulse writes nothing", ownDeleteOfAPulseWritesNothing)
|
||||
}
|
||||
|
||||
func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
c.t = t0.Add(120 * time.Second)
|
||||
@@ -430,7 +466,7 @@ func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
|
||||
}
|
||||
}
|
||||
|
||||
func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
|
||||
func ownDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
|
||||
w, o, c := setup()
|
||||
w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
|
||||
pk := pulseKeyOf(t, checksKey, "askid1")
|
||||
|
||||
@@ -25,19 +25,28 @@ type state struct {
|
||||
}
|
||||
|
||||
type artifact struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
Language string `json:"language"`
|
||||
System string `json:"system"`
|
||||
From string `json:"from"`
|
||||
Binary string `json:"binary"`
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
Language string `json:"language"`
|
||||
System string `json:"system"`
|
||||
From string `json:"from"`
|
||||
Binary string `json:"binary"`
|
||||
Loads []string `json:"loads"`
|
||||
Context *struct {
|
||||
Repository string `json:"repository"`
|
||||
} `json:"context"`
|
||||
}
|
||||
|
||||
type claim struct {
|
||||
Name string `json:"name"`
|
||||
Scope string `json:"scope"`
|
||||
Serves []string `json:"serves"`
|
||||
}
|
||||
|
||||
type manifest struct {
|
||||
Module string `json:"module"`
|
||||
Claims []claim `json:"claims"`
|
||||
Tools []string `json:"tools"`
|
||||
Moved bool `json:"moved"`
|
||||
Version string `json:"version"`
|
||||
Slug string `json:"slug"`
|
||||
@@ -75,16 +84,24 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
if m.Module != "mesh-delivery-watchman" || m.Moved || m.Slug != "watchman" {
|
||||
t.Errorf("module %q moved %v (it has no template, so it is not moved) slug %q", m.Module, m.Moved, m.Slug)
|
||||
}
|
||||
want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}}
|
||||
want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}}
|
||||
if !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-delivery-assigner.wants", "node-engine.have"}) {
|
||||
if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-controller.lease",
|
||||
"mesh-delivery-assigner.wants", "node-engine.have"}) {
|
||||
t.Errorf("reads = %v", reads)
|
||||
}
|
||||
for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} {
|
||||
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
|
||||
if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) {
|
||||
t.Errorf("claims = %+v", m.Claims)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) {
|
||||
t.Errorf("tools = %v", m.Tools)
|
||||
}
|
||||
for _, k := range []string{"emits", "context", "items", "consumes", "provides"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the watchman has none", k)
|
||||
}
|
||||
@@ -97,25 +114,28 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
}
|
||||
}
|
||||
|
||||
func TestManifestIsWholeInItsRepository(t *testing.T) { // 1135/AC6
|
||||
m, _ := load(t)
|
||||
if len(m.Build.Artifacts) != 1 {
|
||||
t.Fatalf("artifacts = %+v, want the one bundle program", m.Build.Artifacts)
|
||||
// The manifest is whole in its repository: the program, and the tools bundle the runtime loads (1952/WM1).
|
||||
func manifestIsWholeInItsRepository(t *testing.T, m manifest) { // 1135/AC6
|
||||
if len(m.Build.Artifacts) != 2 {
|
||||
t.Fatalf("artifacts = %+v, want the program and the tools bundle", m.Build.Artifacts)
|
||||
}
|
||||
a := m.Build.Artifacts[0]
|
||||
if a.Name != "program" || a.Kind != "bundle" || a.Language != "go" || a.System != "arch" || a.Context != nil {
|
||||
t.Errorf("artifact = %+v", a)
|
||||
}
|
||||
if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
|
||||
t.Errorf("the artifact builds from %q, which is not a directory of this module", a.From)
|
||||
}
|
||||
if a.Binary != "mesh-delivery-watchman" || !strings.HasSuffix(a.From, a.Binary) {
|
||||
t.Errorf("binary %q from %q", a.Binary, a.From)
|
||||
for i, name := range []string{"program", "tools"} {
|
||||
a := m.Build.Artifacts[i]
|
||||
if a.Name != name || a.Kind != "bundle" || a.Language != "go" || a.System != "arch" || a.Context != nil {
|
||||
t.Errorf("artifact = %+v", a)
|
||||
}
|
||||
if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
|
||||
t.Errorf("the artifact builds from %q, which is not a directory of this module", a.From)
|
||||
}
|
||||
if !strings.HasSuffix(a.From, a.Binary) || (name == "tools") != reflect.DeepEqual(a.Loads, []string{a.Binary}) {
|
||||
t.Errorf("binary %q from %q loads %v", a.Binary, a.From, a.Loads)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManifestResources(t *testing.T) { // 1135/AC6
|
||||
m, _ := load(t)
|
||||
manifestIsWholeInItsRepository(t, m)
|
||||
byType := map[string][]map[string]any{}
|
||||
for _, r := range m.Resources {
|
||||
ty, _ := r["type"].(string)
|
||||
|
||||
@@ -2,17 +2,35 @@
|
||||
"module": "mesh-delivery-watchman",
|
||||
"version": "1",
|
||||
"slug": "watchman",
|
||||
"claims": [
|
||||
{
|
||||
"name": "mesh-watchdogs",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"signals",
|
||||
"stuck",
|
||||
"presence",
|
||||
"dead-letters",
|
||||
"resync"
|
||||
]
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-build-image.steps",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-controller.lease",
|
||||
"mesh-delivery-assigner.wants",
|
||||
"node-engine.have"
|
||||
],
|
||||
"tools": [
|
||||
"watchman_findings"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "presence", "history": 20 },
|
||||
{ "name": "pulse", "key-ttl": true },
|
||||
{ "name": "stuck", "history": 20 }
|
||||
{ "name": "stuck", "history": 20 },
|
||||
{ "name": "findings", "history": 5 }
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
@@ -34,6 +52,14 @@
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "socket-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-delivery-watchman-tools",
|
||||
"mode": "0755",
|
||||
"owner": "mesh-delivery-watchman",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "watchman",
|
||||
"type": "process",
|
||||
@@ -60,6 +86,17 @@
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-delivery-watchman",
|
||||
"binary": "mesh-delivery-watchman"
|
||||
},
|
||||
{
|
||||
"name": "tools",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-delivery-watchman-tools",
|
||||
"binary": "mesh-delivery-watchman-tools",
|
||||
"loads": [
|
||||
"mesh-delivery-watchman-tools"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user