diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go index 196855de..13d2eb5e 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holder.go @@ -18,6 +18,12 @@ package main // - **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. +// - **Retiring is judged from a whole picture only** (issue 940, met before as 937 and 1332): no pass judges a +// retirement before every watched read has loaded once, nor while a module on the node has no composition published +// at its build; a key whose consumer still has a want or a have on the node, and no plan read in this pass, is left +// as it is; a pass that would retire most of a node's consumers at once retires nothing and says it as a fault of +// the mesh (provisioner_health), never as a question; at most askOpenAtMost asks stand open at once, and an ask the +// router does not hold is raised again. import ( "bytes" @@ -84,7 +90,10 @@ type Composer func(ctx context.Context, node string, keptAt map[string]string) C // Composition is a node's plans at its wanted builds and at its newest, and where the time went. type Composition struct { AtWants, Newest HeldPlans - Took []Phase + // Unpublished is each module whose composition at the build it is read at is not published yet: the node's picture + // is not whole, so no retirement is judged on it (issue 940). + Unpublished []string + Took []Phase } // HeldPlans is one scope's plans by module, the modules held with why, and a failure that is no single module's. @@ -114,8 +123,13 @@ type Holder struct { Asks Asker Now func() time.Time Logf func(format string, args ...any) + // Loaded says which watched reads have not loaded once yet, "" when all have (nil: all have). A pass before then + // judges nothing (issue 940: a first read that was not whole retired 21 routes of running apps). + Loaded func() string mu sync.Mutex + // faults is, by node, the retirement a pass held back as a fault of the mesh (issue 940), said by provisioner_health. + faults map[string]string } // Composed is what one node's composition did. @@ -149,7 +163,13 @@ func TookSaid(took []Took) string { } // ComposeNode composes one node: its plans and its provisions. -func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, err error) { +func (h *Holder) ComposeNode(ctx context.Context, node string) (Composed, error) { + return h.ComposeNodeWith(ctx, node, false) +} + +// ComposeNodeWith composes one node; retireMany lets this one pass retire more of the node's consumers at once than +// the bound (manyAtOnce) allows, each still asked: the operator's word that a large removal is meant. +func (h *Holder) ComposeNodeWith(ctx context.Context, node string, retireMany bool) (out Composed, err error) { h.mu.Lock() defer h.mu.Unlock() out = Composed{Node: node} @@ -163,13 +183,29 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er start = now } defer func() { mark("whole", time.Since(began)) }() + if h.Loaded != nil { + if missing := h.Loaded(); missing != "" { + return out, fmt.Errorf("%s is not composed: %s has not loaded once yet, and a pass over a part judges nothing", node, missing) + } + } wanted, err := h.wantedOn(node) if err != nil { return out, err } + running, err := h.runningOn(node) + if err != nil { + return out, err + } + // present is every module with a want or a have on the node: one of them with no plan read in this pass is not + // unneeded (issue 940). + present := map[string]bool{} + for module := range wanted { + present[module] = true + } // A module held on the node with no want (the bus on the control-node) is composed at the build its have says it // runs, so it has a plan too and its client keeps filling it (review of 2026-10-30). - for module, fp := range h.runningOn(node) { + for module, fp := range running { + present[module] = true if _, ok := wanted[module]; !ok { wanted[module] = fp } @@ -200,7 +236,22 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er if err := h.writeRefused(node, heldAtWants); err != nil { out.Problems = append(out.Problems, err.Error()) } - if err := h.writePlans(node, atWants, heldAtWants, &out); err != nil { + // partial is why the node's picture is not whole in this pass: then nothing is removed or retired on it. + partial := "" + if len(c.Unpublished) > 0 { + partial = fmt.Sprintf("the composition of %s on %s is not published yet", strings.Join(c.Unpublished, ", "), node) + } + keep := heldAtWants + if partial != "" { + keep = map[string]string{} + for m, why := range heldAtWants { + keep[m] = why + } + for _, m := range c.Unpublished { + keep[m] = partial + } + } + if err := h.writePlans(node, atWants, keep, partial != "", &out); err != nil { out.Problems = append(out.Problems, err.Error()) } lap("plans written") @@ -259,8 +310,21 @@ func (h *Holder) ComposeNode(ctx context.Context, node string) (out Composed, er } } } + // A module with a want or a have on the node whose plan this pass did not read is left as it is: its keys are not + // judged unneeded (issue 940). + untouched := map[string]string{} + for m, why := range heldAtWants { + untouched[m] = why + } + for m := range present { + _, a := atWants[m] + _, n := newest[m] + if _, u := untouched[m]; !a && !n && !u { + untouched[m] = "it has a want or a have on " + node + " and no plan of it was read" + } + } h.judgeElsewhere(elsewhere, &out) - h.judgeNode(node, needed, ahead, heldAtWants, &out) + h.judgeNode(node, needed, ahead, untouched, partial, retireMany, &out) lap("provisions judged") return out, nil } @@ -419,11 +483,11 @@ func (h *Holder) wantedOn(node string) (map[string]string, error) { // runningOn is the fingerprint each module's have on a node says it runs, by module: what a module held without a want // is composed at. -func (h *Holder) runningOn(node string) map[string]string { +func (h *Holder) runningOn(node string) (map[string]string, error) { out := map[string]string{} keys, err := h.Store.Keys(HaveRead) if err != nil { - return out + return nil, fmt.Errorf("the haves cannot be read: %w", err) } for _, k := range keys { n, module, isNode, err := delivery.ParseHaveKey(k) @@ -438,12 +502,13 @@ func (h *Holder) runningOn(node string) map[string]string { out[module] = have.FP } } - return out + return out, nil } // writePlans writes each plan that differs and removes the node's plans of modules no longer composed there; a held -// module's plan is kept as it was (issue 774). -func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held map[string]string, out *Composed) error { +// module's plan is kept as it was (issue 774), and in a pass whose picture is partial no plan is removed (issue 940). +func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held map[string]string, partial bool, + out *Composed) error { var problems []error for _, module := range sortedKeys(plans) { key, err := provisions.PlanKey(node, module) @@ -477,7 +542,7 @@ func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held m if _, still := plans[module]; still { continue } - if _, kept := held[module]; kept { + if _, kept := held[module]; kept || partial { continue } if err := h.Store.Delete(PlansState, k); err != nil { @@ -489,14 +554,19 @@ func (h *Holder) writePlans(node string, plans map[string]catalogue.Plan, held m return errors.Join(problems...) } -// judgeNode brings the node's provision keys to what is needed and what each provider said. A key of a module held on -// the node (heldModules) that its composition no longer names is left as it is: not composed is not unneeded. +// judgeNode brings the node's provision keys to what is needed and what each provider said. A key of a module left +// untouched (held on the node, or with a want or a have and no plan read in this pass) that its composition no longer +// names is left as it is: not composed is not unneeded. While partial says why the node's picture is not whole, no key +// turns retiring; and a pass that would retire most of the node's consumers at once retires none, said as a fault of +// the mesh (issue 940), unless retireMany. func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, ahead map[string]string, - heldModules map[string]string, out *Composed) { + untouched map[string]string, partial string, retireMany bool, out *Composed) { held := h.held(node) + consumers := map[string]bool{} for key, have := range held { + consumers[have.Consumer.Module] = true if _, isNeeded := needed[key]; !isNeeded && have.Consumer.Node == node { - if _, notComposed := heldModules[have.Consumer.Module]; notComposed { + if _, left := untouched[have.Consumer.Module]; left { delete(held, key) } } @@ -510,6 +580,39 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, } } } + // The keys this pass would turn retiring, and whose they are. + var newly []string + leaving := map[string]bool{} + for key, have := range held { + if _, isNeeded := needed[key]; !isNeeded && have.State != provisions.Retiring { + newly = append(newly, key) + leaving[have.Consumer.Module] = true + } + } + sort.Strings(newly) + hold := func(why string) { + for _, key := range newly { + delete(held, key) + } + out.Problems = append(out.Problems, why) + } + switch { + case len(newly) == 0: + h.clearFault(node) + case partial != "": + hold(fmt.Sprintf("%d key(s) of %s are not judged retiring while %s: %s", len(newly), node, partial, + strings.Join(newly, ", "))) + case manyAtOnce(len(leaving), len(consumers)) && !retireMany: + fault := fmt.Sprintf("On %s, one pass would retire the provisions of %d of the %d consumers that hold any (%s): "+ + "most likely a read stopped listing them (a node's compositions published in part, a want, a have or the "+ + "placement not read whole), not their removal. Nothing is retired and nothing is asked; the keys stay as they "+ + "were. If the removal is meant, provisioner_compose with retire-many retires them, each asked.", + node, len(leaving), len(consumers), strings.Join(sortedKeys(leaving), ", ")) + hold(fault) + h.setFault(node, fault) + default: + h.clearFault(node) + } for _, key := range unionKeys(needed, held) { want, isNeeded := needed[key] have, isHeld := held[key] @@ -532,12 +635,12 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, } case have.State == provisions.Retiring: next = have + if next.Ask != "" && next.Warrant == nil && !h.askHeld(next) { + out.Problems = append(out.Problems, "the router holds no ask "+next.Ask+" of "+key+": it is raised again") + next.Ask = "" + } 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) - } + h.raise(&next, key, out) } if gone, why := h.retired(key, have); gone { if err := h.Store.Delete(ProvisionsState, key); err != nil { @@ -552,11 +655,7 @@ func (h *Holder) judgeNode(node string, needed map[string]provisions.Provision, 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) - } + h.raise(&next, key, out) } if isHeld && same(have, next) { continue @@ -878,7 +977,33 @@ 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. +// askOpenAtMost is how many of its asks mesh-provisioner keeps open at once: the router shows at most three per asker +// and refuses the rest (issue 940: 18 of 21 retiring asks were refused so, yet counted asked). The next is raised as +// one ends (Decide). +const askOpenAtMost = 3 + +// askHeldWithin is how long the router may take to hold an ask before an ask it holds no record of is raised again. +const askHeldWithin = time.Minute + +// askWaits is what a retiring key's why says while its ask waits for an open one to end. +const askWaits = "its ask waits until fewer than 3 of mesh-provisioner's asks stand open, the router showing at most 3 per asker" + +// errAskWaits is an ask held back by askOpenAtMost: no fault, raised once one ends. +var errAskWaits = errors.New(askWaits) + +// raise asks a retiring key and says so in out: asked, waiting for an open ask to end, or a problem. +func (h *Holder) raise(p *provisions.Provision, key string, out *Composed) { + switch err := h.ask(p, key); { + case err == nil: + out.Asked = append(out.Asked, key) + case errors.Is(err, errAskWaits): + default: + out.Problems = append(out.Problems, err.Error()) + } +} + +// ask raises the retiring ask of a key, under mesh-provisioner's own name; one already raised stands. While +// askOpenAtMost asks stand open it raises none and the key's why says so. func (h *Holder) ask(p *provisions.Provision, key string) error { if p.Ask != "" { return nil @@ -886,6 +1011,12 @@ func (h *Holder) ask(p *provisions.Provision, key string) error { if h.Asks == nil { return errors.New("the operator cannot be asked from here, so " + key + " is retiring and nothing is removed") } + if h.openAsks() >= askOpenAtMost { + if !strings.Contains(p.Why, askWaits) { + p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ") + } + return errAskWaits + } now := h.Now().UTC() id := "retire-" + strings.ReplaceAll(key, ".", "-") + "-" + fmt.Sprint(now.Unix()) a := retiringAsk(key, *p, now, id) @@ -896,9 +1027,37 @@ func (h *Holder) ask(p *provisions.Provision, key string) error { return fmt.Errorf("the retiring ask of %s could not be published: %w", key, err) } p.Ask = id + p.Why = strings.TrimSuffix(strings.TrimSuffix(p.Why, askWaits), "; ") return nil } +// openAsks is how many retiring keys hold an ask with no answer yet. +func (h *Holder) openAsks() int { + n := 0 + keys, _ := h.Store.Keys(ProvisionsState) + 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.State == provisions.Retiring && p.Ask != "" && + p.Warrant == nil { + n++ + } + } + } + return n +} + +// askHeld says the router holds a key's ask, or may not have taken it yet: false only when it has no record of it +// askHeldWithin after it was raised (issue 940: an ask the router refused was counted asked for good). +func (h *Holder) askHeld(p provisions.Provision) bool { + if h.Asks == nil || p.Ask == "" { + return true + } + if _, found, err := h.Asks.Record(p.Ask); err != nil || found { + return true + } + return h.Now().Sub(p.At) < askHeldWithin +} + // 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 { @@ -922,6 +1081,14 @@ func (h *Holder) Decide() []string { continue } raw, found, err := h.Asks.Record(p.Ask) + if err == nil && !found && h.Now().Sub(p.At) >= askHeldWithin { + // The router holds no record of it: not asked, raised again below once an open ask allows. + p.Ask = "" + p.Why = strings.TrimPrefix(p.Why+"; "+askWaits, "; ") + h.put(k, p) + said = append(said, k+": the router holds no ask of it; raised again") + continue + } if err != nil || !found { continue } @@ -963,9 +1130,74 @@ func (h *Holder) Decide() []string { h.put(k, p) said = append(said, k+": "+option+" by "+by) } + return append(said, h.raiseWaiting()...) +} + +// raiseWaiting raises the asks that wait for an open one to end, as many as askOpenAtMost allows: an answer, which +// wakes Decide, is what frees a place. +func (h *Holder) raiseWaiting() []string { + keys, err := h.Store.Keys(ProvisionsState) + if err != nil { + return nil + } + 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 || !strings.Contains(p.Why, askWaits) { + continue + } + switch err := h.ask(&p, k); { + case errors.Is(err, errAskWaits): + return said + case err != nil: + said = append(said, k+": "+err.Error()) + default: + h.put(k, p) + said = append(said, k+": asked") + } + } return said } +// manyAtOnce says a pass retiring the provisions of leaving consumers, of all the consumers holding a key on the node, +// retires most of them at once: more than 3, or more than half and at least 2 (one consumer leaving is a removal, not +// a fault). +func manyAtOnce(leaving, all int) bool { + return leaving > 3 || (leaving >= 2 && 2*leaving > all) +} + +func (h *Holder) setFault(node, why string) { + if h.faults == nil { + h.faults = map[string]string{} + } + if h.faults[node] != why { + h.Logf("[%s] %s", Module, why) + } + h.faults[node] = why +} + +func (h *Holder) clearFault(node string) { + if _, ok := h.faults[node]; ok { + h.Logf("[%s] the retirement held back on %s is no longer judged", Module, node) + delete(h.faults, node) + } +} + +// Faults is, by node, each retirement a pass held back as a fault of the mesh. +func (h *Holder) Faults() map[string]string { + h.mu.Lock() + defer h.mu.Unlock() + out := make(map[string]string, len(h.faults)) + for n, why := range h.faults { + out[n] = why + } + return out +} + func (h *Holder) put(key string, p provisions.Provision) { p.At = h.Now().UTC() if raw, err := provisions.Encode(p); err == nil { @@ -1123,7 +1355,9 @@ func orSaid(words ...string) string { // builds a module is read at the fingerprint keptAt names, else at the indexer's newest (FP2); at the newest, each placed // module at FP2. A key whose engine refused the module holds it with the engine's words; a key not published yet holds // it at its wanted build (its last plan kept, as for a refusal) and leaves it out of the newest, whose build the engine -// writes once it is done (ADR 0593 §6). A plan absent from a key is a module that hands the provisioner nothing. +// writes once it is done (ADR 0593 §6), and says it unpublished, so the pass judges no retirement on the node (issue +// 940). A placed module no build of which is known is held too. A plan absent from a key is a module that hands the +// provisioner nothing. func FromCompositions(s Store) Composer { return func(_ context.Context, node string, keptAt map[string]string) Composition { start := time.Now() @@ -1134,7 +1368,11 @@ func FromCompositions(s Store) Composer { at.Err, newest.Err = err, err return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} } - placed, fp2 := placedModules(s, node) + placed, fp2, err := placedModules(s, node) + if err != nil { + at.Err, newest.Err = err, err + return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} + } atWants := map[string]string{} for _, m := range placed { atWants[m] = fp2[m] @@ -1142,14 +1380,19 @@ func FromCompositions(s Store) Composer { for m, fp := range keptAt { atWants[m] = fp } + var unpublished []string for _, m := range sortedKeys(atWants) { fp := atWants[m] if fp == "" { + // Placed, with no want, no have and no newest build the indexer names: no build of it is known, which is + // not a module handing nothing (issue 940). + at.Held[m] = fmt.Sprintf("no build of %s on %s is known: no want, no have, and the indexer names no newest build", m, node) continue } c, ok := published[composedAt(node, m, fp)] switch { case !ok: + unpublished = append(unpublished, m) at.Held[m] = fmt.Sprintf("the composition of %s at %s is not published by the node-engine of %s yet", m, fp, node) case c.Refused != "": at.Held[m] = c.Refused @@ -1170,7 +1413,8 @@ func FromCompositions(s Store) Composer { newest.Plans[m] = *c.Plan } } - return Composition{AtWants: at, Newest: newest, Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} + return Composition{AtWants: at, Newest: newest, Unpublished: unpublished, + Took: []Phase{{Name: "compositions read", Took: time.Since(start)}}} } } @@ -1216,16 +1460,18 @@ func publishedOn(s Store, node string) (map[string]publishedPlan, error) { } // placedModules is every module placed on node, sorted, and the indexer's newest fingerprint (FP2) of each. -func placedModules(s Store, node string) ([]string, map[string]string) { +func placedModules(s Store, node string) ([]string, map[string]string, error) { raw := map[string][]byte{} - if keys, err := s.Keys(PlacementRead); err == nil { - for _, k := range keys { - if n, _, _, err := delivery.ParsePlacementKey(k); err != nil || n != node { - continue - } - if e, err := s.Get(PlacementRead, k); err == nil && e != nil { - raw[k] = e.Value - } + keys, err := s.Keys(PlacementRead) + if err != nil { + return nil, nil, fmt.Errorf("the placement cannot be read: %w", err) + } + for _, k := range keys { + if n, _, _, err := delivery.ParsePlacementKey(k); err != nil || n != node { + continue + } + if e, err := s.Get(PlacementRead, k); err == nil && e != nil { + raw[k] = e.Value } } placements, _ := delivery.PlacementsOf(raw) @@ -1244,7 +1490,7 @@ func placedModules(s Store, node string) ([]string, map[string]string) { } } sort.Strings(placed) - return placed, fp2 + return placed, fp2, nil } // WatchedReads are the states the holder reads whole, each from the watched copy (ADR 0580 clause 2): the placement, diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go index 0ba670fc..a1148ba1 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holder_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "fmt" "go/ast" "go/parser" "go/token" @@ -19,14 +20,19 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" ) -// router is the operator's channel as the holder asks on it. +// router is the operator's channel as the holder asks on it: it holds an ask open, or, like the router past its open +// asks per asker, holds no record of it (drop). type router struct { asked map[string]asks.Ask records map[string][]byte + drop bool } func (r *router) Publish(_ string, body any, id string) error { r.asked[id] = body.(asks.Ask) + if !r.drop { + r.records[id], _ = json.Marshal(map[string]any{"state": "open", "ask": body}) + } return nil } @@ -58,20 +64,22 @@ type mesh struct { atWant map[string]map[string]catalogue.Plan atNewest map[string]map[string]catalogue.Plan heldAt map[string]map[string]string // by node, the modules its composition at the wanted builds holds - composed []string + // unpublished is, by node, the modules whose composition is not published yet. + unpublished map[string][]string + 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{}, - heldAt: map[string]map[string]string{}} + heldAt: map[string]map[string]string{}, unpublished: map[string][]string{}} 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) Composition { m.composed = append(m.composed, node) return Composition{AtWants: HeldPlans{Plans: m.atWant[node], Held: m.heldAt[node]}, - Newest: HeldPlans{Plans: m.atNewest[node]}} + Newest: HeldPlans{Plans: m.atNewest[node]}, Unpublished: m.unpublished[node]} }} // What each provider's manifest declares: postgres, keycloak and the vault keep entries, the others report by their // have. @@ -566,3 +574,153 @@ func TestAKeyAheadOfItsWantCarriesItsPlanEntryAndSaysWhy(t *testing.T) { t.Errorf("the planned key: %+v", pg) } } + +// Issue 940 (met before as 937 and 1332): a retirement is judged from a whole picture only. At 2026-11-02 22:55:58 a +// first pass after a new membership read a part, counted the modules missing from it as handing nothing, and asked the +// operator to remove the routes of 21 running web apps; 3 s later the next pass put every plan back. +func TestARetirementIsJudgedOnlyFromAWholePicture(t *testing.T) { + fp := "src2:" + strings.Repeat("ab", 32) + route := func(n int) catalogue.PlanProvision { + return catalogue.PlanProvision{Provision: "route", Provider: "novox", ProviderModule: "keycloak", Local: fmt.Sprint(n)} + } + apps := []string{"de-spiegel", "invoicing", "link2pay", "nextcloud", "photos", "website"} + for _, row := range []struct { + name string + consumer int // how many of apps hold a key, each one + gone int // how many of them the pass no longer composes + change func(m *mesh) // what else the pass sees + many bool // the operator's retire-many + retiring int // keys retiring after the pass + asked int // asks raised by the pass + problem string // a part of a problem the pass says + fault string // a part of the health's why, "" for healthy + after func(*testing.T, *mesh) // a further check + }{ + {name: "a read not loaded once: nothing judged", consumer: 6, gone: 6, + change: func(m *mesh) { m.h.Loaded = func() string { return WantsRead } }}, + {name: "a module's composition not published: nothing retires", consumer: 2, gone: 1, + change: func(m *mesh) { m.unpublished["novox"] = []string{"gitea"} }, problem: "not published yet"}, + {name: "a consumer that still has a want: not retired", consumer: 2, gone: 1, + change: func(m *mesh) { + m.s.write(t, WantsRead, "want.novox.de-spiegel", `{"v":1,"fp":"`+fp+`","self":true}`) + }}, + {name: "a consumer that still has a have: not retired", consumer: 2, gone: 1, + change: func(m *mesh) { + m.s.write(t, HaveRead, "novox.de-spiegel", `{"v":1,"fp":"`+fp+`","phase":"running","health":"ok"}`) + }}, + {name: "one consumer of two leaving: retired and asked", consumer: 2, gone: 1, retiring: 1, asked: 1}, + {name: "most consumers at once: a fault, not a question", consumer: 6, gone: 6, problem: "a read stopped listing them", + fault: "6 of the 6 consumers"}, + {name: "more than 3 at once, though fewer than half", consumer: 6, gone: 4, fault: "4 of the 6"}, + {name: "the operator's retire-many: retired, at most 3 asked at once", consumer: 6, gone: 6, many: true, + retiring: 6, asked: askOpenAtMost, after: func(t *testing.T, m *mesh) { + var waiting int + for _, a := range apps { + if strings.Contains(m.state(t, "novox."+a+".route.1").Why, askWaits) { + waiting++ + } + } + if waiting != 6-askOpenAtMost { + t.Errorf("%d keys say their ask waits, want %d", waiting, 6-askOpenAtMost) + } + }}, + } { + t.Run(row.name, func(t *testing.T) { + m := newMesh() + plans := map[string]catalogue.Plan{} + for i := range row.consumer { + plans[apps[i]] = planOf(route(1)) + } + m.atWant["novox"], m.atNewest["novox"] = plans, plans + m.compose(t, "novox") + left := map[string]catalogue.Plan{} + for i := row.gone; i < row.consumer; i++ { + left[apps[i]] = plans[apps[i]] + } + m.atWant["novox"], m.atNewest["novox"] = left, left + if row.change != nil { + row.change(m) + } + c, err := m.h.ComposeNodeWith(context.Background(), "novox", row.many) + if m.h.Loaded != nil { + if err == nil || !strings.Contains(err.Error(), WantsRead) { + t.Errorf("a pass before every read loaded: %v", err) + } + } else if err != nil { + t.Fatal(err) + } + retiring := 0 + for i := range row.consumer { + if m.state(t, "novox."+apps[i]+".route.1").State == provisions.Retiring { + retiring++ + } + } + if retiring != row.retiring || len(c.Asked) != row.asked || len(m.r.asked) != row.asked { + t.Errorf("%d retiring, %d asked (%d published), want %d and %d: %+v", retiring, len(c.Asked), + len(m.r.asked), row.retiring, row.asked, c) + } + if row.problem != "" && !strings.Contains(strings.Join(c.Problems, " "), row.problem) { + t.Errorf("the problems %v do not say %q", c.Problems, row.problem) + } + if a := m.h.health(); (row.fault == "") != a.Healthy || !strings.Contains(a.Why, row.fault) || + (row.fault != "" && !strings.Contains(a.Why, "retire-many")) { + t.Errorf("the health is %+v, want a fault saying %q", a, row.fault) + } + if row.fault != "" { + // The read whole again: the fault clears. + m.atWant["novox"], m.atNewest["novox"] = plans, plans + m.compose(t, "novox") + if a := m.h.health(); !a.Healthy { + t.Errorf("the fault stands once the picture is whole: %+v", a) + } + } + if row.after != nil { + row.after(t, m) + } + }) + } + // A placed module no build of which is known is held, not counted as handing nothing; one unpublished says so. + var c today + readJSON(t, "testdata/today-composed.json", &c) + s := goldenMesh(t, c) + published(t, s, c) + s.write(t, PlacementRead, "placement."+c.Node, `{"v":1,"node":"`+c.Node+`","decided":{},"modules":[{"module":"store"},`+ + `{"module":"postgres"},{"module":"mesh-provisioner-client"},{"module":"restic"},{"module":"de-spiegel"}]}`) + got := FromCompositions(s)(context.Background(), c.Node, map[string]string{"photos": fp}) + if !strings.Contains(got.AtWants.Held["de-spiegel"], "no build") || !slices.Equal(got.Unpublished, []string{"photos"}) { + t.Errorf("held %v, unpublished %v", got.AtWants.Held, got.Unpublished) + } +} + +// Issue 940: an ask the router does not hold is not counted asked; it is raised again once an open ask ends. +func TestAnAskTheRouterDoesNotHoldIsRaisedAgain(t *testing.T) { + m := newMesh() + key := "g14.gitea.postgres-database" + m.atWant["g14"] = map[string]catalogue.Plan{"gitea": planOf(pgDB), "dunst": planOf(oidc)} + m.compose(t, "g14") + m.atWant["g14"] = map[string]catalogue.Plan{"dunst": planOf(oidc)} + m.r.drop = true + m.compose(t, "g14") + first := m.state(t, key).Ask + if first == "" { + t.Fatal("not asked") + } + m.r.drop = false + steps := []struct { + name string + later time.Duration + held bool // the router holds an ask of the key after the step + }{ + {"within a minute the router may not have taken it yet", 0, false}, + {"a minute on, with no record: raised again", 2 * time.Minute, true}, + } + for _, s := range steps { + now := m.h.Now() + m.h.Now = func() time.Time { now = now.Add(time.Second); return now.Add(s.later) } + m.h.Decide() + p := m.state(t, key) + if _, held := m.r.records[p.Ask]; held != s.held || (s.held && p.Ask == first) { + t.Errorf("%s: the key's ask is %q, the router holds it: %v", s.name, p.Ask, held) + } + } +} diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go b/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go index 12b4c906..a617e3d0 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/holdertools.go @@ -1,7 +1,8 @@ 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, making one node's plans now, and reading the operator's answers now. +// `plans`, one key whole, each state's fill, making one node's plans now (retiring more at once on the operator's +// word), reading the operator's answers now, and its health: a retirement held back as a fault of the mesh (issue 940). import ( "context" @@ -10,6 +11,7 @@ import ( "strings" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/provisions" ) @@ -19,6 +21,7 @@ const ( PlansTool = "provisioner_plans" ComposeTool = "provisioner_compose" DecideTool = "provisioner_decide" + HealthTool = "provisioner_health" ) func str(d string) map[string]any { return map[string]any{"type": "string", "description": d} } @@ -57,22 +60,44 @@ func holderTools(h *Holder) []stdio.Tool { {Name: ComposeTool, Description: "Make one node's plans and provisions now from the node-engine's published compositions and the " + "providers' outcomes, as a change event would (it composes nothing), and say what was written, removed, " + - "changed and asked, and where the time went.", - Input: map[string]any{"type": "object", "properties": map[string]any{"node": str("the node")}, + "changed and asked, and where the time went. A pass that would retire the provisions of most of the " + + "node's consumers at once retires none and says it as a fault (provisioner_health); retire-many: true " + + "is the operator's word that such a removal is meant, and retires them in this pass, each asked.", + Input: map[string]any{"type": "object", "properties": map[string]any{"node": str("the node"), + "retire-many": map[string]any{"type": "boolean", "description": "retire past the bound in this pass"}}, "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")) + many, _ := a["retire-many"].(bool) + return h.ComposeNodeWith(context.Background(), arg(a, "node"), many) }}, {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 }}, + {Name: HealthTool, + Description: "The provisioner's runtime health, as the node-engine asks it (novox/hq ADR 0366): unhealthy while " + + "a pass held back a retirement of most of a node's consumers at once as a fault of the mesh (issue 940), " + + "naming the node, the consumers and the likely cause; healthy once a pass over that node judges none.", + Run: func(map[string]any) (any, error) { return h.health(), nil }}, } } +// health is the holder's runtime health: unhealthy while a retirement is held back as a fault of the mesh. +func (h *Holder) health() health.Answer { + faults := h.Faults() + if len(faults) == 0 { + return health.Healthy("No pass holds back a retirement of most of a node's consumers.") + } + var said []string + for _, n := range sortedKeys(faults) { + said = append(said, faults[n]) + } + return health.Unhealthy(strings.Join(said, " ")) +} + // 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 != "" { diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index ad007f0d..7ea6bfe4 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -10,6 +10,7 @@ import ( "context" "fmt" "os" + "strings" "sync" "time" @@ -68,6 +69,52 @@ func (runtimeAsks) Record(id string) ([]byte, bool, error) { return e.Value, true, nil } +// loads is which of the reads a pass composes from have loaded once: their watch took every current value. +type loads struct { + mu sync.Mutex + want []string + loaded map[string]bool +} + +func newLoads(states ...string) *loads { return &loads{want: states, loaded: map[string]bool{}} } + +func (l *loads) done(state string) { + l.mu.Lock() + defer l.mu.Unlock() + l.loaded[state] = true +} + +// missing names the reads not loaded yet, "" once all are. +func (l *loads) missing() string { + l.mu.Lock() + defer l.mu.Unlock() + var out []string + for _, s := range l.want { + if !l.loaded[s] { + out = append(out, s) + } + } + return strings.Join(out, ", ") +} + +// heldBack is, once every read has loaded, every node the placement names: the passes their loads held back. +func (l *loads) heldBack(h *Holder) []string { + if l.missing() != "" { + return nil + } + keys, err := h.Store.Keys(PlacementRead) + if err != nil { + return nil + } + nodes := map[string]bool{} + for _, k := range keys { + for _, n := range NodesOfPlacement(k) { + nodes[n] = true + } + } + return sortedKeys(nodes) +} + // 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 @@ -128,14 +175,25 @@ func main() { h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8) w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)} + // No pass judges before every read it composes from has loaded once (issue 940); each load wakes every node it + // held back, through its follower's handler. + loads := newLoads(PlacementRead, WantsRead, ComposedRead, HaveRead, ModulesRead) + h.Loaded = loads.missing go w.run(h) go starting(h) // Each watch's handler runs after the copy took its change, so a composition it wakes reads the copy that change made. - go ws.Follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) - go ws.Follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) - go ws.Follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) - go ws.Follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) - go ws.Follow(ModulesRead, func(c stdio.StateChange) error { + follow := func(state string, then func(stdio.StateChange) error) { + go func() { + ws.Follow(state, then) + loads.done(state) + w.add(loads.heldBack(h)...) + }() + } + follow(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) + follow(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) + follow(ComposedRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) + follow(HaveRead, func(c stdio.StateChange) error { w.add(NodesOfNodeKey(c.Key)...); return nil }) + follow(ModulesRead, func(c stdio.StateChange) error { if module, err := delivery.ParseModulesKey(c.Key); err == nil { w.add(h.NodesOfModule(module)...) } diff --git a/modules/mesh-provisioner/module.json b/modules/mesh-provisioner/module.json index 05a747a8..e14cb7e3 100644 --- a/modules/mesh-provisioner/module.json +++ b/modules/mesh-provisioner/module.json @@ -33,7 +33,8 @@ "provisioner_provisions", "provisioner_plans", "provisioner_compose", - "provisioner_decide" + "provisioner_decide", + "provisioner_health" ], "state": [ "registry", @@ -48,6 +49,14 @@ "broker": "${dir:state}/broker" }, "secrets-owner": "mesh-provisioner", + "runtime-health": { + "kind": "tool", + "tool": "provisioner_health", + "interval": "60s", + "timeout": "10s", + "looks": 2, + "grace": "60s" + }, "resources": [ { "id": "account",