From 78f03b7422db6209bf2ec70235f06dc65aec3a6e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 22:37:39 +0100 Subject: [PATCH] Watchman: say needs-operator from mesh-provisioner's provisions in waits, so the row the haves could not carry has its facts on the bus (issue 1954 D2) Pins mesh-sdk 4a96793. --- .../cmd/mesh-delivery-watchman/main.go | 29 ++++++++++ modules/mesh-delivery-watchman/go.mod | 2 +- modules/mesh-delivery-watchman/go.sum | 6 +-- .../internal/watch/findings_test.go | 19 +++++++ .../internal/watch/signals.go | 54 +++++++++++++++++-- .../internal/watch/watch.go | 4 +- .../mesh-delivery-watchman/manifest_test.go | 2 +- modules/mesh-delivery-watchman/module.json | 1 + 8 files changed, 107 insertions(+), 10 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 4d50197..62b84a8 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provisions" "mesh-delivery-watchman/internal/socket" "mesh-delivery-watchman/internal/watch" @@ -297,6 +298,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := w.Rebuild(ask); err != nil { return err } + // The seat mesh-provisioner's provisions (ADR 0570 §7): a provision in waits is needs-operator. Read before the findings are + // reconciled when the bucket is there; bound once it appears when it is not (mesh-provisioner comes in its own delivery). + provisionsWatch := func(kv jetstream.KeyValue) (jetstream.KeyWatcher, error) { + return settle(ctx, kv, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) + } + var provWatch jetstream.KeyWatcher + if kv, err := js.KeyValue(ctx, provisions.ProvisionsBucket); err == nil { + if provWatch, err = provisionsWatch(kv); err != nil { + process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again") + } + } else { + process.AwaitBucket(ctx, js, conn, provisions.ProvisionsBucket, "no provision waiting for the operator is said", + logger.Printf).Ready(func(kv jetstream.KeyValue) { + pw, err := provisionsWatch(kv) + if err != nil { + process.NotWatched(logger.Printf, provisions.ProvisionsBucket, err, "no provision waiting for the operator is said until the watchman starts again") + return + } + go func() { + _ = follow(ctx, pw, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) + }() + }) + } dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf} w.SetRecount(dead.count) dead.count() @@ -309,6 +333,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } errs := make(chan error, 8) + if provWatch != nil { + go func() { + errs <- follow(ctx, provWatch, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) }) + }() + } if kvs.lease != nil { lw, err := kvs.lease.Watch(ctx, buscompose.ControllerLeaseKey) if err != nil { diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index d5fd93f..8842356 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-watchman go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-watchman/go.sum b/modules/mesh-delivery-watchman/go.sum index d215b52..b279ec6 100644 --- a/modules/mesh-delivery-watchman/go.sum +++ b/modules/mesh-delivery-watchman/go.sum @@ -1,7 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7 h1:/XEAhD7/8QoMjqI7XiX9HM4HkpKubWheQosqgfe2WrU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368 h1:LaBxsyU3HeTIPKYmvpJuw+Wm2plpf6O7qHO91O+mAi0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213215-4a967933b368/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index f31b704..182c5ca 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -6,6 +6,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provisions" ) func haveOf(t *testing.T, phase delivery.Phase, health delivery.Health, detail string) []byte { @@ -154,6 +155,24 @@ func TestTheWatchmansFindings(t *testing.T) { expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter) w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "a part waits for the secret token"), Put) // needs-operator is not observed here }, map[string]string{"module.web.ace.unhealthy": Urgent}}, + {"1954/D2: a provision in waits is its consumer's needs-operator; provided, it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + enc := func(state, why string) []byte { + b, err := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"}, + Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: state, Why: why, At: t0}) + if err != nil { + t.Fatal(err) + } + return b + } + w.Provision("ace.photos.oidc", enc(provisions.Waits, "the operator gives the client's redirect"), Put) + w.Provision("ace.photos.db", enc(provisions.Provided, ""), Put) + if f := o.findings["module.photos.ace.needs-operator"]; !strings.Contains(f.Summary, "redirect") || f.Machine != "ace" { + t.Fatalf("findings = %v", o.findings) + } + w.Provision("ace.photos.oidc", enc(provisions.Provided, ""), Put) + w.Provision("ace.web.oidc", enc(provisions.Waits, "x"), Put) + w.Provision("ace.web.oidc", nil, Delete) + }, map[string]string{}}, {"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put) w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put) diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 4c3a067..45c2881 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -9,6 +9,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provisions" ) // The rows of the signals table the watchman holds (novox/hq ADR 0564 §4, Limits 3 and 11). Each is a key of the pulse @@ -458,9 +459,8 @@ func ToolsNode(subject string) string { // unhealthy (two writes in a row, urgent after four hours), and where the controller runs (S1's control node). A have // removed or deleted ends its part of them. // -// needs-operator is not observed here: the controller excuses a wait only after checking it against the module's manifest -// and against whether a person gave the secret on that machine (its operator_wait.go), and neither each resource's wait nor -// what a person gave is on the bus (the have carries the node's words only). The row waits for those facts' home. +// needs-operator is not observed from the haves: a module's part that waits for the operator is a provision in the state +// `waits` of the seat mesh-provisioner (Provision, ADR 0570 §7). func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Have) { unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy") if module == ControllerModule { @@ -511,6 +511,53 @@ func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Hav // ControllerModule is the module whose have names a control node. const ControllerModule = "mesh-controller" +// Provision is a change of a key of mesh-provisioner's `provisions` (ADR 0570 §7): a provision in `waits` is a part of its +// consumer waiting for the operator, said as `module...needs-operator` naming each waiting provision and +// why; the finding ends when no provision of that consumer on that node waits. A key deleted, or one that cannot be read, +// waits no more. +func (w *Watchman) Provision(key string, value []byte, op Op) { + node, consumer, _, _, err := provisions.ParseKey(key) + if err != nil { + return + } + w.mu.Lock() + defer w.mu.Unlock() + id := consumer + "." + node + if w.waiting[id] == nil { + w.waiting[id] = map[string]string{} + } + delete(w.waiting[id], key) + if op == Put { + if p, err := provisions.Decode(value); err != nil { + w.say("the provision %s cannot be read, and is taken as not waiting: %v", key, err) + } else if p.State == provisions.Waits { + w.waiting[id][key] = fmt.Sprintf("%s from %s on %s: %s", p.Provision, p.Provider.Module, p.Provider.Node, p.Why) + } + } + fkey := FindingKey(ScopeModule, id, KindNeedsOperator) + if len(w.waiting[id]) == 0 { + delete(w.waiting, id) + w.unfind(fkey) + return + } + var said []string + for _, k := range sortedKeys(w.waiting[id]) { + said = append(said, w.waiting[id][k]) + } + w.find(Finding{Scope: ScopeModule, ID: id, Kind: KindNeedsOperator, Machine: node, Severity: Warning, Source: "provisions", + Summary: fmt.Sprintf("%s on %s waits for the operator: %s", consumer, node, strings.Join(said, "; ")), + Said: strings.Join(said, "; ")}) +} + +func sortedKeys(m map[string]string) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + // ProvReport is a machine's provisioner client's word on one module's files (`prov.applied..`, ADR 0362 §3). type ProvReport struct { Node string `json:"node"` @@ -786,6 +833,7 @@ var SignalRows = []struct{ Row, Signal, Bound, Kind string }{ {"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent}, {"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing}, {"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy}, + {"provisions", "a provision of mesh-provisioner in waits (ADR 0570 §7)", "at once", KindNeedsOperator}, } // Signal is one signal's pulse as `signals` shows it. diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 0861f54..84e1ef0 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -84,6 +84,8 @@ type Watchman struct { provFailed map[string]map[string]ProvReport closures map[string]map[string]time.Time recount func() + // waiting is, by `.`, each of its provisions in waits, with its words. + waiting map[string]map[string]string // stepBound is how long a build step may run once a node took the ask; a test shortens it. stepBound time.Duration } @@ -94,7 +96,7 @@ func New(out Out, now func() time.Time) *Watchman { pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound, findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{}, haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}, - control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}} + control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{}} } // Logf sets where a write that failed is said; it is silent until set. diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 3b8a9a0..21a1ad5 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -91,7 +91,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 reads := slices.Clone(m.Reads) slices.Sort(reads) if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-controller.lease", - "mesh-delivery-assigner.wants", "node-engine.have"}) { + "mesh-delivery-assigner.wants", "mesh-provisioner.provisions", "node-engine.have"}) { t.Errorf("reads = %v", reads) } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 45e46eb..291908c 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -21,6 +21,7 @@ "mesh-build-node.steps", "mesh-controller.lease", "mesh-delivery-assigner.wants", + "mesh-provisioner.provisions", "node-engine.have" ], "tools": [