G findings4 assigner: the assigner's findings are the seat's generated finding written through views.NewHeldWriter, its catalog halves adopted through the views reader
Builds: design 47 G A-29, A-26, A-22, A-3, A-14, §689, §687, §461
This commit is contained in:
@@ -18,7 +18,6 @@ import (
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
plc "mesh-delivery-assigner/internal/assignment"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"mesh-delivery-assigner/internal/puts"
|
||||
)
|
||||
@@ -119,11 +118,7 @@ func (cw callWait) keep(ctx context.Context, id, callee string, call func() (jso
|
||||
if said || clock().Sub(since) < mcc.CallUnansweredAfter || cw.heard == nil || !cw.heard() || cw.findings.Put == nil {
|
||||
continue
|
||||
}
|
||||
o := mcc.CallUnanswered(id, callee, "", err.Error(), "assigner."+mcc.KindCallUnanswered, since)
|
||||
value, eerr := mcc.EncodeFinding(o.Finding(plc.Module, "", clock().UTC()))
|
||||
if eerr == nil {
|
||||
eerr = cw.findings.Put(key, value)
|
||||
}
|
||||
eerr := cw.findings.Put(assign.CallUnanswered(id, callee, "", err.Error(), "assigner."+mcc.KindCallUnanswered, since).Finding(clock()))
|
||||
if eerr != nil {
|
||||
logf("the finding %s is not written: %v", key, eerr)
|
||||
continue
|
||||
|
||||
@@ -9,99 +9,64 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/sdk/graph"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
// halves follows the catalog half of each watched branch through the seat mesh-catalog (design 47 G §14, §184, §195): one
|
||||
// meshcatalogclient.Catalog per branch, a first view of the verb `revision` and then each `catalog-revision` of that branch,
|
||||
// each entry taken into the world as the value and the graph node the rules read. Of the view the assigner reads only the catalog
|
||||
// half, never its own output back (§195).
|
||||
// meshcatalogclient.Catalog per branch, adopted through the module's views reader (A-26), which asks the first view of the verb
|
||||
// `revision`, hears each `catalog-revision` of that branch and raises and clears the follower's findings; each entry is taken
|
||||
// into the world as the value and the graph node the rules read. Of the view the assigner reads only the catalog half, never its
|
||||
// own output back (§195).
|
||||
type halves struct {
|
||||
w *world
|
||||
call mcc.Call
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
ctx context.Context
|
||||
w *world
|
||||
call mcc.Call
|
||||
reader *views.ProcessReader
|
||||
wake func()
|
||||
logf func(string, ...any)
|
||||
ctx context.Context
|
||||
|
||||
mu sync.Mutex
|
||||
cats map[string]*mcc.Catalog
|
||||
// started is called with each catalog made, for the part that reads their status.
|
||||
started func(*mcc.Catalog)
|
||||
// settled is told once when a branch's first view is answered, for the first pass to wait on (the trunk's alone).
|
||||
settled chan<- struct{}
|
||||
}
|
||||
|
||||
// follow makes the catalog of branch and asks its first view, once; true when it was made now.
|
||||
// follow adopts the catalog of branch through the reader, once; true when it was made now.
|
||||
func (h *halves) follow(branch string) bool {
|
||||
h.mu.Lock()
|
||||
if h.cats[branch] != nil {
|
||||
h.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
c, err := mcc.NewCatalog(h.call, branch)
|
||||
c, _, err := mcc.WireCatalogProcess(h.reader, h.call, branch, func(module string, entry json.RawMessage, deleted bool, _ time.Time) {
|
||||
h.changed(branch, module, entry, deleted)
|
||||
})
|
||||
if err != nil {
|
||||
h.mu.Unlock()
|
||||
h.logf("the catalog half of branch %q is not followed: %v", branch, err)
|
||||
return false
|
||||
}
|
||||
c.Follower.Logf = h.logf
|
||||
c.Follower.Changed = func(module string, entry json.RawMessage, deleted bool, _ time.Time) {
|
||||
h.changed(branch, module, entry, deleted)
|
||||
}
|
||||
if h.cats == nil {
|
||||
h.cats = map[string]*mcc.Catalog{}
|
||||
}
|
||||
h.cats[branch] = c
|
||||
h.mu.Unlock()
|
||||
if h.started != nil {
|
||||
h.started(c)
|
||||
}
|
||||
go func() {
|
||||
if err := c.Start(h.ctx); err != nil {
|
||||
if h.ctx.Err() == nil {
|
||||
h.logf("the first view of the catalog half of branch %q is not answered: %v", branch, err)
|
||||
if branch == trunkBranch && h.settled != nil {
|
||||
go func() {
|
||||
select {
|
||||
case <-c.Ready():
|
||||
h.settled <- struct{}{}
|
||||
case <-h.ctx.Done():
|
||||
}
|
||||
return
|
||||
}
|
||||
if branch == trunkBranch && h.settled != nil {
|
||||
h.settled <- struct{}{}
|
||||
}
|
||||
}()
|
||||
}()
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// hear gives a catalog-revision to every catalog: each takes its own branch's and passes over another's.
|
||||
func (h *halves) hear(m jetstream.Msg) {
|
||||
h.mu.Lock()
|
||||
cats := make([]*mcc.Catalog, 0, len(h.cats))
|
||||
for _, c := range h.cats {
|
||||
cats = append(cats, c)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
for _, c := range cats {
|
||||
if err := c.Hear(h.ctx, m.Data()); err != nil {
|
||||
h.logf("%s on branch %q: %v", m.Subject(), c.Branch, err)
|
||||
}
|
||||
}
|
||||
_ = m.Ack()
|
||||
}
|
||||
|
||||
// followers are the first views the catalogs hold.
|
||||
func (h *halves) followers() []*mcc.Catalog {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
out := make([]*mcc.Catalog, 0, len(h.cats))
|
||||
for _, c := range h.cats {
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// changed takes one entry of branch's half into the world, or a module taken out of it.
|
||||
func (h *halves) changed(branch, module string, raw json.RawMessage, deleted bool) {
|
||||
defer h.wake()
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
"io"
|
||||
"maps"
|
||||
@@ -52,7 +53,6 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -179,6 +179,10 @@ func startBed(t *testing.T, leaveOut string) *bed {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The assigner's own stream of its findings' events (design 47 G §383), which its finding writer says into.
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_FINDINGS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.finding-written.*", "mesh.stream.mesh-delivery-assigner.event.finding-cleared.*", "mesh.stream.mesh-delivery-assigner.event.findings-stated.*"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The assigner's durable consumer on EVENTS, as the grants part makes it from the manifest's consumes: a node-engine seat's
|
||||
// event names the engine's machine last, so the filter is the event with every machine (mesh-grants buscompose
|
||||
// machineEvent; the grants part's own test holds the derivation).
|
||||
@@ -508,7 +512,8 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
"seat-holders": {"n1.node-tools"},
|
||||
"wants": {"n1.mesh-grants"},
|
||||
})
|
||||
graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, assignment.Module+"."+meshcatalog.Seat+"."+views.BranchState(meshcatalog.StateCatalogRevisions, trunkBranch), views.UnansweredKind)
|
||||
// The reader raises the finding of the first view through the module's finding writer (A-26), its id escaped as a key's name.
|
||||
graphKey := views.FindingKey(views.FindingScope, keys.EscapeName(assignment.Module+"."+meshcatalog.Seat+"."+views.BranchState(meshcatalog.StateCatalogRevisions, trunkBranch)), views.UnansweredKind)
|
||||
graphRaised := func() bool {
|
||||
e, err := b.buckets[assign.FindingsBucket].Get(b.ctx, graphKey)
|
||||
return err == nil && strings.Contains(string(e.Value()), "mesh-catalog.revision")
|
||||
@@ -855,12 +860,12 @@ func (b *bed) conditionsSaid(t *testing.T) map[string]oc.Condition {
|
||||
c := out[e.Key()]
|
||||
c.Key, c.State = e.Key(), occ.StateCleared
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
f, err := meshconditionsclient.DecodeFinding(e.Value())
|
||||
if err != nil {
|
||||
var f mcseat.Finding
|
||||
if err := json.Unmarshal(e.Value(), &f); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c = oc.Condition{Key: e.Key(), State: occ.StateRaised, Severity: string(f.Severity), Explanation: f.Explanation,
|
||||
Resolved: f.Resolved, Subject: oc.ConditionSubject{Scope: string(f.Subject.Scope), ID: f.Subject.ID, Machine: f.Subject.Machine}}
|
||||
Subject: oc.ConditionSubject{Scope: string(f.Subject.Scope), ID: f.Subject.ID, Machine: f.Subject.Machine}}
|
||||
}
|
||||
out[e.Key()] = c
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
@@ -18,6 +17,7 @@ import (
|
||||
"os"
|
||||
"os/signal"
|
||||
"slices"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -25,7 +25,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
@@ -110,7 +110,7 @@ var keptStates = []announce.State{
|
||||
{Name: "differences", Readable: true},
|
||||
{Name: "puts", Readable: true},
|
||||
{Name: "wants", Readable: true},
|
||||
{Name: mcc.FindingsState, Readable: true},
|
||||
{Name: assign.FindingsState, Readable: true},
|
||||
{Name: buckets.PoliciesState, Readable: true},
|
||||
{Name: mma.StateSeatHolders, Readable: true},
|
||||
{Name: views.MarksState, Readable: true}, // what its views family said each state up to (views.go)
|
||||
@@ -235,11 +235,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
thisNode = account.Node
|
||||
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
||||
findings := findingsOf(ctx, kvs, logger.Printf)
|
||||
findings, writer := findingsOf(ctx, account, conn, js, said, kvs, logger.Printf)
|
||||
if writer != nil {
|
||||
defer writer.Close()
|
||||
}
|
||||
// Each change of the states its readers read is said on its views family (ADR 0615 §5, ADR 0619 §2); a watch of it
|
||||
// that ends stops the program, as a watch of what a pass reads does.
|
||||
viewErrs := make(chan error, len(viewedStates))
|
||||
startViews(ctx, js, conn, account, kvs, findings, viewErrs, logger.Printf)
|
||||
startViews(ctx, js, conn, account, kvs, wireOf(writer), viewErrs, logger.Printf)
|
||||
go func() {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -248,27 +251,52 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
conn.Close()
|
||||
}
|
||||
}()
|
||||
iv := &catalogViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf,
|
||||
iv := &catalogViews{js: js, account: account, call: said.Call, writer: wireOf(writer), logf: logger.Printf,
|
||||
status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)}
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger)
|
||||
}
|
||||
|
||||
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
|
||||
// condition keeper carries (novox/hq ADR 0590 §3). Without it the conditions are said straight to the router.
|
||||
func findingsOf(ctx context.Context, kvs kvSet, logf func(string, ...any)) assign.Findings {
|
||||
func findingsOf(ctx context.Context, account process.Account, conn *nats.Conn, js jetstream.JetStream, said *process.Conn, kvs kvSet, logf func(string, ...any)) (assign.Findings, *heldWriter) {
|
||||
var f assign.Findings
|
||||
if kv := kvs[assign.FindingsBucket]; kv != nil {
|
||||
f.Put = func(key string, value []byte) error { _, err := kv.Put(ctx, key, value); return err }
|
||||
f.Delete = func(key string) error { return kv.Delete(ctx, key) }
|
||||
f.Keys = func() ([]string, error) {
|
||||
keys, err := kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return keys, err
|
||||
}
|
||||
if kvs[assign.FindingsBucket] == nil {
|
||||
return f, nil
|
||||
}
|
||||
return f
|
||||
state := said.State(assign.FindingsState)
|
||||
w, err := views.NewHeldWriter(views.WriterConfig[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent]{
|
||||
Module: account.Module, State: state, Node: account.Node, Asked: mcseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf})
|
||||
if err != nil {
|
||||
logf("the assigner's finding writer cannot be made, so its findings are not written: %v", err)
|
||||
return f, nil
|
||||
}
|
||||
// A writer that cannot start (the stream of its asks is not there yet) is said; its findings are then kept in the
|
||||
// state alone and said by the next start (as the node-engine's is).
|
||||
if err := w.Start(ctx); err != nil {
|
||||
logf("the finding writer cannot start, its findings are kept in the state alone: %v", err)
|
||||
}
|
||||
f.Put = func(x mcseat.Finding) error { return w.Write(ctx, x) }
|
||||
f.Delete = func(key string) error {
|
||||
scope, rest, _ := strings.Cut(key, ".")
|
||||
i := strings.LastIndex(rest, ".")
|
||||
if i < 0 {
|
||||
return fmt.Errorf("%q is no finding's key", key)
|
||||
}
|
||||
return w.Clear(ctx, scope, rest[:i], rest[i+1:], "the assigner no longer observes it")
|
||||
}
|
||||
f.Keys = state.Keys
|
||||
return f, w
|
||||
}
|
||||
|
||||
// heldWriter is the assigner's finding writer: its findings are raised as they happen, so the writer holds them (design 47 G A-29).
|
||||
type heldWriter = views.HeldWriter[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent]
|
||||
|
||||
// wireOf is w as the views parts' finding writer; nil when the assigner has none, so no part writes through a nil writer.
|
||||
func wireOf(w *heldWriter) views.WireWriter {
|
||||
if w == nil {
|
||||
return nil
|
||||
}
|
||||
return w
|
||||
}
|
||||
|
||||
// thisNode is the node this process runs on, as its credential names it: said with each of the seat's events.
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
plc "mesh-delivery-assigner/internal/assignment"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
@@ -31,8 +34,7 @@ import (
|
||||
// (`<module>.<state>` read as the bucket `<module>_<state>`), so it is granted what it reads and reads nothing else.
|
||||
func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
// Review round 2: the assigner's views emitter writes its finding view-unsaid (Emitter.Findings set) when it keeps findings.
|
||||
findings := assign.Findings{Put: func(string, []byte) error { return nil }, Delete: func(string) error { return nil }}
|
||||
if viewsEmitter(nil, process.Account{}, nil, findings, nil, t.Logf).Findings == nil {
|
||||
if viewsEmitter(nil, process.Account{}, nil, &heldWriter{}, t.Logf).Findings == nil {
|
||||
t.Error("the assigner's views emitter writes no view-unsaid finding")
|
||||
}
|
||||
// 1297/B2a: without a credential the program does not start, and says why.
|
||||
@@ -197,3 +199,12 @@ func (e kvEntry) Created() time.Time { return time.Time{} }
|
||||
func (w *watchKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) {
|
||||
return nil, jetstream.ErrNoKeysFound
|
||||
}
|
||||
|
||||
// putInto is an assign.Findings Put that keeps each finding in into, as the finding writer would, by its key and its words.
|
||||
func putInto(into map[string][]byte) func(mcseat.Finding) error {
|
||||
return func(f mcseat.Finding) error {
|
||||
raw, err := json.Marshal(f)
|
||||
into[views.FindingKey(f.Scope, f.ID, f.Kind)] = raw
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -149,13 +149,9 @@ func (p *passer) unansweredReads(known []string, online map[string]bool) {
|
||||
if p.calls == nil || p.calls.Keeper != p.reads {
|
||||
p.calls = &enginereads.ReadsCalls{Keeper: p.reads, Bound: mcc.CallUnansweredAfter, Now: p.clock, Logf: p.logf,
|
||||
Raise: func(_ context.Context, u enginereads.UnansweredCall) error {
|
||||
o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "assigner."+mcc.KindCallUnanswered, u.Since)
|
||||
o := assign.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "assigner."+mcc.KindCallUnanswered, u.Since)
|
||||
o.Needs = "assigner_engine_reads with ask_again asks the node again now and shows its statement."
|
||||
value, err := mcc.EncodeFinding(o.Finding(plc.Module, "", u.At))
|
||||
if err == nil {
|
||||
err = p.findings.Put(o.Key(), value)
|
||||
}
|
||||
return err
|
||||
return p.findings.Put(o.Finding(u.At))
|
||||
},
|
||||
Clear: func(_ context.Context, node string) error {
|
||||
return p.findings.Delete(mcc.Key(mcc.ScopeCall, enginereads.CallID(node), mcc.KindCallUnanswered))
|
||||
@@ -1126,9 +1122,9 @@ func (p *passer) provisionFindings(failed map[string]meshprovisionerclient.Provi
|
||||
if p.failedProvisions[key] {
|
||||
continue
|
||||
}
|
||||
value, err := assign.ProvisionFailedFinding(key, pr, now())
|
||||
f, err := assign.ProvisionFailedFinding(key, pr, now())
|
||||
if err == nil {
|
||||
err = p.findings.Put(assign.ProvisionFailedKey(key), value)
|
||||
err = p.findings.Put(f)
|
||||
}
|
||||
if err != nil {
|
||||
p.logf("the failed provision %s is not said: %v", key, err)
|
||||
@@ -1178,9 +1174,9 @@ func (p *passer) provisionsMissing(ctx context.Context, missing map[string]bool)
|
||||
if p.missingSaid[key] || p.findings.Put == nil {
|
||||
continue
|
||||
}
|
||||
value, err := assign.ProvisionMissingFinding(key, MissingBound, p.clock())
|
||||
f, err := assign.ProvisionMissingFinding(key, MissingBound, p.clock())
|
||||
if err == nil {
|
||||
err = p.findings.Put(assign.ProvisionMissingKey(key), value)
|
||||
err = p.findings.Put(f)
|
||||
}
|
||||
if err != nil {
|
||||
p.logf("the missing provision %s is not said: %v", key, err)
|
||||
@@ -1226,9 +1222,9 @@ func (p *passer) unplacedFindings() {
|
||||
if p.unplaced[node] {
|
||||
continue
|
||||
}
|
||||
value, err := assign.UnplacedFinding(node, p.clock())
|
||||
f, err := assign.UnplacedFinding(node, p.clock())
|
||||
if err == nil {
|
||||
err = p.findings.Put(assign.UnplacedKey(node), value)
|
||||
err = p.findings.Put(f)
|
||||
}
|
||||
if err != nil {
|
||||
p.logf("the unplaced node %s is not said: %v", node, err)
|
||||
|
||||
@@ -209,7 +209,7 @@ func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) {
|
||||
for _, heard := range []bool{false, true} {
|
||||
clock, findings := now, map[string][]byte{}
|
||||
cw := callWait{heard: func() bool { return heard }, now: func() time.Time { return clock },
|
||||
findings: assign.Findings{Put: func(k string, v []byte) error { findings[k] = v; return nil },
|
||||
findings: assign.Findings{Put: putInto(findings),
|
||||
Delete: func(k string) error { delete(findings, k); return nil }}}
|
||||
tries, during := 0, 0
|
||||
answer, err := cw.keep(context.Background(), asksCallID, channelAsksVerb, func() (json.RawMessage, error) {
|
||||
@@ -332,7 +332,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
|
||||
return nil, context.DeadlineExceeded
|
||||
}
|
||||
q := &passer{w: w, logf: t.Logf, now: func() time.Time { return clock }, findings: assign.Findings{
|
||||
Put: func(k string, v []byte) error { held[k] = v; return nil },
|
||||
Put: putInto(held),
|
||||
Delete: func(k string) error { delete(held, k); return nil },
|
||||
Keys: func() ([]string, error) { return slices.Collect(maps.Keys(held)), nil }},
|
||||
reads: &enginereads.Keeper{Call: silent, Sync: true, Now: func() time.Time { return clock }}}
|
||||
|
||||
@@ -24,7 +24,6 @@ import (
|
||||
|
||||
mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -147,6 +146,9 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
for _, stream := range raw0.Streams {
|
||||
if strings.HasSuffix(stream.Name, "-findings") { // the part's own finding events, not a state of the seat
|
||||
continue
|
||||
}
|
||||
if !reads["stream "+stream.Name] {
|
||||
t.Errorf("no verb reads the seat's stream %s", stream.Name)
|
||||
}
|
||||
@@ -283,7 +285,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
}
|
||||
findings := map[string][]byte{}
|
||||
p := &passer{w: w, logf: func(string, ...any) {}, findings: assign.Findings{
|
||||
Put: func(k string, v []byte) error { findings[k] = v; return nil },
|
||||
Put: putInto(findings),
|
||||
Delete: func(k string) error { delete(findings, k); return nil }}}
|
||||
p.unplacedFindings()
|
||||
if _, ok := findings[assign.UnplacedKey("g14")]; !ok || len(findings) != 1 {
|
||||
@@ -471,7 +473,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
t.Run("1954/D1: a failed provision is the assigner's finding until it is provided", func(t *testing.T) {
|
||||
findings := map[string][]byte{}
|
||||
p := &passer{w: newWorld(), logf: func(string, ...any) {}, kvs: kvSet{buckets.Provisions: nil}, findings: assign.Findings{
|
||||
Put: func(k string, v []byte) error { findings[k] = v; return nil },
|
||||
Put: putInto(findings),
|
||||
Delete: func(k string) error { delete(findings, k); return nil }}}
|
||||
failed, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "db", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
|
||||
Provider: meshprovisionerclient.End{Module: "postgres", Node: "novox"}, State: meshprovisionerclient.Failed, Why: "the role exists already"})
|
||||
@@ -612,7 +614,7 @@ func wantsVerbs(t *testing.T) {
|
||||
state, watches[state], watches[state+" by key"], indexer.asks[state])
|
||||
}
|
||||
}
|
||||
for _, state := range []string{haveState, mcc.FindingsState} {
|
||||
for _, state := range []string{haveState, assign.FindingsState} {
|
||||
if watches[state] != 1 {
|
||||
t.Errorf("%s was watched %d times, want once", state, watches[state])
|
||||
}
|
||||
|
||||
@@ -6,23 +6,19 @@ import (
|
||||
"mesh-delivery-assigner/internal/assignment"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
|
||||
mcat "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
plc "mesh-delivery-assigner/internal/assignment"
|
||||
@@ -47,7 +43,7 @@ var readTools = []viewedTool{
|
||||
{"assigner_read_assignments", mma.StateAssignments, "where each module is placed, by module."},
|
||||
{"assigner_read_seat_holders", mma.StateSeatHolders, "who holds which seat on which node."},
|
||||
{"assigner_read_wants", "wants", "what each node should run: want.<node>.<module>."},
|
||||
{"assigner_read_findings", mcc.FindingsState, "the assigner's findings, which the condition keeper raises and clears."},
|
||||
{"assigner_read_findings", assign.FindingsState, "the assigner's findings, which the condition keeper raises and clears."},
|
||||
{"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."},
|
||||
{"assigner_read_differences", "differences", "every difference between a want and what a node runs."},
|
||||
{"assigner_read_copies", copiesState, "the copies the assigner made of a retired state: when the copy into assignments ended and how many keys it copied."},
|
||||
@@ -58,7 +54,7 @@ var viewedStates = map[string]string{
|
||||
mma.StateAssignments: plc.AssignmentsBucket,
|
||||
mma.StateSeatHolders: plc.SeatHoldersBucket,
|
||||
"wants": buckets.Wants,
|
||||
mcc.FindingsState: assign.FindingsBucket,
|
||||
assign.FindingsState: assign.FindingsBucket,
|
||||
"asks": buckets.Asks,
|
||||
"differences": buckets.Differences,
|
||||
copiesState: copiesBucket,
|
||||
@@ -83,18 +79,10 @@ func keptSource(name string) views.Source { return views.KeptSource{State: stdio
|
||||
|
||||
// startViews says each change of the assigner's viewed states on its views family, once its marks state is bound,
|
||||
// each state as soon as its bucket is (kvs, or bound once it appears). A watch that ends is sent on errs.
|
||||
func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs kvSet, findings assign.Findings, errs chan<- error, logf func(string, ...any)) {
|
||||
tooLarge := func(state, key string, size int) {
|
||||
if findings.Put == nil {
|
||||
return
|
||||
}
|
||||
if err := viewFindings(findings).Put(views.TooLargeFinding(assignment.Module, state, key, size, time.Now())); err != nil {
|
||||
logf("views: the finding of %s.%s too large is not written: %v", state, key, err)
|
||||
}
|
||||
}
|
||||
func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs kvSet, writer views.WireWriter, errs chan<- error, logf func(string, ...any)) {
|
||||
process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) {
|
||||
em := viewsEmitter(js, account, mk, findings, tooLarge, logf)
|
||||
for _, state := range []string{"asks", "differences", mcc.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants", copiesState} {
|
||||
em := viewsEmitter(js, account, mk, writer, logf)
|
||||
for _, state := range []string{"asks", "differences", assign.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants", copiesState} {
|
||||
state := state
|
||||
start := func(kv jetstream.KeyValue) {
|
||||
if err := em.Start(ctx, state, kv, errs); err != nil {
|
||||
@@ -111,35 +99,27 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
|
||||
}
|
||||
|
||||
// viewsEmitter is the assigner's emitter of its views family: a change not said past views.Bound is its finding
|
||||
// view-unsaid, in its findings (review round 2); with no findings state it is said in the log only.
|
||||
func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, findings assign.Findings, tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter {
|
||||
em := views.Emitter{Owner: assignment.Module, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf}
|
||||
if findings.Put != nil && findings.Delete != nil {
|
||||
em.Findings = viewFindings(findings)
|
||||
}
|
||||
return em
|
||||
}
|
||||
|
||||
// viewFindings is the assigner's findings as the views' FindingWriter, in the seat mesh-conditions' layout.
|
||||
func viewFindings(f assign.Findings) views.FindingWriter {
|
||||
return sdkfindings.WriterOver(f.Put, f.Delete)
|
||||
// view-unsaid, in its findings, written through its finding writer (review round 2); with none it is said in the log only.
|
||||
func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, writer views.WireWriter, logf func(string, ...any)) views.Emitter {
|
||||
return views.Emitter{Owner: assignment.Module, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, Findings: writer, Logf: logf}
|
||||
}
|
||||
|
||||
// firstViewsBucket is the assigner's own state of its first views (views.StatusState).
|
||||
var firstViewsBucket = assignment.Module + "_" + views.StatusState
|
||||
|
||||
// catalogViews is the assigner's first view of the catalog half of each watched branch (design 47 G §14, §184, §195): asked
|
||||
// at the start through the seat mesh-catalog's verb `revision` (the trunk's, then each branch a registration names), then
|
||||
// followed on its event `catalog-revision`, asking once more on a gap; past the bound its finding names the seat and the verb,
|
||||
// cleared by the answer. Its status is read by assigner_first_views and asked again by assigner_first_view_again.
|
||||
// catalogViews is the assigner's first view of the catalog half of each watched branch (design 47 G §14, §184, §195, A-26): the
|
||||
// module's views reader adopts each branch's follower, so it asks the seat mesh-catalog's verb `revision` (the trunk's, then
|
||||
// each branch a registration names), follows its event `catalog-revision`, asks once more on a gap, and past the bound raises
|
||||
// the finding naming the seat and the verb, cleared by the answer, in the assigner's findings. Its status is read by
|
||||
// assigner_first_views and asked again by assigner_first_view_again.
|
||||
type catalogViews struct {
|
||||
js jetstream.JetStream
|
||||
account process.Account
|
||||
call func(string, any) (json.RawMessage, error)
|
||||
findings assign.Findings
|
||||
status *process.AwaitedBucket
|
||||
logf func(string, ...any)
|
||||
hub *process.EventHub
|
||||
js jetstream.JetStream
|
||||
account process.Account
|
||||
call func(string, any) (json.RawMessage, error)
|
||||
writer views.WireWriter
|
||||
status *process.AwaitedBucket
|
||||
logf func(string, ...any)
|
||||
hub *process.EventHub
|
||||
|
||||
h *halves
|
||||
}
|
||||
@@ -149,70 +129,31 @@ type catalogViews struct {
|
||||
// seconds). A test shortens them.
|
||||
var firstViewBound, firstViewWait time.Duration
|
||||
|
||||
// start follows the trunk's catalog half and hears the seat's catalog-revision; it answers how many settles to wait for.
|
||||
// start adopts the trunk's catalog half through the reader and starts the reader; it answers how many settles to wait for.
|
||||
func (iv *catalogViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int {
|
||||
finding := func(s views.Status, raised bool) {
|
||||
if iv.findings.Put == nil {
|
||||
return
|
||||
}
|
||||
f := views.UnansweredFinding(assignment.Module, s, time.Now())
|
||||
var err error
|
||||
if raised {
|
||||
err = viewFindings(iv.findings).Put(f)
|
||||
} else {
|
||||
err = viewFindings(iv.findings).Delete(f)
|
||||
}
|
||||
if err != nil {
|
||||
iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err)
|
||||
}
|
||||
reader := &views.ProcessReader{Module: assignment.Module, Call: iv.call, Logf: iv.logf, Hub: iv.hub, Bound: firstViewBound, Wait: firstViewWait}
|
||||
if iv.writer != nil {
|
||||
reader.Findings = func() views.WireWriter { return iv.writer }
|
||||
}
|
||||
var statusKV jetstream.KeyValue
|
||||
var mu sync.Mutex
|
||||
iv.h = &halves{w: w, call: iv.call, wake: wake, logf: iv.logf, ctx: ctx, settled: settled}
|
||||
iv.h.started = func(c *mcat.Catalog) {
|
||||
f := c.Follower
|
||||
f.Bound, f.Wait = firstViewBound, firstViewWait
|
||||
f.Raise, f.Clear = func(s views.Status) { finding(s, true) }, func(s views.Status) { finding(s, false) }
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if statusKV != nil {
|
||||
f.Said = views.KeepStatus(ctx, statusKV, iv.logf)
|
||||
}
|
||||
}
|
||||
if iv.status != nil {
|
||||
iv.status.Ready(func(kv jetstream.KeyValue) {
|
||||
mu.Lock()
|
||||
statusKV = kv
|
||||
mu.Unlock()
|
||||
var all []*views.Follower
|
||||
for _, c := range iv.h.followers() {
|
||||
c.Follower.Said = views.KeepStatus(ctx, kv, iv.logf)
|
||||
all = append(all, c.Follower)
|
||||
}
|
||||
if err := views.HearAgain(ctx, kv, all, iv.logf); err != nil {
|
||||
iv.logf("views: assigner_first_view_again is not heard (%v) until the assigner starts again", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
// The hub says when the bus made its consumer: a picture asked before then is asked again (views.Follower.ConsumerMade).
|
||||
iv.hub.OnBound(func(created time.Time) {
|
||||
for _, c := range iv.h.followers() {
|
||||
c.Follower.ConsumerMade(ctx, created)
|
||||
}
|
||||
})
|
||||
// Not heard, the catalog-revision is the finding family-not-heard until it binds (ADR 0615 §5): the handler is registered
|
||||
// on the module's hub, which the assigner starts once every handler is.
|
||||
_ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the catalog half is the first views' alone until the assigner starts again", func() error {
|
||||
return iv.hub.Handle(meshcatalog.SubjectCatalogRevisionAll, iv.h.hear)
|
||||
})
|
||||
iv.h = &halves{w: w, call: iv.call, reader: reader, wake: wake, logf: iv.logf, ctx: ctx, settled: settled}
|
||||
iv.h.follow(trunkBranch)
|
||||
reader.Start(ctx, iv.js, iv.account, iv.status)
|
||||
return 1
|
||||
}
|
||||
|
||||
// notHeardWriter writes the assigner's finding of a family it does not hear into its findings (heard.Writer); with no
|
||||
// findings state it is said in the log alone.
|
||||
func notHeardWriter(findings assign.Findings, logf func(string, ...any)) heard.Writer {
|
||||
w := heard.Writer{Observer: assignment.Module, Logf: logf, Put: findings.Put}
|
||||
w := heard.Writer{Observer: assignment.Module, Logf: logf}
|
||||
if findings.Put != nil {
|
||||
w.Put = func(_ string, value []byte) error {
|
||||
var f mcseat.Finding
|
||||
if err := json.Unmarshal(value, &f); err != nil {
|
||||
return err
|
||||
}
|
||||
return findings.Put(f)
|
||||
}
|
||||
}
|
||||
if findings.Delete != nil {
|
||||
w.Delete = func(key string) error {
|
||||
if keys, err := findings.Keys(); err == nil && !slices.Contains(keys, key) {
|
||||
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -170,7 +169,7 @@ func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) {
|
||||
out["published"] = published(values)
|
||||
}
|
||||
// The assigner's findings, its said conditions as the condition keeper reads them (§15).
|
||||
if found, err := s.values(mcc.FindingsState); err != nil {
|
||||
if found, err := s.values(assign.FindingsState); err != nil {
|
||||
out["findings-unread"] = err.Error()
|
||||
} else {
|
||||
keys := []string{}
|
||||
|
||||
@@ -16,10 +16,10 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
// 1297/B9 (the assigner's half): the ask about a held fingerprint is a valid ask of the router and of mesh-sdk's layout at
|
||||
@@ -292,11 +292,10 @@ func saidConditionsAreFindings(t *testing.T) {
|
||||
{SeatWaitsCondition([]string{"assignments"}), SeatWaitsCondition(nil), "module.mesh-delivery-assigner.seat-states"},
|
||||
{BackupCondition("ace", "mail", fNew, "disk full", true), BackupCondition("ace", "mail", fNew, "", false), "module.ace.mail." + strings.TrimPrefix(fNew, meshcatalogclient.SourcePrefix2) + ".backup"},
|
||||
} {
|
||||
puts, deletes, said := map[string]mcseat.FindingsByKeyValue{}, []string{}, 0
|
||||
f := Findings{Put: func(k string, v []byte) error {
|
||||
got, err := mcc.DecodeFinding(v)
|
||||
puts[k] = got
|
||||
return err
|
||||
puts, deletes, said := map[string]mcseat.Finding{}, []string{}, 0
|
||||
f := Findings{Put: func(f mcseat.Finding) error {
|
||||
puts[views.FindingKey(f.Scope, f.ID, f.Kind)] = f
|
||||
return nil
|
||||
}, Delete: func(k string) error { deletes = append(deletes, k); return nil }, Now: func() time.Time { return at }}
|
||||
say := f.Say(func(oc.Condition) error { said++; return nil })
|
||||
if err := say(c.raised); err != nil {
|
||||
@@ -314,7 +313,7 @@ func saidConditionsAreFindings(t *testing.T) {
|
||||
t.Fatalf("%s: with no findings bucket said %d (%v), want straight once", c.key, said, err)
|
||||
}
|
||||
}
|
||||
if _, _, _, err := FindingOf(oc.Condition{Key: "another.x", State: occ.StateRaised}, at); err == nil {
|
||||
if _, _, err := FindingOf(oc.Condition{Key: "another.x", State: occ.StateRaised}, at); err == nil {
|
||||
t.Error("a condition not the assigner's is a finding")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
// The assigner's said conditions become findings (novox/hq ADR 0564 §2, issue 1952 WM6, note 16): every condition it says
|
||||
@@ -21,15 +22,66 @@ import (
|
||||
// assigner no longer says it straight (never both).
|
||||
|
||||
// FindingsBucket is the assigner's findings.
|
||||
var FindingsBucket = mcc.FindingsBucket(assignment.Module)
|
||||
// FindingsState is the assigner's state of its findings, and FindingsBucket its bucket.
|
||||
const FindingsState = "findings"
|
||||
|
||||
// FindingOf is a said condition as a finding: its key `<scope>.<id>.<kind>`, from the condition's own key
|
||||
// `mesh-delivery-assigner.<kind>.<id…>` (the subject's id when the key names none), its words unchanged. raised is false for
|
||||
// a clear, whose key is to be deleted; value is nil then.
|
||||
func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised bool, err error) {
|
||||
var FindingsBucket = assignment.Module + "_" + FindingsState
|
||||
|
||||
// KeyID is an id as the parts of a key: each dot-separated part made safe for a bus key, as the key of a condition is
|
||||
// (mcc.Key), so a finding's key keeps the shape its condition had.
|
||||
func KeyID(id string) string {
|
||||
key := mcc.Key("x", id, "x")
|
||||
return strings.TrimSuffix(strings.TrimPrefix(key, "x."), ".x")
|
||||
}
|
||||
|
||||
// Observed is what the assigner observes wrong, before it is the seat mesh-conditions' generated finding.
|
||||
type Observed struct {
|
||||
Scope, ID, Kind, Machine string
|
||||
Also []string
|
||||
Severity string
|
||||
Summary, Said string
|
||||
Headline, Explanation string
|
||||
Needs, Source string
|
||||
Actions []mcseat.Action
|
||||
}
|
||||
|
||||
// Finding is o as the seat's generated finding at at (design 47 G §686, A-22): its id made safe as a key's part, the resolved
|
||||
// words not a field of it, its observer the assigner; the writer completes its `since`.
|
||||
func (o Observed) Finding(at time.Time) mcseat.Finding {
|
||||
at = at.UTC()
|
||||
id := KeyID(o.ID)
|
||||
said := o.Said
|
||||
if said == "" {
|
||||
said = o.Summary
|
||||
}
|
||||
sev := o.Severity
|
||||
if sev == "" {
|
||||
sev = mcseat.FindingSeverityWarning
|
||||
}
|
||||
return mcseat.Finding{Scope: o.Scope, ID: id, Kind: o.Kind, Severity: sev, Summary: o.Summary, Headline: o.Headline,
|
||||
Explanation: o.Explanation, Needs: o.Needs, Actions: o.Actions, Source: o.Source, Observer: assignment.Module, At: at,
|
||||
Evidence: []mcseat.Evidence{{At: at, Said: said}},
|
||||
Subject: mcseat.Subject{Scope: o.Scope, ID: id, Machine: o.Machine, Also: o.Also}}
|
||||
}
|
||||
|
||||
// CallUnanswered is the observation of a kept call to callee (a seat verb's or a tool's key) about id (what the assigner waits
|
||||
// on: a node, a key), unanswered since since, its last fault said. Its key is `call.<id>.call-unanswered`.
|
||||
func CallUnanswered(id, callee, machine, fault, source string, since time.Time) Observed {
|
||||
return Observed{Scope: mcc.ScopeCall, ID: id, Kind: mcc.KindCallUnanswered, Machine: machine, Severity: mcseat.FindingSeverityWarning,
|
||||
Summary: fmt.Sprintf("the call %s about %s is unanswered since %s", callee, id, since.UTC().Format(time.RFC3339)),
|
||||
Said: fault, Headline: "A call whose answer is kept is not answered",
|
||||
Explanation: fmt.Sprintf("The call %s is made again on every transport fault and on the callee's event; it has "+
|
||||
"not been answered since %s. Its last fault: %s.", callee, since.UTC().Format(time.RFC3339), fault),
|
||||
Source: source}
|
||||
}
|
||||
|
||||
// FindingOf is a said condition as a finding: from the condition's own key `mesh-delivery-assigner.<kind>.<id…>` (the
|
||||
// subject's id when the key names none), its words unchanged but the resolved ones. raised is false for a clear, whose
|
||||
// finding names only its scope, id and kind.
|
||||
func FindingOf(c oc.Condition, at time.Time) (f mcseat.Finding, raised bool, err error) {
|
||||
rest, ok := strings.CutPrefix(c.Key, assignment.Module+".")
|
||||
if !ok || rest == "" {
|
||||
return "", nil, false, fmt.Errorf("the condition %q is not the assigner's", c.Key)
|
||||
return f, false, fmt.Errorf("the condition %q is not the assigner's", c.Key)
|
||||
}
|
||||
kind, id, _ := strings.Cut(rest, ".")
|
||||
if id == "" {
|
||||
@@ -39,12 +91,11 @@ func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised b
|
||||
if scope == "" {
|
||||
scope = mcc.ScopeModule
|
||||
}
|
||||
key = mcc.Key(scope, id, kind)
|
||||
if c.State == occ.StateCleared {
|
||||
return key, nil, false, nil
|
||||
return mcseat.Finding{Scope: scope, ID: KeyID(id), Kind: kind}, false, nil
|
||||
}
|
||||
if c.State != occ.StateRaised {
|
||||
return "", nil, false, fmt.Errorf("the condition %s is neither raised nor cleared", c.Key)
|
||||
return f, false, fmt.Errorf("the condition %s is neither raised nor cleared", c.Key)
|
||||
}
|
||||
summary := c.Summary
|
||||
if summary == "" {
|
||||
@@ -54,16 +105,14 @@ func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised b
|
||||
for _, a := range c.Actions {
|
||||
actions = append(actions, mcseat.Action{Label: a.Label, Verb: a.Verb, Machine: a.Machine, Arguments: a.Arguments, Level: string(a.Level)})
|
||||
}
|
||||
f := mcc.Observation{Scope: scope, ID: id, Kind: kind, Machine: c.Subject.Machine, Severity: string(c.Severity), Summary: summary,
|
||||
Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: actions,
|
||||
Source: "assigner." + kind}.Finding(assignment.Module, "", at)
|
||||
value, err = mcc.EncodeFinding(f)
|
||||
return key, value, true, err
|
||||
return Observed{Scope: scope, ID: id, Kind: kind, Machine: c.Subject.Machine, Severity: string(c.Severity), Summary: summary,
|
||||
Headline: c.Headline, Explanation: c.Explanation, Needs: c.Needs, Actions: actions, Source: "assigner." + kind}.Finding(at), true, nil
|
||||
}
|
||||
|
||||
// Findings is where the assigner's findings go: the bucket's put and delete.
|
||||
// Findings is where the assigner's findings go: its finding writer's write and clear (design 47 G §689, A-29), by the
|
||||
// finding's key `<scope>.<id>.<kind>`.
|
||||
type Findings struct {
|
||||
Put func(key string, value []byte) error
|
||||
Put func(f mcseat.Finding) error
|
||||
Delete func(key string) error
|
||||
Keys func() ([]string, error)
|
||||
Now func() time.Time
|
||||
@@ -80,12 +129,13 @@ func (f Findings) Say(say func(oc.Condition) error) func(oc.Condition) error {
|
||||
if f.Now != nil {
|
||||
now = f.Now
|
||||
}
|
||||
key, value, raised, err := FindingOf(c, now())
|
||||
finding, raised, err := FindingOf(c, now())
|
||||
key := views.FindingKey(finding.Scope, finding.ID, finding.Kind)
|
||||
switch {
|
||||
case err != nil:
|
||||
errs = append(errs, err)
|
||||
case raised:
|
||||
if err := f.Put(key, value); err != nil {
|
||||
if err := f.Put(finding); err != nil {
|
||||
errs = append(errs, fmt.Errorf("the finding %s cannot be written: %w", key, err))
|
||||
}
|
||||
default:
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"time"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -65,19 +66,17 @@ func ProvisionFailedKey(key string) string {
|
||||
}
|
||||
|
||||
// ProvisionFailedFinding is the finding of a failed provision, naming its consumer, its provider and why.
|
||||
func ProvisionFailedFinding(key string, p meshprovisionerclient.Provision, at time.Time) ([]byte, error) {
|
||||
func ProvisionFailedFinding(key string, p meshprovisionerclient.Provision, at time.Time) (mcseat.Finding, error) {
|
||||
summary := fmt.Sprintf("%s for %s on %s failed at %s on %s: %s, so %s is not moved there until it is provided", p.Provision,
|
||||
p.Consumer.Module, p.Consumer.Node, p.Provider.Module, p.Provider.Node, p.Why, p.Consumer.Module)
|
||||
return encode(mcc.Observation{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionFailed, Machine: p.Consumer.Node, Also: also(p.Provider.Node, p.Consumer.Node),
|
||||
return encode(Observed{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionFailed, Machine: p.Consumer.Node, Also: also(p.Provider.Node, p.Consumer.Node),
|
||||
Summary: summary, Said: p.Why, Headline: clipHeadline(p.Consumer.Module + " waits for a provision that failed"),
|
||||
Explanation: occ.NothingToDo + " " + summary + ". It clears once the provider provides it.",
|
||||
Resolved: p.Provision + " for " + p.Consumer.Module + " is provided", Source: "assigner." + KindProvisionFailed}, at)
|
||||
Source: "assigner." + KindProvisionFailed}, at)
|
||||
}
|
||||
|
||||
// encode is the observation as the assigner's finding at at, in the seat's layout.
|
||||
func encode(o mcc.Observation, at time.Time) ([]byte, error) {
|
||||
return mcc.EncodeFinding(o.Finding(assignment.Module, "", at))
|
||||
}
|
||||
// encode is the observation as the assigner's finding at at.
|
||||
func encode(o Observed, at time.Time) (mcseat.Finding, error) { return o.Finding(at), nil }
|
||||
|
||||
func also(provider, consumer string) []string {
|
||||
if provider == "" || provider == consumer {
|
||||
@@ -95,18 +94,18 @@ func ProvisionMissingKey(key string) string {
|
||||
}
|
||||
|
||||
// ProvisionMissingFinding is the finding of a required provision with no key past bound (key `<node>.<module>.<provision>`).
|
||||
func ProvisionMissingFinding(key string, bound time.Duration, at time.Time) ([]byte, error) {
|
||||
func ProvisionMissingFinding(key string, bound time.Duration, at time.Time) (mcseat.Finding, error) {
|
||||
parts := strings.SplitN(key, ".", 3)
|
||||
if len(parts) != 3 {
|
||||
return nil, fmt.Errorf("%q is not <node>.<module>.<provision>", key)
|
||||
return mcseat.Finding{}, fmt.Errorf("%q is not <node>.<module>.<provision>", key)
|
||||
}
|
||||
node, module, provision := parts[0], parts[1], parts[2]
|
||||
summary := fmt.Sprintf("%s on %s requires %s and mesh-provisioner has composed no key for it in %s, so %s is not moved there: "+
|
||||
"no provider of it may be placed", module, node, provision, bound, module)
|
||||
return encode(mcc.Observation{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionMissing, Machine: node,
|
||||
return encode(Observed{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionMissing, Machine: node,
|
||||
Summary: summary, Headline: clipHeadline(module + " on " + node + " lacks " + provision),
|
||||
Explanation: occ.NothingToDo + " " + summary + ". It clears once the provision has its key.",
|
||||
Resolved: provision + " for " + module + " on " + node + " has its key", Source: "assigner." + KindProvisionMissing}, at)
|
||||
Source: "assigner." + KindProvisionMissing}, at)
|
||||
}
|
||||
|
||||
// KindUnplaced is, from the lease write, a node that runs but no assignments names (ADR 0571 §6).
|
||||
@@ -118,13 +117,13 @@ func UnplacedKey(node string) string {
|
||||
}
|
||||
|
||||
// UnplacedFinding is the finding of a node no assignments names since the lease write.
|
||||
func UnplacedFinding(node string, at time.Time) ([]byte, error) {
|
||||
func UnplacedFinding(node string, at time.Time) (mcseat.Finding, error) {
|
||||
summary := fmt.Sprintf("%s runs, but no assignments names it: since the controller handed its lease away nothing composes it, "+
|
||||
"so the assigner writes it no want", node)
|
||||
return encode(mcc.Observation{Scope: mcc.ScopeMachine, ID: node, Kind: KindUnplaced, Machine: node, Summary: summary,
|
||||
return encode(Observed{Scope: mcc.ScopeMachine, ID: node, Kind: KindUnplaced, Machine: node, Summary: summary,
|
||||
Headline: clipHeadline(node + " has no assignments"), Needs: "place it with mesh-module-assigner.propose.",
|
||||
Explanation: occ.NeedsYou + "place it with mesh-module-assigner.propose. " + summary + ".",
|
||||
Resolved: node + " is placed", Source: "assigner." + KindUnplaced}, at)
|
||||
Source: "assigner." + KindUnplaced}, at)
|
||||
}
|
||||
|
||||
func clipHeadline(s string) string {
|
||||
|
||||
@@ -117,8 +117,9 @@ func TestManifest(t *testing.T) {
|
||||
// ADR 0635 §5: its views family is its ordinary events, `views.<state>` among its emits, on no stream of its own.
|
||||
// The seat's four events are said on the seat's own subject (the seat's family), so the module keeps no stream for them
|
||||
// and lists none among its emits.
|
||||
if len(m.Streams) != 0 {
|
||||
t.Errorf("streams = %+v, want none: the seat's events ride the shared events stream", m.Streams)
|
||||
// Its one stream is its own findings' events (design 47 G A-30, §383), which its finding writer says.
|
||||
if len(m.Streams) != 1 || m.Streams[0].Name != "mesh-delivery-assigner-findings" || !slices.Equal(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("streams = %+v, want its findings' stream alone: the seat's events ride the shared events stream", m.Streams)
|
||||
}
|
||||
for _, e := range meshmoduleassigner.Events() {
|
||||
if slices.Contains(m.Emits, e) {
|
||||
@@ -127,7 +128,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
|
||||
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
|
||||
@@ -42,6 +42,7 @@
|
||||
"operator-channel"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-conditions.findings-asked",
|
||||
"mesh-catalog.catalog-revision",
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
@@ -66,6 +67,14 @@
|
||||
"seat:node-engine.manifest-reads",
|
||||
"seat:operator-channel.asks"
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
"name": "mesh-delivery-assigner-findings",
|
||||
"events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"],
|
||||
"max-age-days": 7,
|
||||
"max-mib": 16
|
||||
}
|
||||
],
|
||||
"emits": [
|
||||
"assignments-copied",
|
||||
"put",
|
||||
|
||||
Reference in New Issue
Block a user