jschoubben
c75d726205
The try of the on-change mirror failed with every test passed: goroutines a served test started outlived its serve and met the next test's globals and its closed stores. Under -race the served tests show it as a race on conditionsFrom (watchTheMesh writing it while settlingPending, already running, read it), and a take-in cut by the stop raised fingerprint-not-composed, a kind with no plain words, which fails the package after its PASS line. - serve counts every goroutine it and its parts start (serving_group.go) and ends and waits for them before anything else is undone: the lease's watch, the pending settler, the status summary, the facts, the mirrors and the new declared-packages keeper with its two bucket watches, the declaration sweep, the read-call counter, the watches' closers, the have watches, the vault move on a report, and the watchdogs, doctor, healers and asker under their own group. - The condition store opens before anything that reads conditionsFrom starts; askerFrom is an atomic pointer, read from the keeper's goroutines. - A done key's take-in cut by the stop raises nothing, and fingerprint-not-composed has plain words.
456 lines
16 KiB
Go
456 lines
16 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/compose"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go"
|
|
"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/recordmirror"
|
|
)
|
|
|
|
// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
|
|
// and each registered module's declared config, which mesh-config resolves a module's config values from.
|
|
//
|
|
// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
|
|
// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
|
|
// so no one call site sees every change; what every change has in common is that it is committed to the
|
|
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
|
|
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
|
|
// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
|
|
// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
|
|
|
|
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
|
|
const mirrorEvery = 10 * time.Second
|
|
|
|
// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
|
|
func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
|
|
api, err := jetstream.New(conn)
|
|
if err != nil {
|
|
fmt.Printf("the record mirrors are not kept: %v\n", err)
|
|
return
|
|
}
|
|
holds := func() bool {
|
|
epoch, err := theLease.epoch(ctx)
|
|
return err == nil && epoch > 0
|
|
}
|
|
epoch := func() uint64 {
|
|
e, _ := theLease.epoch(ctx)
|
|
return e
|
|
}
|
|
sources := map[string]func(context.Context) (map[string][]byte, error){
|
|
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
|
recordmirror.Predefined.Name: inv.PredefinedMirror,
|
|
}
|
|
packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
|
|
held := false
|
|
writers := map[string]*recordmirror.Writer{}
|
|
failing := map[string]string{}
|
|
tick := time.NewTicker(mirrorEvery)
|
|
defer tick.Stop()
|
|
events := &assignmentEvents{}
|
|
for {
|
|
now := holds()
|
|
if now && !held {
|
|
// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
|
|
packages.poke()
|
|
}
|
|
held = now
|
|
if now {
|
|
// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
|
|
// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
|
|
keepingAssignments(ctx, inv, announce, epoch, events, failing)
|
|
for _, m := range recordmirror.Mirrors() {
|
|
if sources[m.Name] == nil {
|
|
continue // the declared packages, kept on a change (keepingDeclaredPackages)
|
|
}
|
|
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
|
|
if why != failing[m.Name] {
|
|
if why != "" {
|
|
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
|
|
} else {
|
|
fmt.Printf("the %s mirror is kept again\n", m.Name)
|
|
}
|
|
failing[m.Name] = why
|
|
}
|
|
}
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-tick.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
|
|
func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
|
|
source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
|
|
holds func() bool) string {
|
|
reading, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
defer cancel()
|
|
w := writers[m.Name]
|
|
if w == nil {
|
|
kv, err := api.KeyValue(reading, m.Bucket)
|
|
if err != nil {
|
|
return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
|
|
}
|
|
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
|
|
writers[m.Name] = w
|
|
}
|
|
want, err := source(reading)
|
|
if err != nil {
|
|
return fmt.Sprintf("the store cannot be read: %v", err)
|
|
}
|
|
lost := w.LostEvents
|
|
r, err := w.Sync(reading, want)
|
|
if r.Written+r.Removed > 0 {
|
|
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
|
|
}
|
|
if w.LostEvents > lost {
|
|
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
|
|
m.Name, w.LostEvents-lost)
|
|
}
|
|
if err != nil {
|
|
return err.Error()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// assignmentEvents is the assignments as the bucket held them at the last look, to say what changed since.
|
|
type assignmentEvents struct {
|
|
known map[string]inventory.KeptValue
|
|
}
|
|
|
|
// keepingAssignments is the tick's work for the assignments once they are on the bus: the provisioner's plan written into
|
|
// the values it belongs to, and one `assignment-changed` event for each key that is new, changed or ended since the last
|
|
// look — whoever wrote it, a verb in this process or a command in another. False when they are not on the bus yet.
|
|
func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce recordmirror.Announcer, epoch func() uint64,
|
|
events *assignmentEvents, failing map[string]string) {
|
|
reading, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
defer cancel()
|
|
const name = "assignments"
|
|
if _, _, err := inv.KeepAssignmentPlans(reading); err != nil {
|
|
if failing[name] != err.Error() {
|
|
failing[name] = err.Error()
|
|
fmt.Printf("the provisioner's plans are not kept in the assignments: %v\n", err)
|
|
}
|
|
} else if failing[name] != "" {
|
|
fmt.Printf("the provisioner's plans are kept in the assignments again\n")
|
|
failing[name] = ""
|
|
}
|
|
now, err := inv.KeptAssignments(reading)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if events.known != nil {
|
|
keys := make([]string, 0, len(now)+len(events.known))
|
|
for key := range now {
|
|
keys = append(keys, key)
|
|
}
|
|
for key := range events.known {
|
|
if _, still := now[key]; !still {
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
sort.Strings(keys)
|
|
for _, key := range keys {
|
|
was, had := events.known[key]
|
|
is, has := now[key]
|
|
e := recordmirror.Event{Mirror: recordmirror.Assignments.Name, Key: key, Revision: is.Revision, Epoch: epoch(),
|
|
At: time.Now().UTC()}
|
|
switch {
|
|
case has && !had:
|
|
e.Change = recordmirror.Added
|
|
case has && had && was.Revision != is.Revision:
|
|
e.Change, e.Fields = recordmirror.Changed, recordmirror.ChangedFields(was.Value, is.Value)
|
|
case !has && had:
|
|
e.Change, e.Revision = recordmirror.Removed, 0
|
|
default:
|
|
continue
|
|
}
|
|
body, err := json.Marshal(e)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
for try := 0; try < 2; try++ {
|
|
if err = announce.PublishSeatEvent(reading, broker.ControllerSeat, recordmirror.Assignments.Event, body); err == nil {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
events.known = now
|
|
}
|
|
|
|
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
|
|
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
|
|
return func(ctx context.Context) (map[string][]byte, error) {
|
|
in, err := inputFor(ctx, open)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return declaredPackagesFrom(ctx, in)
|
|
}
|
|
}
|
|
|
|
// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
|
|
// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
|
|
// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
|
|
func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
|
|
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
|
|
asked := in.Input
|
|
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
|
|
for i := range asked.Nodes {
|
|
asked.Nodes[i].ReadsModules = true
|
|
}
|
|
out := map[string][]byte{}
|
|
for _, n := range in.sortedNodes() {
|
|
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
|
|
if compose.IsUnresolvable(err) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
|
|
}
|
|
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
|
|
// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
|
|
// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
|
|
// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
|
|
// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
|
|
// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
|
|
// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
|
|
// change no watched key says, and with an unchanged digest composes nothing either.
|
|
const (
|
|
packagesSafetyEvery = 10 * time.Minute
|
|
packagesGap = 30 * time.Second
|
|
)
|
|
|
|
// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
|
|
// starts of two passes.
|
|
type packagesTiming struct{ safety, gap time.Duration }
|
|
|
|
// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
|
|
// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
|
|
type packagesKeeper struct {
|
|
read func(ctx context.Context) (*meshInput, error)
|
|
compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
|
|
sync func(ctx context.Context, want map[string][]byte) error
|
|
holds func() bool
|
|
wake chan struct{}
|
|
// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
|
|
// next pass composes.
|
|
last string
|
|
}
|
|
|
|
func newPackagesKeeper(read func(context.Context) (*meshInput, error),
|
|
compose func(context.Context, *meshInput) (map[string][]byte, error),
|
|
sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
|
|
return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
|
|
}
|
|
|
|
// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
|
|
func (k *packagesKeeper) poke() {
|
|
select {
|
|
case k.wake <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
|
|
func (k *packagesKeeper) pass(ctx context.Context) error {
|
|
if !k.holds() {
|
|
k.last = ""
|
|
return nil
|
|
}
|
|
in, err := k.read(ctx)
|
|
if err != nil {
|
|
k.last = ""
|
|
return fmt.Errorf("the store cannot be read: %w", err)
|
|
}
|
|
digest := inputDigest(in)
|
|
if digest != "" && digest == k.last {
|
|
return nil
|
|
}
|
|
want, err := k.compose(ctx, in)
|
|
if err != nil {
|
|
k.last = ""
|
|
return err
|
|
}
|
|
if err := k.sync(ctx, want); err != nil {
|
|
k.last = ""
|
|
return err
|
|
}
|
|
k.last = digest
|
|
return nil
|
|
}
|
|
|
|
// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
|
|
// and an empty line when one succeeds again.
|
|
func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
|
|
safety := time.NewTicker(at.safety)
|
|
defer safety.Stop()
|
|
failing := ""
|
|
var began time.Time
|
|
for {
|
|
// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
|
|
select {
|
|
case <-k.wake:
|
|
default:
|
|
}
|
|
began = time.Now()
|
|
why := ""
|
|
if err := k.pass(ctx); err != nil {
|
|
why = err.Error()
|
|
}
|
|
if why != failing {
|
|
said(why)
|
|
failing = why
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-k.wake:
|
|
case <-safety.C:
|
|
}
|
|
if rest := at.gap - time.Since(began); rest > 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(rest):
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
|
|
// for the same Input. Empty when it cannot be made, and then the pass composes.
|
|
func inputDigest(in *meshInput) string {
|
|
refused := map[string]string{}
|
|
for node, err := range in.refused {
|
|
if err != nil {
|
|
refused[node] = err.Error()
|
|
}
|
|
}
|
|
bus := ""
|
|
if in.busRefused != nil {
|
|
bus = in.busRefused.Error()
|
|
}
|
|
body, err := json.Marshal(struct {
|
|
Input compose.Input
|
|
Refused map[string]string
|
|
BusRefused string
|
|
}{in.Input, refused, bus})
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
sum := sha256.Sum256(body)
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
|
|
// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
|
|
func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
|
|
epoch func() uint64, holds func() bool) *packagesKeeper {
|
|
m := recordmirror.DeclaredPackages
|
|
var w *recordmirror.Writer
|
|
read := func(ctx context.Context) (*meshInput, error) {
|
|
reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
|
defer cancel()
|
|
return readComposeInput(reading, open)
|
|
}
|
|
sync := func(ctx context.Context, want map[string][]byte) error {
|
|
writing, cancel := context.WithTimeout(ctx, 30*time.Second)
|
|
defer cancel()
|
|
if w == nil {
|
|
kv, err := api.KeyValue(writing, m.Bucket)
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
|
|
}
|
|
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
|
|
}
|
|
lost := w.LostEvents
|
|
r, err := w.Sync(writing, want)
|
|
if r.Written+r.Removed > 0 {
|
|
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
|
|
}
|
|
if w.LostEvents > lost {
|
|
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
|
|
m.Name, w.LostEvents-lost)
|
|
}
|
|
return err
|
|
}
|
|
k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
|
|
for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
|
|
goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
|
|
}
|
|
goServing(ctx, func() {
|
|
k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
|
|
if why != "" {
|
|
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
|
|
} else {
|
|
fmt.Printf("the %s mirror is kept again\n", m.Name)
|
|
}
|
|
})
|
|
})
|
|
return k
|
|
}
|
|
|
|
// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
|
|
// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
|
|
func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
|
|
for {
|
|
if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
|
|
fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(time.Minute):
|
|
}
|
|
}
|
|
}
|
|
|
|
// watchBucketOnce watches the bucket until its watch ends or ctx does.
|
|
func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
|
|
kv, err := api.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = watch.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case e, ok := <-watch.Updates():
|
|
if !ok {
|
|
return fmt.Errorf("the watch ended")
|
|
}
|
|
if e != nil {
|
|
wake()
|
|
}
|
|
}
|
|
}
|
|
}
|