Files
mesh-controller/cmd/mesh-control/main.go
T
jschoubben f563ababa1 The mesh keeps a copy of what each node owns
novox/hq 09-the-node-lifecycle asks for this and it was missing: the host
reports what it owns and the mesh keeps the last report. A backup, never a
source -- nothing decides anything from it, 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 applied: nothing on the machine knows those resources were 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.

Never reported and reported nothing are kept apart, and that is the whole care
in it. A node that applied nothing holds nothing; a node that has never spoken
is unknown -- and handing back an empty list for the second would tell a
rebuilding node it owns nothing and have it remove whatever it found.

The age comes back with the answer rather than being left for the caller to go
and find. 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.

A refusal or a partial failure moves last_seen and nothing else: neither is an
account of what the machine holds, and recording one as though it were would
tell a rebuilding node to remove what it still has.
2026-08-29 16:51:54 +02:00

462 lines
14 KiB
Go

// Command mesh-control is the control plane: everything that needs to know about more than one
// node (novox/hq ADR 0006).
//
// It runs as one process holding several contexts, each owning its own store. Today it holds one,
// `inventory`, and does one thing with it — brings its schema up to date, which is step 3 of the
// bootstrap in novox/hq 07-the-substrate and the step the first node cannot get past without.
package main
import (
"context"
"errors"
"flag"
"fmt"
"os"
"os/signal"
"syscall"
"time"
"github.com/novox/mesh-control/internal/broker"
"github.com/novox/mesh-control/internal/identity"
"github.com/novox/mesh-control/internal/inventory"
"github.com/novox/mesh-control/internal/link"
"github.com/novox/mesh-control/internal/store"
"github.com/novox/mesh-control/internal/token"
)
// version is stamped at link time. Unset in a development build, and it says so rather than
// claiming a number.
var version = "development build"
// held is a context this process was granted, and the schema it carries.
//
// novox/hq ADR 0006 names seven. One is built. The list is short because the others do not exist
// yet, not because they are optional.
var held = []struct {
name string
migrations func() ([]store.Migration, error)
}{
{inventory.Name, inventory.Migrations},
{identity.Name, identity.Migrations},
}
func main() {
if err := run(); err != nil {
fmt.Fprintf(os.Stderr, "mesh-control: %v\n", err)
os.Exit(1)
}
}
func run() error {
args := os.Args[1:]
if len(args) == 0 {
usage()
return fmt.Errorf("no command given")
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
switch args[0] {
case "migrate":
return migrate(ctx)
case "node":
return nodeCommand(ctx, args[1:])
case "token":
return tokenCommand(ctx, args[1:])
case "identity":
return identityCommand(ctx, args[1:])
case "broker":
return brokerCommand(args[1:])
case "serve":
return serve(ctx)
case "declare":
return declare(ctx, args[1:])
case "version":
fmt.Println(version)
return nil
case "help", "-h", "--help":
usage()
return nil
default:
usage()
return fmt.Errorf("%q is not a command", args[0])
}
}
func usage() {
fmt.Fprint(os.Stderr, `mesh-control — the control plane
migrate bring each context's schema up to date
node add <name> create a node record
node list the nodes this mesh knows about
token issue --node <name> a one-time right to join, for an existing record
token issue --new <name> create the record and issue for it
identity show this control plane's signing key
broker show where the broker is, and what to expect there
serve consume what nodes say, and answer
declare <node> <file> send a node a signed declaration
version what this binary is
Each context reaches its own store through its own credential (novox/hq ADR 0008), named
`+store.Variable("<context>")+`. This process holds:
`)
for _, c := range held {
fmt.Fprintf(os.Stderr, " %-12s database %-12s from %s\n",
c.name, store.Database(c.name), store.Variable(c.name))
}
fmt.Fprintln(os.Stderr)
}
// migrate brings every held context's schema up to date.
//
// Reported per context and per migration, because this runs during a bootstrap on a machine with
// nothing else on it — the output is the only account of what happened, and "migrated" is not one.
func migrate(ctx context.Context) error {
for _, c := range held {
migrations, err := c.migrations()
if err != nil {
return err
}
s, err := store.Open(ctx, c.name)
if err != nil {
return err
}
defer s.Close()
// The bootstrap raises PostgreSQL moments before this runs, and a container that is
// running is not a database that will answer — a distinction this project has already
// paid for once, when a crash-looping database reported itself as up between restarts.
if err := s.Ready(ctx, 60*time.Second); err != nil {
return err
}
done, err := s.Migrate(ctx, migrations)
for _, m := range done {
fmt.Printf("%s: applied %04d-%s\n", c.name, m.Number, m.Name)
}
if err != nil {
return err
}
if len(done) == 0 {
applied, err := s.AppliedMigrations(ctx)
if err != nil {
return err
}
fmt.Printf("%s: already up to date — %d migration(s)\n", c.name, len(applied))
}
}
return nil
}
// openInventory connects and waits, the way every command that touches it needs to.
func openInventory(ctx context.Context) (*inventory.Inventory, error) {
inv, err := inventory.Open(ctx)
if err != nil {
return nil, err
}
if err := inv.Ready(ctx, 30*time.Second); err != nil {
inv.Close()
return nil, err
}
return inv, nil
}
func nodeCommand(ctx context.Context, args []string) error {
if len(args) == 0 {
return errors.New("node add <name>, or node list")
}
inv, err := openInventory(ctx)
if err != nil {
return err
}
defer inv.Close()
switch args[0] {
case "add":
if len(args) != 2 {
return errors.New("node add <name>")
}
node, err := inv.AddNode(ctx, args[1])
if err != nil {
return err
}
fmt.Printf("added %s (%s)\n", node.Name, node.ID)
return nil
case "list":
nodes, err := inv.Nodes(ctx)
if err != nil {
return err
}
if len(nodes) == 0 {
// Said rather than printed as nothing: an empty list and a failed read must never
// look the same, and this command answering "none" is only honest because getting
// here means the store answered.
fmt.Println("this mesh has no node records yet")
return nil
}
for _, n := range nodes {
fmt.Printf("%-20s %s added %s\n", n.Name, n.ID, n.Created.Format(time.RFC3339))
}
return nil
default:
return fmt.Errorf("node has no %q; it has add and list", args[0])
}
}
func tokenCommand(ctx context.Context, args []string) error {
if len(args) == 0 || args[0] != "issue" {
return errors.New("token issue --node <name>, or token issue --new <name>")
}
set := flag.NewFlagSet("token issue", flag.ContinueOnError)
existing := set.String("node", "", "issue for a node record that already exists")
fresh := set.String("new", "", "create the node record, then issue for it")
validFor := set.Duration("for", time.Hour, "how long the token may be used")
if err := set.Parse(args[1:]); err != nil {
return err
}
// Exactly one, because the difference is what the token binds to. A command that guessed
// would sometimes create a second record for a machine that already has one.
if (*existing == "") == (*fresh == "") {
return errors.New("give exactly one of --node <name> or --new <name>: the first is a " +
"machine the mesh already has a record for, the second is one it has never seen")
}
inv, err := openInventory(ctx)
if err != nil {
return err
}
defer inv.Close()
name := *existing
if *fresh != "" {
node, err := inv.AddNode(ctx, *fresh)
if err != nil {
return err
}
name = node.Name
}
issued, err := inv.IssueToken(ctx, name, *validFor)
if err != nil {
return err
}
// Assembled from two contexts by the process that holds both grants. Neither reads the
// other's store (novox/hq ADR 0008) — each is asked for its own part.
ident, err := openIdentity(ctx)
if err != nil {
return err
}
defer ident.Close()
key, err := ident.Establish(ctx)
if err != nil {
return err
}
// The account is created before the token is handed over, which is what removes the
// chicken-and-egg entirely: the mesh runs the broker, so a joining node's credentials can
// exist before it does. The one-time secret IS the password, so a node's first connection is
// already authenticated and enrolment is what happens over it.
if management, err := broker.ManagementFromEnvironment(); err == nil {
if err := management.CreateNodeAccount(ctx, issued.Node.Name, issued.Secret); err != nil {
return err
}
fmt.Printf("broker account %s created, scoped to %s and the %s exchange\n\n",
issued.Node.Name, link.QueueFor(issued.Node.Name), link.Exchange)
} else if !errors.Is(err, broker.ErrNotConfigured) {
return err
}
made := token.Token{Signer: key.Public, Secret: issued.Secret}
// Absent is a state, not a failure: a control plane can hold records and a key before it has
// a broker. What it cannot do is issue a token anybody could use, and Missing() says so.
known, err := broker.FromEnvironment()
switch {
case err == nil:
made.Broker, made.Fingerprint = known.Address, known.Fingerprint
case errors.Is(err, broker.ErrNotConfigured):
default:
return err
}
encoded, err := made.Encode()
if err != nil {
return err
}
fmt.Printf("token for %s, usable once, until %s\n\n %s\n\n",
issued.Node.Name, issued.Expires.Format(time.RFC3339), encoded)
fmt.Println("This is the only time it is shown. What is stored is a hash of the secret.")
if missing := made.Missing(); len(missing) > 0 {
fmt.Printf("\nINCOMPLETE — this token cannot be used to join anything yet. Missing:\n")
for _, m := range missing {
fmt.Printf(" - %s\n", m)
}
fmt.Printf("\nSet %s and %s once the broker is raised.\n",
broker.AddressVar, broker.CertificateVar)
}
return nil
}
func openIdentity(ctx context.Context) (*identity.Identity, error) {
ident, err := identity.Open(ctx)
if err != nil {
return nil, err
}
if err := ident.Ready(ctx, 30*time.Second); err != nil {
ident.Close()
return nil, err
}
return ident, nil
}
func identityCommand(ctx context.Context, args []string) error {
if len(args) == 0 || args[0] != "show" {
return errors.New("identity show")
}
ident, err := openIdentity(ctx)
if err != nil {
return err
}
defer ident.Close()
// Establish rather than read: a control plane asked for its identity before it has one should
// get one, not an error. Generating it is idempotent, so this is safe to run at any time.
key, err := ident.Establish(ctx)
if err != nil {
return err
}
fmt.Printf("signing key %s\n", key.ID)
fmt.Printf("fingerprint %s\n", key.Fingerprint())
fmt.Printf("created %s\n", key.Created.Format(time.RFC3339))
fmt.Printf("\nThe public half of this travels in every enrolment token. A node believes a\n" +
"declaration because it carries a signature this key made (novox/hq ADR 0004).\n")
return nil
}
func brokerCommand(args []string) error {
if len(args) == 0 || args[0] != "show" {
return errors.New("broker show")
}
known, err := broker.FromEnvironment()
if errors.Is(err, broker.ErrNotConfigured) {
fmt.Printf("no broker configured. Set %s and %s.\n\n"+
"Until then tokens carry the signing key and the one-time secret, and say what they\n"+
"are missing. They cannot be used to join.\n",
broker.AddressVar, broker.CertificateVar)
return nil
}
if err != nil {
return err
}
fmt.Printf("address %s\n", known.Address)
fmt.Printf("fingerprint %s\n", known.Fingerprint)
fmt.Print("\nThe fingerprint is computed from the certificate on disk, never configured. A\n" +
"node checks it before sending anything (novox/hq ADR 0004).\n")
return nil
}
// serve is the control plane running: one connection to the broker, one queue, one consumer.
func serve(ctx context.Context) error {
inv, err := openInventory(ctx)
if err != nil {
return err
}
defer inv.Close()
ident, err := openIdentity(ctx)
if err != nil {
return err
}
defer ident.Close()
// Established at start rather than on first use. A control plane that cannot sign is one
// whose declarations every node correctly refuses, and that should be a startup failure
// rather than something discovered at the first declaration.
key, err := ident.Establish(ctx)
if err != nil {
return err
}
fmt.Printf("signing as %s\n", key.Fingerprint()[:16])
management, err := broker.ManagementFromEnvironment()
if err != nil && !errors.Is(err, broker.ErrNotConfigured) {
return err
}
// Where the broker is and what to expect there, so a node can be told how to come back
// without a person and a new token.
known, err := broker.FromEnvironment()
if err != nil && !errors.Is(err, broker.ErrNotConfigured) {
return err
}
if errors.Is(err, broker.ErrNotConfigured) {
fmt.Printf("no broker address configured, so enrolled nodes will not be told how to "+
"reconnect. Set %s and %s.\n", broker.AddressVar, broker.CertificateVar)
}
work := link.Enrolment{Inventory: inv, Identity: ident, Management: management, Broker: known}
server, err := link.Connect(work, work)
if err != nil {
return err
}
defer server.Close()
return server.Serve(ctx)
}
// declare sends one node a declaration, signed.
//
// Signed here rather than trusted from the broker: a node connects to the broker and takes
// instruction from the control plane behind it, and those are two identities. If a node believed
// whatever arrived on its queue, a compromised broker could forge declarations — and since the
// host applies whatever the link delivers, that is the whole machine (novox/hq ADR 0004).
func declare(ctx context.Context, args []string) error {
if len(args) != 2 {
return errors.New("declare <node> <declaration.json>")
}
node, path := args[0], args[1]
raw, err := os.ReadFile(path)
if err != nil {
return err
}
ident, err := openIdentity(ctx)
if err != nil {
return err
}
defer ident.Close()
// The node has to exist before it can be told anything. Publishing to a queue nobody consumes
// would sit there looking like success.
inv, err := openInventory(ctx)
if err != nil {
return err
}
defer inv.Close()
if _, err := inv.NodeByName(ctx, node); err != nil {
return err
}
server, err := link.Connect(nil, nil)
if err != nil {
return err
}
defer server.Close()
if err := link.Declare(ctx, server.Channel(), ident, node, raw, 15*time.Second); err != nil {
return err
}
fmt.Printf("sent %s a signed declaration (%d bytes)\n", node, len(raw))
return nil
}