diff --git a/cmd/mesh-control/main.go b/cmd/mesh-control/main.go index cb9b4e7..9a794fc 100644 --- a/cmd/mesh-control/main.go +++ b/cmd/mesh-control/main.go @@ -403,8 +403,8 @@ func serve(ctx context.Context) error { "reconnect. Set %s and %s.\n", broker.AddressVar, broker.CertificateVar) } - server, err := link.Connect(link.Enrolment{ - Inventory: inv, Identity: ident, Management: management, Broker: known}) + work := link.Enrolment{Inventory: inv, Identity: ident, Management: management, Broker: known} + server, err := link.Connect(work, work) if err != nil { return err } @@ -447,7 +447,7 @@ func declare(ctx context.Context, args []string) error { return err } - server, err := link.Connect(nil) + server, err := link.Connect(nil, nil) if err != nil { return err } diff --git a/internal/inventory/inventory_test.go b/internal/inventory/inventory_test.go index 8b03597..188439c 100644 --- a/internal/inventory/inventory_test.go +++ b/internal/inventory/inventory_test.go @@ -290,3 +290,87 @@ func TestAnUnknownSecretIsRefusedTheSameWayAsAnExpiredOne(t *testing.T) { t.Errorf("the two are distinguishable:\n unknown: %v\n expired: %v", unknownErr, expiredErr) } } + +func TestNeverReportedIsNotTheSameAsReportedNothing(t *testing.T) { + // The distinction that makes this safe to hand back. A node that applied nothing holds + // nothing; a node that has never spoken is unknown — and returning an empty list for the + // second would tell a rebuilding node it owns nothing, and have it remove whatever it found + // on the machine. + inv := fresh(t) + node, err := inv.AddNode(t.Context(), "laptop") + if err != nil { + t.Fatal(err) + } + + owned, reported, err := inv.Owned(t.Context(), node.ID) + if err != nil { + t.Fatal(err) + } + if owned != nil { + t.Errorf("a node that never reported came back owning %v", owned) + } + if !reported.IsZero() { + t.Error("a node that never reported has a report time") + } + + if err := inv.RecordOwned(t.Context(), node.ID, []string{}); err != nil { + t.Fatal(err) + } + owned, reported, err = inv.Owned(t.Context(), node.ID) + if err != nil { + t.Fatal(err) + } + if owned == nil { + t.Error("a node that reported holding nothing is indistinguishable from one that never spoke") + } + if reported.IsZero() { + t.Error("a report that happened has no time on it") + } +} + +func TestWhatANodeOwnsIsReplacedNotAccumulated(t *testing.T) { + // The question this answers is what is on that machine now. A node that stopped owning + // something and had it remembered would be handed it back on a rebuild and put it there again. + 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", "c"}); err != nil { + t.Fatal(err) + } + if err := inv.RecordOwned(t.Context(), node.ID, []string{"a"}); err != nil { + t.Fatal(err) + } + + owned, _, err := inv.Owned(t.Context(), node.ID) + if err != nil { + t.Fatal(err) + } + if len(owned) != 1 || owned[0] != "a" { + t.Errorf("after reporting a, the mesh believes the node owns %v", owned) + } +} + +func TestAnAnswerAboutAMachineCarriesItsAge(t *testing.T) { + // This repository has already been bitten by a cache with no age on it: a node running from + // one looked identical to a node running from the database. An answer about a machine is + // worth much less without knowing how old it is, so the age comes back with it. + inv := fresh(t) + node, err := inv.AddNode(t.Context(), "laptop") + if err != nil { + t.Fatal(err) + } + before := time.Now().Add(-time.Second) + if err := inv.RecordOwned(t.Context(), node.ID, []string{"a"}); err != nil { + t.Fatal(err) + } + + _, reported, err := inv.Owned(t.Context(), node.ID) + if err != nil { + t.Fatal(err) + } + if reported.Before(before) { + t.Errorf("the report time is %s, which is before the report", reported) + } +} diff --git a/internal/inventory/migrations/0003-what-a-node-owns.sql b/internal/inventory/migrations/0003-what-a-node-owns.sql new file mode 100644 index 0000000..d6a08c7 --- /dev/null +++ b/internal/inventory/migrations/0003-what-a-node-owns.sql @@ -0,0 +1,18 @@ +-- The last thing a node said it owns. +-- +-- novox/hq 09-the-node-lifecycle: the host reports what it owns and the mesh keeps the last +-- report. **This is a backup, never a source.** The host never reads it to decide anything; it is +-- handed back only when a node has lost its own store, and a node that disagrees with it wins, +-- because the node is the one that can see the machine. +-- +-- Its point is the orphans. A node that loses its state file currently strands whatever it had +-- applied: nothing on the machine knows those resources were ever the mesh's doing, so nothing +-- removes them. With this, a rebuilt node receives both the declaration and the record of what it +-- previously owned, and can take away what is no longer declared. + +alter table node add column owned jsonb; + +-- When that report arrived. Separate from last_seen, which moves on any word from the node at +-- all: a node can be plainly alive for weeks without applying anything, and reading one as the +-- other would make a quiet node look like a stale one. +alter table node add column owned_reported timestamptz; diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index ee69991..a24ba0d 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -233,3 +233,53 @@ func (i *Inventory) Seen(ctx context.Context, node string) error { _, err := i.store.Pool().Exec(ctx, `update node set last_seen = now() where id = $1`, node) return err } + +// RecordOwned keeps the last account a node gave of what it holds. +// +// A copy for recovery and never a source (novox/hq 09-the-node-lifecycle). Nothing here decides +// anything from it; it is handed back to a node that has lost its own store, and if that node +// then disagrees, the node wins — it is the one that can see the machine. +// +// Replaced rather than appended. A history of what a node used to own answers a question nobody +// asks, and the one question this does answer — what is on that machine now — is only answered by +// the latest. +func (i *Inventory) RecordOwned(ctx context.Context, node string, owned []string) error { + raw, err := json.Marshal(owned) + if err != nil { + return err + } + _, err = i.store.Pool().Exec(ctx, + `update node set owned = $2, owned_reported = now(), last_seen = now() where id = $1`, + node, raw) + return err +} + +// Owned is what a node last said it holds, and when it said so. +// +// The age is returned with it rather than left to the caller to look up, because an answer about +// a machine is worth much less without one — and this repository has already been bitten by a +// cache with no age on it. +func (i *Inventory) Owned(ctx context.Context, node string) ([]string, time.Time, error) { + var raw []byte + var reported *time.Time + err := i.store.Pool().QueryRow(ctx, + `select owned, owned_reported from node where id = $1`, node).Scan(&raw, &reported) + if errors.Is(err, pgx.ErrNoRows) { + return nil, time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, node) + } + if err != nil { + return nil, time.Time{}, err + } + if len(raw) == 0 || reported == nil { + // Never reported is not the same as reported nothing. A node that has applied nothing + // holds nothing; a node that has never spoken is unknown, and handing back an empty list + // as though it were a report would tell a rebuilding node it owns nothing and have it + // remove whatever it found. + return nil, time.Time{}, nil + } + var owned []string + if err := json.Unmarshal(raw, &owned); err != nil { + return nil, time.Time{}, err + } + return owned, *reported, nil +} diff --git a/internal/link/enrolment.go b/internal/link/enrolment.go index 4490f24..480e7be 100644 --- a/internal/link/enrolment.go +++ b/internal/link/enrolment.go @@ -106,3 +106,25 @@ var _ Enroller = Enrolment{} // ErrNoBrokerManagement is returned when an account cannot be made because nothing was configured. var ErrNoBrokerManagement = errors.New("no broker management configured") + +// Heard records what a node reported about itself. +// +// A node states; the owning context writes (novox/hq ADR 0006). What a node says it applied is +// its own account of its own machine, kept as a copy for recovery — so this writes it down and +// decides nothing from it. +func (e Enrolment) Heard(ctx context.Context, report Report) error { + if report.Node == "" { + return errors.New("a report named no node") + } + node, err := e.Inventory.NodeByName(ctx, report.Node) + 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 { + return e.Inventory.Seen(ctx, node.ID) + } + return e.Inventory.RecordOwned(ctx, node.ID, report.Applied) +} diff --git a/internal/link/serve.go b/internal/link/serve.go index 3a952e3..c42b495 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -28,15 +28,22 @@ type Enroller interface { } // Server consumes what nodes say. +// Listener is what the control plane does with a report. Separate from Enroller so the two can +// be given independently, and so a server that only sends declarations needs neither. +type Listener interface { + Heard(ctx context.Context, report Report) error +} + type Server struct { conn *amqp.Connection channel *amqp.Channel enroller Enroller + listener Listener log *log.Logger } // Connect opens the control plane's own connection to the broker. -func Connect(enroller Enroller) (*Server, error) { +func Connect(enroller Enroller, listener Listener) (*Server, error) { url := strings.TrimSpace(os.Getenv(AMQPVar)) if url == "" { return nil, fmt.Errorf( @@ -79,7 +86,7 @@ func Connect(enroller Enroller) (*Server, error) { } } - return &Server{conn: conn, channel: channel, enroller: enroller, + return &Server{conn: conn, channel: channel, enroller: enroller, listener: listener, log: log.New(os.Stdout, "", log.LstdFlags)}, nil } @@ -161,6 +168,14 @@ func (s *Server) handleReport(delivery amqp.Delivery) { _ = delivery.Reject(false) return } + if s.listener != nil { + if err := s.listener.Heard(context.Background(), report); err != nil { + // Said rather than swallowed. A report the mesh heard and failed to write down is a + // node whose recovery copy is silently older than it looks. + s.log.Printf("could not record %s's report: %v", report.Node, err) + } + } + switch { case report.Refused != "": s.log.Printf("%s refused a declaration: %s", report.Node, report.Refused)