sultan/pkg/registry/registry.go

// Package registry ist der Dynamic Config Aggregator des Servers (Spec
// 5.2): Er hält den zuletzt gemeldeten Zustand (service_state) jedes
// Knotens, übersetzt ihn über traefik.FromLabels, wendet die Policy an,
// führt alle Knoten zusammen und liefert das Ergebnis an Traefik aus
// (GET /v1/traefik-config).
//
// Neu berechnet wird bei jedem Update/Remove und bei Refresh (z. B.
// periodisch oder nach Änderungen an Mandant/Lizenz, und weil Stufen wie
// Ablauf oder Mahnstufe zeitabhängig sind).
package registry

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"net/http"
	"slices"
	"sync"

	"sultan/pkg/license"
	"sultan/pkg/policy"
	"sultan/pkg/traefik"
)

// ErrUnknownNode meldet das Directory für Knoten, die es nicht (mehr) gibt.
var ErrUnknownNode = errors.New("registry: unbekannter Knoten")

// Zusätzliche Verstoßarten, die vor der Policy entstehen.
const (
	// ViolationLabels: Die Labels eines Workloads ließen sich nicht übersetzen.
	ViolationLabels policy.ViolationKind = "labels"
	// ViolationNameConflict: Zwei Workloads eines Knotens nutzen denselben
	// Namen (Workload oder Router/Service/Middleware).
	ViolationNameConflict policy.ViolationKind = "name_conflict"
)

// Workload ist ein Container bzw. ein Dienst aus /etc/sultan.d mit seinen
// Traefik-Labels. Host ist die vom Agenten ermittelte Backend-Adresse
// (LAN- oder NetBird-IP).
type Workload struct {
	Name   string            `json:"name"`
	Host   string            `json:"host"`
	Labels map[string]string `json:"labels"`
}

// NodeState ist der vollständige Zustand eines Knotens (Payload von
// service_state). Er ersetzt den vorherigen komplett.
type NodeState struct {
	Workloads []Workload `json:"workloads"`
	// DownServices: Services (Originalnamen), deren Healthcheck fehlschlägt.
	DownServices []string `json:"down_services,omitempty"`
}

// NodeContext ist, was die Policy über einen Knoten wissen muss.
type NodeContext struct {
	Tenant  policy.Tenant
	License license.License
	Drained bool
}

// Directory liefert den NodeContext; für gelöschte/unbekannte Knoten
// ErrUnknownNode (ggf. verpackt).
type Directory interface {
	NodeContext(ctx context.Context, nodeID string) (NodeContext, error)
}

// Violation ist ein Verstoß mit Herkunft, für das Audit-Log.
type Violation struct {
	NodeID   string
	Workload string // leer bei Verstößen aus der Policy
	Kind     policy.ViolationKind
	Router   string
	Detail   string
	Observed bool
}

// Registry hält den Zustand aller Knoten. Sie ist für parallele Nutzung
// geeignet; die Felder dürfen nach der ersten Verwendung nicht mehr
// geändert werden.
type Registry struct {
	Directory Directory
	Policy    *policy.Engine
	// OnViolation wird für jeden neu aufgetretenen Verstoß einmal aufgerufen
	// (nicht bei jeder Neuberechnung erneut).
	OnViolation func(Violation)

	mu     sync.Mutex
	nodes  map[string]*nodeEntry
	config *traefik.Configuration
	json   []byte
	yaml   []byte
}

type nodeEntry struct {
	state      NodeState
	config     *traefik.Configuration
	violations map[Violation]bool
}

// Update übernimmt den gemeldeten Zustand eines Knotens. Schlägt die
// Auswertung fehl (Directory, Policy), bleibt der vorherige Stand aktiv.
func (r *Registry) Update(ctx context.Context, nodeID string, state NodeState) error {
	if nodeID == "" {
		return fmt.Errorf("registry: Knoten-ID fehlt")
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	entry, err := r.evaluate(ctx, nodeID, state)
	if err != nil {
		return err
	}
	r.store(nodeID, entry)
	return r.rebuild()
}

// Remove entfernt einen Knoten (z. B. Stream getrennt, Knoten gelöscht).
func (r *Registry) Remove(nodeID string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if _, ok := r.nodes[nodeID]; !ok {
		return
	}
	delete(r.nodes, nodeID)
	r.rebuild()
}

// Refresh wertet alle Knoten neu aus. Unbekannte Knoten werden entfernt;
// bei anderen Fehlern bleibt der letzte gültige Stand des Knotens erhalten
// (fail-static) und der Fehler wird zurückgegeben.
func (r *Registry) Refresh(ctx context.Context) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	var errs []error
	for _, id := range r.nodeIDs() {
		entry, err := r.evaluate(ctx, id, r.nodes[id].state)
		switch {
		case errors.Is(err, ErrUnknownNode):
			delete(r.nodes, id)
		case err != nil:
			errs = append(errs, fmt.Errorf("Knoten %s: %w", id, err))
		default:
			r.store(id, entry)
		}
	}
	if err := r.rebuild(); err != nil {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

// Config liefert die aktuell ausgelieferte Konfiguration (nie nil).
func (r *Registry) Config() *traefik.Configuration {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.config == nil {
		return &traefik.Configuration{}
	}
	return r.config
}

// Nodes liefert die Knoten mit gespeichertem Zustand, sortiert.
func (r *Registry) Nodes() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.nodeIDs()
}

// Handler liefert die Konfiguration für Traefiks HTTP-Provider aus
// (?format=json|yaml, Default JSON).
func (r *Registry) Handler() http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		if req.Method != http.MethodGet && req.Method != http.MethodHead {
			w.Header().Set("Allow", "GET, HEAD")
			http.Error(w, "nur GET", http.StatusMethodNotAllowed)
			return
		}
		format, err := traefik.ParseFormat(req.URL.Query().Get("format"))
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		r.mu.Lock()
		body := r.json
		if format == traefik.FormatYAML {
			body = r.yaml
		}
		r.mu.Unlock()
		if body == nil { // noch nie berechnet
			body, _ = (&traefik.Configuration{}).Marshal(format)
		}
		w.Header().Set("Content-Type", format.ContentType())
		w.Header().Set("Cache-Control", "no-store")
		w.Write(body)
	})
}

// evaluate berechnet das Ergebnis eines Knotens, ohne den Zustand zu ändern.
func (r *Registry) evaluate(ctx context.Context, nodeID string, state NodeState) (*nodeEntry, error) {
	nc, err := r.Directory.NodeContext(ctx, nodeID)
	if err != nil {
		return nil, err
	}
	entry := &nodeEntry{state: state, violations: map[Violation]bool{}}
	add := func(v Violation) { v.NodeID = nodeID; entry.violations[v] = true }

	workloads := slices.Clone(state.Workloads)
	slices.SortStableFunc(workloads, func(a, b Workload) int { return cmp.Compare(a.Name, b.Name) })
	merged := &traefik.Configuration{}
	seen := map[string]bool{}
	for _, w := range workloads {
		if seen[w.Name] {
			add(Violation{Workload: w.Name, Kind: ViolationNameConflict, Detail: "Workload-Name doppelt"})
			continue
		}
		seen[w.Name] = true
		c, err := traefik.FromLabels(w.Host, w.Labels)
		if err != nil {
			add(Violation{Workload: w.Name, Kind: ViolationLabels, Detail: err.Error()})
			continue
		}
		m, err := traefik.Merge(merged, c)
		if err != nil {
			add(Violation{Workload: w.Name, Kind: ViolationNameConflict, Detail: err.Error()})
			continue
		}
		merged = m
	}

	down := map[string]bool{}
	for _, s := range state.DownServices {
		down[s] = true
	}
	res, err := r.Policy.Apply(policy.Input{
		Tenant:       nc.Tenant,
		License:      nc.License,
		NodeID:       nodeID,
		NodeDrained:  nc.Drained,
		Config:       merged,
		DownServices: down,
	})
	if err != nil {
		return nil, err
	}
	for _, v := range res.Violations {
		add(Violation{Kind: v.Kind, Router: v.Router, Detail: v.Detail, Observed: v.Observed})
	}
	entry.config = res.Config
	return entry, nil
}

// store übernimmt entry und meldet nur Verstöße, die vorher nicht bestanden.
func (r *Registry) store(nodeID string, entry *nodeEntry) {
	if r.nodes == nil {
		r.nodes = map[string]*nodeEntry{}
	}
	var previous map[Violation]bool
	if old := r.nodes[nodeID]; old != nil {
		previous = old.violations
	}
	r.nodes[nodeID] = entry
	if r.OnViolation == nil {
		return
	}
	var fresh []Violation
	for v := range entry.violations {
		if !previous[v] {
			fresh = append(fresh, v)
		}
	}
	slices.SortFunc(fresh, func(a, b Violation) int {
		return cmp.Or(cmp.Compare(a.Workload, b.Workload), cmp.Compare(a.Router, b.Router), cmp.Compare(a.Kind, b.Kind))
	})
	for _, v := range fresh {
		r.OnViolation(v)
	}
}

// rebuild führt alle Knoten zusammen und rendert die Ausgabe vor.
func (r *Registry) rebuild() error {
	ids := r.nodeIDs()
	configs := make([]*traefik.Configuration, 0, len(ids))
	for _, id := range ids {
		configs = append(configs, r.nodes[id].config)
	}
	merged, err := traefik.Merge(configs...)
	if err != nil {
		// Durch den Namensraum der Policy nicht zu erwarten; die bisherige
		// Ausgabe bleibt dann stehen.
		return fmt.Errorf("registry: Zusammenführen: %w", err)
	}
	j, err := merged.Marshal(traefik.FormatJSON)
	if err != nil {
		return err
	}
	y, err := merged.Marshal(traefik.FormatYAML)
	if err != nil {
		return err
	}
	r.config, r.json, r.yaml = merged, j, y
	return nil
}

func (r *Registry) nodeIDs() []string {
	ids := make([]string, 0, len(r.nodes))
	for id := range r.nodes {
		ids = append(ids, id)
	}
	slices.Sort(ids)
	return ids
}