diff --git a/modules/mesh-provisioner/.gitignore b/modules/mesh-provisioner/.gitignore new file mode 100644 index 00000000..b51e2c5c --- /dev/null +++ b/modules/mesh-provisioner/.gitignore @@ -0,0 +1,2 @@ +/mesh-provisioner +/cmd/mesh-provisioner/mesh-provisioner diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/busread.go b/modules/mesh-provisioner/cmd/mesh-provisioner/busread.go new file mode 100644 index 00000000..5de3c58c --- /dev/null +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/busread.go @@ -0,0 +1,204 @@ +package main + +// The bus as the holder composes from it (frombus.Reader), read through this module's runtime with its own `reads`: +// placement, each node's record in its own have key, the indexer's modules and manifests, the builder's builds and +// mesh-config's resolved values. A plan names no value, so no own secret and no credential is read (compose.Planning). + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net/http" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// busReader reads the bus for frombus through a Store. +type busReader struct { + s Store + fetch func(ctx context.Context, url string) ([]byte, error) +} + +func (r busReader) each(read string, visit func(key string, value []byte)) error { + keys, err := r.s.Keys(read) + if err != nil { + return fmt.Errorf("%s: %w", read, err) + } + for _, k := range keys { + e, err := r.s.Get(read, k) + if err != nil { + return fmt.Errorf("%s %s: %w", read, k, err) + } + if e != nil { + visit(k, e.Value) + } + } + return nil +} + +func (r busReader) Placements(context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + err := r.each(PlacementRead, func(k string, v []byte) { + node, err := delivery.ParsePlacementKey(k) + if err != nil { + return + } + if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node { + out[node] = p + } + }) + return out, err +} + +func (r busReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + out := map[string]delivery.NodeRecord{} + err := r.each(HaveRead, func(k string, v []byte) { + node, _, isNode, err := delivery.ParseHaveKey(k) + if err != nil || !isNode { + return + } + if h, err := delivery.DecodeHaveNode(v); err == nil && h.Record != nil { + out[node] = *h.Record + } + }) + return out, err +} + +func (r busReader) Modules(context.Context) (map[string]delivery.Module, error) { + out := map[string]delivery.Module{} + err := r.each(ModulesRead, func(k string, v []byte) { + module, err := delivery.ParseModulesKey(k) + if err != nil { + return + } + if m, err := delivery.DecodeModule(v); err == nil { + out[module] = m + } + }) + return out, err +} + +func (r busReader) Manifest(_ context.Context, digest string) ([]byte, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return nil, err + } + e, err := r.s.Get(ManifestsRead, key) + if err != nil { + return nil, err + } + if e == nil { + return nil, fmt.Errorf("the indexer holds no manifest %s", digest) + } + m, err := delivery.DecodeManifest(e.Value) + if err != nil { + return nil, err + } + return []byte(m.JSON), nil +} + +func (r busReader) Values(_ context.Context, node, module string) (map[string]any, error) { + var out map[string]any + prefix := node + "." + module + "." + err := r.each(ResolvedRead, func(k string, v []byte) { + name, ok := strings.CutPrefix(k, prefix) + if !ok || strings.Contains(name, ".") { + return + } + if value, laid := catalogue.LaidValue(v); laid { + if out == nil { + out = map[string]any{} + } + out[name] = value + } + }) + return out, err +} + +func (busReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { return nil, nil } + +func (busReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil +} + +func (r busReader) Build(_ context.Context, module, fp string) (*delivery.BuilderBuild, error) { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return nil, err + } + e, err := r.s.Get(BuildsRead, key) + if err != nil || e == nil { + return nil, err + } + b, err := delivery.DecodeBuilderBuild(e.Value) + if err != nil { + return nil, err + } + return &b, nil +} + +// archiveWithin bounds one read of a templates archive. +const archiveWithin = 30 * time.Second + +func (r busReader) Archive(ctx context.Context, url, digest string) ([]byte, error) { + if r.fetch == nil { + return nil, fmt.Errorf("this holder reads no archive (%s)", url) + } + fetching, cancel := context.WithTimeout(ctx, archiveWithin) + defer cancel() + body, err := r.fetch(fetching, url) + if err != nil { + return nil, err + } + sum := sha256.Sum256(body) + if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest { + return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got) + } + return body, nil +} + +func (r busReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) { + all, err := r.Values(ctx, node, module) + var out map[string]any + for _, k := range keys { + if v, ok := all[k]; ok { + if out == nil { + out = map[string]any{} + } + out[k] = v + } + } + return out, err +} + +// httpFetch reads an archive from the artifact store over HTTP. +func httpFetch(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%s answered %s", url, resp.Status) + } + return io.ReadAll(io.LimitReader(resp.Body, 256<<20)) +} + +// FromBus is the holder's composer over the bus: frombus's plans of a node, at the builds keptAt names. +func FromBus(s Store) Composer { + r := busReader{s: s, fetch: httpFetch} + return func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, error) { + return frombus.Plans(ctx, r, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt}) + } +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go new file mode 100644 index 00000000..04c2fb3e --- /dev/null +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go @@ -0,0 +1,617 @@ +package main + +// The seat's holder (novox/hq ADR 0570 clauses 1 to 5, 7; issue 1954 H1-H3): mesh-provisioner composes every placed +// module's provisioning plan and every provision from bus state, on change events and never on a timer, and keeps them +// in its states `plans` (`.`) and `provisions` (`..[.]`), which it alone +// writes. +// +// - **Composed per node.** A change of a node's placement or of one of its wants composes that node; a change of the +// indexer's module composes the nodes the module is placed on. Each node is composed twice: at the builds its wants +// name (its plans, what its client applies) and at the newest build of each placed module, so a build that adds a +// requirement has its key before its want (clause 2). A provision is needed while either needs it (clause 5). +// - **A key's state comes from its provider's outcome alone** (clause 3, 7): the provider's own entry for that +// consumer, `_bound` under the same key: present is provided (the vault's generation carried into the +// reference, clause 6), refused is failed with why, waits is waits with what. The holder never sets provided by +// itself, but for a provider that keeps no entries at all (its manifest declares no `bound`), which makes nothing per +// consumer. +// - **Retiring is asked** (clause 5): a key no longer needed is `retiring`, one approve ask under mesh-provisioner's +// name says what goes from which provider on which node; the warrant goes into the key and the provider acts on it; +// a decline is kept and not asked again while nothing changes; a key needed again returns to its outcome's state. + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provisions" +) + +// The states it keeps, beside Registry, Needs, Applied and Collect. +const ( + ProvisionsState = provisions.ProvisionsState + PlansState = provisions.PlansState +) + +// The reads it composes from. +const ( + PlacementRead = delivery.PlacementRead + WantsRead = delivery.AssignerModule + ".wants" + ModulesRead = delivery.IndexerModule + ".modules" + ManifestsRead = delivery.IndexerModule + ".manifests" + BuildsRead = delivery.BuilderModule + ".builds" + HaveRead = "node-engine.have" + ResolvedRead = "mesh-config.resolved" +) + +// Providers are the providing modules that keep an entry per consumer, `_bound`, read as `.bound`: the +// outcome of each provision they make. A provider not named here keeps none. +var Providers = []string{"gitea", "keycloak", "postgres", "route-adapter", "mesh-vault"} + +// boundRead is a provider's entries as this module reads them. +func boundRead(module string) string { return module + ".bound" } + +// Composer is each module's provisioning plan on a node, composed from the bus: at the builds keptAt names +// (module → fingerprint), the newest build of every other placed module. +type Composer func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, error) + +// Asker is the operator's channel as this module asks on it: publish an ask under its own name, and read the router's +// record of it. +type Asker interface { + Publish(subject string, body any, id string) error + Record(id string) ([]byte, bool, error) +} + +// Holder is the seat's holder. +type Holder struct { + Store Store + Compose Composer + Asks Asker + Now func() time.Time + Logf func(format string, args ...any) + + mu sync.Mutex +} + +// Composed is what one node's composition did. +type Composed struct { + Node string `json:"node"` + Plans []string `json:"plans-written,omitempty"` + Gone []string `json:"plans-removed,omitempty"` + Changed []string `json:"provisions-changed,omitempty"` + Asked []string `json:"retiring-asked,omitempty"` + Problems []string `json:"problems,omitempty"` +} + +// ComposeNode composes one node: its plans and its provisions. +func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error) { + h.mu.Lock() + defer h.mu.Unlock() + out := Composed{Node: node} + wanted, err := h.wantedOn(node) + if err != nil { + return out, err + } + atWants, err := h.Compose(ctx, node, wanted) + if err != nil { + return out, fmt.Errorf("the plans of %s cannot be composed: %w", node, err) + } + newest, err := h.Compose(ctx, node, nil) + if err != nil { + return out, fmt.Errorf("the plans of %s at the newest builds cannot be composed: %w", node, err) + } + if err := h.writePlans(node, atWants, &out); err != nil { + out.Problems = append(out.Problems, err.Error()) + } + needed := map[string]provisions.Provision{} + for _, plans := range []map[string]catalogue.Plan{atWants, newest} { + for module, plan := range plans { + for _, p := range plan.Provisions { + pv := provisions.Provision{Provision: p.Provision, Local: p.Local, + Consumer: provisions.End{Module: module, Node: node}, + Provider: provisions.End{Module: p.ProviderModule, Node: p.Provider}} + key, err := pv.Key() + if err != nil { + out.Problems = append(out.Problems, err.Error()) + continue + } + if _, seen := needed[key]; !seen { + needed[key] = pv + } + } + } + } + h.judgeNode(node, needed, &out) + return out, nil +} + +// wantedOn is the fingerprint each want on a node names, by module. +func (h *Holder) wantedOn(node string) (map[string]string, error) { + out := map[string]string{} + keys, err := h.Store.Keys(WantsRead) + if err != nil { + return nil, fmt.Errorf("the wants cannot be read: %w", err) + } + for _, k := range keys { + n, module, err := delivery.ParseWantKey(k) + if err != nil || n != node { + continue + } + e, err := h.Store.Get(WantsRead, k) + if err != nil || e == nil { + continue + } + if w, err := delivery.DecodeWant(e.Value); err == nil && !w.Removed && w.FP != "" { + out[module] = w.FP + } + } + return out, nil +} + +// writePlans writes each plan that differs and removes the node's plans of modules no longer composed there. +func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, out *Composed) error { + var problems []error + for _, module := range sortedKeys(plans) { + key, err := provisions.PlanKey(node, module) + if err != nil { + problems = append(problems, err) + continue + } + raw, err := provisions.EncodePlan(provisions.Plan{Node: node, Module: module, Plan: plans[module]}) + if err != nil { + problems = append(problems, err) + continue + } + if e, err := h.Store.Get(PlansState, key); err == nil && e != nil && bytes.Equal(e.Value, raw) { + continue + } + if err := h.Store.Put(PlansState, key, json.RawMessage(raw)); err != nil { + problems = append(problems, fmt.Errorf("the plan %s: %w", key, err)) + continue + } + out.Plans = append(out.Plans, key) + } + keys, err := h.Store.Keys(PlansState) + if err != nil { + return errors.Join(append(problems, err)...) + } + for _, k := range keys { + n, module, ok := strings.Cut(k, ".") + if !ok || n != node { + continue + } + if _, still := plans[module]; still { + continue + } + if err := h.Store.Delete(PlansState, k); err != nil { + problems = append(problems, err) + continue + } + out.Gone = append(out.Gone, k) + } + return errors.Join(problems...) +} + +// judgeNode brings the node's provision keys to what is needed and what each provider said. +func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, out *Composed) { + held := h.held(node) + for _, key := range unionKeys(needed, held) { + want, isNeeded := needed[key] + have, isHeld := held[key] + var next provisions.Provision + switch { + case isNeeded: + next = want + if isHeld { + next.Reference = have.Reference + } + h.outcome(key, &next) + case have.State == provisions.Retiring: + next = have + if next.Ask == "" && next.Warrant == nil { + if err := h.ask(&next, key); err != nil { + out.Problems = append(out.Problems, err.Error()) + } else { + out.Asked = append(out.Asked, key) + } + } + if gone, why := h.retired(key, have); gone { + if err := h.Store.Delete(ProvisionsState, key); err != nil { + out.Problems = append(out.Problems, err.Error()) + } else { + out.Changed = append(out.Changed, key+" (retired: "+why+")") + } + continue + } + default: + next = have + next.State, next.Why = provisions.Retiring, fmt.Sprintf("no build wanted or placed on %s requires %s of %s on %s "+ + "any more", node, have.Provision, have.Provider.Module, have.Provider.Node) + next.Warrant = nil + if err := h.ask(&next, key); err != nil { + out.Problems = append(out.Problems, err.Error()) + } else { + out.Asked = append(out.Asked, key) + } + } + if isHeld && same(have, next) { + continue + } + next.At = h.Now().UTC() + raw, err := provisions.Encode(next) + if err != nil { + out.Problems = append(out.Problems, err.Error()) + continue + } + if err := h.Store.Put(ProvisionsState, key, json.RawMessage(raw)); err != nil { + out.Problems = append(out.Problems, fmt.Sprintf("%s: %v", key, err)) + continue + } + out.Changed = append(out.Changed, key+" "+next.State) + } +} + +// held is the node's provision keys as kept. +func (h *Holder) held(node string) map[string]provisions.Provision { + out := map[string]provisions.Provision{} + keys, err := h.Store.Keys(ProvisionsState) + if err != nil { + return out + } + for _, k := range keys { + if !strings.HasPrefix(k, node+".") { + continue + } + e, err := h.Store.Get(ProvisionsState, k) + if err != nil || e == nil { + continue + } + if p, err := provisions.Decode(e.Value); err == nil { + out[k] = p + } + } + return out +} + +// same says two provisions say the same, their time aside. +func same(a, b provisions.Provision) bool { + a.At, b.At, a.V, b.V = time.Time{}, time.Time{}, 0, 0 + ra, _ := json.Marshal(a) + rb, _ := json.Marshal(b) + return bytes.Equal(ra, rb) +} + +// entry is a provider's outcome for one consumer, as its entry says it: the provider helper's Entry, read for what +// the holder needs and nothing more. +type entry struct { + Refused string `json:"refused,omitempty"` + Waits string `json:"waits,omitempty"` + VaultName string `json:"vault-name,omitempty"` + Generation int `json:"generation,omitempty"` +} + +// outcome sets a needed provision's state from its provider's entry: provided, failed or waits; asked while it has +// none. A provider that keeps no entries makes nothing per consumer: provided, said so. +func (h *Holder) outcome(key string, p *provisions.Provision) { + p.Ask, p.Warrant = "", nil + if !keepsEntries(p.Provider.Module) { + p.State, p.Why = provisions.Provided, p.Provider.Module+" makes nothing per consumer: what it provides is its "+ + "placement and served values" + return + } + e, err := h.Store.Get(boundRead(p.Provider.Module), key) + if err != nil { + p.State, p.Why = provisions.Asked, "the provider's entries cannot be read: "+err.Error() + return + } + if e == nil { + p.State, p.Why = provisions.Asked, "" + return + } + var en entry + if err := json.Unmarshal(e.Value, &en); err != nil { + p.State, p.Why = provisions.Asked, "the provider's entry is not readable" + return + } + switch { + case en.Refused != "": + p.State, p.Why = provisions.Failed, en.Refused + case en.Waits != "": + p.State, p.Why = provisions.Waits, en.Waits + default: + p.State, p.Why = provisions.Provided, "" + if en.VaultName != "" || en.Generation > 0 { + p.Reference = &provisions.Reference{ID: en.VaultName, Generation: en.Generation} + } + } +} + +func keepsEntries(module string) bool { + for _, m := range Providers { + if m == module { + return true + } + } + return false +} + +// retired says a retiring key the operator approved is retired: its provider's entry is gone. +func (h *Holder) retired(key string, p provisions.Provision) (bool, string) { + if p.Warrant == nil || p.Warrant.Option != OptionApprove { + return false, "" + } + if !keepsEntries(p.Provider.Module) { + return true, "its provider keeps nothing for it" + } + e, err := h.Store.Get(boundRead(p.Provider.Module), key) + if err != nil || e != nil { + return false, "" + } + return true, "its provider removed its credential on the operator's warrant" +} + +// ---- the retiring ask ---------------------------------------------------------------------------------------------- + +// The options a retiring ask offers. +const ( + OptionApprove = "approve" + OptionDecline = "decline" +) + +// askLasts is how long a retiring ask stays open. +const askLasts = asks.ApproveLasts - 10*time.Minute + +// retiringAsk is the ask of one retiring provision, composed from the key alone. +func retiringAsk(key string, p provisions.Provision, at time.Time, id string) asks.Ask { + what := fmt.Sprintf("%s's access to %s on %s, for %s on %s", p.Consumer.Module, p.Provision, p.Provider.Node, + p.Consumer.Module, p.Consumer.Node) + a := asks.Ask{ID: id, Headline: "Remove " + p.Consumer.Module + "'s " + p.Provision + " access?", + Explanation: fmt.Sprintf("Nothing wanted or placed on %s needs %s any more. On Approve, %s removes %s: what it owns "+ + "is handed to the resource's owner and its credential goes; the resource and its data stay.", p.Consumer.Node, + p.Provision, p.Provider.Module, what), + Who: asks.Operator, Expires: at.Add(askLasts), OnExpiry: "nothing is removed; it is asked again when it changes", + About: "provision." + key, Details: "The provision " + key + " of mesh-provisioner's state provisions."} + for _, o := range []struct{ id, label, does string }{ + {OptionApprove, "Remove", "the provider removes the credential; the resource and its data stay"}, + {OptionDecline, "Keep", "nothing is removed, and it is not asked again while nothing changes"}, + } { + binds, _ := asks.ActDigest(retireAct(key, id, o.id)) + a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds}) + } + return a +} + +func retireAct(key, id, option string) asks.Act { + return asks.Act{"verb": "mesh-provisioner.retire", "provision": key, "ask": id, "option": option} +} + +// ask raises the retiring ask of a key, under mesh-provisioner's own name; one already raised stands. +func (h *Holder) ask(p *provisions.Provision, key string) error { + if p.Ask != "" { + return nil + } + if h.Asks == nil { + return errors.New("the operator cannot be asked from here, so " + key + " is retiring and nothing is removed") + } + now := h.Now().UTC() + id := "retire-" + strings.ReplaceAll(key, ".", "-") + "-" + fmt.Sprint(now.Unix()) + a := retiringAsk(key, *p, now, id) + if err := a.Check(now); err != nil { + return err + } + if err := h.Asks.Publish(asks.AskSubject(Module), a, id); err != nil { + return fmt.Errorf("the retiring ask of %s could not be published: %w", key, err) + } + p.Ask = id + return nil +} + +// Decide reads the router's record of each retiring key's ask and writes the warrant into the key (clause 5): the +// provider acts on an Approve; a Decline is kept, and not asked again while nothing changes. +func (h *Holder) Decide() []string { + h.mu.Lock() + defer h.mu.Unlock() + if h.Asks == nil { + return nil + } + keys, err := h.Store.Keys(ProvisionsState) + if err != nil { + return []string{"the provisions cannot be read: " + err.Error()} + } + var said []string + for _, k := range keys { + e, err := h.Store.Get(ProvisionsState, k) + if err != nil || e == nil { + continue + } + p, err := provisions.Decode(e.Value) + if err != nil || p.State != provisions.Retiring || p.Ask == "" || p.Warrant != nil { + continue + } + raw, found, err := h.Asks.Record(p.Ask) + if err != nil || !found { + continue + } + var rec struct { + State string `json:"state"` + Ask *asks.Ask `json:"ask"` + Warrant *asks.Warrant `json:"warrant"` + } + if json.Unmarshal(raw, &rec) != nil || rec.State == "open" || rec.Warrant == nil || rec.Ask == nil { + continue + } + w := *rec.Warrant + if w.Asker != Module || w.Ask != p.Ask { + continue + } + option := "" + if w.Outcome == asks.OutcomeChosen { + o, err := w.For(Module, *rec.Ask) + if err == nil { + err = o.Performs(retireAct(k, p.Ask, o.ID)) + } + if err != nil { + said = append(said, k+": the answer is refused: "+err.Error()) + continue + } + option = o.ID + } else { + // Ended with no choice: asked again at the next change. + p.Ask = "" + h.put(k, p) + said = append(said, k+": the ask "+string(w.Outcome)+"; asked again when it changes") + continue + } + by := "the operator" + if w.By != nil { + by = fmt.Sprintf("the %s, as %s identity %s", w.By.Who, w.By.Kind, w.By.Identity) + } + p.Warrant = &provisions.Warrant{Ask: p.Ask, Option: option, By: by, At: w.At.UTC()} + h.put(k, p) + said = append(said, k+": "+option+" by "+by) + } + return said +} + +func (h *Holder) put(key string, p provisions.Provision) { + p.At = h.Now().UTC() + if raw, err := provisions.Encode(p); err == nil { + if err := h.Store.Put(ProvisionsState, key, json.RawMessage(raw)); err != nil { + h.Logf("[%s] %s is not written: %v", Module, key, err) + } + } +} + +// ---- what a change touches -------------------------------------------------------------------------------------- + +// NodesOfPlacement is the node a placement key names. +func NodesOfPlacement(key string) []string { + if n, err := delivery.ParsePlacementKey(key); err == nil { + return []string{n} + } + return nil +} + +// NodesOfWant is the node a want key names. +func NodesOfWant(key string) []string { + if n, _, err := delivery.ParseWantKey(key); err == nil { + return []string{n} + } + return nil +} + +// NodesOfModule is every node a module is placed on, and every node whose plans name it as a provider: what a change +// of the module's manifest touches. +func (h *Holder) NodesOfModule(module string) []string { + nodes := map[string]bool{} + if keys, err := h.Store.Keys(PlacementRead); err == nil { + for _, k := range keys { + n, err := delivery.ParsePlacementKey(k) + if err != nil { + continue + } + e, err := h.Store.Get(PlacementRead, k) + if err != nil || e == nil { + continue + } + if p, err := delivery.DecodePlacement(e.Value); err == nil { + for _, m := range p.Modules { + if m.Module == module { + nodes[n] = true + } + } + } + } + } + if keys, err := h.Store.Keys(ProvisionsState); err == nil { + for _, k := range keys { + if e, err := h.Store.Get(ProvisionsState, k); err == nil && e != nil { + if p, err := provisions.Decode(e.Value); err == nil && p.Provider.Module == module { + nodes[p.Consumer.Node] = true + } + } + } + } + return sortedKeys(nodes) +} + +// NodeOfEntry is the consumer's node a provider's entry key names. +func NodeOfEntry(key string) []string { + if n, _, _, _, err := provisions.ParseKey(key); err == nil { + return []string{n} + } + return nil +} + +func sortedKeys[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +func unionKeys[A, B any](a map[string]A, b map[string]B) []string { + seen := map[string]bool{} + for k := range a { + seen[k] = true + } + for k := range b { + seen[k] = true + } + return sortedKeys(seen) +} + +// Rejudge sets one key's state again from its provider's outcome: what a change of a provider's entry touches. A +// retiring key whose provider removed its entry on the operator's warrant goes. +func (h *Holder) Rejudge(key string) (string, error) { + h.mu.Lock() + defer h.mu.Unlock() + e, err := h.Store.Get(ProvisionsState, key) + if err != nil || e == nil { + return "", err + } + have, err := provisions.Decode(e.Value) + if err != nil { + return "", err + } + if have.State == provisions.Retiring { + if gone, why := h.retired(key, have); gone { + return key + " retired: " + why, h.Store.Delete(ProvisionsState, key) + } + return "", nil + } + next := have + h.outcome(key, &next) + if same(have, next) { + return "", nil + } + h.put(key, next) + return key + " " + next.State, nil +} + +// RetireNeeds deletes every key of the retired state `needs`, which the clients no longer write (clause 1): run once at +// the start. +func (h *Holder) RetireNeeds() (int, error) { + keys, err := h.Store.Keys(Needs) + if err != nil { + return 0, err + } + n := 0 + for _, k := range keys { + if err := h.Store.Delete(Needs, k); err != nil { + return n, err + } + n++ + } + return n, nil +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go new file mode 100644 index 00000000..1164f728 --- /dev/null +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go @@ -0,0 +1,257 @@ +package main + +import ( + "context" + "encoding/json" + "go/ast" + "go/parser" + "go/token" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/provisions" +) + +// router is the operator's channel as the holder asks on it. +type router struct { + asked map[string]asks.Ask + records map[string][]byte +} + +func (r *router) Publish(_ string, body any, id string) error { + r.asked[id] = body.(asks.Ask) + return nil +} + +func (r *router) Record(id string) ([]byte, bool, error) { + raw, ok := r.records[id] + return raw, ok, nil +} + +// answer is the operator's answer to an ask. +func (r *router) answer(t *testing.T, id, option string) { + t.Helper() + a := r.asked[id] + o, ok := a.Option(option) + if !ok { + t.Fatalf("the ask %s offers no %s", id, option) + } + w := asks.Warrant{Ask: id, Asker: Module, About: a.About, Outcome: asks.OutcomeChosen, Option: o.ID, Label: o.Label, + Level: o.Level, By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42"}, Channel: "telegram", + At: time.Date(2026, 10, 29, 23, 0, 0, 0, time.UTC), AskDigest: a.Digest()} + r.records[id], _ = json.Marshal(map[string]any{"state": "closed", "ask": a, "warrant": w}) +} + +// mesh is the holder over memory, its composer answering each node from tables the test sets: the plans at the wanted +// builds and at the newest. +type mesh struct { + h *Holder + s *memStore + r *router + atWant map[string]map[string]catalogue.Plan + atNewest map[string]map[string]catalogue.Plan + composed []string +} + +func newMesh() *mesh { + m := &mesh{s: newStore(), r: &router{asked: map[string]asks.Ask{}, records: map[string][]byte{}}, + atWant: map[string]map[string]catalogue.Plan{}, atNewest: map[string]map[string]catalogue.Plan{}} + now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC) + m.h = &Holder{Store: m.s, Asks: m.r, Logf: func(string, ...any) {}, + Now: func() time.Time { now = now.Add(time.Second); return now }, + Compose: func(_ context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, error) { + m.composed = append(m.composed, node) + if keptAt != nil { + return m.atWant[node], nil + } + return m.atNewest[node], nil + }} + return m +} + +// needs makes a module on a node require provisions in its wanted build (and the same in its newest unless set). +func planOf(provs ...catalogue.PlanProvision) catalogue.Plan { + return catalogue.Plan{Provisions: provs} +} + +var ( + pgDB = catalogue.PlanProvision{Provision: "postgres-database", Provider: "ace", ProviderModule: "postgres"} + oidc = catalogue.PlanProvision{Provision: "oidc-client", Provider: "novox", ProviderModule: "keycloak"} + split = catalogue.PlanProvision{Provision: "split-dns", Provider: "g14", ProviderModule: "systemd-resolved"} +) + +func (m *mesh) state(t *testing.T, key string) provisions.Provision { + t.Helper() + e, _ := m.s.Get(ProvisionsState, key) + if e == nil { + return provisions.Provision{} + } + p, err := provisions.Decode(e.Value) + if err != nil { + t.Fatal(err) + } + return p +} + +func (m *mesh) compose(t *testing.T, node string) Composed { + t.Helper() + c, err := m.h.ComposeNode(context.Background(), node) + if err != nil { + t.Fatal(err) + } + return c +} + +// A node composes its plans and provisions from what is wanted and placed; a key's state follows its provider's entry +// alone; a provider that keeps no entries makes nothing per consumer; the vault's generation reaches the reference; a +// quiet pass writes nothing (ADR 0570 clauses 1-3, 6, 7; issue 1954 H1, H2). +func TestAKeysStateFollowsItsProvidersOutcome(t *testing.T) { + m := newMesh() + m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB, split)} + m.atNewest["g14"] = m.atWant["g14"] + key := "g14.gitea.postgres-database" + steps := []struct { + name string + entry string // the provider's entry written before the step, "-" to delete + state string + why string + }{ + {"composed: asked", "", provisions.Asked, ""}, + {"the provider waits for a person", `{"waits":"the database's owner to accept"}`, provisions.Waits, "owner"}, + {"the provider refused", `{"refused":"no room on ace"}`, provisions.Failed, "no room"}, + {"the provider made it", `{"vault-name":"pair/gitea-postgres","generation":3}`, provisions.Provided, ""}, + } + for _, s := range steps { + if s.entry != "" { + m.s.write(t, "postgres.bound", key, s.entry) + } + m.compose(t, "g14") + p := m.state(t, key) + if p.State != s.state || !strings.Contains(p.Why, s.why) || p.Provider.Module != "postgres" || p.Provider.Node != "ace" { + t.Errorf("%s: %+v", s.name, p) + } + } + if p := m.state(t, key); p.Reference == nil || p.Reference.ID != "pair/gitea-postgres" || p.Reference.Generation != 3 { + t.Errorf("the vault's generation does not reach the reference: %+v", p.Reference) + } + if p := m.state(t, "g14.gitea.split-dns"); p.State != provisions.Provided || !strings.Contains(p.Why, "makes nothing") { + t.Errorf("a provider keeping no entries: %+v", p) + } + if e, _ := m.s.Get(PlansState, "g14.gitea"); e == nil { + t.Error("the plan is not written") + } + if c := m.compose(t, "g14"); len(c.Plans)+len(c.Changed)+len(c.Asked) != 0 { + t.Errorf("a quiet pass wrote %+v", c) + } + // An entry's change judges its key alone, no composition. + before := len(m.composed) + m.s.write(t, "postgres.bound", key, `{"refused":"the login is locked"}`) + if said, err := m.h.Rejudge(key); err != nil || !strings.Contains(said, provisions.Failed) || len(m.composed) != before { + t.Errorf("rejudged %q, %v, composed %d more", said, err, len(m.composed)-before) + } +} + +// A provision no build wanted or placed needs is retiring with one ask naming provider, node and what goes; one composed +// ahead of its want raises none; a change removing one requirement and adding another retires only the removed one; the +// warrant goes into the key and the key goes once the provider removed its entry; a decline is kept and not asked again; +// a provision needed again returns from retiring (ADR 0570 clause 5; issue 1954 H3). +func TestRetiringIsAskedAndActsOnlyOnTheWarrant(t *testing.T) { + m := newMesh() + pg, kc := "g14.gitea.postgres-database", "g14.gitea.oidc-client" + m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB)} + m.atNewest["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB, oidc)} // the newest build adds oidc ahead of its want + m.s.write(t, "postgres.bound", pg, `{}`) + steps := []struct { + name string + change func() + asked []string + states map[string]string + removed bool + }{ + {"a key composed ahead of its want is asked, not retiring", func() {}, nil, + map[string]string{pg: provisions.Provided, kc: provisions.Asked}, false}, + {"one requirement removed and another added: only the removed retires", func() { + m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)} + m.atNewest["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)} + }, []string{pg}, map[string]string{pg: provisions.Retiring, kc: provisions.Asked}, false}, + {"composed again: no second ask", func() {}, nil, map[string]string{pg: provisions.Retiring}, false}, + {"needed again before the warrant: back to its outcome", func() { + m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc, pgDB)} + }, nil, map[string]string{pg: provisions.Provided}, false}, + {"gone again: asked again", func() { m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(oidc)} }, + []string{pg}, map[string]string{pg: provisions.Retiring}, false}, + } + for _, s := range steps { + s.change() + c := m.compose(t, "g14") + if !slices.Equal(c.Asked, s.asked) { + t.Errorf("%s: asked %v, want %v", s.name, c.Asked, s.asked) + } + for k, st := range s.states { + if p := m.state(t, k); p.State != st { + t.Errorf("%s: %s is %s, want %s (%s)", s.name, k, p.State, st, p.Why) + } + } + } + p := m.state(t, pg) + a := m.r.asked[p.Ask] + if !strings.Contains(a.Explanation, "postgres") || !strings.Contains(a.Explanation, "ace") || + !strings.Contains(a.Explanation, "data stay") { + t.Errorf("the ask does not say what goes from which provider on which node: %+v", a) + } + // Nothing before the warrant: the provider's entry stands and the key stays. + if said := m.h.Decide(); len(said) != 0 || m.state(t, pg).Warrant != nil { + t.Errorf("decided before an answer: %v", said) + } + m.r.answer(t, p.Ask, OptionApprove) + m.h.Decide() + if w := m.state(t, pg).Warrant; w == nil || w.Option != OptionApprove || !strings.Contains(w.By, "42") { + t.Fatalf("the warrant is not in the key: %+v", w) + } + if said, _ := m.h.Rejudge(pg); said != "" || m.state(t, pg).State != provisions.Retiring { + t.Errorf("retired while the provider's entry stands: %q", said) + } + _ = m.s.Delete("postgres.bound", pg) + if said, _ := m.h.Rejudge(pg); !strings.Contains(said, "retired") || m.state(t, pg).State != "" { + t.Errorf("not retired once the provider removed its entry: %q", said) + } + // A decline is kept, and nothing is asked again while nothing changes. + d := newMesh() + d.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB)} + d.compose(t, "g14") + d.atWant["g14"] = map[string]catalogue.Plan{} + d.compose(t, "g14") + d.r.answer(t, d.state(t, pg).Ask, OptionDecline) + d.h.Decide() + if c := d.compose(t, "g14"); len(c.Asked) != 0 || d.state(t, pg).Warrant.Option != OptionDecline { + t.Errorf("a decline was asked again: %+v", c) + } +} + +// The holder holds no ticker: it composes on change events alone (ADR 0570 clause 2; issue 1954 H1). +func TestTheHolderHoldsNoTimer(t *testing.T) { + files, _ := filepath.Glob("*.go") + for _, f := range files { + if strings.HasSuffix(f, "_test.go") { + continue + } + file, err := parser.ParseFile(token.NewFileSet(), f, nil, 0) + if err != nil { + t.Fatal(err) + } + ast.Inspect(file, func(n ast.Node) bool { + if sel, ok := n.(*ast.SelectorExpr); ok { + if id, ok := sel.X.(*ast.Ident); ok && id.Name == "time" && + slices.Contains([]string{"NewTicker", "Tick", "AfterFunc", "NewTimer"}, sel.Sel.Name) { + t.Errorf("%s uses time.%s: the holder composes on events alone", f, sel.Sel.Name) + } + } + return true + }) + } +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go new file mode 100644 index 00000000..30a16446 --- /dev/null +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go @@ -0,0 +1,104 @@ +package main + +// The holder's tools (novox/hq ADR 0570 Consequences, ADR 0564 §12's rule): a read of every key of `provisions` and +// `plans`, one key whole, each state's fill, composing one node now, and reading the operator's answers now. + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/provisions" +) + +// The holder's tools. +const ( + ProvisionsTool = "provisioner_provisions" + PlansTool = "provisioner_plans" + ComposeTool = "provisioner_compose" + DecideTool = "provisioner_decide" +) + +func str(d string) map[string]any { return map[string]any{"type": "string", "description": d} } + +func holderTools(h *Holder) []stdio.Tool { + arg := func(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) } + return []stdio.Tool{ + {Name: ProvisionsTool, + Description: "Every provision the seat keeps (novox/hq ADR 0570): its key ..[.], " + + "provider, consumer and state (asked, provided, waits, failed, retiring, with why), narrowed by node and " + + "state, with how many keys and bytes the state holds; or one key whole.", + Input: map[string]any{"type": "object", "properties": map[string]any{"key": str("one key, whole"), + "node": str("a consumer's node"), "state": str("asked, provided, waits, failed or retiring")}}, + Run: func(a map[string]any) (any, error) { + return h.read(ProvisionsState, arg(a, "key"), arg(a, "node"), arg(a, "state")) + }}, + {Name: PlansTool, + Description: "Every module's provisioning plan the seat composed, ., narrowed by node, with how many " + + "keys and bytes the state holds; or one plan whole.", + Input: map[string]any{"type": "object", "properties": map[string]any{"key": str("one plan, ."), + "node": str("a node")}}, + Run: func(a map[string]any) (any, error) { return h.read(PlansState, arg(a, "key"), arg(a, "node"), "") }}, + {Name: ComposeTool, + Description: "Compose one node's plans and provisions now from the bus, as a change event would, and say what " + + "was written, removed, changed and asked.", + Input: map[string]any{"type": "object", "properties": map[string]any{"node": str("the node")}, + "required": []string{"node"}}, + Run: func(a map[string]any) (any, error) { + if arg(a, "node") == "" { + return nil, fmt.Errorf("compose names the node") + } + return h.ComposeNode(context.Background(), arg(a, "node")) + }}, + {Name: DecideTool, + Description: "Read the operator's answers to every retiring provision's ask now, writing each warrant into its " + + "key, as the router's word on an ask would.", + Run: func(map[string]any) (any, error) { return map[string]any{"said": h.Decide()}, nil }}, + } +} + +// read is a state's keys narrowed, or one whole, with the state's fill. +func (h *Holder) read(state, key, node, want string) (any, error) { + if key != "" { + e, err := h.Store.Get(state, key) + if err != nil { + return nil, err + } + if e == nil { + return nil, fmt.Errorf("%s holds no %s", state, key) + } + return json.RawMessage(e.Value), nil + } + keys, err := h.Store.Keys(state) + if err != nil { + return nil, err + } + rows := []map[string]any{} + bytes := 0 + for _, k := range keys { + e, err := h.Store.Get(state, k) + if err != nil || e == nil { + continue + } + bytes += len(e.Value) + if node != "" && !strings.HasPrefix(k, node+".") { + continue + } + row := map[string]any{"key": k} + if state == ProvisionsState { + p, err := provisions.Decode(e.Value) + if err != nil { + row["unreadable"] = err.Error() + } else { + if want != "" && p.State != want { + continue + } + row["state"], row["provider"], row["why"] = p.State, p.Provider.Module+" on "+p.Provider.Node, p.Why + } + } + rows = append(rows, row) + } + return map[string]any{state: rows, "count": len(rows), "fill": map[string]int{"keys": len(keys), "bytes": bytes}}, nil +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index 3955516d..583072b0 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -1,23 +1,28 @@ // mesh-provisioner: the provisioner's server, holder of the mesh-provisioner seat once for the mesh (novox/hq ADR -// 0349 clause 3, design 57 P4). A Go bundle served by a runtime of its own, on its own bus account (`runs-as`, ADR -// 0259 §8), keeping four buckets: mesh-provisioner_registry (which domain answers which kind of placeholder, its -// alone), and mesh-provisioner_needs, _applied and _collect, which each machine's provisioner client writes for its -// own machine (the controller grants the client those keys alone, issue 670). stdout is the MCP channel; what this -// module says, it says on stderr. +// 0349 clause 3, ADR 0570). A Go bundle served by a runtime of its own, on its own bus account (`runs-as`, ADR 0259 +// §8), keeping mesh-provisioner_provisions and _plans (every provision and every plan, composed from the bus on change +// events, its alone), _registry (which domain answers which kind of placeholder, its alone), and _applied and +// _collect, which each machine's provisioner client writes for its own machine; _needs is retired and emptied at its +// start. stdout is the MCP channel; what this module says, it says on stderr. package main import ( + "context" "fmt" "os" + "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // Seat is the role this module holds, once for the mesh, and Verb its one verb. const ( - Seat = "mesh-provisioner" - Verb = "provisioning" + Seat = "mesh-provisioner" + Module = "mesh-provisioner" + Verb = "provisioning" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -42,30 +47,149 @@ func (runtimeStore) Delete(state, key string) error { return stdio.State(state). func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +// runtimeAsks is the operator's channel through the runtime: asks published under this module's own name, and the +// router's record of one read under that name. +type runtimeAsks struct{} + +func (runtimeAsks) Publish(subject string, body any, id string) error { + return stdio.SeatPublish(subject, body, id) +} + +func (runtimeAsks) Record(id string) ([]byte, bool, error) { + e, err := stdio.SeatRecord("", id) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +// work is the nodes a change touched, composed one at a time by one worker: no timer, a change's event is the wake. +type work struct { + mu sync.Mutex + pending map[string]bool + wake chan struct{} +} + +func (w *work) add(nodes ...string) { + if len(nodes) == 0 { + return + } + w.mu.Lock() + for _, n := range nodes { + w.pending[n] = true + } + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *work) run(h *Holder) { + for range w.wake { + w.mu.Lock() + nodes := sortedKeys(w.pending) + w.pending = map[string]bool{} + w.mu.Unlock() + for _, n := range nodes { + c, err := h.ComposeNode(context.Background(), n) + switch { + case err != nil: + logf("[%s] %s is not composed: %v", Module, n, err) + case len(c.Plans)+len(c.Gone)+len(c.Changed)+len(c.Asked)+len(c.Problems) > 0: + logf("[%s] composed %s: %+v", Module, n, c) + } + } + } +} + +// watched asks a watch of the runtime again until it is taken: the runtime answers only once Serve runs. +func watched(state string, handle func(stdio.StateChange) error) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + if err := stdio.State(state).Watch("", handle); err != nil { + logf("[%s] %s is not watched yet (%v); asking again in %s", Module, state, err, wait) + continue + } + return + } +} + func main() { - go publishing(runtimeStore{}) - if err := stdio.Serve("", Tools(runtimeStore{})); err != nil { + s := runtimeStore{} + h := &Holder{Store: s, Compose: FromBus(s), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} + w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)} + go w.run(h) + go starting(h) + go watched(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) + go watched(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) + go watched(ModulesRead, func(c stdio.StateChange) error { + if module, err := delivery.ParseModulesKey(c.Key); err == nil { + w.add(h.NodesOfModule(module)...) + } + return nil + }) + for _, provider := range Providers { + go watched(boundRead(provider), func(c stdio.StateChange) error { + if said, err := h.Rejudge(c.Key); err != nil { + logf("[%s] %s is not judged again: %v", Module, c.Key, err) + } else if said != "" { + logf("[%s] %s", Module, said) + } + return nil + }) + } + go watched(Registry, func(c stdio.StateChange) error { + if c.Op != "put" { + if _, _, err := Publish(s); err != nil { + logf("[%s] the registry is not written back: %v", Module, err) + } + } + return nil + }) + go func() { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + err := stdio.Subscribe(asks.Seat+".decided."+Module, func(stdio.Envelope) error { + for _, said := range h.Decide() { + logf("[%s] %s", Module, said) + } + return nil + }) + if err == nil { + return + } + logf("[%s] does not hear the operator's answers yet (%v); asking again in %s", Module, err, wait) + } + }() + if err := stdio.Serve("", append(Tools(s), holderTools(h)...)); err != nil { logf("%v", err) os.Exit(1) } } -// publishing writes the registry once the runtime answers, and again every hour, so a registry emptied by hand or -// a bucket made again is written back without a restart. The runtime answers only once Serve runs, so the first -// pass is tried again until it lands. -func publishing(s Store) { - for wait := 2 * time.Second; ; { - time.Sleep(wait) - written, deleted, err := Publish(s) - switch { - case err != nil: - logf("[mesh-provisioner] the registry is not written yet (%v); trying again in %s", err, wait) - wait = min(wait*2, time.Minute) - continue - case len(written) > 0 || len(deleted) > 0: - logf("[mesh-provisioner] the registry: wrote %v, deleted %v", written, deleted) +// starting writes the registry, deletes the retired needs and reads the operator's answers once the runtime answers, +// each tried again until it lands. +func starting(h *Holder) { + <-stdio.Served() + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + written, deleted, err := Publish(h.Store) + if err == nil { + if len(written)+len(deleted) > 0 { + logf("[%s] the registry: wrote %v, deleted %v", Module, written, deleted) + } + break } - wait = time.Hour + logf("[%s] the registry is not written yet (%v); trying again in %s", Module, err, wait) + time.Sleep(wait) + } + if n, err := h.RetireNeeds(); err != nil { + logf("[%s] the retired needs are not all deleted: %v", Module, err) + } else if n > 0 { + logf("[%s] deleted %d key(s) of the retired needs", Module, n) + } + for _, said := range h.Decide() { + logf("[%s] %s", Module, said) } } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go index 4661b226..6d48aadb 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go @@ -106,9 +106,9 @@ func TestTheRegistryAnswersEveryKindATemplateMayName(t *testing.T) { for kind, bucket := range map[string]string{"config": "mesh-config_resolved", "collected": "mesh-provisioner_" + Collect, "shell": "mesh-provisioner_" + Collect, "environment": "mesh-provisioner_" + Collect, "contribution": "mesh-provisioner_" + Collect, - "dir": "mesh-controller_assignments", "access": "mesh-controller_assignments", - "port": "mesh-controller_assignments", "seat": "mesh-controller_assignments", - "machine": "mesh-controller_assignments", "backup": "mesh-controller_assignments"} { + "dir": "mesh-provisioner_plans", "access": "mesh-provisioner_plans", + "port": "mesh-provisioner_plans", "seat": "mesh-provisioner_plans", + "machine": "mesh-provisioner_plans", "backup": "mesh-provisioner_plans"} { i := slices.IndexFunc(TheRegistry, func(a Answer) bool { return a.Kind == kind }) if TheRegistry[i].Bucket != bucket { t.Errorf("%s is found in %s, want %s", kind, TheRegistry[i].Bucket, bucket) @@ -288,6 +288,8 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { Serves []string `json:"serves"` } `json:"claims"` State []string `json:"state"` + Reads []string `json:"reads"` + Tools []string `json:"tools"` OwnSecrets map[string]string `json:"own-secrets"` } if err := json.Unmarshal(raw, &m); err != nil { @@ -300,7 +302,39 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { !slices.Equal(m.Claims[0].Serves, []string{Verb}) { t.Errorf("the claim: %+v", m.Claims) } - if want := []string{Registry, Needs, Applied, Collect}; !slices.Equal(m.State, want) { + // needs stays declared this release so its first start can empty it (ADR 0570 clause 1). + if want := []string{Registry, Needs, Applied, Collect, ProvisionsState, PlansState}; !slices.Equal(m.State, want) { t.Errorf("the state is %v, want %v", m.State, want) } + // Every state and act has its tool (ADR 0570 Consequences): the verb reads the clients' states, the holder's tools + // its own and the composition it does on a change. + served := map[string]bool{} + for _, tool := range append(Tools(newStore()), holderTools(&Holder{})...) { + served[tool.Name] = true + } + for what, tool := range map[string]string{Registry: Seat + "." + Verb, Needs: Seat + "." + Verb, + Applied: Seat + "." + Verb, Collect: Seat + "." + Verb, ProvisionsState: ProvisionsTool, PlansState: PlansTool, + "a composition on a change": ComposeTool, "a decision on an answer": DecideTool} { + if !served[tool] { + t.Errorf("%s has no tool: %s is not served", what, tool) + } + } + for _, tool := range m.Tools { + if !served[tool] { + t.Errorf("the manifest lists %s, which is not served", tool) + } + } + for _, r := range []string{PlacementRead, WantsRead, ModulesRead, ManifestsRead, BuildsRead, HaveRead, ResolvedRead} { + if !slices.Contains(m.Reads, r) { + t.Errorf("the manifest does not read %s, which the composition reads", r) + } + } + for _, p := range Providers { + if !slices.Contains(m.Reads, boundRead(p)) { + t.Errorf("the manifest does not read %s's entries", p) + } + } + if slices.Contains(m.Reads, "mesh-controller.assignments") { + t.Error("the manifest reads the controller's assignments (ADR 0570 clause 2)") + } } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/reads-outside b/modules/mesh-provisioner/cmd/mesh-provisioner/reads-outside new file mode 100644 index 00000000..203aee31 --- /dev/null +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/reads-outside @@ -0,0 +1,2 @@ +# What these tests read outside their package, as paths from the repository's root (issue 1936). +modules/mesh-provisioner/module.json diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/registry.go b/modules/mesh-provisioner/cmd/mesh-provisioner/registry.go index 7d7d6f3e..15f0bb3f 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/registry.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/registry.go @@ -48,16 +48,15 @@ type Answer struct { Decision string `json:"decision"` } -// placement is the controller's: where a module runs, its directories, accesses, ports, seats and the machine's -// facts (ADR 0349 clause 6). On the bus, in the provisioner's plan of each module the controller handed to it, which -// its assignment carries (mesh-controller internal/inventory Provisioning, issue 671), or for a module only pulled in -// by another, its value of its own under the same key (issue 765): a client never reads the node's declaration, -// which the node-engine keeps for root alone. +// placement is the seat's: where a module runs, its directories, accesses, ports, seats and the machine's facts (ADR +// 0349 clause 6), in the provisioning plan this seat composes of each module on each node from the bus, its state +// `plans` (novox/hq ADR 0570 clauses 1, 2): a client never reads the node's declaration, which the node-engine keeps for +// root alone, nor the controller's assignments. func placement(kind, what string) Answer { - return Answer{Kind: kind, Domain: "mesh-controller", From: FromBucket, Bucket: "mesh-controller_assignments", + return Answer{Kind: kind, Domain: Module, From: FromBucket, Bucket: Module + "_" + PlansState, Key: ".", - Why: what + ": placement, the controller's, in the provisioner's plan its assignment carries (provisioning." + - "placement)", Decision: "novox/hq ADR 0349 clause 6"} + Why: what + ": placement, in the provisioning plan the seat mesh-provisioner composes of the module on the node " + + "(plans, plan.placement)", Decision: "novox/hq ADR 0570 clause 1"} } // oldCollection is a namespace read as a collection of its own name (novox/hq ADR 0349 clause 13, design 57 L1). diff --git a/modules/mesh-provisioner/go.mod b/modules/mesh-provisioner/go.mod index 9f0b6d60..70df969f 100644 --- a/modules/mesh-provisioner/go.mod +++ b/modules/mesh-provisioner/go.mod @@ -2,4 +2,4 @@ module mesh-provisioner go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d diff --git a/modules/mesh-provisioner/go.sum b/modules/mesh-provisioner/go.sum index bb6beaa1..c728e39b 100644 --- a/modules/mesh-provisioner/go.sum +++ b/modules/mesh-provisioner/go.sum @@ -1,2 +1,14 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 h1:VLir9NZbyi1LdH1YwWDIVRuL+lDC+AYlm6FZPlvmlm4= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d h1:J1QAvzlaIvCCyZGBQX0XGJjYWdNPjLXBF4zaAeIkUOg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= diff --git a/modules/mesh-provisioner/module.json b/modules/mesh-provisioner/module.json index 650425b3..1aba09af 100644 --- a/modules/mesh-provisioner/module.json +++ b/modules/mesh-provisioner/module.json @@ -11,11 +11,36 @@ ] } ], + "uses": [ + "operator-channel" + ], + "reads": [ + "mesh-delivery-assigner.placement", + "mesh-delivery-assigner.wants", + "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.manifests", + "mesh-delivery-builder.builds", + "node-engine.have", + "mesh-config.resolved", + "gitea.bound", + "keycloak.bound", + "postgres.bound", + "route-adapter.bound", + "mesh-vault.bound" + ], + "tools": [ + "provisioner_provisions", + "provisioner_plans", + "provisioner_compose", + "provisioner_decide" + ], "state": [ "registry", "needs", "applied", - "collect" + "collect", + "provisions", + "plans" ], "own-secrets": { "broker": "${dir:state}/broker"