jschoubben
b4a62e9e6e
A node away during the switch finds its last declaration on the stream; once every node's have names the node-engine registered, no node reads one, so the stream and the nodes' consumers on it go, said once, and a raise leaves them out (ADR 0539 clause 11).
342 lines
12 KiB
Go
342 lines
12 KiB
Go
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/inventory"
|
|
"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) {
|
|
sayDeleted(ctx, bucket, "a bucket of the plans' state, which release 6 deleted (novox/hq ADR 0539 clause 13)")
|
|
}
|
|
|
|
// sayDeleted says a deleted bucket or stream, with why, in the log and as the controller's event bucket-deleted.
|
|
func sayDeleted(ctx context.Context, bucket, why string) {
|
|
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)
|
|
}
|
|
if _, err := retireDeclarations(ctx, open); err != nil {
|
|
fmt.Printf("start: whether the declarations' stream may go cannot be read: %v\n", err)
|
|
}
|
|
}
|
|
|
|
// nodeEngineModule is the node-engine's module.
|
|
const nodeEngineModule = "mesh-host"
|
|
|
|
// declarationsDone says every node runs the node-engine at fingerprint current, as its have names: no node still reads
|
|
// a declaration, and none away will need its last one (ADR 0539 clause 11). No node, or no fingerprint of the new
|
|
// form, is not done.
|
|
func declarationsDone(nodes []string, runs map[string]string, current string) bool {
|
|
if len(nodes) == 0 || !strings.HasPrefix(current, sdkdelivery.SourcePrefix2) {
|
|
return false
|
|
}
|
|
for _, n := range nodes {
|
|
if runs[n] != current {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
|
|
// against the fingerprint it is registered at.
|
|
func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
|
|
fps, err := currentFingerprints(ctx, inv)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
nodes, err := inv.Nodes(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
names := make([]string, 0, len(nodes))
|
|
runs := map[string]string{}
|
|
for _, n := range nodes {
|
|
names = append(names, n.Name)
|
|
h, found, err := haveOn(ctx, js, n.Name, nodeEngineModule)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if found {
|
|
runs[n.Name] = h.FP
|
|
}
|
|
}
|
|
return declarationsDone(names, runs, fps[nodeEngineModule]), nil
|
|
}
|
|
|
|
// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
|
|
// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
|
|
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
|
|
if _, err := theLease.epoch(ctx); err != nil {
|
|
return false, err
|
|
}
|
|
bus, err := aBus()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer bus.Close()
|
|
js, err := jetstream.New(bus.Conn())
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
done, err := declarationsRetiredOn(ctx, open.inventory, js)
|
|
if err != nil || !done {
|
|
return done, err
|
|
}
|
|
err = js.DeleteStream(ctx, broker.DeclarationsStream)
|
|
switch {
|
|
case errors.Is(err, jetstream.ErrStreamNotFound):
|
|
return true, nil
|
|
case err != nil:
|
|
return true, fmt.Errorf("deleting the stream %s: %w", broker.DeclarationsStream, err)
|
|
}
|
|
sayDeleted(ctx, broker.DeclarationsStream, "the stream of each node's last declaration: every node runs the new "+
|
|
"node-engine, which reads none (novox/hq ADR 0539 clause 11)")
|
|
return true, nil
|
|
}
|
|
|
|
// replacedController is the controller a control-node ran before this one: its fingerprint and manifest.
|
|
type replacedController struct {
|
|
node, fp string
|
|
manifest catalogue.Manifest
|
|
}
|
|
|
|
// replacedControllers are, for each node the controller is assigned to, the controller it replaced: the newest build of
|
|
// the controller that worked, whose fingerprint is of the new form and not the one the controller is registered at now
|
|
// (the assignments name no build since ADR 0539 §12, so the build records say it). None when there is no such build.
|
|
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
|
|
}
|
|
}
|
|
if len(on) == 0 {
|
|
return nil, nil
|
|
}
|
|
builds, err := inv.Builds(ctx, controllerModule, 20)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
before, found := replacedBuild(builds, fps[controllerModule])
|
|
if !found {
|
|
return nil, nil
|
|
}
|
|
var m catalogue.Manifest
|
|
if err := json.Unmarshal(before.Manifest, &m); err != nil {
|
|
return nil, fmt.Errorf("the manifest of %s's build %s: %w", controllerModule, before.ID, err)
|
|
}
|
|
var out []replacedController
|
|
for _, node := range on {
|
|
out = append(out, replacedController{node: node, fp: before.SourceFingerprint, manifest: m})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// replacedBuild is, of a module's builds newest first, the newest that worked, kept its manifest, and names a
|
|
// fingerprint of the new form other than current.
|
|
func replacedBuild(builds []inventory.Build, current string) (inventory.Build, bool) {
|
|
for _, b := range builds {
|
|
if b.Worked() && b.SourceFingerprint != current && strings.HasPrefix(b.SourceFingerprint, sdkdelivery.SourcePrefix2) &&
|
|
len(b.Manifest) > 0 {
|
|
return b, true
|
|
}
|
|
}
|
|
return inventory.Build{}, false
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
|
|
// store it cannot read is not done, so the stream is asserted as before.
|
|
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
|
|
bus, err := aBus()
|
|
if err != nil {
|
|
return false
|
|
}
|
|
defer bus.Close()
|
|
js, err := jetstream.New(bus.Conn())
|
|
if err != nil {
|
|
return false
|
|
}
|
|
done, err := declarationsRetiredOn(ctx, inv, js)
|
|
return err == nil && done
|
|
}
|