The controller says when it last heard a machine from presence, and no longer hears the alive message (design 47 R2c, issue 1183) #389
@@ -72,8 +72,6 @@ func summarise(ds []inventory.Duration) []durationSummary {
|
|||||||
switch k.kind {
|
switch k.kind {
|
||||||
case inventory.DurationApply:
|
case inventory.DurationApply:
|
||||||
s.Suggests = "S2 bound max(2m, 3×last apply) ≈ " + round(max(2*time.Minute, 3*at(0.9))) + " at the p90"
|
s.Suggests = "S2 bound max(2m, 3×last apply) ≈ " + round(max(2*time.Minute, 3*at(0.9))) + " at the p90"
|
||||||
case inventory.DurationHeartbeatGap:
|
|
||||||
s.Suggests = "S1 bound 3×interval ≈ " + round(3*at(0.5))
|
|
||||||
}
|
}
|
||||||
out = append(out, s)
|
out = append(out, s)
|
||||||
}
|
}
|
||||||
@@ -138,7 +136,7 @@ func durationsCommand(ctx context.Context, args []string) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if len(summary) == 0 {
|
if len(summary) == 0 {
|
||||||
fmt.Printf("nothing measured in the last %d day(s): the serving controller records apply, heartbeat-gap, "+
|
fmt.Printf("nothing measured in the last %d day(s): the serving controller records apply, "+
|
||||||
"plan-tier and build durations as it hears them\n", *days)
|
"plan-tier and build durations as it hears them\n", *days)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -77,15 +77,11 @@ func TestDurationsAreSummarisedPerSubject(t *testing.T) {
|
|||||||
ds = append(ds, inventory.Duration{Kind: inventory.DurationApply, Subject: "anchor",
|
ds = append(ds, inventory.Duration{Kind: inventory.DurationApply, Subject: "anchor",
|
||||||
Took: time.Duration(i) * time.Second})
|
Took: time.Duration(i) * time.Second})
|
||||||
}
|
}
|
||||||
ds = append(ds, inventory.Duration{Kind: inventory.DurationHeartbeatGap, Subject: "anchor", Took: time.Minute})
|
|
||||||
got := summarise(ds)
|
got := summarise(ds)
|
||||||
if len(got) != 2 || got[0].Kind != inventory.DurationApply || got[0].Count != 10 ||
|
if len(got) != 1 || got[0].Kind != inventory.DurationApply || got[0].Count != 10 ||
|
||||||
got[0].Max != "10s" || got[0].Median != "5s" || got[0].P90 != "9s" {
|
got[0].Max != "10s" || got[0].Median != "5s" || got[0].P90 != "9s" {
|
||||||
t.Fatalf("%+v", got)
|
t.Fatalf("%+v", got)
|
||||||
}
|
}
|
||||||
if !strings.Contains(got[1].Suggests, "3m0s") {
|
|
||||||
t.Fatalf("a minute between words suggests %q", got[1].Suggests)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// **A drill has its own verb** — `hand-act drill`, the seat's `drill` — and `hand-act record` refuses
|
// **A drill has its own verb** — `hand-act drill`, the seat's `drill` — and `hand-act record` refuses
|
||||||
|
|||||||
@@ -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,82 @@
|
|||||||
|
//go:build integration
|
||||||
|
|
||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
|
|
||||||
|
"github.com/novox/mesh-controller/internal/broker"
|
||||||
|
"github.com/novox/mesh-controller/internal/inventory"
|
||||||
|
"github.com/novox/mesh-controller/internal/testbus"
|
||||||
|
)
|
||||||
|
|
||||||
|
// 1183/AC5: with the watchman's presence bucket holding a node online and no alive message ever sent, the node reads here
|
||||||
|
// in `nodes` four minutes later by the controller's clock, however old its last report; and an alive message still sent
|
||||||
|
// by a node-engine is published to the same bus without error and changes nothing.
|
||||||
|
func TestAnOnlineNodeReadsHereWithNoAliveMessage(t *testing.T) { // 1183/AC5
|
||||||
|
js, err := broker.Dial(testbus.URL(t))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Cleanup(js.Close)
|
||||||
|
if _, err := broker.RaiseBuckets(js, []broker.Bucket{
|
||||||
|
{Module: sdkdelivery.WatchmanModule, Name: "presence", History: sdkdelivery.PresenceHistory},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
api, err := jetstream.New(js.Conn())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
kv, err := api.KeyValue(ctx, sdkdelivery.PresenceBucket)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
put := func(node string, online bool) {
|
||||||
|
t.Helper()
|
||||||
|
key, err := sdkdelivery.PresenceNodeKey(node)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
body, err := sdkdelivery.EncodePresence(sdkdelivery.Presence{Online: online})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := kv.Put(ctx, key, body); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
put("ace", true)
|
||||||
|
put("laptop", false)
|
||||||
|
// A node-engine still sending its alive message: harmless.
|
||||||
|
if err := js.Conn().Publish("mesh.control.ace.alive", []byte(`{"node":"ace"}`)); err != nil {
|
||||||
|
t.Fatalf("an alive message could not be published: %v", err)
|
||||||
|
}
|
||||||
|
if err := js.Conn().Flush(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
later := time.Now().Add(4 * time.Minute)
|
||||||
|
nodes := []inventory.Node{{Name: "ace", LastSeen: time.Now().Add(-time.Hour)},
|
||||||
|
{Name: "laptop", LastSeen: time.Now().Add(-time.Hour)}}
|
||||||
|
read := func(ctx context.Context, names []string) (map[string]presenceEntry, error) {
|
||||||
|
return readPresence(ctx, js.Conn(), names)
|
||||||
|
}
|
||||||
|
heard, err := nodesHeard(ctx, nodes, read, later)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if got := heardFromAt(heard[0], later); got != "here" {
|
||||||
|
t.Fatalf("a node online by presence reads %q four minutes later", got)
|
||||||
|
}
|
||||||
|
if got := heardFromAt(heard[1], later); got == "here" {
|
||||||
|
t.Fatalf("a node offline by presence reads %q", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,112 @@
|
|||||||
|
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)}
|
||||||
|
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}
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1183/AC4: where the watchman gives no verdict a machine is silent after a fixed three minutes, whatever interval a
|
||||||
|
// node-engine once said.
|
||||||
|
func TestTheSilentBoundIsFixed(t *testing.T) { // 1183/AC4
|
||||||
|
now := time.Now()
|
||||||
|
for _, c := range []struct {
|
||||||
|
ago time.Duration
|
||||||
|
want bool
|
||||||
|
}{{179 * time.Second, false}, {181 * time.Second, true}} {
|
||||||
|
m := machineFacts{name: "laptop", lastHeard: now.Add(-c.ago)}
|
||||||
|
if got := m.silent(now); got != c.want {
|
||||||
|
t.Errorf("heard %v ago: silent %v, want %v", c.ago, got, c.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -318,6 +318,10 @@ func ago(d time.Duration) string {
|
|||||||
return d.Round(time.Minute).String()
|
return d.Round(time.Minute).String()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// silentBound is how long a machine may go unheard, where the watchman gives no verdict: three of the intervals nodes have
|
||||||
|
// always used, fixed now that no machine says its own (issue 1183).
|
||||||
|
const silentBound = heartbeatsMissed * heartbeatEvery
|
||||||
|
|
||||||
// heartbeatBound is a machine's S1 or S11 bound from the interval it says.
|
// heartbeatBound is a machine's S1 or S11 bound from the interval it says.
|
||||||
func heartbeatBound(every time.Duration) time.Duration {
|
func heartbeatBound(every time.Duration) time.Duration {
|
||||||
if every <= 0 {
|
if every <= 0 {
|
||||||
@@ -333,7 +337,7 @@ func watchHeartbeats(f *signalFacts) []conditions.Observation {
|
|||||||
// Never heard is a machine that has not joined, which status says; asleep is not lost.
|
// Never heard is a machine that has not joined, which status says; asleep is not lost.
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
bound := heartbeatBound(m.every)
|
bound := silentBound
|
||||||
silent := f.now.Sub(m.lastHeard)
|
silent := f.now.Sub(m.lastHeard)
|
||||||
if !m.silent(f.now) {
|
if !m.silent(f.now) {
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ import (
|
|||||||
func calm(now time.Time) *signalFacts {
|
func calm(now time.Time) *signalFacts {
|
||||||
return &signalFacts{now: now, started: now.Add(-time.Hour), host: "anchor", toolsHeardFrom: now.Add(-time.Hour),
|
return &signalFacts{now: now, started: now.Add(-time.Hour), host: "anchor", toolsHeardFrom: now.Add(-time.Hour),
|
||||||
machines: []machineFacts{{name: "anchor", control: true, lastHeard: now.Add(-10 * time.Second),
|
machines: []machineFacts{{name: "anchor", control: true, lastHeard: now.Add(-10 * time.Second),
|
||||||
every: time.Minute, sentAt: now.Add(-time.Hour), reportedCurrent: true, reportedAt: now.Add(-59 * time.Minute),
|
sentAt: now.Add(-time.Hour), reportedCurrent: true, reportedAt: now.Add(-59 * time.Minute),
|
||||||
lastApply: 20 * time.Second, tools: true, toolsHeard: now.Add(-10 * time.Second), toolsEvery: time.Minute}},
|
lastApply: 20 * time.Second, tools: true, toolsHeard: now.Add(-10 * time.Second), toolsEvery: time.Minute}},
|
||||||
plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2,
|
plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2,
|
||||||
entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}},
|
entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}},
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -72,7 +65,7 @@ func (m machineFacts) silent(now time.Time) bool {
|
|||||||
case presenceHeard:
|
case presenceHeard:
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every)
|
return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > silentBound
|
||||||
}
|
}
|
||||||
|
|
||||||
// stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind
|
// stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind
|
||||||
|
|||||||
@@ -72,7 +72,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { /
|
|||||||
if got, err := probeStuck(ctx, d); err != nil || len(got) != 0 {
|
if got, err := probeStuck(ctx, d); err != nil || len(got) != 0 {
|
||||||
t.Fatalf("a stuck key written as moved: %+v %v", got, err)
|
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}}
|
machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour)}}
|
||||||
if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil {
|
if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -60,7 +60,7 @@ func TestEachStuckKeyThatHasNotMovedIsAConditionOfTheDeliveryScope(t *testing.T)
|
|||||||
}
|
}
|
||||||
|
|
||||||
func aMachineHeardAt(now time.Time, ago time.Duration) machineFacts {
|
func aMachineHeardAt(now time.Time, ago time.Duration) machineFacts {
|
||||||
return machineFacts{name: "laptop", lastHeard: now.Add(-ago), every: time.Minute}
|
return machineFacts{name: "laptop", lastHeard: now.Add(-ago)}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1156/AC2: a node's stuck presence key not moved makes the machine silent, unless it sleeps; moved, it is heard; no
|
// 1156/AC2: a node's stuck presence key not moved makes the machine silent, unless it sleeps; moved, it is heard; no
|
||||||
@@ -91,13 +91,13 @@ 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
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
seen := now.Add(-time.Hour)
|
seen := now.Add(-time.Hour)
|
||||||
m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute}
|
m := machineFacts{name: "laptop", lastHeard: seen}
|
||||||
applyPresence(&m, nil, nil, now)
|
applyPresence(&m, nil, nil, now)
|
||||||
if !m.lastHeard.Equal(seen) {
|
if !m.lastHeard.Equal(seen) {
|
||||||
t.Fatalf("with no presence entry 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)
|
||||||
@@ -107,13 +107,13 @@ func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3
|
|||||||
if !m.lastHeard.Equal(offlineAt) {
|
if !m.lastHeard.Equal(offlineAt) {
|
||||||
t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard)
|
t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard)
|
||||||
}
|
}
|
||||||
m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute}
|
m = machineFacts{name: "laptop", lastHeard: now.Add(-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)}
|
||||||
applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now)
|
applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now)
|
||||||
if m.silent(now) {
|
if m.silent(now) {
|
||||||
t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen)
|
t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen)
|
||||||
|
|||||||
@@ -142,9 +142,7 @@ type machineFacts struct {
|
|||||||
control bool
|
control bool
|
||||||
// lastHeard is the store's last word from it, any word; zero when never.
|
// lastHeard is the store's last word from it, any word; zero when never.
|
||||||
lastHeard time.Time
|
lastHeard time.Time
|
||||||
// every is the heartbeat interval it said; zero when it said none.
|
power link.PowerState
|
||||||
every time.Duration
|
|
||||||
power link.PowerState
|
|
||||||
// sentAt is when it was last sent a declaration; reportedCurrent whether it has reported that one.
|
// sentAt is when it was last sent a declaration; reportedCurrent whether it has reported that one.
|
||||||
sentAt time.Time
|
sentAt time.Time
|
||||||
reportedCurrent bool
|
reportedCurrent bool
|
||||||
@@ -457,9 +455,6 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory
|
|||||||
for _, n := range nodes {
|
for _, n := range nodes {
|
||||||
m := machineFacts{name: n.Name, control: slices.Contains(control, n.Name), lastHeard: n.LastSeen,
|
m := machineFacts{name: n.Name, control: slices.Contains(control, n.Name), lastHeard: n.LastSeen,
|
||||||
lastApply: lastApply[n.Name], tools: slices.Contains(tooled, n.Name)}
|
lastApply: lastApply[n.Name], tools: slices.Contains(tooled, n.Name)}
|
||||||
if beat, ok := link.HostBeats.Of(n.Name); ok {
|
|
||||||
m.every = beat.Every
|
|
||||||
}
|
|
||||||
if beat, ok := link.ToolsBeats.Of(n.Name); ok {
|
if beat, ok := link.ToolsBeats.Of(n.Name); ok {
|
||||||
m.toolsHeard, m.toolsEvery = beat.At, beat.Every
|
m.toolsHeard, m.toolsEvery = beat.At, beat.Every
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -450,11 +450,11 @@ var ControllerVerbs = []Verb{
|
|||||||
"recorded — who, why and the cause of each, and which causes repeat: each repeat is a healer the mesh lacks.",
|
"recorded — who, why and the cause of each, and which causes repeat: each repeat is a healer the mesh lacks.",
|
||||||
Input: schema(map[string]string{"days": "how many days back (default 14)"}, nil)},
|
Input: schema(map[string]string{"days": "how many days back (default 14)"}, nil)},
|
||||||
{Name: "durations", Description: "How long things take, as the controller measured them: a send to its machine's " +
|
{Name: "durations", Description: "How long things take, as the controller measured them: a send to its machine's " +
|
||||||
"report (apply), a machine's silence between words (heartbeat-gap), a plan's tier, a build, and each phase of an " +
|
"report (apply), a plan's tier, a build, and each phase of an " +
|
||||||
"ended walk per class (walk-phase, novox/hq ADR 0282) — per machine, repository, module or class/phase, with " +
|
"ended walk per class (walk-phase, novox/hq ADR 0282) — per machine, repository, module or class/phase, with " +
|
||||||
"median, p90 and max. What the core's bounds are set from (novox/hq to-be 45 Phase 0).",
|
"median, p90 and max. What the core's bounds are set from (novox/hq to-be 45 Phase 0).",
|
||||||
Input: schema(map[string]string{
|
Input: schema(map[string]string{
|
||||||
"kind": "one kind: apply, heartbeat-gap, plan-tier, build or walk-phase; every kind when absent",
|
"kind": "one kind: apply, plan-tier, build or walk-phase; every kind when absent",
|
||||||
"days": "how many days back (default 14)",
|
"days": "how many days back (default 14)",
|
||||||
}, nil)},
|
}, nil)},
|
||||||
// What is wrong, and the self-check (novox/hq to-be 45 §2, §4).
|
// What is wrong, and the self-check (novox/hq to-be 45 §2, §4).
|
||||||
|
|||||||
@@ -9,10 +9,9 @@ import (
|
|||||||
|
|
||||||
// The kinds of duration recorded.
|
// The kinds of duration recorded.
|
||||||
const (
|
const (
|
||||||
DurationApply = "apply"
|
DurationApply = "apply"
|
||||||
DurationHeartbeatGap = "heartbeat-gap"
|
DurationPlanTier = "plan-tier"
|
||||||
DurationPlanTier = "plan-tier"
|
DurationBuild = "build"
|
||||||
DurationBuild = "build"
|
|
||||||
// DurationWalkPhase is one phase of an ended walk, per class (novox/hq ADR 0282 decision 6): subject
|
// DurationWalkPhase is one phase of an ended walk, per class (novox/hq ADR 0282 decision 6): subject
|
||||||
// "<class>/<phase>", and "<class>/total" for its merge to every machine running it.
|
// "<class>/<phase>", and "<class>/total" for its merge to every machine running it.
|
||||||
DurationWalkPhase = "walk-phase"
|
DurationWalkPhase = "walk-phase"
|
||||||
@@ -27,7 +26,7 @@ const (
|
|||||||
const ApplyBusy = "busy"
|
const ApplyBusy = "busy"
|
||||||
|
|
||||||
// DurationKinds are every kind, in the order `durations` shows them.
|
// DurationKinds are every kind, in the order `durations` shows them.
|
||||||
var DurationKinds = []string{DurationApply, DurationHeartbeatGap, DurationPlanTier, DurationBuild, DurationWalkPhase,
|
var DurationKinds = []string{DurationApply, DurationPlanTier, DurationBuild, DurationWalkPhase,
|
||||||
DurationBusy}
|
DurationBusy}
|
||||||
|
|
||||||
// DurationsKeptFor is how long a duration is kept: long enough to set a bound from, and to correct it
|
// DurationsKeptFor is how long a duration is kept: long enough to set a bound from, and to correct it
|
||||||
@@ -94,17 +93,6 @@ func (i *Inventory) RecordBusyDuration(ctx context.Context, node, declared strin
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// RecordHeartbeatGap measures the silence before a machine's word: from the last word heard before
|
|
||||||
// it, which the caller read before recording this one.
|
|
||||||
func (i *Inventory) RecordHeartbeatGap(ctx context.Context, node string, before time.Time) error {
|
|
||||||
if before.IsZero() {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
now := time.Now()
|
|
||||||
return i.RecordDuration(ctx, Duration{Kind: DurationHeartbeatGap, Subject: node, Node: node,
|
|
||||||
Ref: before.UTC().Format(time.RFC3339Nano), Started: before, Took: now.Sub(before)})
|
|
||||||
}
|
|
||||||
|
|
||||||
// Durations is every measurement of a kind since a moment, oldest first; every kind when kind is empty.
|
// Durations is every measurement of a kind since a moment, oldest first; every kind when kind is empty.
|
||||||
func (i *Inventory) Durations(ctx context.Context, kind string, since time.Time) ([]Duration, error) {
|
func (i *Inventory) Durations(ctx context.Context, kind string, since time.Time) ([]Duration, error) {
|
||||||
rows, err := i.store.Pool().Query(ctx,
|
rows, err := i.store.Pool().Query(ctx,
|
||||||
|
|||||||
@@ -44,25 +44,6 @@ func TestAnApplyIsMeasuredOncePerSend(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// A machine's silence is the time since its last word, once per word.
|
|
||||||
func TestASilenceIsMeasuredFromTheLastWord(t *testing.T) {
|
|
||||||
inv := ForTest(t)
|
|
||||||
ctx := t.Context()
|
|
||||||
before := time.Now().Add(-90 * time.Second)
|
|
||||||
for range 2 {
|
|
||||||
if err := inv.RecordHeartbeatGap(ctx, "anchor", before); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err := inv.RecordHeartbeatGap(ctx, "anchor", time.Time{}); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
ds, err := inv.Durations(ctx, DurationHeartbeatGap, time.Now().Add(-time.Hour))
|
|
||||||
if err != nil || len(ds) != 1 || ds[0].Took < 90*time.Second || ds[0].Took > 2*time.Minute {
|
|
||||||
t.Fatalf("%v %+v", err, ds)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// A plan's tier is measured when the plan leaves it — moving on, or ending — and only then.
|
// A plan's tier is measured when the plan leaves it — moving on, or ending — and only then.
|
||||||
func TestAPlansTierIsMeasuredWhenItIsLeft(t *testing.T) {
|
func TestAPlansTierIsMeasuredWhenItIsLeft(t *testing.T) {
|
||||||
inv := ForTest(t)
|
inv := ForTest(t)
|
||||||
|
|||||||
@@ -0,0 +1,12 @@
|
|||||||
|
package inventory
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
// 1183/AC3: the heartbeat-gap duration is no kind the mesh measures.
|
||||||
|
func TestNoHeartbeatGapIsMeasured(t *testing.T) { // 1183/AC3
|
||||||
|
for _, k := range DurationKinds {
|
||||||
|
if k == "heartbeat-gap" {
|
||||||
|
t.Fatalf("%s is still a kind of duration: %v", k, DurationKinds)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -67,17 +67,12 @@ const (
|
|||||||
// BuiltSubject is where a build's outcome lands, for results nobody was waiting for.
|
// BuiltSubject is where a build's outcome lands, for results nobody was waiting for.
|
||||||
BuiltSubject = "mesh.control.built"
|
BuiltSubject = "mesh.control.built"
|
||||||
|
|
||||||
// AliveSubjects is every node's heartbeat. Core NATS, never a stream: a lost heartbeat is the
|
|
||||||
// next heartbeat, and a stream of them is the mesh's least valuable message competing for
|
|
||||||
// retention with its most valuable (design 25 §3).
|
|
||||||
AliveSubjects = "mesh.control.*.alive"
|
|
||||||
|
|
||||||
// ToolsAliveSubjects is every machine's node tools saying they are there (novox/hq to-be 45 §3,
|
// ToolsAliveSubjects is every machine's node tools saying they are there (novox/hq to-be 45 §3,
|
||||||
// S11): core NATS like the host's, for the same reason.
|
// S11): core NATS like the host's, for the same reason.
|
||||||
ToolsAliveSubjects = "mesh.control.*.tools-alive"
|
ToolsAliveSubjects = "mesh.control.*.tools-alive"
|
||||||
|
|
||||||
// HealthSubjects is every machine's health statement between its reports (novox/hq ADR 0240): core
|
// HealthSubjects is every machine's health statement between its reports (novox/hq ADR 0240): core
|
||||||
// NATS like the heartbeat, because a statement lost is said again within a minute while anything is
|
// NATS like the node tools' heartbeat, because a statement lost is said again within a minute while anything is
|
||||||
// not healthy, and the next report carries it whatever happens.
|
// not healthy, and the next report carries it whatever happens.
|
||||||
HealthSubjects = "mesh.control.*.health"
|
HealthSubjects = "mesh.control.*.health"
|
||||||
)
|
)
|
||||||
@@ -89,9 +84,6 @@ func HealthSubject(node string) string { return "mesh.control." + node + ".healt
|
|||||||
// message the store-window guarantee is about (ADR 0083).
|
// message the store-window guarantee is about (ADR 0083).
|
||||||
func ReportSubject(node string) string { return "mesh.control." + node + ".report" }
|
func ReportSubject(node string) string { return "mesh.control." + node + ".report" }
|
||||||
|
|
||||||
// AliveSubject is one node's heartbeat.
|
|
||||||
func AliveSubject(node string) string { return "mesh.control." + node + ".alive" }
|
|
||||||
|
|
||||||
// ToolsAliveSubject is one machine's node tools' heartbeat.
|
// ToolsAliveSubject is one machine's node tools' heartbeat.
|
||||||
func ToolsAliveSubject(node string) string { return "mesh.control." + node + ".tools-alive" }
|
func ToolsAliveSubject(node string) string { return "mesh.control." + node + ".tools-alive" }
|
||||||
|
|
||||||
|
|||||||
@@ -32,8 +32,6 @@ var Contracts = map[string]Contract{
|
|||||||
KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
|
KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
|
||||||
"or older than the last look is history, not a second plan (issues 250, 266)",
|
"or older than the last look is history, not a second plan (issues 250, 266)",
|
||||||
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
|
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
|
||||||
KindHeartbeat: {Unordered: "a word that the machine is there: the newest heard is the newest said, and one " +
|
|
||||||
"lost is the next one"},
|
|
||||||
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"},
|
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"},
|
||||||
KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " +
|
KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " +
|
||||||
"than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
|
"than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ import (
|
|||||||
// exists (novox/hq to-be 45 Phase 2, ADR 0227 rule 2).
|
// exists (novox/hq to-be 45 Phase 2, ADR 0227 rule 2).
|
||||||
func TestEveryConsumedKindHasAContract(t *testing.T) {
|
func TestEveryConsumedKindHasAContract(t *testing.T) {
|
||||||
// What the controller can be handed, from the subjects it is granted and the ones it derives.
|
// What the controller can be handed, from the subjects it is granted and the ones it derives.
|
||||||
subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"), AliveSubject("anchor"),
|
subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"),
|
||||||
ToolsAliveSubject("anchor"), HealthSubject("anchor"), "mesh.mod.postgres.event.provisioner.failing"}
|
ToolsAliveSubject("anchor"), HealthSubject("anchor"), "mesh.mod.postgres.event.provisioner.failing"}
|
||||||
subjects = append(subjects, broker.ControllerFollows...)
|
subjects = append(subjects, broker.ControllerFollows...)
|
||||||
subjects = append(subjects, broker.ControllerHears...)
|
subjects = append(subjects, broker.ControllerHears...)
|
||||||
@@ -83,3 +83,17 @@ func TestEveryConsumedKindHasAContract(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 1183/AC3: the controller no longer takes the node-engine's alive message: its subject is no kind, so nothing is
|
||||||
|
// subscribed to it and nothing hears it. The tool runner's tools-alive stays.
|
||||||
|
func TestTheAliveMessageIsNoKindTheControllerHears(t *testing.T) { // 1183/AC3
|
||||||
|
if kind, known := kindOfSubject("mesh.control.anchor.alive"); known {
|
||||||
|
t.Fatalf("the controller still takes the alive message as %s", kind)
|
||||||
|
}
|
||||||
|
if kind, known := kindOfSubject(ToolsAliveSubject("anchor")); !known || kind != KindToolsHeartbeat {
|
||||||
|
t.Fatalf("the tool runner's tools-alive is %q, %v", kind, known)
|
||||||
|
}
|
||||||
|
if _, ok := Contracts["heartbeat"]; ok {
|
||||||
|
t.Fatal("a contract still names the alive message")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -384,9 +384,8 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
|||||||
}
|
}
|
||||||
|
|
||||||
// A bare word that a node is there is not an account of what the machine did or holds: it
|
// A bare word that a node is there is not an account of what the machine did or holds: it
|
||||||
// moves last_seen and touches nothing else. This arrives every minute (link.AliveEvery),
|
// moves last_seen and touches nothing else. Recording one as a report would overwrite the
|
||||||
// while a real report is rare, so recording it as one would overwrite the node's last real
|
// node's last real apply with an empty one — wiping the declaration digest that decides whether the node is
|
||||||
// apply with an empty one — wiping the declaration digest that decides whether the node is
|
|
||||||
// current, the carried ports a push assigns around, and the clean-or-failed outcome — and a
|
// current, the carried ports a push assigns around, and the clean-or-failed outcome — and a
|
||||||
// node that had just caught up would read as behind within the minute. The alive path calls
|
// node that had just caught up would read as behind within the minute. The alive path calls
|
||||||
// this with only a node name; a real report always carries an account (something applied, or
|
// this with only a node name; a real report always carries an account (something applied, or
|
||||||
@@ -394,10 +393,6 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
|||||||
if report.Applied == nil && report.Refused == "" && len(report.Failed) == 0 {
|
if report.Applied == nil && report.Refused == "" && len(report.Failed) == 0 {
|
||||||
if report.Superseded != "" {
|
if report.Superseded != "" {
|
||||||
log.Printf("%s set aside declaration %s for the newer %s", report.Node, report.Declared, report.Superseded)
|
log.Printf("%s set aside declaration %s for the newer %s", report.Node, report.Declared, report.Superseded)
|
||||||
} else if err := e.Inventory.RecordHeartbeatGap(ctx, report.Node, node.LastSeen); err != nil {
|
|
||||||
// The silence before this word, which the machine-silent bound will be set from (novox/hq
|
|
||||||
// to-be 45 Phase 0). A measurement lost is said and costs the report nothing.
|
|
||||||
log.Printf("the silence before %s's word could not be recorded: %v", report.Node, err)
|
|
||||||
}
|
}
|
||||||
return false, e.Inventory.Seen(ctx, node.ID)
|
return false, e.Inventory.Seen(ctx, node.ID)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -27,7 +27,6 @@ const ControlQueue = "control"
|
|||||||
const (
|
const (
|
||||||
KeyEnrol = "enrol"
|
KeyEnrol = "enrol"
|
||||||
KeyReport = "report"
|
KeyReport = "report"
|
||||||
KeyAlive = "alive"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// QueueFor is the queue a node consumes from — the only one it may read.
|
// QueueFor is the queue a node consumes from — the only one it may read.
|
||||||
@@ -134,19 +133,6 @@ type Signed struct {
|
|||||||
Signature []byte `json:"signature"`
|
Signature []byte `json:"signature"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Alive is a node saying nothing except that it is there.
|
|
||||||
//
|
|
||||||
// How long a node has been out of touch is a fact only the mesh can hold — nobody else is
|
|
||||||
// watching — and without it a node running last month's assignments looks exactly like one that
|
|
||||||
// is current.
|
|
||||||
type Alive struct {
|
|
||||||
Node string `json:"node"`
|
|
||||||
// IntervalSeconds is how often the node says it is there (novox/hq to-be 45 §3, S1): the
|
|
||||||
// watchdog's bound is three of them. Zero from a host older than that, which is read as the
|
|
||||||
// interval hosts have always used.
|
|
||||||
IntervalSeconds int `json:"interval_seconds,omitempty"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// ToolsAlive is a machine's node tools saying they are there (novox/hq to-be 45 §3, S11): the
|
// ToolsAlive is a machine's node tools saying they are there (novox/hq to-be 45 §3, S11): the
|
||||||
// runtime every module's tools and every held seat's verbs are served by. Its own word, apart from the
|
// runtime every module's tools and every held seat's verbs are served by. Its own word, apart from the
|
||||||
// host's, because a host heard and a runtime gone is a machine nobody can ask anything.
|
// host's, because a host heard and a runtime gone is a machine nobody can ask anything.
|
||||||
|
|||||||
@@ -27,7 +27,6 @@ import (
|
|||||||
const (
|
const (
|
||||||
KindEnrolment = "enrolment"
|
KindEnrolment = "enrolment"
|
||||||
KindReport = "report"
|
KindReport = "report"
|
||||||
KindHeartbeat = "heartbeat"
|
|
||||||
// KindToolsHeartbeat is a machine's node tools saying they are there (novox/hq to-be 45 S11).
|
// KindToolsHeartbeat is a machine's node tools saying they are there (novox/hq to-be 45 S11).
|
||||||
KindToolsHeartbeat = "tools-heartbeat"
|
KindToolsHeartbeat = "tools-heartbeat"
|
||||||
// KindHealth is a machine's statement of its long-running resources' health (novox/hq ADR 0240).
|
// KindHealth is a machine's statement of its long-running resources' health (novox/hq ADR 0240).
|
||||||
|
|||||||
@@ -114,12 +114,8 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
|
|||||||
// Heartbeats, on core NATS and off any stream (design 25 §3). Their own subscription because
|
// Heartbeats, on core NATS and off any stream (design 25 §3). Their own subscription because
|
||||||
// they are their own guarantee: a lost one is the next one.
|
// they are their own guarantee: a lost one is the next one.
|
||||||
beats := make(chan *nats.Msg, Prefetch)
|
beats := make(chan *nats.Msg, Prefetch)
|
||||||
alive, err := conn.ChanSubscribe(AliveSubjects, beats)
|
// Each machine's node tools, as cheap and lost the same way. The node-engine's alive message is not subscribed to: a
|
||||||
if err != nil {
|
// machine's last word is the watchman's presence (novox/hq ADR 0433 §3, issue 1183).
|
||||||
return fmt.Errorf("subscribing to heartbeats: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = alive.Unsubscribe() }()
|
|
||||||
// And each machine's node tools, on the same channel: as cheap, and lost the same way.
|
|
||||||
toolsAlive, err := conn.ChanSubscribe(ToolsAliveSubjects, beats)
|
toolsAlive, err := conn.ChanSubscribe(ToolsAliveSubjects, beats)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("subscribing to the node tools' heartbeats: %w", err)
|
return fmt.Errorf("subscribing to the node tools' heartbeats: %w", err)
|
||||||
@@ -283,8 +279,6 @@ func kindOfSubject(subject string) (string, bool) {
|
|||||||
switch rest {
|
switch rest {
|
||||||
case "report":
|
case "report":
|
||||||
return KindReport, true
|
return KindReport, true
|
||||||
case "alive":
|
|
||||||
return KindHeartbeat, true
|
|
||||||
case "tools-alive":
|
case "tools-alive":
|
||||||
return KindToolsHeartbeat, true
|
return KindToolsHeartbeat, true
|
||||||
case "health":
|
case "health":
|
||||||
|
|||||||
@@ -157,24 +157,31 @@ func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestNatsAHeartbeatIsHeardAndNothingIsKept(t *testing.T) {
|
// 1183/AC3, AC5: a node-engine still sending its alive message is harmless: it errors nothing, nothing is recorded and
|
||||||
|
// nothing is persisted; the tool runner's tools-alive is still heard.
|
||||||
|
func TestNatsAnAliveMessageIsHarmlessAndToolsAliveIsStillHeard(t *testing.T) { // 1183/AC3
|
||||||
js := aBus(t)
|
js := aBus(t)
|
||||||
store := &counted{}
|
store := &counted{}
|
||||||
_, stop := servingOn(t, js, store)
|
_, stop := servingOn(t, js, store)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
// Given time to subscribe: a core subscription that is not yet up misses what is published,
|
// Given time to subscribe: a core subscription that is not yet up misses what is published.
|
||||||
// which is the guarantee a heartbeat has and not a fault.
|
eventually(t, "the tools-alive subscription coming up", func() bool {
|
||||||
eventually(t, "the heartbeat subscription coming up", func() bool {
|
if err := js.Conn().Publish("mesh.control.anchor.alive", []byte(`{"node":"anchor"}`)); err != nil {
|
||||||
body, _ := json.Marshal(Alive{Node: "anchor"})
|
t.Fatalf("an alive message could not be published: %v", err)
|
||||||
_ = js.Conn().Publish(AliveSubject("anchor"), body)
|
}
|
||||||
|
body, _ := json.Marshal(ToolsAlive{Node: "anchor"})
|
||||||
|
_ = js.Conn().Publish(ToolsAliveSubject("anchor"), body)
|
||||||
_ = js.Conn().Flush()
|
_ = js.Conn().Flush()
|
||||||
return store.count() >= 1
|
_, heard := ToolsBeats.Of("anchor")
|
||||||
|
return heard
|
||||||
})
|
})
|
||||||
|
if n := store.count(); n != 0 {
|
||||||
|
t.Fatalf("an alive message was recorded %d time(s)", n)
|
||||||
|
}
|
||||||
info, err := js.Context().StreamInfo("CONTROL")
|
info, err := js.Context().StreamInfo("CONTROL")
|
||||||
if err != nil || info.State.Msgs != 0 {
|
if err != nil || info.State.Msgs != 0 {
|
||||||
t.Fatalf("a heartbeat was persisted, and the mesh's least valuable message now competes "+
|
t.Fatalf("an alive message was persisted: %+v, %v", info, err)
|
||||||
"for retention with its most valuable: %+v, %v", info, err)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -125,9 +125,6 @@ type enrolsAs struct{ reply EnrolReply }
|
|||||||
|
|
||||||
func (e enrolsAs) Enrol(context.Context, EnrolRequest) (EnrolReply, error) { return e.reply, nil }
|
func (e enrolsAs) Enrol(context.Context, EnrolRequest) (EnrolReply, error) { return e.reply, nil }
|
||||||
|
|
||||||
// A heartbeat is core NATS: it reaches the controller and nothing is persisted, so the stream the
|
|
||||||
// reports live in stays empty.
|
|
||||||
|
|
||||||
// The two events the controller follows arrive over one durable consumer with two filters, and it
|
// The two events the controller follows arrive over one durable consumer with two filters, and it
|
||||||
// can acknowledge them.
|
// can acknowledge them.
|
||||||
//
|
//
|
||||||
|
|||||||
+2
-26
@@ -205,8 +205,8 @@ func (s *Server) Close() {
|
|||||||
|
|
||||||
// Serve acts on what arrives until the context ends.
|
// Serve acts on what arrives until the context ends.
|
||||||
func (s *Server) Serve(ctx context.Context) error {
|
func (s *Server) Serve(ctx context.Context) error {
|
||||||
s.log.Printf("consuming what nodes say: %s, %s, %s, %s",
|
s.log.Printf("consuming what nodes say: %s, %s, %s",
|
||||||
KindEnrolment, KindReport, KindHeartbeat, KindBuilt)
|
KindEnrolment, KindReport, KindBuilt)
|
||||||
if s.upgrader != nil {
|
if s.upgrader != nil {
|
||||||
s.log.Printf("following %s", KindModuleMoved)
|
s.log.Printf("following %s", KindModuleMoved)
|
||||||
}
|
}
|
||||||
@@ -226,8 +226,6 @@ func (s *Server) act(ctx context.Context, m Control) {
|
|||||||
s.enrolling(ctx, m)
|
s.enrolling(ctx, m)
|
||||||
case KindReport:
|
case KindReport:
|
||||||
s.reported(ctx, m)
|
s.reported(ctx, m)
|
||||||
case KindHeartbeat:
|
|
||||||
s.heartbeat(m)
|
|
||||||
case KindToolsHeartbeat:
|
case KindToolsHeartbeat:
|
||||||
s.toolsHeartbeat(m)
|
s.toolsHeartbeat(m)
|
||||||
case KindHealth:
|
case KindHealth:
|
||||||
@@ -324,28 +322,6 @@ func (s *Server) outstanding(ctx context.Context, node string) string {
|
|||||||
return digest
|
return digest
|
||||||
}
|
}
|
||||||
|
|
||||||
// heartbeat records that a node was heard from, and nothing else.
|
|
||||||
//
|
|
||||||
// Deliberately silent: a node saying it is there every minute would fill the log with the ordinary
|
|
||||||
// case, and a log where the ordinary case is loud is a log nobody reads. Not held for the store
|
|
||||||
// either — the next heartbeat is a minute away, and a heartbeat kept for two minutes to be written
|
|
||||||
// late says nothing the one after it will not say better.
|
|
||||||
func (s *Server) heartbeat(m Control) {
|
|
||||||
var alive Alive
|
|
||||||
if err := json.Unmarshal(m.Body(), &alive); err != nil || alive.Node == "" {
|
|
||||||
_ = m.Drop()
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// The interval it says, for the watchdog's bound (novox/hq to-be 45 S1).
|
|
||||||
HostBeats.Heard(alive.Node, time.Now(), time.Duration(alive.IntervalSeconds)*time.Second, "")
|
|
||||||
if s.listener != nil {
|
|
||||||
if _, err := s.listener.Heard(context.Background(), Report{Node: alive.Node}); err != nil {
|
|
||||||
s.log.Printf("could not record that %s is here: %v", alive.Node, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_ = m.Took()
|
|
||||||
}
|
|
||||||
|
|
||||||
// toolsHeartbeat records that a machine's node tools were heard from (novox/hq to-be 45 S11), and
|
// toolsHeartbeat records that a machine's node tools were heard from (novox/hq to-be 45 S11), and
|
||||||
// nothing else: in this process's memory, for the watchdog — the next one is a minute away.
|
// nothing else: in this process's memory, for the watchdog — the next one is a minute away.
|
||||||
func (s *Server) toolsHeartbeat(m Control) {
|
func (s *Server) toolsHeartbeat(m Control) {
|
||||||
|
|||||||
@@ -121,15 +121,3 @@ func TestAReportCarryingWhatOnlyTheNodeKnowsIsActedOnHoweverOldItIs(t *testing.T
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// A heartbeat is not held for the store: the next one is a minute away, and one kept for two
|
|
||||||
// minutes to be written late says nothing the one after it will not say better.
|
|
||||||
func TestAHeartbeatIsNotHeldForTheStore(t *testing.T) {
|
|
||||||
s, in := serving()
|
|
||||||
s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("starting up"))}
|
|
||||||
to := &settled{}
|
|
||||||
s.act(context.Background(), in.sends(t, to, KindHeartbeat, Alive{Node: "anchor"}))
|
|
||||||
if !to.acked || len(in.held) != 0 {
|
|
||||||
t.Fatalf("a heartbeat was held for the store: %+v, %d held", *to, len(in.held))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -21,8 +21,7 @@ import (
|
|||||||
//
|
//
|
||||||
// **In memory, on purpose.** A heartbeat is the least valuable message the mesh sends — the next one
|
// **In memory, on purpose.** A heartbeat is the least valuable message the mesh sends — the next one
|
||||||
// is a minute away — and what a watchdog needs of it is the newest moment and the interval it said.
|
// is a minute away — and what a watchdog needs of it is the newest moment and the interval it said.
|
||||||
// A machine's last word is kept in the store as well (`last_seen`), which is what S1 reads; these are
|
// What the store does not keep: the node tools' word and its interval, and when the event loop last took
|
||||||
// what the store does not keep: the interval, the node tools' word, and when the event loop last took
|
|
||||||
// a message.
|
// a message.
|
||||||
|
|
||||||
// Beat is one emitter's newest heartbeat.
|
// Beat is one emitter's newest heartbeat.
|
||||||
@@ -43,11 +42,9 @@ type Beats struct {
|
|||||||
// NewBeats is an empty record, started now.
|
// NewBeats is an empty record, started now.
|
||||||
func NewBeats() *Beats { return &Beats{started: time.Now(), heard: map[string]Beat{}} }
|
func NewBeats() *Beats { return &Beats{started: time.Now(), heard: map[string]Beat{}} }
|
||||||
|
|
||||||
// HostBeats are the node-engines' heartbeats (S1); ToolsBeats the node tools' (S11).
|
// ToolsBeats are the node tools' heartbeats (S11). The node-engine's alive message is not heard: a machine's last word is
|
||||||
var (
|
// the watchman's presence (novox/hq ADR 0433 §3).
|
||||||
HostBeats = NewBeats()
|
var ToolsBeats = NewBeats()
|
||||||
ToolsBeats = NewBeats()
|
|
||||||
)
|
|
||||||
|
|
||||||
// Heard records one heartbeat.
|
// Heard records one heartbeat.
|
||||||
func (b *Beats) Heard(node string, at time.Time, every time.Duration, version string) {
|
func (b *Beats) Heard(node string, at time.Time, every time.Duration, version string) {
|
||||||
|
|||||||
Reference in New Issue
Block a user