feat: umfangreiches UI+API-Polish (v1.1.36–1.1.42)
Backend: - Audit-Log: Search-Endpoint mit ILIKE-Filter (actor/action/subject/date) - NTP: /ntp/status via chronyc tracking (Stratum, Offset, Quelle) - System: /service-restart mit Allowlist (haproxy/squid/unbound/chrony/scheduler) - Domain-Response-Headers + Rate-Limit (Migration 0024) - Join-Tokens (Migration 0025), Cluster-mTLS, Aggregator-Fan-Out - apt-Service für Update-Banner (apt-get update + Versionsprüfung) - Backup-Retry mit exponential backoff (retry_apt 3×) - publish.sh fail-fast + cleanup-old.sh (max 10 Versionen) Frontend: - Audit-Log-Page (/audit) mit Filter + Pagination - ErrorBoundary an React-Root + Vite build-target festgenagelt (iOS 15+) - Storage-Schema-Stamp: auto-wipe bei Versions-Mismatch (blank-page-Fix) - EmptyState-Komponente überall ausgerollt - SSL: Aggregate-Karte (total/expiring/expired/errors) - Backups: Aggregate-Karte (letzter Backup/Größe/Fehlschläge 24h) + Backup-Now - NTP: Sync-Status-Karte (chronyc tracking live) - Domains: Backend-UP/DOWN-Chip aus HAProxy-Stats - Backends: HAProxy-Status-Spalte (UP/DEGRADED/DOWN) - Settings: Service-Neustart-Karte (haproxy/squid/unbound/chrony/scheduler) - Settings: Upgrade-Status-Card, Wartungsmodus, Auto-Update, Retention - Dashboard: Recent-Alerts, Cluster-Health, License-Chip, Onboarding-Hint - System-Regeln im Firewall als eigener Tab Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
187
internal/aggregator/aggregator.go
Normal file
187
internal/aggregator/aggregator.go
Normal file
@@ -0,0 +1,187 @@
|
||||
// Package aggregator führt parallele Cluster-Reads gegen alle Peer-Nodes
|
||||
// via mTLS aus.
|
||||
//
|
||||
// Pattern: ein Aggregator-Endpoint auf der Main-API (z.B.
|
||||
// /api/v1/cluster/system/load) ruft Aggregator.FanOut() — das verteilt
|
||||
// die Request parallel an alle Peers' Agent-Listener (:8443 mTLS) und
|
||||
// sammelt die Antworten in einer Map[node_id]→Ergebnis. Timeouts pro
|
||||
// Peer (3s default) verhindern dass ein hängender Peer die ganze Antwort
|
||||
// blockt; partielle Ergebnisse + per-Peer-Fehler werden zurückgegeben.
|
||||
//
|
||||
// mTLS-Auth: ClientTLSConfig aus clustertls.Store. CA muss auf beiden
|
||||
// Seiten dieselbe sein — sonst RequireAndVerifyClientCert lehnt ab.
|
||||
package aggregator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.netcell-it.de/projekte/edgeguard-native/internal/cluster"
|
||||
"git.netcell-it.de/projekte/edgeguard-native/internal/models"
|
||||
)
|
||||
|
||||
// DefaultAgentPort: alle Peers exposen ihren mTLS-Listener auf diesem
|
||||
// Port. api_url in ha_nodes zeigt typischerweise auf den Public-3443-
|
||||
// Port — wir derive'n den Agent-Port daraus, statt eine zweite Spalte
|
||||
// in ha_nodes zu führen.
|
||||
const DefaultAgentPort = 8443
|
||||
|
||||
// DefaultPeerTimeout: pro-Peer-Timeout. Aggregat-Caller sollten eine
|
||||
// Obergrenze von max(N×PeerTimeout/parallel) im Kopf haben; in der
|
||||
// Praxis ist alles parallel, also bestimmt der langsamste Peer die
|
||||
// Latenz.
|
||||
const DefaultPeerTimeout = 3 * time.Second
|
||||
|
||||
// Aggregator: dünner Wrapper mit ClientTLSConfig + http.Client.
|
||||
type Aggregator struct {
|
||||
HTTPClient *http.Client
|
||||
AgentPort int
|
||||
}
|
||||
|
||||
// New: liefert einen Aggregator der ClientTLSConfig verwendet. Wenn
|
||||
// clientTLS == nil, geht der Client auf normales TLS-Verify zurück —
|
||||
// für Tests nützlich, in Prod aber unsicher (würde Cert-Verify gegen
|
||||
// System-Trust laufen, das den Cluster-CA nicht kennt).
|
||||
func New(clientTLS *tls.Config) *Aggregator {
|
||||
tr := &http.Transport{
|
||||
TLSClientConfig: clientTLS,
|
||||
MaxIdleConns: 16,
|
||||
MaxIdleConnsPerHost: 2,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
TLSHandshakeTimeout: 3 * time.Second,
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
ResponseHeaderTimeout: 5 * time.Second,
|
||||
}
|
||||
return &Aggregator{
|
||||
HTTPClient: &http.Client{
|
||||
Transport: tr,
|
||||
Timeout: DefaultPeerTimeout,
|
||||
},
|
||||
AgentPort: DefaultAgentPort,
|
||||
}
|
||||
}
|
||||
|
||||
// PeerResult kapselt das Ergebnis eines parallelen Fan-Out-Calls.
|
||||
// Wenn Err != nil ist Data leer; sonst enthält Data den raw-JSON-Body
|
||||
// (Aufrufer entscheidet ob es per-Peer typed-unmarshalled oder als
|
||||
// map[string]any belassen wird).
|
||||
type PeerResult struct {
|
||||
NodeID string `json:"node_id"`
|
||||
FQDN string `json:"fqdn"`
|
||||
OK bool `json:"ok"`
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
Err string `json:"error,omitempty"`
|
||||
Duration int64 `json:"duration_ms"`
|
||||
}
|
||||
|
||||
// FanOut: ruft GET <agent-url>/<path> für jeden Peer in `peers` parallel
|
||||
// und sammelt die Ergebnisse in einer slice (stabile Sortierung nach
|
||||
// Peer-FQDN für deterministisches UI-Rendering).
|
||||
//
|
||||
// `path` ist relativ, z.B. "/agent/system/load". `localID` wird als
|
||||
// Marker übergeben damit der Aufrufer den eigenen Node von der Map
|
||||
// ausschließen kann.
|
||||
func (a *Aggregator) FanOut(ctx context.Context, peers []models.HANode, path, localID string) []PeerResult {
|
||||
if !strings.HasPrefix(path, "/") {
|
||||
path = "/" + path
|
||||
}
|
||||
results := make([]PeerResult, len(peers))
|
||||
var wg sync.WaitGroup
|
||||
for i, p := range peers {
|
||||
if p.ID == localID {
|
||||
// Eigener Node nicht über mTLS dial'n — wäre teuer + im
|
||||
// Aufrufer wahrscheinlich der lokale Path
|
||||
results[i] = PeerResult{NodeID: p.ID, FQDN: p.FQDN, OK: false, Err: "skipped: local node"}
|
||||
continue
|
||||
}
|
||||
wg.Add(1)
|
||||
i := i
|
||||
p := p
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
results[i] = a.callPeer(ctx, p, path)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
return results
|
||||
}
|
||||
|
||||
// callPeer macht den Einzel-Call. Wandelt p.APIURL in https://host:8443/
|
||||
// um (Port übersteuert, Pfad ersetzt). Bei Connection-Fehler / Timeout
|
||||
// liefert ein PeerResult mit OK=false zurück.
|
||||
func (a *Aggregator) callPeer(ctx context.Context, p models.HANode, path string) PeerResult {
|
||||
start := time.Now()
|
||||
res := PeerResult{NodeID: p.ID, FQDN: p.FQDN}
|
||||
|
||||
target, err := agentURL(p.APIURL, a.AgentPort, path)
|
||||
if err != nil {
|
||||
res.Err = "bad api_url: " + err.Error()
|
||||
res.Duration = time.Since(start).Milliseconds()
|
||||
return res
|
||||
}
|
||||
reqCtx, cancel := context.WithTimeout(ctx, DefaultPeerTimeout)
|
||||
defer cancel()
|
||||
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, target, nil)
|
||||
if err != nil {
|
||||
res.Err = err.Error()
|
||||
res.Duration = time.Since(start).Milliseconds()
|
||||
return res
|
||||
}
|
||||
resp, err := a.HTTPClient.Do(req)
|
||||
if err != nil {
|
||||
res.Err = err.Error()
|
||||
res.Duration = time.Since(start).Milliseconds()
|
||||
return res
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1 MiB cap
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
res.Err = fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
||||
res.Duration = time.Since(start).Milliseconds()
|
||||
return res
|
||||
}
|
||||
res.OK = true
|
||||
res.Data = body
|
||||
res.Duration = time.Since(start).Milliseconds()
|
||||
return res
|
||||
}
|
||||
|
||||
// agentURL: nimmt z.B. "https://node1.example.com:3443" + port=8443 +
|
||||
// path="/agent/system/load" und liefert "https://node1.example.com:8443/agent/system/load".
|
||||
// Wir tauschen den Port aus, behalten Schema + Host (nur).
|
||||
func agentURL(apiURL string, agentPort int, path string) (string, error) {
|
||||
if apiURL == "" {
|
||||
return "", errors.New("empty api_url")
|
||||
}
|
||||
u, err := url.Parse(apiURL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if u.Scheme == "" {
|
||||
u.Scheme = "https"
|
||||
}
|
||||
host := u.Hostname()
|
||||
if host == "" {
|
||||
return "", errors.New("api_url has no host")
|
||||
}
|
||||
u.Host = net.JoinHostPort(host, fmt.Sprint(agentPort))
|
||||
u.Path = path
|
||||
u.RawQuery = ""
|
||||
return u.String(), nil
|
||||
}
|
||||
|
||||
// Compile-time check dass cluster importiert wird (für Drift-Detection
|
||||
// vom hashSpec — die Aggregator-Resultate werden parallel im Drift-
|
||||
// Banner mitverarbeitet). Nicht runtime-essentiell, aber dokumentiert
|
||||
// die Abhängigkeit.
|
||||
var _ = cluster.ComputeConfigHash
|
||||
Reference in New Issue
Block a user