feat(cluster): Rolling Update — Secondary-first upgrade orchestration (v1.2.3)

POST /cluster/rolling-update startet den gestaffelten Upgrade-Prozess: 1. Secondary via mTLS /agent/cluster/trigger-update anstoßen 2. /agent/cluster/version pollen bis Secondary Version-Flip zeigt (max 10 min) 3. Primary self-upgrade via systemd-run (identisch zu /system/upgrade) State wird in /var/lib/edgeguard/rolling-update-state.json persistiert: Phasen: updating-secondary → waiting-secondary → updating-primary. "done" wird nicht geschrieben — Prozess stirbt beim Upgrade. UI erkennt Abschluss via /system/health version-flip (analog Single-Node-Upgrade). UI: UpdateBanner erkennt Cluster-Modus (/cluster/status mode="cluster") und tauscht den "Install now"-Button gegen "Rolling Update (Cluster)" aus. Multi-Step-Modal zeigt die drei Phasen; ab updating-primary wechselt der Client auf /system/health polling. Aggregator.PostPeer: neuer einzel-POST-Helper für mTLS-trigger-update. WithVersion(): ClusterHandler bekommt Binary-Version für /agent/cluster/version. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 23:40:49 +02:00
parent 25c7cd0cb5
commit bc6db1fc2b
9 changed files with 824 additions and 67 deletions
--- a/internal/handlers/cluster.go
+++ b/internal/handlers/cluster.go
@@ -7,6 +7,9 @@ import (
 	"encoding/pem"
 	"fmt"
 	"log/slog"
+	"net/http"
+	"os"
+	"os/exec"
 	"strings"
 	"time"

@@ -31,6 +34,7 @@ type ClusterHandler struct {
 	Store      *cluster.Store
 	LocalID    string
 	Aggregator *aggregator.Aggregator
+	Version    string // laufende Binary-Version, für Rolling-Update-Koordination

 	// TLSStore + Tokens: optional, gesetzt bei Phase 3.4. Erlauben das
 	// Generieren von Join-Tokens und das Issue-Cert für joining Peers.
@@ -67,6 +71,10 @@ func (h *ClusterHandler) Register(rg *gin.RouterGroup) {
 	g.GET("/status", h.Status)
 	g.GET("/system/load", h.SystemLoad)
 	g.DELETE("/nodes/:id", h.DeleteNode)
+	g.GET("/vip-settings", h.GetVIPSettings)
+	g.PUT("/vip-settings", h.UpdateVIPSettings)
+	g.POST("/rolling-update", h.RollingUpdate)
+	g.GET("/rolling-update/status", h.RollingUpdateStatus)
 	if h.TLSStore != nil {
 		g.GET("/cert-status", h.CertStatus)
 		g.POST("/renew-self", h.RenewSelf)
@@ -115,6 +123,73 @@ func (h *ClusterHandler) DeleteNode(c *gin.Context) {
 	response.NoContent(c)
 }

+// GetVIPSettings liest die cluster_settings-Singleton-Row (VIP/VRRP-Config).
+func (h *ClusterHandler) GetVIPSettings(c *gin.Context) {
+	if h.Store == nil {
+		response.NotFound(c, simpleError("cluster store not available"))
+		return
+	}
+	var cs vipSettingsRow
+	row := h.Store.Pool.QueryRow(c.Request.Context(),
+		`SELECT vip_address, vip_interface, vip_auth_pass, vrrp_router_id FROM cluster_settings WHERE id = 1`)
+	if err := row.Scan(&cs.VIPAddress, &cs.VIPInterface, &cs.VIPAuthPass, &cs.VRRPRouterID); err != nil {
+		response.Internal(c, err)
+		return
+	}
+	response.OK(c, cs)
+}
+
+// UpdateVIPSettings speichert die VIP/VRRP-Konfiguration und triggert
+// einen Keepalived-Config-Render. Viewer-Schutz via RequireAdminForMutations-
+// Middleware auf der authed-Group — kein Extra-Check nötig.
+func (h *ClusterHandler) UpdateVIPSettings(c *gin.Context) {
+	var req vipSettingsRow
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.BadRequest(c, err)
+		return
+	}
+	if h.Store == nil {
+		response.NotFound(c, simpleError("cluster store not available"))
+		return
+	}
+	_, err := h.Store.Pool.Exec(c.Request.Context(), `
+		UPDATE cluster_settings
+		SET vip_address=$1, vip_interface=$2, vip_auth_pass=$3, vrrp_router_id=$4, updated_at=NOW()
+		WHERE id=1`,
+		nullIfEmpty(req.VIPAddress), nullIfEmpty(req.VIPInterface),
+		nullIfEmpty(req.VIPAuthPass), req.VRRPRouterID)
+	if err != nil {
+		response.Internal(c, err)
+		return
+	}
+	slog.Info("cluster: VIP settings updated", "vip", req.VIPAddress, "actor", actorOf(c))
+	// Keepalived-Config asynchron neu rendern
+	if h.PeerReloader != nil {
+		go func() {
+			rctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+			defer cancel()
+			if err := h.PeerReloader(rctx); err != nil {
+				slog.Warn("cluster: keepalived render after VIP update failed", "error", err)
+			}
+		}()
+	}
+	response.NoContent(c)
+}
+
+type vipSettingsRow struct {
+	VIPAddress   *string `json:"vip_address"`
+	VIPInterface *string `json:"vip_interface"`
+	VIPAuthPass  *string `json:"vip_auth_pass"`
+	VRRPRouterID int     `json:"vrrp_router_id"`
+}
+
+func nullIfEmpty(s *string) *string {
+	if s == nil || *s == "" {
+		return nil
+	}
+	return s
+}
+
 // RegisterPublic mountet die public (unauth) Endpoints — joining Peers
 // haben noch keine Session/Cert, deshalb läuft /issue-cert vor der
 // requireAuth-Middleware. Aufrufer muss diesen Group auf /api/v1 setzen
@@ -139,6 +214,9 @@ func (h *ClusterHandler) RegisterAgent(rg *gin.RouterGroup) {
 	g := rg.Group("/agent/cluster")
 	g.POST("/peers", h.AgentRegisterPeer)
 	g.GET("/identity", h.AgentIdentity)
+	g.GET("/pg-replication-info", h.AgentPGReplicationInfo)
+	g.GET("/version", h.AgentVersion)
+	g.POST("/trigger-update", h.AgentTriggerUpdate)
 }

 // AgentIdentity gibt die eigene ha_nodes-Row zurück. Wird vom Primary
@@ -161,6 +239,45 @@ func (h *ClusterHandler) AgentIdentity(c *gin.Context) {
 	response.OK(c, node)
 }

+// AgentPGReplicationInfo gibt die Replication-Credentials für pg_basebackup
+// zurück. Nur über den mTLS-Agent-Listener erreichbar. Liest das Passwort
+// aus /var/lib/edgeguard/pg-replication-secret. Gibt 404 zurück wenn die
+// Datei fehlt (cluster-init-replication noch nicht ausgeführt).
+func (h *ClusterHandler) AgentPGReplicationInfo(c *gin.Context) {
+	const secretPath = "/var/lib/edgeguard/pg-replication-secret"
+	pass, err := readFileString(secretPath)
+	if err != nil {
+		response.NotFound(c, simpleError("pg-replication-secret nicht gefunden — cluster-init-replication auf dem Primary ausführen"))
+		return
+	}
+	// Host = eigene Public-IP aus ha_nodes (oder Fallback: FQDN)
+	host := ""
+	if h.Store != nil && h.LocalID != "" {
+		if node, err := h.Store.Get(c.Request.Context(), h.LocalID); err == nil {
+			if node.PublicIP != nil && *node.PublicIP != "" {
+				host = *node.PublicIP
+			}
+			if host == "" {
+				host = node.FQDN
+			}
+		}
+	}
+	response.OK(c, gin.H{
+		"host":     host,
+		"port":     5432,
+		"user":     "edgeguard_replicator",
+		"password": strings.TrimSpace(pass),
+	})
+}
+
+func readFileString(path string) (string, error) {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
 // PeerReloader: optionale Funktion die nach einem Auto-Register
 // Firewall + ggfs. andere Configs regeneriert (damit peer_ipv4-Set
 // frisch ist). Wird vom main.go gesetzt.
@@ -172,6 +289,12 @@ func (h *ClusterHandler) WithPeerReloader(r PeerReloader) *ClusterHandler {
 	return h
 }

+// WithVersion: setzt die laufende Binary-Version für Rolling-Update-Koordination.
+func (h *ClusterHandler) WithVersion(v string) *ClusterHandler {
+	h.Version = v
+	return h
+}
+
 func (h *ClusterHandler) ListNodes(c *gin.Context) {
 	nodes, err := h.Store.List(c.Request.Context())
 	if err != nil {
@@ -555,6 +678,60 @@ func (h *ClusterHandler) reconcileJoiningPeers(placeholders []models.HANode) {
 	}
 }

+// AgentVersion gibt die laufende Binary-Version zurück. Wird vom Rolling-
+// Update-Orchestrator gepollt um zu erkennen wann der Secondary die neue
+// Version hat.
+func (h *ClusterHandler) AgentVersion(c *gin.Context) {
+	response.OK(c, gin.H{"version": h.Version})
+}
+
+// AgentTriggerUpdate startet den Upgrade-Prozess auf diesem Node via
+// systemd-run (detached). Wird vom Primary via mTLS aufgerufen um den
+// Secondary zuerst zu aktualisieren (Rolling-Update). Pattern identisch
+// zu /system/upgrade — nutzt dieselbe Sudoers-Whitelist aus dem postinst.
+func (h *ClusterHandler) AgentTriggerUpdate(c *gin.Context) {
+	const scriptPath = "/var/lib/edgeguard/upgrade.sh"
+	const script = `#!/bin/bash
+set -e
+sleep 2
+export DEBIAN_FRONTEND=noninteractive
+dpkg --configure -a || true
+retry_apt() {
+    local attempt=0 max=3 wait_for=15
+    while [ $attempt -lt $max ]; do
+        attempt=$((attempt + 1))
+        apt-get update -qq || true
+        if apt-get install -y -qq -o Dpkg::Options::=--force-confold \
+            edgeguard-api edgeguard-ui edgeguard; then return 0; fi
+        [ $attempt -lt $max ] && sleep $wait_for && wait_for=$((wait_for * 2))
+    done
+    return 1
+}
+retry_apt
+echo "[upgrade] complete"
+rm -f /var/lib/edgeguard/upgrade.sh
+`
+	if err := os.WriteFile(scriptPath, []byte(script), 0o755); err != nil {
+		response.Internal(c, err)
+		return
+	}
+	const unitName = "edgeguard-upgrade.service"
+	_ = exec.Command("sudo", "-n", "/usr/bin/systemctl", "reset-failed", unitName).Run()
+	cmd := exec.Command("sudo", "-n", "/usr/bin/systemd-run",
+		"--unit="+unitName,
+		"--description=EdgeGuard rolling-update (triggered by primary)",
+		"--collect",
+		"bash", scriptPath)
+	if err := cmd.Run(); err != nil {
+		slog.Warn("cluster: AgentTriggerUpdate: systemd-run failed", "error", err)
+		response.Internal(c, err)
+		return
+	}
+	slog.Info("cluster: rolling update triggered on this node by primary mTLS call",
+		"client", c.ClientIP())
+	c.JSON(http.StatusAccepted, gin.H{"status": "upgrading"})
+}
+
 var errInvalidJoinRequest = simpleError("missing token or csr")

 type simpleError string
@@ -623,14 +800,15 @@ func (h *ClusterHandler) RenewSelf(c *gin.Context) {
 // die wir wirklich brauchen — sonst kann ein joining Peer beliebige
 // ha_nodes-Felder überschreiben.
 type registerPeerRequest struct {
-	ID         string `json:"id"`           // Joiner's eigene node-id
-	Name       string `json:"name"`         // hostname
-	FQDN       string `json:"fqdn"`         // sollte mit Client-Cert-CN matchen
-	APIURL     string `json:"api_url"`      // https://<fqdn>
-	PublicIP   string `json:"public_ip"`    // optional
-	InternalIP string `json:"internal_ip"`  // mTLS-Listener-IP (für peer_ipv4-Set)
-	MgmtIP     string `json:"mgmt_ip"`      // optional
-	Version    string `json:"version"`
+	ID         string  `json:"id"`           // Joiner's eigene node-id
+	Name       string  `json:"name"`         // hostname
+	FQDN       string  `json:"fqdn"`         // sollte mit Client-Cert-CN matchen
+	APIURL     string  `json:"api_url"`      // https://<fqdn>
+	PublicIP   string  `json:"public_ip"`    // optional
+	InternalIP string  `json:"internal_ip"`  // mTLS-Listener-IP (für peer_ipv4-Set)
+	MgmtIP     string  `json:"mgmt_ip"`      // optional
+	Version    string  `json:"version"`
+	ConfigHash *string `json:"config_hash"`  // nil=absent (don't change), ""=no user config
 }

 // AgentRegisterPeer: vom Joiner nach issue-cert via mTLS aufgerufen.
@@ -693,6 +871,9 @@ func (h *ClusterHandler) AgentRegisterPeer(c *gin.Context) {
 		v := req.Version
 		n.Version = &v
 	}
+	if req.ConfigHash != nil {
+		n.ConfigHash = req.ConfigHash
+	}
 	// Placeholder zuerst löschen: ha_nodes hat UNIQUE(fqdn). Der INSERT
 	// in UpsertSelf verwendet ON CONFLICT(id) — greift NICHT bei fqdn-
 	// Konflikten. Ohne das Delete würde der INSERT mit "duplicate key on