129 lines
4.6 KiB
Go
129 lines
4.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"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"
|
|
)
|
|
|
|
// **Where the controller reads config values once the move has run** (novox/hq ADR 0348 §9, design 57 K5, issue 664):
|
|
// mesh-config_resolved, the bucket mesh-config alone writes, keyed `<node>.<module>.<key>`, each value with the
|
|
// source it came from. Read only for a module the switch reaches (inventory.ReadsConfig), so a controller before the
|
|
// move never asks the bus for config. The exception to ADR 0348 §9 ("the controller holds no config value") that
|
|
// ADR 0349 clause 17's old way needs, until design 57's R1 hands every module to the provisioner.
|
|
|
|
// resolvedEntry is one key of mesh-config_resolved as the controller reads it: the value and where it came from.
|
|
type resolvedEntry struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value,omitempty"`
|
|
Source string `json:"source"`
|
|
}
|
|
|
|
// The sources a resolved value is laid over a module's files from: mesh config, node config and the module's own
|
|
// value. Its manifest's default is composition's already, and a taken key's default is never laid over its file;
|
|
// a required value nothing sets (`unset`) is refused by composition by name, as it is today.
|
|
var laidSources = map[string]bool{"mesh": true, "node": true, "module": true}
|
|
|
|
// configLaid is the values of one module on one node that are laid over its files, from the resolved entries.
|
|
func configLaid(entries []resolvedEntry, node, module string) (map[string]any, error) {
|
|
out := map[string]any{}
|
|
for _, e := range entries {
|
|
if e.Node != node || e.Module != module || !laidSources[e.Source] {
|
|
continue
|
|
}
|
|
var v any
|
|
if err := json.Unmarshal(e.Value, &v); err != nil {
|
|
return nil, fmt.Errorf("mesh-config's value %s of %s on %s cannot be read: %w", e.Key, module, node, err)
|
|
}
|
|
out[e.Key] = v
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// resolvedRead is what one read of the whole bucket held, and when it was read.
|
|
type resolvedRead struct {
|
|
mu sync.Mutex
|
|
at time.Time
|
|
entries []resolvedEntry
|
|
}
|
|
|
|
// resolvedFresh is how long one read of the bucket answers: a plan reads every module of a machine, and one read
|
|
// serves them all; a value set since is read by the next composition.
|
|
const resolvedFresh = 5 * time.Second
|
|
|
|
var theResolved resolvedRead
|
|
|
|
// readConfigValues is where an opened inventory reads config values once the switch is on; a test replaces it.
|
|
var readConfigValues inventory.ConfigValues = busConfigValues
|
|
|
|
// busConfigValues reads one module's config values on one node from mesh-config_resolved (inventory.ConfigValues).
|
|
func busConfigValues(ctx context.Context, node, module string) (map[string]any, error) {
|
|
theResolved.mu.Lock()
|
|
defer theResolved.mu.Unlock()
|
|
if theResolved.entries == nil || time.Since(theResolved.at) > resolvedFresh {
|
|
entries, err := readResolved(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
theResolved.entries, theResolved.at = entries, time.Now()
|
|
}
|
|
return configLaid(theResolved.entries, node, module)
|
|
}
|
|
|
|
// readResolved is every entry of mesh-config_resolved, read on the bus.
|
|
func readResolved(ctx context.Context) ([]resolvedEntry, error) {
|
|
entries := []resolvedEntry{}
|
|
err := onTheBus(func(conn *nats.Conn) error {
|
|
js, err := jetstream.New(conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
|
|
defer cancel()
|
|
kv, err := js.KeyValue(ctx, broker.ConfigResolvedBucket)
|
|
if err != nil {
|
|
return fmt.Errorf("%s cannot be opened: %w", broker.ConfigResolvedBucket, err)
|
|
}
|
|
w, err := kv.WatchAll(ctx, jetstream.IgnoreDeletes())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = w.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return fmt.Errorf("%s was not read whole: %w", broker.ConfigResolvedBucket, ctx.Err())
|
|
case e, open := <-w.Updates():
|
|
if !open {
|
|
return errors.New(broker.ConfigResolvedBucket + " stopped answering")
|
|
}
|
|
if e == nil {
|
|
return nil
|
|
}
|
|
var r resolvedEntry
|
|
if err := json.Unmarshal(e.Value(), &r); err != nil {
|
|
return fmt.Errorf("%s holds %q, which is not a resolved value: %w", broker.ConfigResolvedBucket,
|
|
e.Key(), err)
|
|
}
|
|
if r.Node == "" || r.Module == "" || r.Key == "" || !strings.HasPrefix(e.Key(), r.Node+"."+r.Module+".") {
|
|
return fmt.Errorf("%s holds %q, whose value names another key", broker.ConfigResolvedBucket, e.Key())
|
|
}
|
|
entries = append(entries, r)
|
|
}
|
|
}
|
|
})
|
|
return entries, err
|
|
}
|