mesh-tools
97a2d841c7
The controller plans config, templates and moved providers' provisions, hands bus accounts to the vault and judges the new manifest words, on the feature's SDK and node-engine validator (feature 57, config and provisioning).
107 lines
3.7 KiB
Go
107 lines
3.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"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.
|
|
|
|
// 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, 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.Assignments.Name: inv.AssignmentsMirror,
|
|
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
|
recordmirror.Predefined.Name: inv.PredefinedMirror,
|
|
}
|
|
writers := map[string]*recordmirror.Writer{}
|
|
failing := map[string]string{}
|
|
tick := time.NewTicker(mirrorEvery)
|
|
defer tick.Stop()
|
|
for {
|
|
if holds() {
|
|
for _, m := range recordmirror.Mirrors() {
|
|
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 ""
|
|
}
|