package main import ( "context" "crypto/sha256" "encoding/json" "errors" "flag" "fmt" "os" "slices" "sort" "strings" "sync" "time" "github.com/novox/mesh-host/validate" "github.com/novox/mesh-controller/internal/artifacts" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/builder" "github.com/novox/mesh-controller/internal/catalogue" snapshot "github.com/novox/mesh-controller/internal/facts" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) // The facts snapshot (novox/hq to-be 45 §9, ADR 0227 rule 9): what a merge check needs to judge a change // against the mesh that runs, written by the controller to the artifact store, where the build seat reads // it. internal/facts says what it holds and what it never holds; this composes it from the store and // keeps it current. // factsEvery is how often the snapshot is composed. It is kept when it moved — a machine, an // assignment, a seat, a setting, a build — or once a day when nothing did, so its age says the // controller is still writing it (S14). var factsEvery = 10 * time.Minute // factsDaily is how old a snapshot of an unchanged mesh may grow before it is written again. const factsDaily = 24 * time.Hour // factsStaleAfter is S14's bound: a snapshot older than this is one no check should be fed. const factsStaleAfter = 48 * time.Hour // factsExport is what this controller knows of the snapshot it keeps: when the newest was taken, its // content, and the last attempt's error. type factsExport struct { mu sync.Mutex taken time.Time content string digest string err error began time.Time } // exportedFacts is this process's export, read by S14. var exportedFacts = &factsExport{} func (e *factsExport) last() (taken time.Time, digest string, began time.Time, err error) { e.mu.Lock() defer e.mu.Unlock() return e.taken, e.digest, e.began, e.err } func (e *factsExport) kept(f snapshot.Facts, content, digest string) { e.mu.Lock() defer e.mu.Unlock() e.taken, e.content, e.digest, e.err = f.Taken, content, digest, nil } func (e *factsExport) failed(err error) { e.mu.Lock() defer e.mu.Unlock() e.err = err } // exportingFacts keeps the snapshot current for as long as this controller holds the lease: ctx ends // when it stops acting. func exportingFacts(ctx context.Context, open *stores, busVersion func() string) { exportedFacts.mu.Lock() exportedFacts.began = time.Now() exportedFacts.mu.Unlock() // What the store holds already, so a restarted controller neither writes an unchanged snapshot again // nor reads its age as zero. if address, err := factsStore(ctx, open); err == nil { if body, digest, err := (artifacts.Store{Address: address}).GetTagged(ctx, snapshot.Repository, snapshot.Tag); err == nil { if f, err := snapshot.Decode(body); err == nil { content, _ := f.Content() exportedFacts.kept(f, content, digest) } } } failing := "" first := time.NewTimer(time.Minute) defer first.Stop() tick := time.NewTicker(factsEvery) defer tick.Stop() for { select { case <-ctx.Done(): return case <-first.C: case <-tick.C: } wrote, err := exportFacts(ctx, open, busVersion(), false) why := "" if err != nil { why = err.Error() exportedFacts.failed(err) } if why != failing { if why != "" { fmt.Printf("the facts snapshot cannot be kept: %s\n", why) } else { fmt.Println("the facts snapshot is kept again") } failing = why } if wrote != "" { fmt.Printf("the facts snapshot moved and is kept as %s\n", short(strings.TrimPrefix(wrote, "sha256:"))) } } } // exportFacts composes the snapshot and keeps it when it moved, or when the one kept is a day old, or // when told to. Answers the digest it kept, empty when it kept nothing. func exportFacts(ctx context.Context, open *stores, busVersion string, force bool) (string, error) { f, err := gatherFacts(ctx, open, busVersion) if err != nil { return "", err } content, err := f.Content() if err != nil { return "", err } exportedFacts.mu.Lock() unchanged := content == exportedFacts.content && time.Since(exportedFacts.taken) < factsDaily exportedFacts.mu.Unlock() if unchanged && !force { return "", nil } body, err := f.Encode() if err != nil { return "", err } address, err := factsStore(ctx, open) if err != nil { return "", err } digest, err := (artifacts.Store{Address: address}).PutTagged(ctx, snapshot.Repository, snapshot.Tag, snapshot.MediaType, body) if err != nil && digest == "" { return "", err } exportedFacts.kept(f, content, digest) return digest, err } // factsStore is the artifact store as this controller reaches it. func factsStore(ctx context.Context, open *stores) (string, error) { shelf, err := open.inventory.Catalogue(ctx) if err != nil { return "", err } address, err := artifactStoreAddress(ctx, open.inventory, shelf, "") if err != nil { return "", err } if address == "" { return "", errors.New("the artifact store is not on the private network, so there is nowhere to keep the facts") } return address, nil } // gatherFacts composes one snapshot from the store: read only, nothing made, nothing sent. func gatherFacts(ctx context.Context, open *stores, busVersion string) (snapshot.Facts, error) { inv := open.inventory f := snapshot.Facts{Format: snapshot.Format, Taken: time.Now().UTC(), Controller: snapshot.Build{Version: version}} f.Versions.Bus = busVersion storeVersion, err := inv.ServerVersion(ctx) if err != nil { return snapshot.Facts{}, fmt.Errorf("the store will not say its version: %w", err) } f.Versions.Store = storeVersion nodes, err := inv.Nodes(ctx) if err != nil { return snapshot.Facts{}, err } overlays, err := inv.Overlays(ctx) if err != nil { return snapshot.Facts{}, err } place := map[string]inventory.Overlay{} for _, o := range overlays { place[o.Name] = o } entries, err := inv.Catalogued(ctx) if err != nil { return snapshot.Facts{}, err } shelf := map[string]catalogue.Manifest{} for _, e := range entries { shelf[e.Manifest.Module] = e.Manifest } current, err := inv.CurrentBuilds(ctx) if err != nil { return snapshot.Facts{}, err } read, err := inv.ReadRepositories(ctx) if err != nil { return snapshot.Facts{}, err } edges, err := inv.Dependencies(ctx) if err != nil { return snapshot.Facts{}, err } holdings, err := inv.Holdings(ctx) if err != nil { return snapshot.Facts{}, err } // **Every name first**, so text read afterwards — a setting naming a machine, a problem naming a // site — has it replaced wherever it appears. scrub := snapshot.NewScrubber() domains := map[string]string{} for _, n := range nodes { scrub.Machine(n.Name) if n.Account != "" && n.Account != "root" { scrub.Account(n.Account) } d, err := inv.PublicDomainOf(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } domains[n.Name] = scrub.Domain(d) } for _, o := range overlays { scrub.Site(o.Site) } // What a merge check runs in and reads beside it, as the build seat would be asked for it. if held, err := inv.Held(ctx); err == nil { for language, reference := range builder.ToolchainsOf(held) { if f.Versions.Toolchains == nil { f.Versions.Toolchains = map[string]string{} } f.Versions.Toolchains[language] = catalogue.Recorded(reference) } } else { return snapshot.Facts{}, err } for _, e := range entries { if dir, core := coreModules[e.Manifest.Module]; core && !e.Provided && e.Source.Repository != "" { for d, ref := range besideRefs(dir, current[e.Manifest.Module].Commit) { if f.Beside == nil { f.Beside = map[string]string{} } f.Beside[d] = ref } } } if c, ok := current["mesh-controller"]; ok { f.Controller.Commit = c.Commit } gens, gensErr := generators(ctx, open) hostShelf := shelf[hostModule] meshWide := map[string]bool{} engines := map[string]bool{} for _, n := range nodes { m := snapshot.Machine{Name: scrub.Machine(n.Name), Length: len(n.Name), Adopted: n.Adopted, AccountHome: scrub.Text(n.AccountHome), NodeEngine: n.HostVersion, PublicDomain: domains[n.Name]} switch n.Account { case "", "root": m.Account = n.Account default: m.Account = scrub.Account(n.Account) } if n.HostVersion != "" { engines[n.HostVersion] = true } m.System = systemOf(hostShelf, n.HostVersion) m.Libc = libcOf(m.System) reported, err := inv.DescribedOf(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } m.Architecture, m.Kernel = reported.Architecture, reported.Kernel outward, err := inv.OutwardLinksOf(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } for _, l := range outward { m.OutwardLinks = append(m.OutwardLinks, scrub.Text(l)) } capabilities, err := inv.Profile(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } for _, c := range capabilities { kept := snapshot.Capability{Name: c.Name, Present: c.Present} // The detail only where it is a version: everything else a detector says — a ruleset, a // device, a path — is the machine's own business and no check reads it. if c.Present && (c.Name == "container-runtime" || c.Name == "package-manager") { kept.Detail = scrub.Text(c.Detail) } m.Capabilities = append(m.Capabilities, kept) } if o, ok := place[n.Name]; ok { m.Site, m.Hub, m.Public, m.OnNetwork = scrub.Site(o.Site), o.Hub, o.Endpoint != "", o.Address != "" } assigned, err := inv.Assigned(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } m.Assigned = assigned sent, known, err := inv.SentBuilds(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } if known && slices.Contains(assigned, broker.RuntimeModule) { m.NodeTools = sent[broker.RuntimeModule] } pins, err := inv.PinsFor(ctx, n.Name) if err != nil { return snapshot.Facts{}, err } for provision, c := range pins { m.Pins = append(m.Pins, snapshot.Pin{Provision: provision, Machine: scrub.Machine(c.Node), Module: c.Module}) } for _, module := range assigned { held, err := inv.SecretsOf(ctx, n.Name, module) if err != nil { return snapshot.Facts{}, err } for _, h := range held { if h.Origin == inventory.OriginAccepted { m.Accepted = append(m.Accepted, snapshot.Accepted{Module: module, Name: h.Name, Provider: scrub.Machine(h.Provider), Local: h.Local}) } } layers, err := inv.SettingsFor(ctx, n.Name, module) if err != nil { return snapshot.Facts{}, err } for _, layer := range layers { if layer.From == catalogue.MeshWideLayer { if !meshWide[module] { meshWide[module] = true f.Settings = append(f.Settings, snapshot.Settings{Module: module, Values: scrub.Values(layer.Values)}) } continue } m.Settings = append(m.Settings, snapshot.Settings{Module: module, Values: scrub.Values(layer.Values)}) } } m.Declaration = declarationFacts(ctx, open, n.Name, gens, gensErr, scrub) if ctx.Err() != nil { return snapshot.Facts{}, ctx.Err() } f.Machines = append(f.Machines, m) } for e := range engines { f.Versions.NodeEngines = append(f.Versions.NodeEngines, e) } // Seats and their holders, and from them the roles a machine is named by. roles := map[string][]string{} seats := map[string]*snapshot.Seat{} for _, h := range holdings { key := h.Claim + "\x00" + h.Scope s, ok := seats[key] if !ok { s = &snapshot.Seat{Name: h.Claim, Scope: h.Scope} seats[key] = s } s.Holders = append(s.Holders, snapshot.Holder{Machine: scrub.Machine(h.Node), Module: h.Module}) if h.Scope == catalogue.ScopeMesh { role := "holds " + h.Claim if h.Claim == catalogue.ControllerSeatName { role = "the control node" } roles[h.Node] = append(roles[h.Node], role) } } for _, s := range seats { f.Seats = append(f.Seats, *s) } for i := range f.Machines { for _, n := range nodes { if scrub.Machine(n.Name) != f.Machines[i].Name { continue } f.Machines[i].Roles = roles[n.Name] if f.Machines[i].Hub { f.Machines[i].Roles = append(f.Machines[i].Roles, "the hub") } } } // Every module, as the mesh holds it, and where it is built from. newest := map[string]inventory.Source{} count := map[string]int{} for _, e := range entries { raw, err := json.Marshal(e.Manifest) if err != nil { return snapshot.Facts{}, err } mod := snapshot.Module{Name: e.Manifest.Module, Repository: snapshot.RepositoryName(e.Source.Repository), Path: e.Source.Path, Commit: e.Source.BuiltFrom, Provided: e.Provided, RollOut: current[e.Manifest.Module].RollOut, Manifest: raw} for _, r := range read[e.Manifest.Module] { mod.Reads = append(mod.Reads, snapshot.RepositoryName(r.Repository)) } f.Modules = append(f.Modules, mod) if e.Provided || e.Source.Repository == "" { continue } count[e.Source.Repository]++ if was, ok := newest[e.Source.Repository]; !ok || e.Source.Seen.After(was.Seen) { newest[e.Source.Repository] = e.Source } } for repository, s := range newest { commit := s.Head if commit == "" { commit = s.BuiltFrom } f.Sources = append(f.Sources, snapshot.Source{Repository: snapshot.RepositoryName(repository), Commit: commit, Modules: count[repository]}) } for _, e := range edges { f.Edges = append(f.Edges, snapshot.Edge{From: e.From, To: e.To, Kind: e.Kind}) } f.Sorted() return f, nil } // declarationFacts is how one machine's declaration composes now, as the next push would compose it and // without making anything (D1's composition), and whether the node-engine's validator takes it. func declarationFacts(ctx context.Context, open *stores, node string, gens map[string]catalogue.Generator, gensErr error, scrub *snapshot.Scrubber) snapshot.Declaration { var d snapshot.Declaration if gensErr != nil { d.Problems = []string{scrub.Text("the private network cannot be computed: " + oneLine(gensErr.Error()))} return d } declared, problems, err := composedAndValidated(ctx, open, node, gens, Foreseeing) if err != nil { d.Problems = []string{scrub.Text(oneLine(err.Error()))} return d } for _, p := range problems { d.Problems = append(d.Problems, scrub.Text(p)) } d.Composes = len(problems) == 0 if body, err := declared.Body(); err == nil { d.Digest = fmt.Sprintf("sha256:%x", sha256.Sum256(body)) } // Scrubbed like every other word: a resource is named after the machine a grant is for. for _, r := range resourceNames(declared.Resources) { d.Resources = append(d.Resources, scrub.Text(r)) } for module, why := range declared.leftOutWhy { if d.LeftOut == nil { d.LeftOut = map[string]string{} } d.LeftOut[module] = scrub.Text(why) } for _, o := range declared.withheld { d.Withheld = append(d.Withheld, scrub.Text(o.String())) } for _, u := range declared.unbound { d.Unbound = append(d.Unbound, scrub.Text(u.String())) } sort.Strings(d.Withheld) sort.Strings(d.Unbound) return d } // composedAndValidated composes one machine's declaration — as a push would (Allocating) or as the next // push will without making anything (Foreseeing) — with the order it was last sent, and runs the // node-engine's own validator over the body. An error is that it did not compose; problems are what the // validator refuses. func composedAndValidated(ctx context.Context, open *stores, node string, gens map[string]catalogue.Generator, choosing Choosing) (sendable, []string, error) { plan, settings, err := planFor(ctx, open, node) if err != nil { return sendable{}, nil, err } declared, err := declarationWith(ctx, open, node, plan, settings, gens, choosing) if err != nil { return sendable{}, nil, err } record, err := open.inventory.NodeByName(ctx, node) if err != nil { return sendable{}, nil, err } if declared.Sequence, err = open.inventory.Sequence(ctx, record.ID); err != nil { return sendable{}, nil, err } if declared.Epoch, err = open.inventory.SentEpoch(ctx, record.ID); err != nil { return sendable{}, nil, err } body, err := declared.Body() if err != nil { return sendable{}, nil, err } return declared, validate.Declaration(body), nil } // resourceNames are a declaration's resources as `type:id`, sorted. func resourceNames(resources []map[string]any) []string { out := make([]string, 0, len(resources)) for _, r := range resources { kind, _ := r["type"].(string) id, _ := r["id"].(string) out = append(out, kind+":"+id) } sort.Strings(out) return out } // systemOf is the system a node-engine of that version was built for, read from the build the mesh // holds: the artifact a resource delivered into `versions/` came from is named for its system // (`host-arch`). Empty when the version is not a delivered one — an engine placed by hand. func systemOf(host catalogue.Manifest, version string) string { if version == "" { return "" } for _, r := range host.Resources { path, _ := r["path"].(string) if !strings.HasSuffix(path, "/versions/"+version) { continue } source, _ := r["source"].(string) for _, part := range strings.Split(source, "/") { if system, ok := strings.CutPrefix(part, "host-"); ok && system != "" { return system } } } return "" } // libcOf is the C library of a system the node-engine is built for. func libcOf(system string) string { switch system { case "arch": return "glibc" case "alpine": return "musl" case "android": return "bionic" } return "" } // factsCommand is `facts`: what the controller keeps, and keeping it now. // // facts the snapshot the artifact store holds: when, which, how many machines // facts show the same, whole, as JSON // facts export compose and keep one now // facts compose compose one and print it, keeping nothing func factsCommand(ctx context.Context, args []string) error { set := flag.NewFlagSet("facts", flag.ContinueOnError) rest, err := parseAround(set, args) if err != nil { return err } what := "" if len(rest) > 0 { what = rest[0] } open, err := openStores(ctx) if err != nil { return err } defer open.Close() switch what { case "", "show": address, err := factsStore(ctx, open) if err != nil { return err } body, digest, err := (artifacts.Store{Address: address}).GetTagged(ctx, snapshot.Repository, snapshot.Tag) if err != nil { return err } if what == "show" { _, err := os.Stdout.Write(body) return err } f, err := snapshot.Decode(body) if err != nil { return err } fmt.Printf("the facts snapshot kept as %s:%s is %s, taken %s (%s ago) by controller %s\n", snapshot.Repository, snapshot.Tag, short(strings.TrimPrefix(digest, "sha256:")), f.Taken.Format(time.RFC3339), ago(time.Since(f.Taken)), orNone(f.Controller.Commit)) fmt.Printf(" %d machine(s), %d module(s), %d seat(s); the longest machine name is %d characters\n", len(f.Machines), len(f.Modules), len(f.Seats), f.Longest()) fmt.Printf(" the bus runs %s, the store %s\n", orNone(f.Versions.Bus), orNone(f.Versions.Store)) for _, m := range f.Machines { state := "composes" if !m.Declaration.Composes { state = "does NOT compose: " + strings.Join(m.Declaration.Problems, "; ") } fmt.Printf(" %-12s %s; %d module(s); %s\n", m.Name, m.Described(), len(m.Assigned), state) } return nil case "export", "compose": busVersion := "" if server, err := connectLink(ctx, nil, nil, nil); err == nil { busVersion = busVersionOf(server) server.Close() } if what == "compose" { f, err := gatherFacts(ctx, open, busVersion) if err != nil { return err } body, err := f.Encode() if err != nil { return err } _, err = os.Stdout.Write(append(body, '\n')) return err } digest, err := exportFacts(ctx, open, busVersion, true) if err != nil { return err } fmt.Printf("the facts snapshot is kept as %s:%s, %s\n", snapshot.Repository, snapshot.Tag, digest) return nil } return fmt.Errorf("facts [show|export|compose], not %q", what) } // busVersionOf is the bus server's release, as it told this connection. func busVersionOf(server *link.Server) string { if bus, ok := server.Bus().(link.OverNATS); ok && bus.Conn != nil { return bus.Conn.ConnectedServerVersion() } return "" }