edgeguard-native/internal/aggregator/aggregator.go

// 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 func() { _ = 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
	// Agent-Endpoints liefern die Standard-API-Envelope zurück
	// ({"data": {...}, "error": null, "message": "ok"}). Wir entpacken
	// das `data`-Feld so dass der Aufrufer direkt die Nutzlast bekommt —
	// konsistent mit dem Lokal-Pfad (der marshalt die Struct direkt ohne
	// Envelope).
	var env struct {
		Data json.RawMessage `json:"data"`
	}
	if err := json.Unmarshal(body, &env); err == nil && len(env.Data) > 0 {
		res.Data = env.Data
	} else {
		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
}

// PostPeer sendet einen POST-Request an einen einzelnen Peer.
// Wird vom Rolling-Update-Orchestrator genutzt um /agent/cluster/trigger-update
// auf dem Secondary auszulösen.
func (a *Aggregator) PostPeer(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()
		return res
	}
	reqCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, target, nil)
	if err != nil {
		res.Err = err.Error()
		return res
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := a.HTTPClient.Do(req)
	if err != nil {
		res.Err = err.Error()
		res.Duration = time.Since(start).Milliseconds()
		return res
	}
	defer func() { _ = resp.Body.Close() }()
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted {
		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.Duration = time.Since(start).Milliseconds()
	return res
}

// PostPeerWithBody sendet einen POST-Request mit JSON-Body an einen Peer.
// Wird für VIP-Schwenk-Tests genutzt (/agent/cluster/vip-cmd).
func (a *Aggregator) PostPeerWithBody(ctx context.Context, p models.HANode, path string, body []byte) 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()
		return res
	}
	reqCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, target, strings.NewReader(string(body)))
	if err != nil {
		res.Err = err.Error()
		return res
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := a.HTTPClient.Do(req)
	if err != nil {
		res.Err = err.Error()
		res.Duration = time.Since(start).Milliseconds()
		return res
	}
	defer func() { _ = resp.Body.Close() }()
	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusNoContent {
		res.Err = fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
		res.Duration = time.Since(start).Milliseconds()
		return res
	}
	res.OK = true
	res.Data = respBody
	res.Duration = time.Since(start).Milliseconds()
	return res
}

// 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