Take a registered module's new done key in, compose it and grant it; at start delete the plans' buckets and compose every node and the replaced controller (issue 1945, B3, B5, B13)
With no plan left to take a done key in, a write of the builder's keys registers the new fingerprint, composes the module's nodes and the bus's user list, and writes its grant, so the assigner's rules can name it (ADR 0539 clauses 5 and 7). The new controller composes every node at its start, and the controller it replaced at its own fingerprint, the operator's way back (clause 10); the merge window's, silent sends' and building plans' buckets are no longer asserted and are deleted once, each said as the event bucket-deleted (clause 13). The assigner may read the compositions' digests, and a node-engine hears its backup holder's backed-up.
This commit is contained in:
@@ -0,0 +1,225 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/novox/mesh-controller/internal/builderkeys"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7,
|
||||
// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's
|
||||
// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the
|
||||
// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's
|
||||
// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now
|
||||
// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone,
|
||||
// written into its want at once by the assigner), and writes `grants.<module>.<fp>`. The assigner's sixth and seventh
|
||||
// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every
|
||||
// module once (B5's start pass).
|
||||
|
||||
// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for.
|
||||
type doneKeySteps struct {
|
||||
// held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go).
|
||||
held func(ctx context.Context, module string) (bool, error)
|
||||
// trunk is the indexer's trunk value of the module.
|
||||
trunk func(ctx context.Context, module string) (sdkdelivery.Module, error)
|
||||
// registeredAt says the module is registered at the commit.
|
||||
registeredAt func(ctx context.Context, module, commit string) (bool, error)
|
||||
// takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when
|
||||
// the key is not done yet.
|
||||
takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error
|
||||
// nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none.
|
||||
nodesOf func(ctx context.Context, module string) ([]string, error)
|
||||
busHolder func(ctx context.Context) (string, error)
|
||||
// compose writes the nodes' compositions; grants writes the grants issued.
|
||||
compose func(ctx context.Context, nodes []string) error
|
||||
grants func(ctx context.Context) error
|
||||
}
|
||||
|
||||
// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was
|
||||
// nothing to do.
|
||||
func (s doneKeySteps) take(ctx context.Context, module string) (string, error) {
|
||||
held, err := s.held(ctx, module)
|
||||
if err != nil || !held {
|
||||
return "", err
|
||||
}
|
||||
trunk, err := s.trunk(ctx, module)
|
||||
if errors.Is(err, builderkeys.ErrNoValue) {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) {
|
||||
return "", nil
|
||||
}
|
||||
at, err := s.registeredAt(ctx, module, trunk.Commit)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
said := ""
|
||||
if !at {
|
||||
if err := s.takeIn(ctx, module, trunk); err != nil {
|
||||
return "", err
|
||||
}
|
||||
said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit))
|
||||
}
|
||||
nodes, err := s.nodesOf(ctx, module)
|
||||
if err != nil {
|
||||
return said, err
|
||||
}
|
||||
if len(nodes) == 0 {
|
||||
return said, nil
|
||||
}
|
||||
// The module's nodes first: composing them makes what the new manifest needs, its credentials among it.
|
||||
if err := s.compose(ctx, nodes); err != nil {
|
||||
return said, err
|
||||
}
|
||||
// Then the bus's user list, which carries the module's permissions at its new fingerprint.
|
||||
holder, err := s.busHolder(ctx)
|
||||
if err != nil {
|
||||
return said, err
|
||||
}
|
||||
if holder != "" && !inList(nodes, holder) {
|
||||
if err := s.compose(ctx, []string{holder}); err != nil {
|
||||
return said, err
|
||||
}
|
||||
}
|
||||
// Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7).
|
||||
if err := s.grants(ctx); err != nil {
|
||||
return said, err
|
||||
}
|
||||
if said == "" {
|
||||
said = fmt.Sprintf("%s at %s", module, short(trunk.FP2))
|
||||
}
|
||||
return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil
|
||||
}
|
||||
|
||||
func inList(list []string, s string) bool {
|
||||
for _, v := range list {
|
||||
if v == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
|
||||
var doneKeysMu sync.Map
|
||||
|
||||
// takeDoneKey is a woken module's take-in on the live stores and buckets, said.
|
||||
func takeDoneKey(ctx context.Context, open *stores, module string) {
|
||||
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
|
||||
mu.(*sync.Mutex).Lock()
|
||||
defer mu.(*sync.Mutex).Unlock()
|
||||
keys, closeKeys, err := openBuilderKeys()
|
||||
if err != nil {
|
||||
fmt.Printf("done keys: %s: the builder's keys cannot be read: %v\n", module, err)
|
||||
return
|
||||
}
|
||||
defer closeKeys()
|
||||
said, err := liveDoneKeySteps(open, keys).take(ctx, module)
|
||||
switch {
|
||||
case errors.Is(err, errKeyWaits):
|
||||
// Said by nobody: the key's next write wakes this again.
|
||||
case err != nil:
|
||||
fmt.Printf("done keys: %s: %v\n", module, err)
|
||||
case said != "":
|
||||
fmt.Printf("done keys: %s\n", said)
|
||||
}
|
||||
}
|
||||
|
||||
// liveDoneKeySteps are the steps on the controller's own stores and the bus.
|
||||
func liveDoneKeySteps(open *stores, keys builderkeys.Reader) doneKeySteps {
|
||||
inv := open.inventory
|
||||
return doneKeySteps{
|
||||
held: func(ctx context.Context, module string) (bool, error) {
|
||||
_, err := inv.SourceOf(ctx, module)
|
||||
if errors.Is(err, inventory.ErrNoSuchModule) {
|
||||
return false, nil
|
||||
}
|
||||
return err == nil, err
|
||||
},
|
||||
trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) {
|
||||
return keys.ModuleAt(ctx, module, "")
|
||||
},
|
||||
registeredAt: func(ctx context.Context, module, commit string) (bool, error) {
|
||||
src, err := inv.SourceOf(ctx, module)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return sameCommitID(src.BuiltFrom, commit), nil
|
||||
},
|
||||
takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error {
|
||||
_, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}})
|
||||
if errors.Is(err, inventory.ErrSuperseded) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
},
|
||||
nodesOf: func(ctx context.Context, module string) ([]string, error) {
|
||||
entries, err := inv.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.Manifest.Module == module {
|
||||
return e.On, nil
|
||||
}
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
busHolder: func(ctx context.Context) (string, error) {
|
||||
holders, err := seatHolders(ctx, inv)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return holders[catalogue.BrokerSeat].Node, nil
|
||||
},
|
||||
compose: func(ctx context.Context, nodes []string) error {
|
||||
_, err := composeModules(ctx, open, nodes)
|
||||
return err
|
||||
},
|
||||
grants: func(ctx context.Context) error {
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
grantsIssued(ctx, inv, js, "done keys: ")
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx
|
||||
// ends: never on the watch's own goroutine.
|
||||
func watchDoneKeys(ctx context.Context, open *stores) {
|
||||
keys, closeKeys, err := openBuilderKeys()
|
||||
if err != nil {
|
||||
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
|
||||
inBackground(func() { takeDoneKey(ctx, open, module) })
|
||||
}); err != nil {
|
||||
closeKeys()
|
||||
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
closeKeys()
|
||||
}()
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/novox/mesh-controller/internal/builderkeys"
|
||||
)
|
||||
|
||||
// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to
|
||||
// nodes, the bus held on holder; every act is written to acts in order.
|
||||
func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps {
|
||||
return doneKeySteps{
|
||||
held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil },
|
||||
trunk: func(context.Context, string) (sdkdelivery.Module, error) {
|
||||
if fp == "" {
|
||||
return sdkdelivery.Module{}, builderkeys.ErrNoValue
|
||||
}
|
||||
return sdkdelivery.Module{Commit: commit, FP2: fp}, nil
|
||||
},
|
||||
registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil },
|
||||
takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error {
|
||||
*acts = append(*acts, "take in "+module+" at "+trunk.Commit)
|
||||
return nil
|
||||
},
|
||||
nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil },
|
||||
busHolder: func(context.Context) (string, error) { return holder, nil },
|
||||
compose: func(_ context.Context, ns []string) error {
|
||||
*acts = append(*acts, "compose "+strings.Join(ns, ","))
|
||||
return nil
|
||||
},
|
||||
grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil },
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are
|
||||
// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist
|
||||
// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is
|
||||
// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left
|
||||
// alone.
|
||||
func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) {
|
||||
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64)
|
||||
var acts []string
|
||||
said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" {
|
||||
t.Errorf("the acts were %q", got)
|
||||
}
|
||||
if !strings.Contains(said, "registered from the builder's done key") {
|
||||
t.Errorf("said %q", said)
|
||||
}
|
||||
|
||||
acts = nil
|
||||
if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := strings.Join(acts, "; "); got != "compose novox; grants" {
|
||||
t.Errorf("a module registered at its trunk already: %q", got)
|
||||
}
|
||||
|
||||
for name, s := range map[string]doneKeySteps{
|
||||
"assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"),
|
||||
"no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"),
|
||||
} {
|
||||
acts = nil
|
||||
if _, err := s.take(context.Background(), "mesh-mail"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" {
|
||||
t.Errorf("%s: %v", name, acts)
|
||||
}
|
||||
if name == "no trunk value" && len(acts) != 0 {
|
||||
t.Errorf("%s: %v", name, acts)
|
||||
}
|
||||
}
|
||||
acts = nil
|
||||
if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 {
|
||||
t.Errorf("a module the catalogue does not hold: %v %v", acts, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC4: the bus's user list is a file of the bus module's composition (ADR 0539 clause 7): a change of the composed
|
||||
// users is a new revision of that composition alone, every other composition of the node keeping its digest.
|
||||
func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testing.T) {
|
||||
body, fps := capturedBody(t, "g14")
|
||||
fps["nats"] = sdkdelivery.SourcePrefix2 + strings.Repeat("e", 64)
|
||||
withUsers := func(users string) map[string]sdkdelivery.Composed {
|
||||
var d map[string]any
|
||||
if err := json.Unmarshal(body, &d); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// As the composer writes it for the module holding mesh-broker (mesh-sdk catalogue: BusUsersID, owned by the module).
|
||||
d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file",
|
||||
"module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"})
|
||||
raw, _ := json.Marshal(d)
|
||||
comps, _, err := compositionsOf("g14", raw, fps, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return comps
|
||||
}
|
||||
before, after := withUsers("authorization { users = [ {user: a} ] }\n"), withUsers("authorization { users = [ {user: a}, {user: b} ] }\n")
|
||||
natsKey, _ := sdkdelivery.ComposedKey("g14", "nats", fps["nats"])
|
||||
for key, c := range before {
|
||||
d1, _ := sdkdelivery.ComposedDigestOf(c)
|
||||
d2, _ := sdkdelivery.ComposedDigestOf(after[key])
|
||||
if changed := d1 != d2; changed != (key == natsKey) {
|
||||
t.Errorf("%s changed %v; only the bus module's composition changes with its user list", key, changed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// What the controller does at its start under the lease (novox/hq ADR 0539 clauses 10 and 13):
|
||||
//
|
||||
// - it deletes the buckets of the plans' state this release deleted, each said once in its log and as its event
|
||||
// `bucket-deleted`: a later start finds none and says nothing;
|
||||
// - it composes every node, so every assigned module has its composition at the fingerprint it is registered at
|
||||
// and each node its own (ADR 0539 clause 11: the new controller writes every composition at its start);
|
||||
// - and it composes, on the control-node, the controller it replaced at that controller's own fingerprint, from that
|
||||
// build's manifest, so the operator's way back is one key at the terminal (clause 10).
|
||||
|
||||
// planStateBucketDeleter is what deleting the plans' buckets needs of the bus: jetstream.JetStream, or a test's.
|
||||
type planStateBucketDeleter interface {
|
||||
DeleteKeyValue(ctx context.Context, bucket string) error
|
||||
}
|
||||
|
||||
// deletePlanStateBuckets deletes each bucket of the plans' state the bus still holds and says each it deleted; it
|
||||
// answers those it deleted. A bucket not there is no error and is not said.
|
||||
func deletePlanStateBuckets(ctx context.Context, js planStateBucketDeleter, say func(bucket string)) ([]string, error) {
|
||||
var deleted []string
|
||||
var errs []error
|
||||
for _, bucket := range broker.PlanStateBuckets {
|
||||
err := js.DeleteKeyValue(ctx, bucket)
|
||||
switch {
|
||||
case errors.Is(err, jetstream.ErrBucketNotFound), errors.Is(err, jetstream.ErrStreamNotFound):
|
||||
continue
|
||||
case err != nil:
|
||||
errs = append(errs, fmt.Errorf("deleting %s: %w", bucket, err))
|
||||
continue
|
||||
}
|
||||
deleted = append(deleted, bucket)
|
||||
say(bucket)
|
||||
}
|
||||
return deleted, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// BucketDeleted is the body of the controller's event `bucket-deleted`.
|
||||
type BucketDeleted struct {
|
||||
Bucket string `json:"bucket"`
|
||||
Why string `json:"why"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// sayBucketDeleted says a deleted bucket in the log and as the controller's event.
|
||||
func sayBucketDeleted(ctx context.Context, bucket string) {
|
||||
why := "a bucket of the plans' state, which release 6 deleted (novox/hq ADR 0539 clause 13)"
|
||||
fmt.Printf("start: deleted %s: %s\n", bucket, why)
|
||||
bus := checkEvents
|
||||
if bus == nil {
|
||||
return
|
||||
}
|
||||
body, err := json.Marshal(BucketDeleted{Bucket: bucket, Why: why, At: time.Now().UTC()})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
saying, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer stop()
|
||||
if err := bus.PublishSeatEvent(saying, link.MeshControllerSeat, "bucket-deleted", body); err != nil {
|
||||
fmt.Printf("start: that %s was deleted could not be said as the event bucket-deleted: %v\n", bucket, err)
|
||||
}
|
||||
}
|
||||
|
||||
// composeAtStart is the start's pass: the plans' buckets deleted, every node composed, the replaced controller
|
||||
// composed. Said, never fatal: what is not composed now is composed by the next act that touches it.
|
||||
func composeAtStart(ctx context.Context, open *stores) {
|
||||
if bus, err := aBus(); err != nil {
|
||||
fmt.Printf("start: the plans' buckets are not looked at: %v\n", err)
|
||||
} else {
|
||||
if js, err := jetstream.New(bus.Conn()); err == nil {
|
||||
if _, err := deletePlanStateBuckets(ctx, js, func(b string) { sayBucketDeleted(ctx, b) }); err != nil {
|
||||
fmt.Printf("start: %v\n", err)
|
||||
}
|
||||
}
|
||||
bus.Close()
|
||||
}
|
||||
nodes, err := open.inventory.Nodes(ctx)
|
||||
if err != nil {
|
||||
fmt.Printf("start: nothing is composed: %v\n", err)
|
||||
return
|
||||
}
|
||||
names := make([]string, 0, len(nodes))
|
||||
for _, n := range nodes {
|
||||
names = append(names, n.Name)
|
||||
}
|
||||
sort.Strings(names)
|
||||
if _, err := composeModules(ctx, open, names); err != nil {
|
||||
fmt.Printf("start: %v\n", err)
|
||||
}
|
||||
if err := composeReplacedController(ctx, open); err != nil {
|
||||
fmt.Printf("start: the controller this one replaced is not composed: %v\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
// replacedController is the controller a control-node ran before this one: the build the mesh last sent it and that
|
||||
// build's fingerprint and manifest.
|
||||
type replacedController struct {
|
||||
node, fp string
|
||||
manifest catalogue.Manifest
|
||||
}
|
||||
|
||||
// replacedControllers are, for each node the controller is assigned to, the controller build it was last sent when its
|
||||
// fingerprint is of the new form and not the one the controller is registered at now.
|
||||
func replacedControllers(ctx context.Context, open *stores) ([]replacedController, error) {
|
||||
inv := open.inventory
|
||||
fps, err := currentFingerprints(ctx, inv)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries, err := inv.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var on []string
|
||||
for _, e := range entries {
|
||||
if e.Manifest.Module == controllerModule {
|
||||
on = e.On
|
||||
}
|
||||
}
|
||||
var out []replacedController
|
||||
for _, node := range on {
|
||||
sent, known, err := inv.SentBuilds(ctx, node)
|
||||
if err != nil || !known || sent[controllerModule] == "" {
|
||||
continue
|
||||
}
|
||||
build, found, err := inv.BuildByID(ctx, sent[controllerModule])
|
||||
if err != nil || !found || build.SourceFingerprint == fps[controllerModule] ||
|
||||
!strings.HasPrefix(build.SourceFingerprint, sdkdelivery.SourcePrefix2) || len(build.Manifest) == 0 {
|
||||
continue
|
||||
}
|
||||
var m catalogue.Manifest
|
||||
if err := json.Unmarshal(build.Manifest, &m); err != nil {
|
||||
return nil, fmt.Errorf("the manifest of %s's build %s: %w", controllerModule, build.ID, err)
|
||||
}
|
||||
out = append(out, replacedController{node: node, fp: build.SourceFingerprint, manifest: m})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// composeReplacedController writes, on each control-node, the controller's composition at the fingerprint of the one
|
||||
// it replaced: the node composed with that build's manifest in place of the controller's registered one, in the
|
||||
// reading mode (nothing made), and only the controller's composition of it kept.
|
||||
func composeReplacedController(ctx context.Context, open *stores) error {
|
||||
if _, err := theLease.epoch(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
replaced, err := replacedControllers(ctx, open)
|
||||
if err != nil || len(replaced) == 0 {
|
||||
return err
|
||||
}
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sign, err := controllerSigner(ctx, open)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var errs []error
|
||||
for _, r := range replaced {
|
||||
in, err := inputFor(ctx, open)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c, err := replacedComposition(in.Input, r)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", r.node, err))
|
||||
continue
|
||||
}
|
||||
at, err := openComposedPlace(ctx, js, bus.Conn(), r.node)
|
||||
if err != nil {
|
||||
errs = append(errs, err)
|
||||
continue
|
||||
}
|
||||
at.sign = sign
|
||||
key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp)
|
||||
if wrote, err := writeComposed(ctx, at, key, c); err != nil {
|
||||
errs = append(errs, err)
|
||||
} else if wrote {
|
||||
fmt.Printf("start: composed %s, the controller this one replaced, on %s: the way back is want.%s.%s naming it\n",
|
||||
short(r.fp), r.node, r.node, controllerModule)
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// replacedComposition is the controller's composition on r's node at r's fingerprint: in composed with r's manifest
|
||||
// in place of the controller's registered one.
|
||||
func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Composed, error) {
|
||||
shelf := make(map[string]catalogue.Manifest, len(in.Catalogue))
|
||||
for k, v := range in.Catalogue {
|
||||
shelf[k] = v
|
||||
}
|
||||
shelf[controllerModule] = r.manifest
|
||||
in.Catalogue = shelf
|
||||
all, _, err := compose.Compositions(in, r.node, map[string]string{controllerModule: r.fp}, compose.Reading)
|
||||
if err != nil {
|
||||
return sdkdelivery.Composed{}, err
|
||||
}
|
||||
key, _ := sdkdelivery.ComposedKey(r.node, controllerModule, r.fp)
|
||||
c, ok := all[key]
|
||||
if !ok {
|
||||
return sdkdelivery.Composed{}, fmt.Errorf("%s composes no %s with the replaced build's manifest", r.node, controllerModule)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// fakeBuckets is the bus's buckets by name, for deleting.
|
||||
type fakeBuckets map[string]bool
|
||||
|
||||
func (f fakeBuckets) DeleteKeyValue(_ context.Context, bucket string) error {
|
||||
if !f[bucket] {
|
||||
return jetstream.ErrBucketNotFound
|
||||
}
|
||||
delete(f, bucket)
|
||||
return nil
|
||||
}
|
||||
|
||||
// 1945/AC14 (unit half): the first start deletes the merge window's, the silent sends' and the building plans' buckets
|
||||
// and says each once; a later start finds none and says nothing; a bucket of anything else is left alone.
|
||||
func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) {
|
||||
bus := fakeBuckets{"mesh-controller_merge-window": true, "mesh-controller_rollout-silent": true,
|
||||
"mesh-controller_plans-building": true, "mesh-controller_grants": true}
|
||||
var said []string
|
||||
deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) })
|
||||
if err != nil || len(deleted) != 3 || strings.Join(said, " ") != strings.Join(broker.PlanStateBuckets, " ") {
|
||||
t.Fatalf("first start: deleted %v, said %v, %v", deleted, said, err)
|
||||
}
|
||||
if !bus["mesh-controller_grants"] || len(bus) != 1 {
|
||||
t.Errorf("the bus holds %v after the first start", bus)
|
||||
}
|
||||
said = nil
|
||||
if deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }); err != nil || len(deleted)+len(said) != 0 {
|
||||
t.Errorf("a later start deleted %v and said %v (%v)", deleted, said, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC6 (unit half): the controller this one replaced is composed on its node from its own build's manifest at its
|
||||
// own fingerprint: the composition holds what that manifest declares, not what the controller registered now declares.
|
||||
func TestTheReplacedControllerIsComposedFromItsOwnManifest(t *testing.T) {
|
||||
now := catalogue.Manifest{Module: controllerModule, Version: "6", Resources: []map[string]any{
|
||||
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "six\n"}}}
|
||||
before := catalogue.Manifest{Module: controllerModule, Version: "5", Resources: []map[string]any{
|
||||
{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": "five\n"}}}
|
||||
in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now},
|
||||
Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true}}}
|
||||
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("5", 64)
|
||||
c, err := replacedComposition(in, replacedController{node: "anchor", fp: fp, manifest: before})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if c.Module != controllerModule || c.FP != fp || c.Node != "anchor" || len(c.Resources) != 1 {
|
||||
t.Fatalf("composed %s at %s on %s with %d resources", c.Module, c.FP, c.Node, len(c.Resources))
|
||||
}
|
||||
var f struct{ Content string }
|
||||
_ = json.Unmarshal(c.Resources[0], &f)
|
||||
if f.Content != "five\n" {
|
||||
t.Errorf("the replaced controller's composition holds %q, its own manifest says five", f.Content)
|
||||
}
|
||||
if in.Catalogue[controllerModule].Version != "6" {
|
||||
t.Errorf("composing the replaced controller changed the Input's catalogue")
|
||||
}
|
||||
}
|
||||
@@ -203,6 +203,11 @@ func watchNewModules(ctx context.Context, open *stores) {
|
||||
inBackground(func() {
|
||||
takeNewModules(ctx, open)
|
||||
})
|
||||
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
|
||||
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
|
||||
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
|
||||
inBackground(func() { composeAtStart(ctx, open) })
|
||||
watchDoneKeys(ctx, open)
|
||||
keys, closeKeys, err := openBuilderKeys()
|
||||
if err != nil {
|
||||
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
||||
|
||||
Reference in New Issue
Block a user