// 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" "strconv" "strings" "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 Maintenance *policy.Maintenance // Wartung des Knotens (nil = keine) // Disconnected: Knoten getrennt, Routen werden (noch) gehalten – sie // zeigen die Wartungsseite. Disconnected bool } // Labels, mit denen ein Workload (Container, sultan.d-Dienst) seine Services // in Wartung schickt. const ( LabelMaintenance = "sultan.maintenance" LabelMaintenancePublicReason = "sultan.maintenance.public_reason" ) // 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{} serviceMaint := map[string]policy.Maintenance{} 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 if maint, ok := workloadMaintenance(w.Labels); ok { if c.HTTP != nil { for svc := range c.HTTP.Services { serviceMaint[svc] = maint } } if c.TCP != nil { for svc := range c.TCP.Services { serviceMaint[svc] = maint } } } } 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, Maintenance: nc.Maintenance, ServiceMaintenance: serviceMaint, Disconnected: nc.Disconnected, }) 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 } // workloadMaintenance liest sultan.maintenance(=true) und den optionalen // öffentlichen Grund; Schlüssel ohne Beachtung der Groß-/Kleinschreibung. func workloadMaintenance(labels map[string]string) (policy.Maintenance, bool) { var on bool var m policy.Maintenance for k, v := range labels { switch strings.ToLower(k) { case LabelMaintenance: on, _ = strconv.ParseBool(strings.TrimSpace(v)) case LabelMaintenancePublicReason: m.PublicReason = strings.TrimSpace(v) } } return m, on } // 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 }