diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index cb32676..c7ac65a 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -45,7 +45,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8 ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() kvs := map[string]jetstream.KeyValue{} - for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} { + for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} { kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5}) if err != nil { t.Fatal(err) @@ -122,3 +122,89 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8 put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK)) rowsMatch(func(s view.Snapshot) bool { r := byKey(s, "a", "x"); return r.Phase == "running" && r.Health == "ok" }) } + +// TestProcessWholeShowsWhatTheWatchmanSays starts the program whole and writes an offline presence and a stuck key to the +// watchman's buckets, sees both shown, then writes them online and moved and sees both gone. +func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5 + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + kvs := map[string]jetstream.KeyValue{} + for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5}) + if err != nil { + t.Fatal(err) + } + kvs[b] = kv + } + presence := func(key string, online bool) { + t.Helper() + b, _ := delivery.EncodePresence(delivery.Presence{Online: online}) + if _, err := kvs[delivery.PresenceBucket].Put(ctx, key, b); err != nil { + t.Fatal(err) + } + } + stuck := func(moved bool) { + t.Helper() + b, _ := delivery.EncodeStuck(delivery.Stuck{Bucket: delivery.HaveBucket, Key: "a.x", Waited: "a report from a", Since: time.Now(), Moved: moved}) + if _, err := kvs[delivery.StuckBucket].Put(ctx, delivery.HaveBucket+".a.x", b); err != nil { + t.Fatal(err) + } + } + presence("a", false) + presence("part.mesh-delivery-view.i1", false) + stuck(false) + + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + broker := filepath.Join(dir, "broker") + cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + socket := filepath.Join(dir, "run", "cli.sock") + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) + + wait := func(check func(view.Snapshot) bool) { + t.Helper() + var last view.Snapshot + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + if s, err := view.Ask(socket); err == nil { + last = s + if check(s) { + return + } + } + } + t.Fatalf("the socket never showed the expected lists; last: %+v", last) + } + wait(func(s view.Snapshot) bool { return len(s.Offline) == 2 && len(s.Stuck) == 1 && s.Stuck[0].Key == "a.x" }) + presence("a", true) + presence("part.mesh-delivery-view.i1", true) + stuck(true) + wait(func(s view.Snapshot) bool { return len(s.Offline) == 0 && len(s.Stuck) == 0 }) +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 040a7ca..ae339e4 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -136,7 +136,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} { + for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} { go follow(ctx, js, bucket, model, logger) } select { diff --git a/modules/mesh-delivery-view/internal/view/serve.go b/modules/mesh-delivery-view/internal/view/serve.go index 97d584b..5bf3247 100644 --- a/modules/mesh-delivery-view/internal/view/serve.go +++ b/modules/mesh-delivery-view/internal/view/serve.go @@ -143,13 +143,44 @@ func Table(w io.Writer, s Snapshot) { row([]string{n.Node, fmt.Sprint(n.Windows), fmt.Sprint(n.Busy), fmt.Sprint(n.Strays), since(n.Since)}) } } + if len(s.Offline) > 0 { + fmt.Fprintln(tw) + row([]string{"OFFLINE", "NAME", "STATE", "SINCE"}) + for _, o := range s.Offline { + row([]string{o.Kind, o.name(), "offline", since(o.Since)}) + } + } + if len(s.Stuck) > 0 { + fmt.Fprintln(tw) + row([]string{"STUCK", "KEY", "WAITED FOR", "SINCE"}) + for _, k := range s.Stuck { + row([]string{k.Bucket, k.Key, k.Waited, since(k.Since)}) + } + } + if len(s.Unreadable) > 0 { + fmt.Fprintln(tw) + row([]string{"UNREADABLE", "KEY", "WHY"}) + for _, u := range s.Unreadable { + row([]string{u.Bucket, u.Key, u.Error}) + } + } tw.Flush() } +func (o OfflineRow) name() string { + if o.Kind == "part" { + return o.Part + "." + o.Instance + } + return o.Node +} + var page = template.Must(template.New("p").Parse(` Delivery

What each node runs

{{range .Head}}{{end}} -{{range .Rows}}{{range .}}{{end}}{{end}}
{{.}}
{{.}}
`)) +{{range .Rows}}{{range .}}{{.}}{{end}}{{end}} +{{if .Offline}}

OFFLINE

{{range .Offline}}{{range .}}{{end}}{{end}}
{{.}}
{{end}} +{{if .Stuck}}

STUCK

{{range .Stuck}}{{range .}}{{end}}{{end}}
BUCKETKEYWAITED FORSINCE
{{.}}
{{end}} +{{if .Unreadable}}

UNREADABLE

{{range .Unreadable}}{{range .}}{{end}}{{end}}
{{.}}
{{end}}`)) // Web is the read-only web view of the same rows, refreshed by the browser. func Web(m *Model) http.Handler { @@ -168,6 +199,17 @@ func Web(m *Model) http.Handler { rows = append(rows, rowText(row)) } w.Header().Set("Content-Type", "text/html; charset=utf-8") - page.Execute(w, map[string]any{"Head": head, "Rows": rows}) + snap := m.Snapshot() + var off, stuck, bad [][]string + for _, o := range snap.Offline { + off = append(off, []string{o.Kind, o.name(), "offline", since(o.Since)}) + } + for _, k := range snap.Stuck { + stuck = append(stuck, []string{k.Bucket, k.Key, k.Waited, since(k.Since)}) + } + for _, u := range snap.Unreadable { + bad = append(bad, []string{u.Bucket, u.Key, u.Error}) + } + page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Offline": off, "Stuck": stuck, "Unreadable": bad}) }) } diff --git a/modules/mesh-delivery-view/internal/view/serve_test.go b/modules/mesh-delivery-view/internal/view/serve_test.go index 5baa74a..704a5c3 100644 --- a/modules/mesh-delivery-view/internal/view/serve_test.go +++ b/modules/mesh-delivery-view/internal/view/serve_test.go @@ -110,3 +110,44 @@ func TestWebIsTheSameRowsAndReadOnly(t *testing.T) { // 1134/AC6 t.Fatalf("POST = %d, want 405", rec.Code) } } + +func watchmanSnapshot(t *testing.T) *Model { + t.Helper() + m := New() + fill(m, presenceE(t, "a", false, t0), stuckE(t, "node-engine_have", "a.x", "a report from a", t0, false), + Entry{Bucket: delivery.PresenceBucket, Key: "c", Value: []byte(`{"v":9}`), At: t0}) + return m +} + +func TestTableAndWebShowTheWatchmansLists(t *testing.T) { // 1168/AC4 + m := watchmanSnapshot(t) + var out strings.Builder + Table(&out, m.Snapshot()) + rec := httptest.NewRecorder() + Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil)) + for name, text := range map[string]string{"table": out.String(), "web": rec.Body.String()} { + for _, w := range []string{"OFFLINE", "STUCK", "UNREADABLE", "node-engine_have", "a.x", "a report from a", "offline", "c"} { + if !strings.Contains(text, w) { + t.Errorf("%s lacks %q:\n%s", name, w, text) + } + } + } +} + +func TestSocketCarriesTheWatchmansLists(t *testing.T) { // 1168/AC4 + m := watchmanSnapshot(t) + dir, _ := os.MkdirTemp(os.Getenv("TMPDIR"), "w") + t.Cleanup(func() { os.RemoveAll(dir) }) + path := filepath.Join(dir, "cli.sock") + l, err := ListenSocket(path) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go ServeSocket(ctx, l, m) + got, err := Ask(path) + if err != nil || len(got.Offline) != 1 || len(got.Stuck) != 1 || len(got.Unreadable) != 1 { + t.Fatalf("socket = %+v, %v", got, err) + } +} diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 945fd36..b838e32 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -51,6 +51,36 @@ type NodeRow struct { type Snapshot struct { Rows []Row `json:"rows"` Nodes []NodeRow `json:"nodes"` + // Offline is every node and part instance whose presence is offline (the watchman's word). + Offline []OfflineRow `json:"offline"` + // Stuck is every key that stopped moving and has not moved again. + Stuck []StuckRow `json:"stuck"` + // Unreadable is every watchman value the view could not read, named by its key: never dropped silently. + Unreadable []UnreadableRow `json:"unreadable"` +} + +// OfflineRow is a node ("node") or a part's instance ("part") that is offline, since the presence entry's time. +type OfflineRow struct { + Kind string `json:"kind"` + Node string `json:"node,omitempty"` + Part string `json:"part,omitempty"` + Instance string `json:"instance,omitempty"` + Since time.Time `json:"since,omitzero"` +} + +// StuckRow is a key that stopped moving: its bucket and key, what it waited for and since when. +type StuckRow struct { + Bucket string `json:"bucket"` + Key string `json:"key"` + Waited string `json:"waited,omitempty"` + Since time.Time `json:"since,omitzero"` +} + +// UnreadableRow is a value of the watchman's buckets that did not decode, with the reader's error. +type UnreadableRow struct { + Bucket string `json:"bucket"` + Key string `json:"key"` + Error string `json:"error"` } // Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values. @@ -61,6 +91,20 @@ type Model struct { newest map[string]string // module -> id haves map[string]haveEntry nodes map[string]nodeEntry + // the watchman's buckets, by key + presence map[string]presenceEntry + stuck map[string]stuckEntry +} + +type presenceEntry struct { + value delivery.Presence + err string + at time.Time +} + +type stuckEntry struct { + value delivery.Stuck + err string } // assignment holds only the fields of the controller's Assignment the view reads. @@ -107,7 +151,8 @@ type nodeEntry struct { // New is an empty model. func New() *Model { return &Model{assigns: map[string]assignment{}, builds: map[string]build{}, newest: map[string]string{}, - haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}} + haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, + presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}} } // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. @@ -136,6 +181,20 @@ func (m *Model) Apply(e Entry) { } h, err := delivery.DecodeHave(e.Value) m.haves[key] = haveEntry{have: h, err: errText(err), at: e.At} + case delivery.PresenceBucket: + if e.Deleted { + delete(m.presence, e.Key) + return + } + p, err := delivery.DecodePresence(e.Value) + m.presence[e.Key] = presenceEntry{value: p, err: errText(err), at: e.At} + case delivery.StuckBucket: + if e.Deleted { + delete(m.stuck, e.Key) + return + } + st, err := delivery.DecodeStuck(e.Value) + m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)} case AssignmentsBucket: if e.Deleted { delete(m.assigns, e.Key) @@ -238,7 +297,47 @@ func (m *Model) Snapshot() Snapshot { Strays: count(n.node.Strays), Since: n.at, Error: n.err}) } sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node }) - return Snapshot{Rows: rows, Nodes: nodes} + offline, stuck, unreadable := []OfflineRow{}, []StuckRow{}, []UnreadableRow{} + for key, p := range m.presence { + if p.err != "" { + unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: p.err}) + continue + } + if p.value.Online { + continue + } + node, part, instance, err := delivery.ParsePresenceKey(key) + if err != nil { + unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: err.Error()}) + continue + } + if part != "" { + offline = append(offline, OfflineRow{Kind: "part", Part: part, Instance: instance, Since: p.at}) + } else { + offline = append(offline, OfflineRow{Kind: "node", Node: node, Since: p.at}) + } + } + for key, st := range m.stuck { + if st.err != "" { + unreadable = append(unreadable, UnreadableRow{Bucket: delivery.StuckBucket, Key: key, Error: st.err}) + continue + } + if st.value.Moved { + continue + } + stuck = append(stuck, StuckRow{Bucket: st.value.Bucket, Key: st.value.Key, Waited: st.value.Waited, Since: st.value.Since}) + } + sort.Slice(offline, func(i, j int) bool { + a, b := offline[i], offline[j] + return a.Kind+a.Node+a.Part+a.Instance < b.Kind+b.Node+b.Part+b.Instance + }) + sort.Slice(stuck, func(i, j int) bool { + return stuck[i].Bucket+"."+stuck[i].Key < stuck[j].Bucket+"."+stuck[j].Key + }) + sort.Slice(unreadable, func(i, j int) bool { + return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key + }) + return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable} } // count is the number of items of a carried JSON array or object, and 0 for anything else. diff --git a/modules/mesh-delivery-view/internal/view/view_test.go b/modules/mesh-delivery-view/internal/view/view_test.go index 29f3a63..d6ce64e 100644 --- a/modules/mesh-delivery-view/internal/view/view_test.go +++ b/modules/mesh-delivery-view/internal/view/view_test.go @@ -220,3 +220,77 @@ func TestNodeRow(t *testing.T) { // 1134/AC1 (the node's own key gives a node ro t.Fatalf("nodes = %+v", n) } } + +func presenceE(t *testing.T, key string, online bool, at time.Time) Entry { + t.Helper() + b, err := delivery.EncodePresence(delivery.Presence{Online: online}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.PresenceBucket, Key: key, Value: b, At: at} +} + +func stuckE(t *testing.T, bucket, key, waited string, since time.Time, moved bool) Entry { + t.Helper() + b, err := delivery.EncodeStuck(delivery.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since, Moved: moved}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.StuckBucket, Key: bucket + "." + key, Value: b, At: since} +} + +func TestOfflineAndStuckAreListed(t *testing.T) { // 1168/AC2 + m := New() + fill(m, base(t)...) + fill(m, + presenceE(t, "a", false, t0), + presenceE(t, "b", true, t0), + presenceE(t, "part.mesh-delivery-view.i1", false, t0.Add(time.Minute)), + stuckE(t, "node-engine_have", "a.x", "a report from a", t0.Add(2*time.Minute), false), + stuckE(t, "node-engine_have", "b.x", "a report from b", t0, true), + ) + s := m.Snapshot() + want := []OfflineRow{{Kind: "node", Node: "a", Since: t0}, {Kind: "part", Part: "mesh-delivery-view", Instance: "i1", Since: t0.Add(time.Minute)}} + if !reflect.DeepEqual(s.Offline, want) { + t.Fatalf("offline = %+v, want %+v", s.Offline, want) + } + wantStuck := []StuckRow{{Bucket: "node-engine_have", Key: "a.x", Waited: "a report from a", Since: t0.Add(2 * time.Minute)}} + if !reflect.DeepEqual(s.Stuck, wantStuck) { + t.Fatalf("stuck = %+v, want %+v", s.Stuck, wantStuck) + } + // back online and moved leave the lists + fill(m, presenceE(t, "a", true, t0.Add(5*time.Minute)), presenceE(t, "part.mesh-delivery-view.i1", true, t0.Add(5*time.Minute)), + stuckE(t, "node-engine_have", "a.x", "a report from a", t0.Add(2*time.Minute), true)) + s = m.Snapshot() + if len(s.Offline) != 0 || len(s.Stuck) != 0 { + t.Fatalf("offline %+v stuck %+v, want both empty", s.Offline, s.Stuck) + } +} + +func TestUnknownWatchmanVersionIsUnreadableNotDropped(t *testing.T) { // 1168/AC3 + m := New() + fill(m, + Entry{Bucket: delivery.PresenceBucket, Key: "c", Value: []byte(`{"v":9,"online":false}`), At: t0}, + Entry{Bucket: delivery.StuckBucket, Key: "node-engine_have.c.x", Value: []byte(`{"v":9}`), At: t0}, + presenceE(t, "a", false, t0), + ) + s := m.Snapshot() + if len(s.Unreadable) != 2 || len(s.Offline) != 1 { + t.Fatalf("unreadable %+v offline %+v", s.Unreadable, s.Offline) + } + _, sdkErr := delivery.DecodePresence([]byte(`{"v":9,"online":false}`)) + var got *UnreadableRow + for i := range s.Unreadable { + if s.Unreadable[i].Key == "c" { + got = &s.Unreadable[i] + } + } + if got == nil || got.Bucket != delivery.PresenceBucket || got.Error != sdkErr.Error() { + t.Fatalf("unreadable = %+v, want key c with the SDK's error %q", s.Unreadable, sdkErr) + } + // a good value later replaces the unreadable one + fill(m, presenceE(t, "c", true, t0)) + if len(m.Snapshot().Unreadable) != 1 { + t.Fatal("a readable value must clear the unreadable mark") + } +} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 141452f..c5d8c05 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -74,7 +74,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-controller.assignments", "mesh-controller.builds", "node-engine.have"} + want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index e79d77a..eb903fc 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -6,6 +6,8 @@ "reads": [ "mesh-controller.assignments", "mesh-controller.builds", + "mesh-delivery-watchman.presence", + "mesh-delivery-watchman.stuck", "node-engine.have" ], "own-secrets": {