1022 lines
34 KiB
Go
1022 lines
34 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// The provisioner's client on one node (novox/hq ADR 0349 clauses 4 and 5, design 57 P5, issue 671).
|
|
//
|
|
// For each module the controller handed to the provisioner on this node, the client reads its templates'
|
|
// placeholders, records them as the module's needs, watches each owner's keys for this node (a watch delivers the
|
|
// current values first), fills the templates only once every value is there — otherwise it keeps the files there
|
|
// are and reports what it waits for — has the node-engine write each changed file through its narrow local verb,
|
|
// which makes the program take the change as the file declares, and reports `prov.applied` (applied, or failed
|
|
// with the reason). It does so again on every change it watches, writes every file again on a drift pass so the
|
|
// node-engine puts an owned file back, and when its node comes back it starts from the current values. A module
|
|
// taken off the node loses its needs and its report; the node-engine removes the files the declaration no longer
|
|
// names.
|
|
//
|
|
// The owners it fills from are placement (the controller's, in the module's assignment), config
|
|
// (mesh-config_resolved) and the vault, asked for each secret the module's plan names (secrets.go), and for each
|
|
// consumer's credential a provider here lists (consumers.go); and each provider's entries for a consumer here. Every
|
|
// key it writes is its own machine's, and every key it reads but the assignments mirror's, which it reads whole for
|
|
// who binds a provider here (ADR 0373): its bus account is granted no other.
|
|
|
|
// The buckets the client reads and writes, and the subject it reports on.
|
|
const (
|
|
AssignmentsBucket = "mesh-controller_assignments"
|
|
ConfigBucket = "mesh-config_resolved"
|
|
RegistryBucket = "mesh-provisioner_registry"
|
|
NeedsBucket = "mesh-provisioner_needs"
|
|
AppliedBucket = "mesh-provisioner_applied"
|
|
)
|
|
|
|
// AppliedSubject is where the client says what it applied for one module (ADR 0349 clauses 4 and 15).
|
|
func AppliedSubject(node, module string) string { return "prov.applied." + node + "." + module }
|
|
|
|
// What a report says of a module.
|
|
const (
|
|
StatusApplied = "applied"
|
|
StatusWaiting = "waiting"
|
|
StatusFailed = "failed"
|
|
StatusRemoved = "removed"
|
|
)
|
|
|
|
// Report is what the client says of one module on its node, on prov.applied and in mesh-provisioner_applied.
|
|
type Report struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Status string `json:"status"`
|
|
// Reason is why it failed.
|
|
Reason string `json:"reason,omitempty"`
|
|
// WaitsFor is each value it waits for, `<kind>:<name>`, and why.
|
|
WaitsFor []string `json:"waits-for,omitempty"`
|
|
// Templates is the digest of the templates it filled from: which build of the module this is.
|
|
Templates string `json:"templates,omitempty"`
|
|
Files []FileApplied `json:"files,omitempty"`
|
|
// Items are the collections the module's items are recorded in on this node (design 57 L1).
|
|
Items []string `json:"items,omitempty"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// FileApplied is one file as the node-engine wrote it.
|
|
type FileApplied struct {
|
|
Path string `json:"path"`
|
|
Action string `json:"action,omitempty"`
|
|
Digest string `json:"digest,omitempty"`
|
|
Takes string `json:"takes,omitempty"`
|
|
}
|
|
|
|
// NeedRecord is one need a module's templates name, as mesh-provisioner_needs keeps it at
|
|
// `<node>.<module>.<kind>.<name>`: the key's name in a form the bus takes, and the name as written here.
|
|
type NeedRecord struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Kind string `json:"kind"`
|
|
Name string `json:"name"`
|
|
Files []string `json:"files"`
|
|
}
|
|
|
|
// Answer is a registry's entry, as mesh-provisioner keeps it: which domain answers a kind, and where.
|
|
type Answer struct {
|
|
Kind string `json:"kind"`
|
|
Domain string `json:"domain"`
|
|
From string `json:"from"`
|
|
Bucket string `json:"bucket,omitempty"`
|
|
Key string `json:"key,omitempty"`
|
|
// Ask is a seat's verb asked on request and reply, `<seat>.<verb>`: the vault's get for a secret.
|
|
Ask string `json:"ask,omitempty"`
|
|
}
|
|
|
|
// resolved is one value of mesh-config_resolved, as far as filling reads it.
|
|
type resolved struct {
|
|
Value json.RawMessage `json:"value,omitempty"`
|
|
}
|
|
|
|
// Client is one node's provisioner client.
|
|
type Client struct {
|
|
Node string
|
|
JS jetstream.JetStream
|
|
// Conn is where the reports are published.
|
|
Conn *nats.Conn
|
|
Fill Filler
|
|
// Secrets hands the module's secrets from the vault (secrets.go); nil fails every secret placeholder.
|
|
Secrets Secrets
|
|
// Templates fetches and keeps the templates archives.
|
|
Templates *templateStore
|
|
// Drift is how often every applied module's files are written again, so an owned file changed on disk is put
|
|
// back; zero is a minute.
|
|
Drift time.Duration
|
|
// Now dates the reports; nil is time.Now.
|
|
Now func() time.Time
|
|
|
|
// watchBound watches a provider's bucket of entries for this machine's keys, once per run (consumers.go); set by
|
|
// Run.
|
|
watchBound func(bucket string) error
|
|
|
|
// What it last had written, by path: the content's digest. Kept across runs of Run, so a node coming back
|
|
// writes only what moved — and the drift pass writes everything again anyway.
|
|
written map[string][32]byte
|
|
}
|
|
|
|
// state is what one run holds, from its watches.
|
|
type state struct {
|
|
registry map[string]Answer
|
|
plans map[string]Plan
|
|
refused map[string]string // a module whose plan cannot be acted on, and why
|
|
config map[string]map[string]json.RawMessage
|
|
needs map[string][]byte // key → value recorded
|
|
applied map[string]bool // module → a report is kept
|
|
reported map[string][]byte // module → the last report said, undated
|
|
secrets map[string]heldSecret // the vault's name → what this run holds of it; never written anywhere
|
|
// items is every item recorded for this node, by collection and contributor: what a collector is filled from.
|
|
items map[string]map[string]itemValue
|
|
// recorded is each item's key and its value as kept: what this client compares before it records again.
|
|
recorded map[string][]byte
|
|
assigned map[string]bool // every module assigned to this node, with a plan or not (asks.go)
|
|
// binds is every machine's assignment by its key, as far as whom it binds (consumers.go): what a provider here is
|
|
// filled with its consumers from.
|
|
binds map[string]binding
|
|
// providing is each module here whose templates name its consumers (consumers.go), and leftOut what was last said
|
|
// of the consumers left out of one of its files.
|
|
providing map[string]bool
|
|
leftOut map[string]string
|
|
// bound is each provider's bucket of entries this client watches, by key: this machine's consumers' entries alone;
|
|
// boundReady is a bucket whose current entries are all read, and boundUsers the modules filled from each.
|
|
bound map[string]map[string][]byte
|
|
boundReady map[string]bool
|
|
boundUsers map[string]map[string]bool
|
|
}
|
|
|
|
type update struct {
|
|
bucket string
|
|
entry jetstream.KeyValueEntry // nil: the watch delivered every current value
|
|
// changed is what the vault said of a secret that changed (secret-changed): never a value. Set, entry is nil.
|
|
changed []byte
|
|
// asked is the tool runner asking for a module's value (asks.go). Set, entry is nil.
|
|
asked *nats.Msg
|
|
}
|
|
|
|
// Run serves until ctx ends or a watch stops, which it returns: run again, it starts from the current values.
|
|
func (c *Client) Run(ctx context.Context) error {
|
|
if c.written == nil {
|
|
c.written = map[string][32]byte{}
|
|
}
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
kvs := map[string]jetstream.KeyValue{}
|
|
for _, b := range []string{AssignmentsBucket, ConfigBucket, RegistryBucket, NeedsBucket, AppliedBucket,
|
|
CollectBucket} {
|
|
kv, err := c.JS.KeyValue(ctx, b)
|
|
if err != nil {
|
|
return fmt.Errorf("the bucket %s: %w", b, err)
|
|
}
|
|
kvs[b] = kv
|
|
}
|
|
updates := make(chan update, 256)
|
|
stopped := make(chan error, 64)
|
|
watch := func(bucket, keys string) error {
|
|
w, err := kvs[bucket].Watch(ctx, keys)
|
|
if err != nil {
|
|
return fmt.Errorf("watching %s %s: %w", bucket, keys, err)
|
|
}
|
|
go func() {
|
|
defer func() { _ = w.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
stopped <- fmt.Errorf("the watch of %s stopped", bucket)
|
|
return
|
|
}
|
|
select {
|
|
case updates <- update{bucket: bucket, entry: e}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
mine := c.Node + ".>"
|
|
// The registry whole, and every machine's assignment: a provider here is bound by consumers on any machine, in
|
|
// their plans (ADR 0373). Every other bucket, this machine's keys alone.
|
|
for _, b := range []string{RegistryBucket, AssignmentsBucket} {
|
|
if err := watch(b, ">"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, b := range []string{ConfigBucket, NeedsBucket, AppliedBucket, CollectBucket} {
|
|
if err := watch(b, mine); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// A provider's bucket of entries, watched once a plan here names that provider (consumers.go): this machine's
|
|
// consumers' keys alone. A bucket not there yet is tried again at the next fill.
|
|
watchedBound := map[string]bool{}
|
|
c.watchBound = func(bucket string) error {
|
|
if watchedBound[bucket] {
|
|
return nil
|
|
}
|
|
kv, err := c.JS.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
kvs[bucket] = kv
|
|
if err := watch(bucket, mine); err != nil {
|
|
return err
|
|
}
|
|
watchedBound[bucket] = true
|
|
return nil
|
|
}
|
|
defer func() { c.watchBound = nil }()
|
|
// What the vault says changed: a secret this run holds is asked for again, so a rotation reaches its file
|
|
// without a new declaration.
|
|
said, err := c.Conn.Subscribe(SecretChangedSubject, func(m *nats.Msg) {
|
|
select {
|
|
case updates <- update{bucket: SecretChangedSubject, changed: m.Data}:
|
|
case <-ctx.Done():
|
|
}
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("listening for %s: %w", SecretChangedSubject, err)
|
|
}
|
|
defer func() { _ = said.Unsubscribe() }()
|
|
// This machine's tool runner asking for a module's config value (asks.go): answered from this run's state, in
|
|
// its loop, so an answer is never read from half an update.
|
|
asks, err := c.Conn.Subscribe(AskSubject(c.Node), func(m *nats.Msg) {
|
|
select {
|
|
case updates <- update{asked: m}:
|
|
case <-ctx.Done():
|
|
}
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("listening for %s: %w", AskSubject(c.Node), err)
|
|
}
|
|
defer func() { _ = asks.Unsubscribe() }()
|
|
|
|
s := &state{registry: map[string]Answer{}, plans: map[string]Plan{}, refused: map[string]string{},
|
|
config: map[string]map[string]json.RawMessage{}, needs: map[string][]byte{}, applied: map[string]bool{},
|
|
reported: map[string][]byte{}, secrets: map[string]heldSecret{}, items: map[string]map[string]itemValue{},
|
|
recorded: map[string][]byte{}, assigned: map[string]bool{}, binds: map[string]binding{},
|
|
providing: map[string]bool{}, leftOut: map[string]string{}, bound: map[string]map[string][]byte{},
|
|
boundReady: map[string]bool{}, boundUsers: map[string]map[string]bool{}}
|
|
// Every current value first: nothing is filled from half of them.
|
|
current := map[string]bool{}
|
|
var early []*nats.Msg
|
|
for len(current) < 6 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case err := <-stopped:
|
|
return err
|
|
case u := <-updates:
|
|
if u.asked != nil {
|
|
// Answered once every current value is in: never from half of them.
|
|
early = append(early, u.asked)
|
|
continue
|
|
}
|
|
if u.changed != nil {
|
|
// Nothing is held yet: every secret is asked for when it is first filled.
|
|
continue
|
|
}
|
|
if u.entry == nil {
|
|
current[u.bucket] = true
|
|
continue
|
|
}
|
|
c.take(s, u)
|
|
}
|
|
}
|
|
// A module that left while this client was away: its needs and its report go.
|
|
for _, m := range c.leftModules(s) {
|
|
c.leave(ctx, kvs, s, m)
|
|
}
|
|
// Contributors first, so a collector starting with its node is filled from the items there are now.
|
|
for _, m := range contributorsFirst(s.plans) {
|
|
c.reconcile(ctx, kvs, s, m, false)
|
|
}
|
|
for _, m := range sortedKeys(s.refused) {
|
|
c.report(ctx, kvs, s, Report{Module: m, Status: StatusFailed, Reason: s.refused[m]})
|
|
}
|
|
// What was asked meanwhile, and every module's config said to have moved: a watch held across this client's
|
|
// absence is asked again from the current values.
|
|
for _, m := range early {
|
|
c.respond(s, m)
|
|
}
|
|
for _, m := range sortedKeys(s.config) {
|
|
c.tell(m, Changed{Kind: AskConfig})
|
|
}
|
|
|
|
drift := c.Drift
|
|
if drift <= 0 {
|
|
drift = time.Minute
|
|
}
|
|
tick := time.NewTicker(drift)
|
|
defer tick.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case err := <-stopped:
|
|
return err
|
|
case <-tick.C:
|
|
for _, m := range sortedKeys(s.plans) {
|
|
c.reconcile(ctx, kvs, s, m, true)
|
|
}
|
|
case u := <-updates:
|
|
// What moved together is filled together: every update waiting is taken before anything is filled.
|
|
dirty := map[string]bool{}
|
|
for {
|
|
switch {
|
|
case u.asked != nil:
|
|
c.respond(s, u.asked)
|
|
case u.changed != nil:
|
|
for _, m := range c.secretChanged(s, u.changed) {
|
|
dirty[m] = true
|
|
}
|
|
case u.entry != nil:
|
|
for _, m := range c.take(s, u) {
|
|
dirty[m] = true
|
|
}
|
|
c.told(u)
|
|
case strings.HasSuffix(u.bucket, boundSuffix):
|
|
// A provider's bucket's current entries are all read: what waited on them is filled.
|
|
s.boundReady[u.bucket] = true
|
|
for m := range s.boundUsers[u.bucket] {
|
|
dirty[m] = true
|
|
}
|
|
}
|
|
select {
|
|
case u = <-updates:
|
|
continue
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
break
|
|
}
|
|
for _, m := range contributorsFirst(onlyThese(s.plans, dirty)) {
|
|
switch {
|
|
case s.refused[m] != "":
|
|
c.report(ctx, kvs, s, Report{Module: m, Status: StatusFailed, Reason: s.refused[m]})
|
|
case hasPlan(s, m):
|
|
c.reconcile(ctx, kvs, s, m, false)
|
|
default:
|
|
c.leave(ctx, kvs, s, m)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func hasPlan(s *state, m string) bool { _, ok := s.plans[m]; return ok }
|
|
|
|
// told says what an update moved of a module's config, or of its placement, which decides whether it is answered
|
|
// at all (asks.go).
|
|
func (c *Client) told(u update) {
|
|
rest, ok := strings.CutPrefix(u.entry.Key(), c.Node+".")
|
|
if !ok {
|
|
return
|
|
}
|
|
switch u.bucket {
|
|
case ConfigBucket:
|
|
if module, name, ok := strings.Cut(rest, "."); ok {
|
|
c.tell(module, Changed{Kind: AskConfig, Key: name})
|
|
}
|
|
case AssignmentsBucket:
|
|
c.tell(rest, Changed{Kind: AskConfig})
|
|
}
|
|
}
|
|
|
|
// onlyThese is the modules of dirty, each with its plan where it has one: what contributorsFirst orders.
|
|
func onlyThese(plans map[string]Plan, dirty map[string]bool) map[string]Plan {
|
|
out := map[string]Plan{}
|
|
for m := range dirty {
|
|
out[m] = plans[m]
|
|
}
|
|
return out
|
|
}
|
|
|
|
// contributorsFirst is the modules in name order, those contributing an item before the others: a contributor's item
|
|
// recorded first is in its collector's file the first time it is filled.
|
|
func contributorsFirst(plans map[string]Plan) []string {
|
|
var first, then []string
|
|
for _, m := range sortedKeys(plans) {
|
|
if len(plans[m].Items) > 0 {
|
|
first = append(first, m)
|
|
} else {
|
|
then = append(then, m)
|
|
}
|
|
}
|
|
return append(first, then...)
|
|
}
|
|
|
|
// take folds one update into the state, and answers the modules it may change.
|
|
func (c *Client) take(s *state, u update) []string {
|
|
e := u.entry
|
|
gone := e.Operation() != jetstream.KeyValuePut
|
|
key := e.Key()
|
|
switch u.bucket {
|
|
case RegistryBucket:
|
|
if gone {
|
|
delete(s.registry, key)
|
|
} else {
|
|
var a Answer
|
|
if err := json.Unmarshal(e.Value(), &a); err != nil {
|
|
logf("[provisioner-client] the registry's %s cannot be read, so the kind is not answered: %v", key, err)
|
|
delete(s.registry, key)
|
|
} else {
|
|
s.registry[key] = a
|
|
}
|
|
}
|
|
// A kind's answer moves what every module waits on.
|
|
return sortedKeys(s.plans)
|
|
case AssignmentsBucket:
|
|
// Whom it binds, whichever machine's it is: a provider here listing its consumers is filled again.
|
|
var providers []string
|
|
if takeBinds(s, key, e.Value(), gone) {
|
|
providers = sortedKeys(s.providing)
|
|
}
|
|
module, mine := strings.CutPrefix(key, c.Node+".")
|
|
if !mine {
|
|
return providers
|
|
}
|
|
delete(s.plans, module)
|
|
delete(s.refused, module)
|
|
if gone {
|
|
delete(s.assigned, module)
|
|
return append(providers, module)
|
|
}
|
|
s.assigned[module] = true
|
|
m, p, err := planOf(c.Node, key, e.Value())
|
|
switch {
|
|
case errors.Is(err, errNoPlan):
|
|
case err != nil:
|
|
s.refused[module] = err.Error()
|
|
logf("[provisioner-client] %v", err)
|
|
default:
|
|
s.plans[m] = p
|
|
}
|
|
return append(providers, module)
|
|
case ConfigBucket:
|
|
rest := strings.TrimPrefix(key, c.Node+".")
|
|
module, name, ok := strings.Cut(rest, ".")
|
|
if !ok {
|
|
return nil
|
|
}
|
|
if s.config[module] == nil {
|
|
s.config[module] = map[string]json.RawMessage{}
|
|
}
|
|
var r resolved
|
|
if gone || json.Unmarshal(e.Value(), &r) != nil || len(r.Value) == 0 {
|
|
delete(s.config[module], name)
|
|
} else {
|
|
s.config[module][name] = r.Value
|
|
}
|
|
return []string{module}
|
|
case NeedsBucket:
|
|
if gone {
|
|
delete(s.needs, key)
|
|
} else {
|
|
s.needs[key] = e.Value()
|
|
}
|
|
case AppliedBucket:
|
|
module := strings.TrimPrefix(key, c.Node+".")
|
|
s.applied[module] = !gone
|
|
case CollectBucket:
|
|
collection, contributor, ok := itemOfKey(c.Node, key)
|
|
if !ok {
|
|
logf("[provisioner-client] %s in %s is not <node>.<collection>.<contributor>, and is not read", key,
|
|
CollectBucket)
|
|
return nil
|
|
}
|
|
var it ItemRecord
|
|
if gone {
|
|
delete(s.recorded, key)
|
|
delete(s.items[collection], contributor)
|
|
} else if err := json.Unmarshal(e.Value(), &it); err != nil {
|
|
logf("[provisioner-client] the item %s cannot be read, so it is left out: %v", key, err)
|
|
delete(s.recorded, key)
|
|
delete(s.items[collection], contributor)
|
|
} else {
|
|
s.recorded[key] = e.Value()
|
|
if s.items[collection] == nil {
|
|
s.items[collection] = map[string]itemValue{}
|
|
}
|
|
s.items[collection][contributor] = itemValue{text: it.Text, attributes: it.Attributes}
|
|
}
|
|
// Every collector of it on this node is filled again.
|
|
var collectors []string
|
|
for _, m := range sortedKeys(s.plans) {
|
|
if s.plans[m].collects(collection) {
|
|
collectors = append(collectors, m)
|
|
}
|
|
}
|
|
return collectors
|
|
}
|
|
if strings.HasSuffix(u.bucket, boundSuffix) {
|
|
// A provider's entry for one of this machine's consumers: each module filled from that bucket again.
|
|
if s.bound[u.bucket] == nil {
|
|
s.bound[u.bucket] = map[string][]byte{}
|
|
}
|
|
if gone {
|
|
delete(s.bound[u.bucket], key)
|
|
} else {
|
|
s.bound[u.bucket][key] = e.Value()
|
|
}
|
|
return sortedKeys(s.boundUsers[u.bucket])
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// leftModules are the modules with needs or a report kept and no assignment with a plan.
|
|
func (c *Client) leftModules(s *state) []string {
|
|
left := map[string]bool{}
|
|
for key := range s.needs {
|
|
if parts := strings.SplitN(key, ".", 3); len(parts) == 3 && !hasPlan(s, parts[1]) && s.refused[parts[1]] == "" {
|
|
left[parts[1]] = true
|
|
}
|
|
}
|
|
for m, kept := range s.applied {
|
|
if kept && !hasPlan(s, m) && s.refused[m] == "" {
|
|
left[m] = true
|
|
}
|
|
}
|
|
for key := range s.recorded {
|
|
if _, m, ok := itemOfKey(c.Node, key); ok && !hasPlan(s, m) && s.refused[m] == "" {
|
|
left[m] = true
|
|
}
|
|
}
|
|
return sortedKeys(left)
|
|
}
|
|
|
|
// leave removes a module's needs and its report: it is no longer the provisioner's on this node.
|
|
func (c *Client) leave(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string) {
|
|
prefix := c.Node + "." + module + "."
|
|
removed := 0
|
|
for _, key := range sortedKeys(s.needs) {
|
|
if strings.HasPrefix(key, prefix) {
|
|
if err := kvs[NeedsBucket].Delete(ctx, key); err != nil {
|
|
logf("[provisioner-client] the need %s could not be removed: %v", key, err)
|
|
continue
|
|
}
|
|
delete(s.needs, key)
|
|
removed++
|
|
}
|
|
}
|
|
// Its items go: an item is in its collection only while its contributor is placed here.
|
|
removed += c.forgetItems(ctx, kvs, s, module, nil)
|
|
if !s.applied[module] && removed == 0 && s.reported[module] == nil {
|
|
return
|
|
}
|
|
if err := kvs[AppliedBucket].Delete(ctx, c.Node+"."+module); err != nil {
|
|
logf("[provisioner-client] the report of %s could not be removed: %v", module, err)
|
|
}
|
|
s.applied[module] = false
|
|
delete(s.reported, module)
|
|
c.publish(Report{Node: c.Node, Module: module, Status: StatusRemoved, At: c.now()})
|
|
logf("[provisioner-client] %s left %s: its %d need(s) and item(s) and its report are removed", module, c.Node,
|
|
removed)
|
|
}
|
|
|
|
// reconcile fills one module's files from the current values, and says what came of it. force writes every file
|
|
// again, whether it moved or not: the drift pass.
|
|
func (c *Client) reconcile(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string,
|
|
force bool) {
|
|
p := s.plans[module]
|
|
r := Report{Module: module, Templates: p.Templates.Digest}
|
|
texts, err := c.Templates.load(ctx, p.Templates)
|
|
if err != nil {
|
|
r.Status, r.Reason = StatusFailed, err.Error()
|
|
c.report(ctx, kvs, s, r)
|
|
return
|
|
}
|
|
// The needs, file by file: each template, and each file holding one secret alone.
|
|
targets, wrong := fillTargets(p, texts)
|
|
if wrong != "" {
|
|
r.Status, r.Reason = StatusFailed, wrong
|
|
c.report(ctx, kvs, s, r)
|
|
return
|
|
}
|
|
items, wrong := itemTargets(p, texts)
|
|
if wrong != "" {
|
|
r.Status, r.Reason = StatusFailed, wrong
|
|
c.report(ctx, kvs, s, r)
|
|
return
|
|
}
|
|
byFile := map[string]string{}
|
|
needs := map[Need][]string{}
|
|
var order []Need
|
|
named := func(id, text string) {
|
|
for _, n := range needsOf(text) {
|
|
if needs[n] == nil {
|
|
order = append(order, n)
|
|
}
|
|
needs[n] = append(needs[n], id)
|
|
}
|
|
}
|
|
for _, f := range targets {
|
|
byFile[f.Path] = f.text
|
|
named(f.ID, f.text)
|
|
}
|
|
for _, it := range items {
|
|
named(it.id(), it.text)
|
|
}
|
|
c.recordNeeds(ctx, kvs, s, module, order, needs)
|
|
|
|
values := map[Need]string{}
|
|
var failed []string
|
|
for _, n := range order {
|
|
v, wait, fail := c.valueOf(ctx, s, module, p, n)
|
|
switch {
|
|
case fail != "":
|
|
failed = append(failed, fail)
|
|
case wait != "":
|
|
r.WaitsFor = append(r.WaitsFor, wait)
|
|
default:
|
|
values[n] = v
|
|
}
|
|
}
|
|
if len(failed) > 0 {
|
|
r.Status, r.Reason = StatusFailed, strings.Join(failed, "; ")
|
|
c.report(ctx, kvs, s, r)
|
|
return
|
|
}
|
|
if len(r.WaitsFor) > 0 {
|
|
// Every value or none: the files there are stay as they are.
|
|
r.Status = StatusWaiting
|
|
c.report(ctx, kvs, s, r)
|
|
return
|
|
}
|
|
// The items first: recorded, they fill the collectors, this module's own among them.
|
|
for _, it := range items {
|
|
if err := c.record(ctx, kvs, s, module, p, it.PlannedItem, render(it.text, values)); err != nil {
|
|
failed = append(failed, fmt.Sprintf("item %s: %v", it.key(), err))
|
|
continue
|
|
}
|
|
r.Items = append(r.Items, it.key())
|
|
}
|
|
keep := map[string]bool{}
|
|
for _, it := range p.Items {
|
|
keep[it.key()] = true
|
|
}
|
|
c.forgetItems(ctx, kvs, s, module, keep)
|
|
moved := false
|
|
for _, f := range targets {
|
|
content := []byte(render(byFile[f.Path], values))
|
|
sum := sha256.Sum256(content)
|
|
fa := FileApplied{Path: f.Path, Takes: f.Takes}
|
|
if !force && c.written[f.Path] == sum {
|
|
fa.Action = "unchanged"
|
|
r.Files = append(r.Files, fa)
|
|
continue
|
|
}
|
|
a, err := c.Fill.Fill(ctx, f.Path, content)
|
|
if err == nil {
|
|
err = errFill(a)
|
|
}
|
|
if err != nil {
|
|
delete(c.written, f.Path)
|
|
failed = append(failed, fmt.Sprintf("%s: %v", f.Path, err))
|
|
continue
|
|
}
|
|
c.written[f.Path] = sum
|
|
fa.Action, fa.Digest = a.Action, a.Digest
|
|
if a.Action != "unchanged" && a.Action != "kept" {
|
|
moved = true
|
|
logf("[provisioner-client] %s: %s %s (%s)", module, a.Action, f.Path, a.Digest)
|
|
}
|
|
r.Files = append(r.Files, fa)
|
|
}
|
|
if len(failed) > 0 {
|
|
r.Status, r.Reason = StatusFailed, strings.Join(failed, "; ")
|
|
} else {
|
|
r.Status = StatusApplied
|
|
}
|
|
if force && !moved && r.Status == StatusApplied {
|
|
// The drift pass found every file as it should be: nothing to say.
|
|
return
|
|
}
|
|
if force || moved {
|
|
// Said again, even when it reads the same: a file was put back.
|
|
delete(s.reported, module)
|
|
}
|
|
c.report(ctx, kvs, s, r)
|
|
}
|
|
|
|
// valueOf is a need's value, or what it waits for, or why it cannot be filled.
|
|
func (c *Client) valueOf(ctx context.Context, s *state, module string, p Plan, n Need) (value, wait, fail string) {
|
|
answer, known := s.registry[n.Kind]
|
|
switch {
|
|
case slices.Contains(placementKinds, n.Kind):
|
|
if !known {
|
|
return "", n.String() + " (no domain answers " + n.Kind + " in the registry yet)", ""
|
|
}
|
|
v, ok := p.Placement[n.String()]
|
|
if !ok {
|
|
return "", "", n.String() + ": the controller's plan carries no value for it"
|
|
}
|
|
return v, "", ""
|
|
case n.Kind == "config":
|
|
if !known {
|
|
return "", n.String() + " (no domain answers config in the registry yet)", ""
|
|
}
|
|
if answer.Bucket != ConfigBucket {
|
|
return "", "", fmt.Sprintf("%s: the registry answers config from %q, which this client does not watch",
|
|
n, answer.Bucket)
|
|
}
|
|
v, ok := s.config[module][n.Name]
|
|
if !ok {
|
|
return "", n.String() + " (mesh-config holds no value for it on this node)", ""
|
|
}
|
|
return configText(v), "", ""
|
|
case collectionKind(n):
|
|
collection, _ := collectionOf(n)
|
|
if !known {
|
|
return "", n.String() + " (no domain answers " + n.Kind + " in the registry yet)", ""
|
|
}
|
|
if answer.Bucket != CollectBucket {
|
|
return "", "", fmt.Sprintf("%s: the registry answers %s from %q, which this client does not watch", n,
|
|
n.Kind, answer.Bucket)
|
|
}
|
|
col, ok := p.collectionNamed(collection)
|
|
if !ok {
|
|
return "", "", fmt.Sprintf("%s: the plan names no collection %s; a module is filled only from the "+
|
|
"collections its plan says it collects", n, collection)
|
|
}
|
|
if renderer, is := environmentRenderer(n); is {
|
|
text, err := environmentText(s.items[collection], renderer)
|
|
if err != nil {
|
|
return "", "", fmt.Sprintf("%s: %v", n, err)
|
|
}
|
|
return text, "", ""
|
|
}
|
|
where, err := selectorOfNeed(n)
|
|
if err != nil {
|
|
return "", "", fmt.Sprintf("%s: %v", n, err)
|
|
}
|
|
return collectionText(s.items[collection], col.Comment, where), "", ""
|
|
case n.Kind == "secret":
|
|
if !known {
|
|
return "", n.String() + " (no domain answers secret in the registry yet)", ""
|
|
}
|
|
if answer.Ask != VaultSeat+"."+GetVerb {
|
|
return "", "", fmt.Sprintf("%s: the registry answers secret from %q, which this client does not ask", n,
|
|
answer.Ask)
|
|
}
|
|
return c.secretOf(ctx, s, module, p, n)
|
|
case n.Kind == ConsumersKind:
|
|
// A provider's consumers' credentials, each the vault's (ADR 0373).
|
|
if !known {
|
|
return "", n.String() + " (no domain answers consumers in the registry yet)", ""
|
|
}
|
|
if answer.Ask != VaultSeat+"."+GetVerb {
|
|
return "", "", fmt.Sprintf("%s: the registry answers consumers from %q, which this client does not ask",
|
|
n, answer.Ask)
|
|
}
|
|
return c.consumersOf(ctx, s, module, n)
|
|
case n.Kind == BoundKind:
|
|
// A consumer's values, in its provider's entry (ADR 0349 clause 12).
|
|
if !known {
|
|
return "", n.String() + " (no domain answers bound in the registry yet)", ""
|
|
}
|
|
return c.boundOf(s, module, p, n, answer)
|
|
}
|
|
return "", "", fmt.Sprintf("%s: this client does not fill %s placeholders yet", n, n.Kind)
|
|
}
|
|
|
|
// recordNeeds makes the module's needs in mesh-provisioner_needs what its templates name: each need written when it
|
|
// is new or names other files, each one no longer named removed.
|
|
func (c *Client) recordNeeds(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string,
|
|
order []Need, files map[Need][]string) {
|
|
want := map[string][]byte{}
|
|
for _, n := range order {
|
|
v, _ := json.Marshal(NeedRecord{Node: c.Node, Module: module, Kind: n.Kind, Name: n.Name, Files: files[n]})
|
|
want[NeedKey(c.Node, module, n)] = v
|
|
}
|
|
for _, key := range sortedKeys(want) {
|
|
if bytes.Equal(s.needs[key], want[key]) {
|
|
continue
|
|
}
|
|
if _, err := kvs[NeedsBucket].Put(ctx, key, want[key]); err != nil {
|
|
logf("[provisioner-client] the need %s could not be recorded: %v", key, err)
|
|
continue
|
|
}
|
|
s.needs[key] = want[key]
|
|
}
|
|
prefix := c.Node + "." + module + "."
|
|
for _, key := range sortedKeys(s.needs) {
|
|
if _, still := want[key]; strings.HasPrefix(key, prefix) && !still {
|
|
if err := kvs[NeedsBucket].Delete(ctx, key); err != nil {
|
|
logf("[provisioner-client] the need %s could not be removed: %v", key, err)
|
|
continue
|
|
}
|
|
delete(s.needs, key)
|
|
}
|
|
}
|
|
}
|
|
|
|
// NeedKey is a need's key in mesh-provisioner_needs, `<node>.<module>.<kind>.<name>`: the name with every
|
|
// character a key cannot hold, and `=`, written `=<hex>`, so `${seat:mesh-broker:4222}` is `seat.mesh-broker=3A4222`.
|
|
func NeedKey(node, module string, n Need) string {
|
|
var b strings.Builder
|
|
for _, r := range []byte(n.Name) {
|
|
switch {
|
|
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '/', r == '.':
|
|
b.WriteByte(r)
|
|
default:
|
|
fmt.Fprintf(&b, "=%02X", r)
|
|
}
|
|
}
|
|
return node + "." + module + "." + n.Kind + "." + b.String()
|
|
}
|
|
|
|
// report says a module's report on prov.applied and keeps it in mesh-provisioner_applied, when it says something
|
|
// other than the last.
|
|
func (c *Client) report(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, r Report) {
|
|
r.Node = c.Node
|
|
undated, _ := json.Marshal(r)
|
|
if bytes.Equal(s.reported[r.Module], undated) {
|
|
return
|
|
}
|
|
r.At = c.now()
|
|
body, err := json.Marshal(r)
|
|
if err != nil {
|
|
logf("[provisioner-client] the report of %s: %v", r.Module, err)
|
|
return
|
|
}
|
|
if _, err := kvs[AppliedBucket].Put(ctx, c.Node+"."+r.Module, body); err != nil {
|
|
logf("[provisioner-client] the report of %s could not be kept: %v", r.Module, err)
|
|
} else {
|
|
s.applied[r.Module] = true
|
|
}
|
|
c.publish(r)
|
|
s.reported[r.Module] = undated
|
|
switch r.Status {
|
|
case StatusFailed:
|
|
logf("[provisioner-client] %s failed: %s", r.Module, r.Reason)
|
|
case StatusWaiting:
|
|
logf("[provisioner-client] %s waits for %s", r.Module, strings.Join(r.WaitsFor, ", "))
|
|
}
|
|
}
|
|
|
|
func (c *Client) publish(r Report) {
|
|
body, err := json.Marshal(r)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if err := c.Conn.Publish(AppliedSubject(c.Node, r.Module), body); err != nil {
|
|
logf("[provisioner-client] prov.applied for %s could not be said: %v", r.Module, err)
|
|
}
|
|
}
|
|
|
|
func (c *Client) now() time.Time {
|
|
if c.Now != nil {
|
|
return c.Now().UTC()
|
|
}
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
// ItemKey is an item among its module's: the collection, and `/<name>` for a named item.
|
|
func ItemKey(collection, name string) string {
|
|
if name == "" {
|
|
return collection
|
|
}
|
|
return collection + "/" + name
|
|
}
|
|
|
|
func sortedKeys[V any](m map[string]V) []string {
|
|
out := make([]string, 0, len(m))
|
|
for k := range m {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// fillTarget is one file the client fills for a module: a template's, or one holding a secret alone.
|
|
type fillTarget struct {
|
|
PlannedFile
|
|
text string
|
|
}
|
|
|
|
// fillTargets is every file a plan fills, in its order: each template file, then each file of a secret of the
|
|
// module's own or of a credential it requires (issue 757), whose text is that secret and nothing else. Or why one
|
|
// cannot be filled.
|
|
func fillTargets(p Plan, texts map[string]string) ([]fillTarget, string) {
|
|
var out []fillTarget
|
|
for _, f := range p.Files {
|
|
text, ok := texts[f.Template]
|
|
if !ok {
|
|
return nil, fmt.Sprintf("%s: the templates %s hold no %s", f.ID, p.Templates.Digest, f.Template)
|
|
}
|
|
out = append(out, fillTarget{PlannedFile: f, text: text})
|
|
}
|
|
for _, sec := range p.Secrets {
|
|
if sec.Path == "" {
|
|
continue
|
|
}
|
|
out = append(out, fillTarget{PlannedFile: PlannedFile{ID: "secret." + sec.Name, Path: sec.Path,
|
|
OwnerAtRuntime: "owned"}, text: "${secret:" + sec.Name + "}"})
|
|
}
|
|
for _, pr := range p.Provisions {
|
|
if pr.Path == "" {
|
|
continue
|
|
}
|
|
out = append(out, fillTarget{PlannedFile: PlannedFile{ID: "secret." + pr.Name, Path: pr.Path,
|
|
OwnerAtRuntime: "owned"}, text: "${secret:" + pr.Name + "}"})
|
|
}
|
|
return out, ""
|
|
}
|
|
|
|
func collectionKind(n Need) bool { _, ok := collectionOf(n); return ok }
|
|
|
|
// itemTarget is one item a module contributes, with its template's text.
|
|
type itemTarget struct {
|
|
PlannedItem
|
|
text string
|
|
}
|
|
|
|
func (it itemTarget) id() string { return "item." + it.key() }
|
|
|
|
// itemTargets is every item a plan contributes, with its template's text, or why one cannot be filled. An item is
|
|
// kept in a bucket, so a template naming a secret is refused: no secret is in a bucket (ADR 0349 clause 7).
|
|
func itemTargets(p Plan, texts map[string]string) ([]itemTarget, string) {
|
|
var out []itemTarget
|
|
for _, it := range p.Items {
|
|
text, ok := texts[it.Template]
|
|
if !ok {
|
|
return nil, fmt.Sprintf("item %s: the templates %s hold no %s", it.key(), p.Templates.Digest,
|
|
it.Template)
|
|
}
|
|
for _, n := range needsOf(text) {
|
|
if n.Kind == "secret" || n.Kind == ConsumersKind {
|
|
return nil, fmt.Sprintf("item %s: its template %s names %s, and an item is kept in a bucket, where no "+
|
|
"secret is (novox/hq ADR 0349 clause 7)", it.key(), it.Template, n)
|
|
}
|
|
}
|
|
out = append(out, itemTarget{PlannedItem: it, text: text})
|
|
}
|
|
return out, ""
|
|
}
|
|
|
|
// record keeps one item of the module's in mesh-provisioner_collect, when it differs from what is kept.
|
|
func (c *Client) record(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string, p Plan,
|
|
planned PlannedItem, text string) error {
|
|
collection, id := planned.Collection, itemID(module, planned.Name)
|
|
key := CollectKey(c.Node, collection, id)
|
|
value, err := json.Marshal(ItemRecord{Node: c.Node, Collection: collection, Contributor: module, Text: text,
|
|
Name: planned.Name, Attributes: planned.Attributes, Templates: p.Templates.Digest})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if bytes.Equal(s.recorded[key], value) {
|
|
return nil
|
|
}
|
|
if _, err := kvs[CollectBucket].Put(ctx, key, value); err != nil {
|
|
return fmt.Errorf("it could not be recorded: %w", err)
|
|
}
|
|
// Taken now, not only when the watch brings it back: the module's own templates may collect it.
|
|
s.recorded[key] = value
|
|
if s.items[collection] == nil {
|
|
s.items[collection] = map[string]itemValue{}
|
|
}
|
|
s.items[collection][id] = itemValue{text: text, attributes: planned.Attributes}
|
|
logf("[provisioner-client] %s: its item %s is recorded", module, planned.key())
|
|
return nil
|
|
}
|
|
|
|
// forgetItems removes the module's items of every collection keep does not name (nil: every one), and answers how
|
|
// many.
|
|
func (c *Client) forgetItems(ctx context.Context, kvs map[string]jetstream.KeyValue, s *state, module string,
|
|
keep map[string]bool) int {
|
|
removed := 0
|
|
for _, key := range sortedKeys(s.recorded) {
|
|
collection, id, ok := itemOfKey(c.Node, key)
|
|
if owner, name := splitID(id); !ok || owner != module || keep[ItemKey(collection, name)] {
|
|
continue
|
|
}
|
|
if err := kvs[CollectBucket].Delete(ctx, key); err != nil {
|
|
logf("[provisioner-client] the item %s could not be removed: %v", key, err)
|
|
continue
|
|
}
|
|
delete(s.recorded, key)
|
|
delete(s.items[collection], id)
|
|
removed++
|
|
}
|
|
return removed
|
|
}
|