mesh-controller/cmd/mesh-controller/config_read.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
}