diff --git a/cmd/mesh-control/main.go b/cmd/mesh-control/main.go index 7791246..546f3d2 100644 --- a/cmd/mesh-control/main.go +++ b/cmd/mesh-control/main.go @@ -206,7 +206,7 @@ func nodeCommand(ctx context.Context, args []string) error { return nil } for _, n := range nodes { - fmt.Printf("%-20s %s added %s\n", n.Name, n.ID, n.Created.Format(time.RFC3339)) + fmt.Printf("%-20s %-14s %s\n", n.Name, heardFrom(n), n.ID) } return nil @@ -633,3 +633,38 @@ func overlayPush(ctx context.Context, inv *inventory.Inventory) error { fmt.Printf("\n%d of %d node(s) told\n", sent, len(nodes)) return nil } + +// SilentFor is how long a node may be quiet before the mesh says so. +// +// A node speaks every minute, so three of them missed is a gap rather than a slow one. The number +// is not the point — being able to say "out of touch" at all is, and nothing could before. +const SilentFor = 3 * time.Minute + +// heardFrom 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 — and until this existed both looked exactly like a node that is current. +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. +func roughly(d time.Duration) string { + switch { + case d < time.Hour: + return fmt.Sprintf("%dm", int(d.Minutes())) + case d < 48*time.Hour: + return fmt.Sprintf("%dh", int(d.Hours())) + default: + return fmt.Sprintf("%dd", int(d.Hours()/24)) + } +} diff --git a/internal/inventory/inventory_test.go b/internal/inventory/inventory_test.go index 188439c..b0a7eae 100644 --- a/internal/inventory/inventory_test.go +++ b/internal/inventory/inventory_test.go @@ -374,3 +374,62 @@ func TestAnAnswerAboutAMachineCarriesItsAge(t *testing.T) { t.Errorf("the report time is %s, which is before the report", reported) } } + +func TestNeverHeardFromIsNotTheSameAsLongAgo(t *testing.T) { + // The distinction the whole thing rests on. A node that has never spoken did not finish + // joining; a node last heard from a month ago is running a month-old picture of the mesh. + // Until this existed both looked exactly like a node that is current. + inv := fresh(t) + node, err := inv.AddNode(t.Context(), "laptop") + if err != nil { + t.Fatal(err) + } + + nodes, err := inv.Nodes(t.Context()) + if err != nil { + t.Fatal(err) + } + if _, ever := nodes[0].Silent(); ever { + t.Error("a node that has never spoken reports a time since it last did") + } + + if err := inv.Seen(t.Context(), node.ID); err != nil { + t.Fatal(err) + } + nodes, err = inv.Nodes(t.Context()) + if err != nil { + t.Fatal(err) + } + silent, ever := nodes[0].Silent() + if !ever { + t.Fatal("a node that has spoken still reports never having done so") + } + if silent > time.Minute { + t.Errorf("a node heard from just now has been silent for %s", silent) + } +} + +func TestBeingHeardFromDoesNotChangeWhatANodeOwns(t *testing.T) { + // A node saying it is there is not an account of what it holds. Treating one as the other + // would replace the recovery copy with an empty list every minute, and a rebuilding node + // would then be told it owns nothing. + inv := fresh(t) + node, err := inv.AddNode(t.Context(), "laptop") + if err != nil { + t.Fatal(err) + } + if err := inv.RecordOwned(t.Context(), node.ID, []string{"a", "b"}); err != nil { + t.Fatal(err) + } + if err := inv.Seen(t.Context(), node.ID); err != nil { + t.Fatal(err) + } + + owned, _, err := inv.Owned(t.Context(), node.ID) + if err != nil { + t.Fatal(err) + } + if len(owned) != 2 { + t.Errorf("after a bare word that the node is here, the mesh believes it owns %v", owned) + } +} diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index 86ca582..1d6dbe4 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -40,6 +40,22 @@ type Node struct { ID string Name string Created time.Time + + // LastSeen is when this node was last heard from, or zero if it never has been. + // + // Zero and long-ago are different answers and are kept different. A node that has never + // spoken has not joined properly; a node that spoke last month is a node running last + // month's assignments, and until this existed those looked the same as a node that is + // current (novox/hq 09-the-node-lifecycle). + LastSeen time.Time +} + +// Silent is how long since this node was last heard from, and whether it ever was. +func (n Node) Silent() (time.Duration, bool) { + if n.LastSeen.IsZero() { + return 0, false + } + return time.Since(n.LastSeen), true } // ErrNoSuchNode is returned when a name matches no record. @@ -78,7 +94,7 @@ func (i *Inventory) AddNode(ctx context.Context, name string) (Node, error) { // Nodes are every node record, oldest first. func (i *Inventory) Nodes(ctx context.Context) ([]Node, error) { rows, err := i.store.Pool().Query(ctx, - `select id, name, created from node order by created, name`) + `select id, name, created, last_seen from node order by created, name`) if err != nil { return nil, err } @@ -87,9 +103,13 @@ func (i *Inventory) Nodes(ctx context.Context) ([]Node, error) { var nodes []Node for rows.Next() { var n Node - if err := rows.Scan(&n.ID, &n.Name, &n.Created); err != nil { + var seen *time.Time + if err := rows.Scan(&n.ID, &n.Name, &n.Created, &seen); err != nil { return nil, err } + if seen != nil { + n.LastSeen = *seen + } nodes = append(nodes, n) } return nodes, rows.Err() diff --git a/internal/link/enrolment.go b/internal/link/enrolment.go index df52638..5705a2b 100644 --- a/internal/link/enrolment.go +++ b/internal/link/enrolment.go @@ -131,10 +131,10 @@ func (e Enrolment) Heard(ctx context.Context, report Report) error { if err != nil { return err } - // A refusal or a failure is not an account of what the machine holds, so it moves last_seen - // and nothing else. Recording a partial list as though it were the whole would tell a - // rebuilding node to remove what it still has. - if report.Refused != "" || len(report.Failed) > 0 { + // A refusal, a failure, or a bare word that the node is there — none of them is an account of + // what the machine holds, so each moves last_seen and nothing else. Recording a partial list + // as though it were the whole would tell a rebuilding node to remove what it still has. + if report.Refused != "" || len(report.Failed) > 0 || report.Applied == nil { return e.Inventory.Seen(ctx, node.ID) } return e.Inventory.RecordOwned(ctx, node.ID, report.Applied) diff --git a/internal/link/protocol.go b/internal/link/protocol.go index e384592..13c5067 100644 --- a/internal/link/protocol.go +++ b/internal/link/protocol.go @@ -17,6 +17,7 @@ const ControlQueue = "control" const ( KeyEnrol = "enrol" KeyReport = "report" + KeyAlive = "alive" ) // QueueFor is the queue a node consumes from — the only one it may read. @@ -59,6 +60,15 @@ type Signed struct { 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"` +} + // Report is what a node states after applying. It states; the owning context writes. type Report struct { Node string `json:"node"` diff --git a/internal/link/serve.go b/internal/link/serve.go index 0a290e2..aeff2e4 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -78,7 +78,7 @@ func Connect(enroller Enroller, listener Listener) (*Server, error) { // accepts, finds no queue for, and drops — the publisher sees success and the consumer sees // nothing. That is exactly what happened to reports: `report` was left unbound while `enrol` // worked, so nodes announced what they had applied into a void for an afternoon. - for _, key := range []string{KeyEnrol, KeyReport} { + for _, key := range []string{KeyEnrol, KeyReport, KeyAlive} { if err := channel.QueueBind(ControlQueue, key, Exchange, false, nil); err != nil { conn.Close() return nil, fmt.Errorf("cannot bind %s to %s/%s: %w", ControlQueue, Exchange, key, err) @@ -121,7 +121,7 @@ func (s *Server) Serve(ctx context.Context) error { } closed := s.conn.NotifyClose(make(chan *amqp.Error, 1)) - s.log.Printf("consuming %s, bound to %s/{%s,%s}", ControlQueue, Exchange, KeyEnrol, KeyReport) + s.log.Printf("consuming %s, bound to %s/{%s,%s,%s}", ControlQueue, Exchange, KeyEnrol, KeyReport, KeyAlive) for { select { @@ -147,6 +147,8 @@ func (s *Server) handle(ctx context.Context, delivery amqp.Delivery) { s.handleEnrol(ctx, delivery) case KeyReport: s.handleReport(delivery) + case KeyAlive: + s.handleAlive(delivery) default: // Rejected without requeue: a message nothing understands will not be understood on the // next attempt either, and requeuing it would spin. @@ -155,6 +157,24 @@ func (s *Server) handle(ctx context.Context, delivery amqp.Delivery) { } } +// handleAlive 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. +func (s *Server) handleAlive(delivery amqp.Delivery) { + var alive Alive + if err := json.Unmarshal(delivery.Body, &alive); err != nil || alive.Node == "" { + _ = delivery.Reject(false) + return + } + 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) + } + } + _ = delivery.Ack(false) +} + // handleReport records what a node says it did. // // A node states; nothing here writes anything the node claimed about itself beyond that it was