jschoubben
cc5c9ae888
inBackground ran the take-ins of done keys (now in #463's slots), the start's look, the given values' replacement and the announced reads beside serve without serve waiting for them: a take-in could outlive its controller as the loops did. It now takes the context and is counted in its serving group as well, and a serve that is stopping starts none.
203 lines
6.9 KiB
Go
203 lines
6.9 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
"log"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// A value given by hand lives only until the module's first good start (novox/hq ADR 0228).
|
|
//
|
|
// The serving controller watches every have; a module's have that is running and healthy is the signal the store asks about
|
|
// (inventory.ReplaceGivenAfterStart). What it replaced is composed at once, said in the log and stated on the bus as the controller seat's `secret-replaced`,
|
|
// so a replacement is never silent. **Not in the hand-act log**: nobody acted by hand, and that log
|
|
// is read as the count of repairs a healer is wanted for.
|
|
|
|
// SecretReplaced is the controller seat's fact that a value given by hand was replaced
|
|
// (link.KeySecretReplaced). Never the value: neither the old one, which the mesh cannot read,
|
|
// nor the new one, sealed to the machine as it was made.
|
|
type SecretReplaced struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Name string `json:"name"`
|
|
Given time.Time `json:"given"`
|
|
// Sent are the machines sent so the module starts again on the new value; Unsent says why
|
|
// they could not be, in which case the next push carries it.
|
|
Sent []string `json:"sent"`
|
|
Unsent string `json:"unsent,omitempty"`
|
|
Why string `json:"why"`
|
|
}
|
|
|
|
// givenEvents is where the serving controller states it; nil in a command.
|
|
var givenEvents link.Bus
|
|
|
|
// replacing keeps one replacement per machine at a time: two reports arriving together find the
|
|
// same rows, and the store's claim makes one of them the replacer, but the sends need not race.
|
|
var replacing sync.Map
|
|
|
|
// startedOn is the signal of a module's first good start: its have is running and healthy, as of the time the node-engine
|
|
// wrote it. The zero time when it is not.
|
|
func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
|
|
if h.Phase == sdkdelivery.PhaseRunning && h.Health == sdkdelivery.HealthOK {
|
|
return written
|
|
}
|
|
return time.Time{}
|
|
}
|
|
|
|
const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"
|
|
|
|
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
|
|
// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
|
|
// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
|
|
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
|
if _, busy := replacing.LoadOrStore(node, true); busy {
|
|
markGivenAgain(node, started)
|
|
return
|
|
}
|
|
for {
|
|
replaceGivenOnce(ctx, open, node, started)
|
|
replacing.Delete(node)
|
|
// Read after letting go: a call meanwhile only marked the node.
|
|
pending := takeGivenMark(node)
|
|
if pending == nil || ctx.Err() != nil {
|
|
return
|
|
}
|
|
if _, busy := replacing.LoadOrStore(node, true); busy {
|
|
markGivenAgain(node, pending) // the one running now takes it
|
|
return
|
|
}
|
|
started = pending
|
|
}
|
|
}
|
|
|
|
// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
|
|
var givenAgain = struct {
|
|
sync.Mutex
|
|
by map[string]map[string]time.Time
|
|
}{by: map[string]map[string]time.Time{}}
|
|
|
|
func markGivenAgain(node string, started map[string]time.Time) {
|
|
givenAgain.Lock()
|
|
defer givenAgain.Unlock()
|
|
m := givenAgain.by[node]
|
|
if m == nil {
|
|
m = map[string]time.Time{}
|
|
givenAgain.by[node] = m
|
|
}
|
|
for module, at := range started {
|
|
if at.After(m[module]) {
|
|
m[module] = at
|
|
}
|
|
}
|
|
}
|
|
|
|
func takeGivenMark(node string) map[string]time.Time {
|
|
givenAgain.Lock()
|
|
defer givenAgain.Unlock()
|
|
m := givenAgain.by[node]
|
|
delete(givenAgain.by, node)
|
|
return m
|
|
}
|
|
|
|
// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
|
|
var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
|
|
replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
|
|
if err != nil {
|
|
log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
|
|
}
|
|
for _, r := range replaced {
|
|
said := SecretReplaced{Node: node, Module: r.Module, Name: r.Name, Given: r.Given.UTC(),
|
|
Sent: r.Machines, Why: givenWhy}
|
|
log.Printf("replaced %q of %s on %s, given %s, with a value the mesh made: %s; composing %s",
|
|
r.Name, r.Module, node, r.Given.UTC().Format(time.RFC3339), givenWhy, strings.Join(r.Machines, ", "))
|
|
// Composed (novox/hq ADR 0539 clause 5): the compositions that use it get a new revision, which each node's
|
|
// want then names.
|
|
if _, err := composeModules(ctx, open, r.Machines); err != nil {
|
|
said.Sent, said.Unsent = nil, err.Error()
|
|
log.Printf("the new %q of %s is sealed and its compositions on %s are not yet written — the next act that "+
|
|
"composes them carries it: %v", r.Name, r.Module, strings.Join(r.Machines, ", "), err)
|
|
}
|
|
stateReplaced(ctx, said)
|
|
}
|
|
}
|
|
|
|
func stateReplaced(ctx context.Context, said SecretReplaced) {
|
|
if givenEvents == nil {
|
|
return
|
|
}
|
|
body, err := json.Marshal(said)
|
|
if err != nil {
|
|
log.Printf("could not say that %s's %q was replaced: %v", said.Module, said.Name, err)
|
|
return
|
|
}
|
|
stating, cancel := context.WithTimeout(ctx, 10*time.Second)
|
|
defer cancel()
|
|
if err := givenEvents.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeySecretReplaced, body); err != nil {
|
|
log.Printf("could not say that %s's %q was replaced: %v", said.Module, said.Name, err)
|
|
}
|
|
}
|
|
|
|
// watchHavesForGiven replaces the given values of a module on a node each time its have is running and healthy, until ctx
|
|
// ends: never on the watch's own goroutine.
|
|
func watchHavesForGiven(ctx context.Context, open *stores) {
|
|
bus, err := aBus()
|
|
if err != nil {
|
|
log.Printf("given values: the haves are not watched: %v", err)
|
|
return
|
|
}
|
|
js, err := jetstream.New(bus.Conn())
|
|
if err != nil {
|
|
bus.Close()
|
|
return
|
|
}
|
|
kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket)
|
|
if err != nil {
|
|
bus.Close()
|
|
log.Printf("given values: the haves are not watched: %v", err)
|
|
return
|
|
}
|
|
w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
|
|
if err != nil {
|
|
bus.Close()
|
|
log.Printf("given values: the haves are not watched: %v", err)
|
|
return
|
|
}
|
|
goServing(ctx, func() {
|
|
defer bus.Close()
|
|
defer w.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return
|
|
}
|
|
if e == nil || e.Operation() != jetstream.KeyValuePut {
|
|
continue
|
|
}
|
|
node, module, isNode, err := sdkdelivery.ParseHaveKey(e.Key())
|
|
if err != nil || isNode {
|
|
continue
|
|
}
|
|
have, err := sdkdelivery.DecodeHave(e.Value())
|
|
if err != nil {
|
|
continue
|
|
}
|
|
at := startedOn(have, e.Created())
|
|
if at.IsZero() {
|
|
continue
|
|
}
|
|
inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
|
|
}
|
|
}
|
|
})
|
|
}
|