diff --git a/cmd/mesh-controller/stuck_probe.go b/cmd/mesh-controller/stuck_probe.go index ec8eeb29..e97a02df 100644 --- a/cmd/mesh-controller/stuck_probe.go +++ b/cmd/mesh-controller/stuck_probe.go @@ -35,10 +35,17 @@ const ( presenceSilent ) -// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none) -// and, when presence. has an entry, its time as when the machine was last heard from. Without a presence entry, -// last heard stays the store's last_seen. -func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, at time.Time) { +// presenceEntry is presence. as the watchman wrote it: whether the machine is online and when the entry was made. +// The watchman writes it only when it changes, so for an online machine the time is when it connected. +type presenceEntry struct { + Online bool + At time.Time +} + +// 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_seen and the entry's time (the time it went offline). +func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) { switch { case stuck == nil: m.presence = presenceUnsaid @@ -47,8 +54,12 @@ func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, at time.Time) { default: m.presence = presenceSilent } - if !at.IsZero() { - m.lastHeard = at + switch { + case entry == nil: + case entry.Online: + m.lastHeard = now + case entry.At.After(m.lastHeard): + m.lastHeard = entry.At } } @@ -137,8 +148,8 @@ func readStuck(ctx context.Context, conn *nats.Conn) (map[string]sdkdelivery.Stu return out, nil } -// readPresenceTimes is, by node, the time of presence.'s entry; a node without an entry is left out. -func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]time.Time, error) { +// readPresence is, by node, presence.'s entry; a node without an entry is left out. +func readPresence(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]presenceEntry, error) { api, err := jetstream.New(conn) if err != nil { return nil, err @@ -150,7 +161,7 @@ func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (ma if err != nil { return nil, fmt.Errorf("the watchman's presence bucket: %w", err) } - out := map[string]time.Time{} + out := map[string]presenceEntry{} for _, node := range nodes { key, err := sdkdelivery.PresenceNodeKey(node) if err != nil { @@ -163,13 +174,18 @@ func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (ma if err != nil { return nil, err } - out[node] = e.Created() + p, err := sdkdelivery.DecodePresence(e.Value()) + if err != nil { + fmt.Printf("the presence key %s is not read: %v\n", key, err) + continue + } + out[node] = presenceEntry{Online: p.Online, At: e.Created()} } return out, nil } // withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time. -func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts) error { +func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts, now time.Time) error { stuck, err := readStuck(ctx, conn) if err != nil { return err @@ -178,7 +194,7 @@ func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts) for i, m := range machines { names[i] = m.name } - times, err := readPresenceTimes(ctx, conn, names) + entries, err := readPresence(ctx, conn, names) if err != nil { return err } @@ -189,7 +205,11 @@ func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts) said = &s } } - applyPresence(&machines[i], said, times[machines[i].name]) + var entry *presenceEntry + if e, ok := entries[machines[i].name]; ok { + entry = &e + } + applyPresence(&machines[i], said, entry, now) } return nil } diff --git a/cmd/mesh-controller/stuck_probe_integration_test.go b/cmd/mesh-controller/stuck_probe_integration_test.go index a3e9bcac..acf070c8 100644 --- a/cmd/mesh-controller/stuck_probe_integration_test.go +++ b/cmd/mesh-controller/stuck_probe_integration_test.go @@ -61,7 +61,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { / t.Fatalf("a stuck key written: %+v %v", got, err) } machines := []machineFacts{{name: "ace", lastHeard: time.Now()}} - if err := withPresence(ctx, js.Conn(), machines); err != nil { + if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil { t.Fatal(err) } if !machines[0].silent(time.Now()) { @@ -73,7 +73,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { / t.Fatalf("a stuck key written as moved: %+v %v", got, err) } machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour), every: time.Minute}} - if err := withPresence(ctx, js.Conn(), machines); err != nil { + if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil { t.Fatal(err) } if machines[0].silent(time.Now()) { diff --git a/cmd/mesh-controller/stuck_probe_test.go b/cmd/mesh-controller/stuck_probe_test.go index d3b258c2..aab335ea 100644 --- a/cmd/mesh-controller/stuck_probe_test.go +++ b/cmd/mesh-controller/stuck_probe_test.go @@ -70,7 +70,7 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 f := calm(now) f.machines = append(f.machines, aMachineHeardAt(now, 10*time.Second)) notMoved := sdkdelivery.Stuck{Bucket: sdkdelivery.PresenceBucket, Key: "laptop", Waited: "its return", Since: now.Add(-3 * time.Minute)} - applyPresence(&f.machines[1], ¬Moved, now.Add(-3*time.Minute)) + applyPresence(&f.machines[1], ¬Moved, nil, now) got := watchHeartbeats(f) if len(got) != 1 || got[0].Machine != "laptop" || got[0].Kind != "silent" { t.Fatalf("a presence key not moved: %+v", got) @@ -85,24 +85,38 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 f.machines[1].lastHeard = now.Add(-2 * time.Hour) moved := notMoved moved.Moved = true - applyPresence(&f.machines[1], &moved, now.Add(-time.Minute)) + applyPresence(&f.machines[1], &moved, nil, now) if got := watchHeartbeats(f); len(got) != 0 { t.Fatalf("a presence key that moved raised %+v", got) } } -// 1156/AC3: a machine's last heard is the presence key's entry time when there is one, and last_seen when there is none. -func TestLastHeardIsThePresenceKeysTimeWhenThereIsOne(t *testing.T) { // 1156/AC3 +// 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; 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 +// 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 now := time.Now() seen := now.Add(-time.Hour) - m := machineFacts{name: "laptop", lastHeard: seen} - applyPresence(&m, nil, time.Time{}) + m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute} + applyPresence(&m, nil, nil, now) if !m.lastHeard.Equal(seen) { - t.Fatalf("with no presence key last heard is %v, want last_seen %v", m.lastHeard, seen) + t.Fatalf("with no presence entry last heard is %v, want last_seen %v", m.lastHeard, seen) } - at := now.Add(-5 * time.Minute) - applyPresence(&m, nil, at) - if !m.lastHeard.Equal(at) { - t.Fatalf("with a presence key last heard is %v, want %v", m.lastHeard, at) + offlineAt := now.Add(-5 * time.Minute) + applyPresence(&m, nil, &presenceEntry{Online: false, At: offlineAt}, now) + if !m.lastHeard.Equal(offlineAt) { + t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard) + } + m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} + applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now) + if !m.lastHeard.Equal(now.Add(-time.Minute)) { + t.Fatalf("an older entry replaced a newer last_seen: %v", m.lastHeard) + } + for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} { + m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute} + applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now) + if m.silent(now) { + t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen) + } } } diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 2e5c2e4e..ce29a2f1 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -483,7 +483,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory // What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.'s time when // it was last heard (ADR 0433 ยง3). Unreadable is said, and the heartbeat rule judges as before. if w.js != nil { - if err := withPresence(ctx, w.js.Conn(), out); err != nil { + if err := withPresence(ctx, w.js.Conn(), out, now); err != nil { fmt.Printf("what the watchman says of the machines' presence cannot be read: %v\n", err) } else { for _, m := range out {