Compose every provision and plan in mesh-provisioner from the bus on change events, its state from the providers' outcomes, and ask before retiring (issue 1954)
The seat holds provisions and plans alone (ADR 0570 §1-§3, §5): a node is composed at its wanted builds and its newest, so a key ahead of its want is never retiring; a key no build needs is retiring with one ask, and the warrant goes into the key for its provider. The registry's placement kinds name plans; needs is emptied at start. Pins mesh-sdk 66ff33d.
This commit is contained in:
@@ -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})
|
||||
}
|
||||
}
|
||||
@@ -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` (`<node>.<module>`) and `provisions` (`<node>.<consumer>.<provision>[.<local>]`), 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, `<provider>_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, `<module>_bound`, read as `<module>.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
|
||||
}
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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 <node>.<consumer>.<provision>[.<local>], " +
|
||||
"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, <node>.<module>, 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>.<module>"),
|
||||
"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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -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: "<node>.<module>",
|
||||
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).
|
||||
|
||||
Reference in New Issue
Block a user