The controller's last-heard word for a machine comes from the watchman's presence in nodes, node, status and the watchdogs (issue 1183, AC1, AC2)
This commit is contained in:
@@ -0,0 +1,120 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/novox/mesh-controller/internal/broker"
|
||||||
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Last heard (novox/hq ADR 0433 §3, issue 1183): one answer to when the controller last heard a machine, read from the
|
||||||
|
// watchman's presence. The alive message that used to keep last_seen fresh is gone; a report, which is rare, is the
|
||||||
|
// answer only where presence has no key for the machine.
|
||||||
|
|
||||||
|
// presenceReader reads presence.<node>'s entry for each node; a node without an entry is left out.
|
||||||
|
type presenceReader func(ctx context.Context, nodes []string) (map[string]presenceEntry, error)
|
||||||
|
|
||||||
|
// lastHeard is when a machine was last heard: now when presence says it is online, the entry's time when it says offline
|
||||||
|
// (the time it went), and the machine's last report only when presence has no entry for it. Zero is never heard.
|
||||||
|
func lastHeard(report time.Time, entry *presenceEntry, now time.Time) time.Time {
|
||||||
|
switch {
|
||||||
|
case entry == nil:
|
||||||
|
return report
|
||||||
|
case entry.Online:
|
||||||
|
return now
|
||||||
|
default:
|
||||||
|
return entry.At
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// nodesHeard returns the nodes with their last-heard time (LastSeen) as lastHeard says it. Unreadable presence is
|
||||||
|
// returned as an error with the nodes unchanged, so a caller says it and shows the reports.
|
||||||
|
func nodesHeard(ctx context.Context, nodes []inventory.Node, read presenceReader, now time.Time) ([]inventory.Node, error) {
|
||||||
|
names := make([]string, len(nodes))
|
||||||
|
for i, n := range nodes {
|
||||||
|
names[i] = n.Name
|
||||||
|
}
|
||||||
|
entries, err := read(ctx, names)
|
||||||
|
if err != nil {
|
||||||
|
return nodes, fmt.Errorf("what the watchman says of the machines' presence cannot be read: %w", err)
|
||||||
|
}
|
||||||
|
out := make([]inventory.Node, len(nodes))
|
||||||
|
for i, n := range nodes {
|
||||||
|
var entry *presenceEntry
|
||||||
|
if e, ok := entries[n.Name]; ok {
|
||||||
|
entry = &e
|
||||||
|
}
|
||||||
|
n.LastSeen = lastHeard(n.LastSeen, entry, now)
|
||||||
|
out[i] = n
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// heardNodes is every node, read with its last-heard time from the watchman's presence over a connection of its own. A bus
|
||||||
|
// that cannot be reached is said on standard error and the reports stand, which reads a machine as quiet rather than well.
|
||||||
|
func (h *stores) heardNodes(ctx context.Context) ([]inventory.Node, error) {
|
||||||
|
nodes, err := h.inventory.Nodes(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return h.withPresence(ctx, nodes), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// heardNode is one node, read the same way.
|
||||||
|
func (h *stores) heardNode(ctx context.Context, name string) (inventory.Node, error) {
|
||||||
|
n, err := h.inventory.NodeByName(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
return h.withPresence(ctx, []inventory.Node{n})[0], nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *stores) withPresence(ctx context.Context, nodes []inventory.Node) []inventory.Node {
|
||||||
|
address, err := broker.BusAddress()
|
||||||
|
var js *broker.JetStream
|
||||||
|
if err == nil {
|
||||||
|
js, err = broker.Dial(address)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
fmt.Fprintf(os.Stderr, "what the watchman says of the machines' presence cannot be read: %v\n", err)
|
||||||
|
return nodes
|
||||||
|
}
|
||||||
|
defer js.Close()
|
||||||
|
read := func(ctx context.Context, names []string) (map[string]presenceEntry, error) {
|
||||||
|
return readPresence(ctx, js.Conn(), names)
|
||||||
|
}
|
||||||
|
heard, err := nodesHeard(ctx, nodes, read, time.Now())
|
||||||
|
if err != nil {
|
||||||
|
fmt.Fprintln(os.Stderr, err)
|
||||||
|
}
|
||||||
|
return heard
|
||||||
|
}
|
||||||
|
|
||||||
|
// heardFromAt says when a node was last heard from, in a form somebody can act on.
|
||||||
|
//
|
||||||
|
// "never" and "an hour ago" are different answers and are kept different. A node that has never spoken did not finish
|
||||||
|
// joining; a node last heard from an hour ago is running an hour-old picture of the mesh.
|
||||||
|
func heardFromAt(n inventory.Node, now time.Time) string {
|
||||||
|
if n.LastSeen.IsZero() {
|
||||||
|
return "never spoken"
|
||||||
|
}
|
||||||
|
if silent := now.Sub(n.LastSeen); silent > SilentFor {
|
||||||
|
return "out of touch " + roughly(silent)
|
||||||
|
}
|
||||||
|
return "here"
|
||||||
|
}
|
||||||
|
|
||||||
|
// notHeardLately is the nodes never heard from or not within the hour: different from failing, for a machine that says
|
||||||
|
// nothing may be new, switched off or unreachable, and none of those is a machine that tried and could not.
|
||||||
|
func notHeardLately(nodes []inventory.Node, now time.Time) []inventory.Node {
|
||||||
|
var out []inventory.Node
|
||||||
|
for _, n := range nodes {
|
||||||
|
if n.LastSeen.IsZero() || now.Sub(n.LastSeen) > time.Hour {
|
||||||
|
out = append(out, n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
@@ -0,0 +1,97 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
|
)
|
||||||
|
|
||||||
|
// aPresenceFake is the watchman's presence bucket as a reader sees it.
|
||||||
|
func aPresenceFake(entries map[string]presenceEntry, err error) presenceReader {
|
||||||
|
return func(_ context.Context, _ []string) (map[string]presenceEntry, error) { return entries, err }
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1183/AC1: one function says when a machine was last heard: online is now, offline is the entry's time, and only where
|
||||||
|
// presence has no key, the last report.
|
||||||
|
func TestLastHeardIsPresenceThenTheReport(t *testing.T) { // 1183/AC1
|
||||||
|
now := time.Now()
|
||||||
|
report := now.Add(-time.Hour)
|
||||||
|
if got := lastHeard(report, nil, now); !got.Equal(report) {
|
||||||
|
t.Fatalf("no presence key: last heard is %v, want the report %v", got, report)
|
||||||
|
}
|
||||||
|
if got := lastHeard(report, &presenceEntry{Online: true, At: now.Add(-9 * time.Hour)}, now); !got.Equal(now) {
|
||||||
|
t.Fatalf("online: last heard is %v, want now", got)
|
||||||
|
}
|
||||||
|
offlineAt := now.Add(-4 * time.Minute)
|
||||||
|
if got := lastHeard(now.Add(-time.Minute), &presenceEntry{At: offlineAt}, now); !got.Equal(offlineAt) {
|
||||||
|
t.Fatalf("offline: last heard is %v, want the entry's time %v, not a report's", got, offlineAt)
|
||||||
|
}
|
||||||
|
if got := lastHeard(time.Time{}, nil, now); !got.IsZero() {
|
||||||
|
t.Fatalf("never heard and no key: %v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The machines facts, the nodes list, node and status all take that one function.
|
||||||
|
func TestEveryReaderOfLastHeardTakesPresence(t *testing.T) { // 1183/AC1
|
||||||
|
now := time.Now()
|
||||||
|
m := machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute}
|
||||||
|
applyPresence(&m, nil, &presenceEntry{At: now.Add(-4 * time.Minute)}, now)
|
||||||
|
if !m.lastHeard.Equal(now.Add(-4 * time.Minute)) {
|
||||||
|
t.Fatalf("machine facts take a report over the offline entry: %v", m.lastHeard)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func twoNodes(now time.Time) []inventory.Node {
|
||||||
|
return []inventory.Node{
|
||||||
|
{Name: "anchor", LastSeen: now.Add(-time.Hour)},
|
||||||
|
{Name: "laptop", LastSeen: now.Add(-30 * time.Second)},
|
||||||
|
{Name: "fresh"},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1183/AC2: a machine online by presence whose last report is an hour old reads here, is not quiet and is not silent; one
|
||||||
|
// offline for 4 minutes reads out of touch.
|
||||||
|
func TestOnlineByPresenceReadsHereWhateverTheReport(t *testing.T) { // 1183/AC2
|
||||||
|
now := time.Now()
|
||||||
|
read := aPresenceFake(map[string]presenceEntry{
|
||||||
|
"anchor": {Online: true, At: now.Add(-6 * time.Hour)},
|
||||||
|
"laptop": {Online: false, At: now.Add(-4 * time.Minute)},
|
||||||
|
}, nil)
|
||||||
|
nodes, err := nodesHeard(context.Background(), twoNodes(now), read, now)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if got := heardFromAt(nodes[0], now); got != "here" {
|
||||||
|
t.Errorf("anchor online with an hour-old report reads %q", got)
|
||||||
|
}
|
||||||
|
if got := heardFromAt(nodes[1], now); got != "out of touch 4m" {
|
||||||
|
t.Errorf("laptop offline 4 minutes reads %q", got)
|
||||||
|
}
|
||||||
|
if got := heardFromAt(nodes[2], now); got != "never spoken" {
|
||||||
|
t.Errorf("a machine with no key and no report reads %q", got)
|
||||||
|
}
|
||||||
|
quiet := notHeardLately(nodes, now)
|
||||||
|
if len(quiet) != 1 || quiet[0].Name != "fresh" {
|
||||||
|
t.Errorf("quiet is %+v: an online machine is not quiet, and a machine offline 4 minutes is not quiet yet", quiet)
|
||||||
|
}
|
||||||
|
f := &machineFacts{name: "anchor", lastHeard: nodes[0].LastSeen, every: time.Minute}
|
||||||
|
if f.silent(now) {
|
||||||
|
t.Errorf("a machine online by presence is silent")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Unreadable presence is said, and the report stands as it was.
|
||||||
|
func TestUnreadablePresenceLeavesTheReport(t *testing.T) { // 1183/AC1
|
||||||
|
now := time.Now()
|
||||||
|
nodes, err := nodesHeard(context.Background(), twoNodes(now), aPresenceFake(nil, errors.New("no bus")), now)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "no bus") {
|
||||||
|
t.Fatalf("an unreadable presence is not said: %v", err)
|
||||||
|
}
|
||||||
|
if !nodes[0].LastSeen.Equal(now.Add(-time.Hour)) {
|
||||||
|
t.Fatalf("the report was replaced: %v", nodes[0].LastSeen)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -50,7 +50,7 @@ func nodeCommand(ctx context.Context, args []string) error {
|
|||||||
return addNode(ctx, inv, args[1:])
|
return addNode(ctx, inv, args[1:])
|
||||||
|
|
||||||
case "list":
|
case "list":
|
||||||
nodes, err := inv.Nodes(ctx)
|
nodes, err := open.heardNodes(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -755,22 +755,8 @@ func consumerReset(ctx context.Context, args []string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// heardFrom says when a node was last heard from, in a form somebody can act on.
|
// heardFrom is heardFromAt for now.
|
||||||
//
|
func heardFrom(n inventory.Node) string { return heardFromAt(n, time.Now()) }
|
||||||
// "never" and "an hour ago" are different answers and are kept different. A node that has never
|
|
||||||
// spoken did not finish joining; a node last heard from an hour ago is running an hour-old
|
|
||||||
// picture of the mesh.
|
|
||||||
func heardFrom(n inventory.Node) string {
|
|
||||||
silent, ever := n.Silent()
|
|
||||||
switch {
|
|
||||||
case !ever:
|
|
||||||
return "never spoken"
|
|
||||||
case silent > SilentFor:
|
|
||||||
return "out of touch " + roughly(silent)
|
|
||||||
default:
|
|
||||||
return "here"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// roughly is a duration a person reads rather than parses.
|
// roughly is a duration a person reads rather than parses.
|
||||||
func roughly(d time.Duration) string {
|
func roughly(d time.Duration) string {
|
||||||
@@ -795,7 +781,7 @@ func roughly(d time.Duration) string {
|
|||||||
// assignment carries `card1-DP-1`, and a person composing settings for that machine needs it.
|
// assignment carries `card1-DP-1`, and a person composing settings for that machine needs it.
|
||||||
func showNode(ctx context.Context, stored *stores, name string) error {
|
func showNode(ctx context.Context, stored *stores, name string) error {
|
||||||
inv := stored.inventory
|
inv := stored.inventory
|
||||||
node, err := inv.NodeByName(ctx, name)
|
node, err := stored.heardNode(ctx, name)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -414,18 +414,11 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return answers{}, err
|
return answers{}, err
|
||||||
}
|
}
|
||||||
out.nodes, err = inv.Nodes(ctx)
|
out.nodes, err = open.heardNodes(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return answers{}, err
|
return answers{}, err
|
||||||
}
|
}
|
||||||
for _, n := range out.nodes {
|
out.quiet = notHeardLately(out.nodes, time.Now())
|
||||||
// Never heard from, or not lately. Different from failing: a machine that says nothing
|
|
||||||
// may be new, switched off, or unreachable, and none of those is a machine that tried
|
|
||||||
// and could not.
|
|
||||||
if n.LastSeen.IsZero() || time.Since(n.LastSeen) > time.Hour {
|
|
||||||
out.quiet = append(out.quiet, n)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
out.behind, err = inv.Behind(ctx)
|
out.behind, err = inv.Behind(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return answers{}, err
|
return answers{}, err
|
||||||
|
|||||||
@@ -43,8 +43,7 @@ type presenceEntry struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none).
|
// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none).
|
||||||
// A presence entry that says online counts the machine as heard now; otherwise last heard is the later of the store's
|
// Last heard is lastHeard's: now when online, the entry's time when offline, the store's last report only without an entry.
|
||||||
// last_seen and the entry's time (the time it went offline).
|
|
||||||
func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) {
|
func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) {
|
||||||
switch {
|
switch {
|
||||||
case stuck == nil:
|
case stuck == nil:
|
||||||
@@ -54,13 +53,7 @@ func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEnt
|
|||||||
default:
|
default:
|
||||||
m.presence = presenceSilent
|
m.presence = presenceSilent
|
||||||
}
|
}
|
||||||
switch {
|
m.lastHeard = lastHeard(m.lastHeard, entry, now)
|
||||||
case entry == nil:
|
|
||||||
case entry.Online:
|
|
||||||
m.lastHeard = now
|
|
||||||
case entry.At.After(m.lastHeard):
|
|
||||||
m.lastHeard = entry.At
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never
|
// silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never
|
||||||
|
|||||||
@@ -91,7 +91,7 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; with none,
|
// 1156/AC3: with a presence entry that says offline, last heard is the entry's time (1183/AC1); with none,
|
||||||
// last_seen. An entry that says online counts as heard now: the watchman writes it only when it changes, so its time is
|
// last_seen. An entry that says online counts as heard now: the watchman writes it only when it changes, so its time is
|
||||||
// when the machine connected, and a machine online for hours with no stuck key is not silent, whatever last_seen says.
|
// when the machine connected, and a machine online for hours with no stuck key is not silent, whatever last_seen says.
|
||||||
func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3
|
func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3
|
||||||
@@ -109,8 +109,8 @@ func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3
|
|||||||
}
|
}
|
||||||
m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute}
|
m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute}
|
||||||
applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now)
|
applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now)
|
||||||
if !m.lastHeard.Equal(now.Add(-time.Minute)) {
|
if !m.lastHeard.Equal(now.Add(-time.Hour)) {
|
||||||
t.Fatalf("an older entry replaced a newer last_seen: %v", m.lastHeard)
|
t.Fatalf("an offline entry's time is last heard, not a report: %v", m.lastHeard)
|
||||||
}
|
}
|
||||||
for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} {
|
for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} {
|
||||||
m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute}
|
m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute}
|
||||||
|
|||||||
Reference in New Issue
Block a user