From 90f0df4c45ab1ad5862071c9b4c75a9b16e1213f Mon Sep 17 00:00:00 2001 From: Debian Date: Thu, 4 Jun 2026 13:18:54 +0200 Subject: [PATCH] =?UTF-8?q?fix(cluster):=20Repair-Rollenerkennung=20=C3=BC?= =?UTF-8?q?ber=20pg=5Fpublication=20statt=20ha=5Fnodes.role=20=E2=80=94=20?= =?UTF-8?q?v1.2.88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug: Dispatch an utm-2 schlug fehl ('dieser Node ist der Cluster-Primary'), weil ha_nodes je Node lokal ist und sich JEDE Node selbst als role=primary markiert. Fix: Primary-Erkennung über pg_publication (edgeguard_shared, für jeden DB-User lesbar) statt role/pg_role. Primary gibt dem Subscriber seine eigene Adresse als primary_host mit (PostPeerWithBody); Agent-Handler vertraut dem mTLS-Dispatch mit Safety-Guard 'läuft nie auf dem Publication-Primary'. Funktioniert auch bei Direktzugriff auf den Subscriber. UI-Gating vereinfacht (Drift + Peer + Admin). Co-Authored-By: Claude Opus 4.8 (1M context) --- VERSION | 2 +- internal/handlers/cluster_repair.go | 212 ++++++++++++---------- management-ui/src/pages/Cluster/index.tsx | 22 +-- 3 files changed, 121 insertions(+), 115 deletions(-) diff --git a/VERSION b/VERSION index fba473b..40dde00 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.2.87 \ No newline at end of file +1.2.88 \ No newline at end of file diff --git a/internal/handlers/cluster_repair.go b/internal/handlers/cluster_repair.go index 0c8b30a..6abeb77 100644 --- a/internal/handlers/cluster_repair.go +++ b/internal/handlers/cluster_repair.go @@ -2,6 +2,7 @@ package handlers import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -26,132 +27,146 @@ import ( // Reparatur baut die Subscription neu auf und kopiert alle geteilten // Tabellen frisch vom Primary (einseitig: Primary = Source of Truth). // -// Der Resync MUSS auf dem Standby/Subscriber laufen (nur der hat eine -// Subscription). Operatoren erreichen die UI aber über die VIP, die immer -// auf den Primary zeigt. Deshalb: +// Rollen-Erkennung: NICHT über ha_nodes.role/pg_role — die sind je Node +// lokal und unzuverlässig (jede Node markiert sich selbst, pg_role bleibt +// 'standalone' bis `promote`). Verlässlich ist die PUBLICATION: nur der +// Primary hat `edgeguard_shared` (pg_publication ist für jeden DB-User +// lesbar). Der Subscriber hat sie nicht → er ist das Resync-Ziel. // -// - Auf dem Primary geklickt → Dispatch via mTLS an den Standby -// (POST /agent/cluster/repair-replication), der dort lokal läuft. -// - Auf dem Standby direkt geklickt → läuft lokal. +// Ablauf: +// - Klick auf dem Primary → Dispatch via mTLS an den Peer +// (POST /agent/cluster/repair-replication) mit der eigenen Adresse als +// primary_host; der Peer resynct von dort. +// - Klick direkt auf dem Subscriber → läuft lokal (Quelle = der Peer). // // Die eigentliche Arbeit läuft — analog zum Rolling-Update — in einer -// transienten systemd-Unit, die das bereits getestete -// `edgeguard-ctl cluster-setup-standby ` ausführt. +// transienten systemd-Unit, die `edgeguard-ctl cluster-setup-standby +// ` ausführt. const ( repairUnitName = "edgeguard-repair-replication.service" repairScriptPath = "/var/lib/edgeguard/repair-replication.sh" repairAgentPath = "/agent/cluster/repair-replication" + repairPubName = "edgeguard_shared" // muss zu cmd/edgeguard-ctl egPubName passen ) // validRepairHost erlaubt nur IPv4/IPv6/Hostnamen — der Wert landet in -// einem Bash-Script das als root läuft, also strikt validieren (defense -// in depth, auch wenn er aus ha_nodes stammt). +// einem Bash-Script das als root läuft, also strikt validieren. var validRepairHost = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,253}$`) -// RepairReplication ist der UI-Endpoint. Läuft der lokale Node als -// Primary, wird der Resync an den Standby-Peer delegiert; auf dem Standby -// selbst läuft er lokal. +// repairDispatchBody ist der Body des Agent-Dispatch: der Primary teilt +// dem Subscriber seine Adresse mit, von der resynct werden soll. +type repairDispatchBody struct { + PrimaryHost string `json:"primary_host"` +} + +// RepairReplication ist der UI-Endpoint. Hat dieser Node die Publication +// (= Primary), wird der Resync an den Peer delegiert; sonst (Subscriber) +// läuft er lokal mit dem Peer als Quelle. func (h *ClusterHandler) RepairReplication(c *gin.Context) { if h.Store == nil { response.Internal(c, errors.New("cluster store unavailable")) return } - all, err := h.Store.List(c.Request.Context()) + ctx := c.Request.Context() + all, err := h.Store.List(ctx) if err != nil { response.Internal(c, err) return } local := findNode(all, h.LocalID) + peer := findOtherPeer(all, h.LocalID) + if peer == nil { + response.BadRequest(c, errors.New("kein Peer-Node im Cluster — nichts zu resyncen")) + return + } - // Primary → an den Subscriber-Peer (Nicht-Primary) delegieren. - if isPrimaryNode(local) { - standby := findSubscriberPeer(all, h.LocalID) + if h.nodeHasPublication(ctx) { + // Primary → an den Subscriber-Peer delegieren, mit eigener Adresse. if h.Aggregator == nil { response.BadRequest(c, errors.New("kein mTLS-Aggregator verfügbar — Resync nicht delegierbar")) return } - if standby == nil { - response.BadRequest(c, errors.New("kein Standby-/Subscriber-Node gefunden, an den der Resync delegiert werden könnte")) + primaryHost := pickPrimaryHost(local) + if primaryHost == "" || !validRepairHost.MatchString(primaryHost) { + response.BadRequest(c, errors.New("eigene Primary-Adresse (Mgmt/Internal/Public-IP/FQDN) fehlt oder ist ungültig")) return } - res := h.Aggregator.PostPeer(c.Request.Context(), *standby, repairAgentPath) + body, _ := json.Marshal(repairDispatchBody{PrimaryHost: primaryHost}) + res := h.Aggregator.PostPeerWithBody(ctx, *peer, repairAgentPath, body) if !res.OK { - response.Internal(c, fmt.Errorf("Resync auf %s anstoßen: %s", standby.FQDN, res.Err)) + response.Internal(c, fmt.Errorf("Resync auf %s anstoßen: %s", peer.FQDN, res.Err)) return } - slog.Info("cluster: replication repair delegated to standby", "standby", standby.FQDN) + slog.Info("cluster: replication repair delegated", "target", peer.FQDN, "primary_host", primaryHost) if h.Audit != nil { - _ = h.Audit.Log(c.Request.Context(), actorOf(c), "cluster.repair-replication", - standby.FQDN, gin.H{"target": "standby", "standby": standby.FQDN}, h.NodeID) + _ = h.Audit.Log(ctx, actorOf(c), "cluster.repair-replication", + peer.FQDN, gin.H{"target": "peer", "peer": peer.FQDN, "primary_host": primaryHost}, h.NodeID) } - response.Accepted(c, gin.H{"dispatched": true, "target": "standby", "standby_fqdn": standby.FQDN}) + response.Accepted(c, gin.H{"dispatched": true, "target": "peer", "peer_fqdn": peer.FQDN}) return } - // Standby (oder Direktzugriff) → lokal ausführen. - host, err := h.runLocalRepair(c.Request.Context(), all) - if err != nil { + // Subscriber → lokal ausführen, Quelle = der Peer (Primary). + host := pickPrimaryHost(peer) + if err := h.startResync(ctx, host); err != nil { response.BadRequest(c, err) return } if h.Audit != nil { - _ = h.Audit.Log(c.Request.Context(), actorOf(c), "cluster.repair-replication", + _ = h.Audit.Log(ctx, actorOf(c), "cluster.repair-replication", host, gin.H{"target": "local", "primary": host}, h.NodeID) } response.Accepted(c, gin.H{"dispatched": true, "target": "local", "primary": host}) } -// AgentRepairReplication wird vom Primary via mTLS auf dem Standby -// aufgerufen und startet dort den lokalen Resync. +// AgentRepairReplication wird vom Primary via mTLS auf dem Subscriber +// aufgerufen und startet dort den lokalen Resync von primary_host. func (h *ClusterHandler) AgentRepairReplication(c *gin.Context) { if h.Store == nil { response.Internal(c, errors.New("cluster store unavailable")) return } - all, err := h.Store.List(c.Request.Context()) - if err != nil { - response.Internal(c, err) - return + ctx := c.Request.Context() + var body repairDispatchBody + _ = c.ShouldBindJSON(&body) // best-effort; Fallback unten + + host := strings.TrimSpace(body.PrimaryHost) + if host == "" { + // Fallback: Quelle aus ha_nodes (der andere Node). + if all, err := h.Store.List(ctx); err == nil { + host = pickPrimaryHost(findOtherPeer(all, h.LocalID)) + } } - host, err := h.runLocalRepair(c.Request.Context(), all) - if err != nil { + if err := h.startResync(ctx, host); err != nil { response.BadRequest(c, err) return } slog.Info("cluster: replication repair triggered by peer", "primary", host, "node", h.LocalID) if h.Audit != nil { - _ = h.Audit.Log(c.Request.Context(), "cluster-peer", "cluster.repair-replication", + _ = h.Audit.Log(ctx, "cluster-peer", "cluster.repair-replication", host, gin.H{"target": "local", "primary": host, "via": "agent"}, h.NodeID) } response.Accepted(c, gin.H{"dispatched": true, "primary": host}) } -// runLocalRepair startet den Resync auf DIESEM Node. Verweigert auf dem -// Primary (kein Subscriber). Gibt den ermittelten Primary-Host zurück. -func (h *ClusterHandler) runLocalRepair(_ context.Context, all []models.HANode) (string, error) { - local := findNode(all, h.LocalID) - primary := findPrimary(all) - - if isPrimaryNode(local) { - return "", errors.New("dieser Node ist der Cluster-Primary — Resync läuft nur auf einem Standby/Subscriber") +// startResync schreibt das Repair-Script und startet die transiente +// systemd-Unit. Safety-Guard: läuft NIE auf dem Publication-Primary. +func (h *ClusterHandler) startResync(ctx context.Context, primaryHost string) error { + primaryHost = strings.TrimSpace(primaryHost) + if primaryHost == "" { + return errors.New("keine Primary-Adresse für den Resync ermittelbar") } - if primary == nil { - return "", errors.New("kein Cluster-Primary gefunden — Resync-Quelle unbekannt") + if !validRepairHost.MatchString(primaryHost) { + return fmt.Errorf("ungültige Primary-Adresse: %q", primaryHost) } - if primary.ID == h.LocalID { - return "", errors.New("der lokale Node ist als Primary markiert — Resync nicht möglich") - } - - host := pickPrimaryHost(primary) - if host == "" { - return "", errors.New("Primary hat keine erreichbare IP/FQDN in ha_nodes") - } - if !validRepairHost.MatchString(host) { - return "", fmt.Errorf("ungültige Primary-Adresse: %q", host) + // Niemals auf dem Primary (Publication-Quelle) resyncen — würde die + // eigene Config mit sich selbst überschreiben bzw. ist sinnlos. + if h.nodeHasPublication(ctx) { + return errors.New("dieser Node ist der Publication-Primary — Resync läuft nur auf einem Subscriber") } if st := repairUnitState(); st == "activating" || st == "active" { - return "", errors.New("Resync läuft bereits") + return errors.New("Resync läuft bereits") } script := fmt.Sprintf(`#!/bin/bash @@ -165,10 +180,10 @@ if [ "$rc" -ne 0 ]; then fi echo "[repair] abgeschlossen — config_hash wird beim nächsten Cluster-Status neu berechnet" rm -f %[2]s -`, host, repairScriptPath) +`, primaryHost, repairScriptPath) if err := os.WriteFile(repairScriptPath, []byte(script), 0o755); err != nil { - return "", fmt.Errorf("write repair script: %w", err) + return fmt.Errorf("write repair script: %w", err) } _ = exec.Command("sudo", "-n", "/usr/bin/systemctl", "reset-failed", repairUnitName).Run() cmd := exec.Command("sudo", "-n", "/usr/bin/systemd-run", @@ -177,10 +192,28 @@ rm -f %[2]s "--collect", "bash", repairScriptPath) if err := cmd.Run(); err != nil { - return "", fmt.Errorf("systemd-run failed: %w", err) + return fmt.Errorf("systemd-run failed: %w", err) } - slog.Info("cluster: replication repair dispatched (local)", "primary", host, "node", h.LocalID) - return host, nil + slog.Info("cluster: replication repair dispatched (local)", "primary", primaryHost, "node", h.LocalID) + return nil +} + +// nodeHasPublication prüft, ob dieser Node die Replikations-Publication +// besitzt — das verlässliche Primary-Signal. pg_publication ist für jeden +// DB-User lesbar (anders als pg_subscription). +func (h *ClusterHandler) nodeHasPublication(ctx context.Context) bool { + if h.Store == nil || h.Store.Pool == nil { + return false + } + cctx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + var exists bool + if err := h.Store.Pool.QueryRow(cctx, + `SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname = $1)`, repairPubName, + ).Scan(&exists); err != nil { + return false + } + return exists } // repairStatusResponse spiegelt den Zustand der transienten Repair-Unit. @@ -195,21 +228,20 @@ type repairStatusResponse struct { } // RepairReplicationStatus liest den Job-Zustand. Auf dem Primary wird der -// Status vom Standby-Peer geholt (dort läuft der Job); sonst lokal. +// Status vom Subscriber-Peer geholt (dort läuft der Job); sonst lokal. func (h *ClusterHandler) RepairReplicationStatus(c *gin.Context) { - if h.Store != nil { - if all, err := h.Store.List(c.Request.Context()); err == nil { - local := findNode(all, h.LocalID) - standby := findSubscriberPeer(all, h.LocalID) - if isPrimaryNode(local) && h.Aggregator != nil && standby != nil { - results := h.Aggregator.FanOut(c.Request.Context(), - []models.HANode{*standby}, repairAgentPath+"/status", h.LocalID) + ctx := c.Request.Context() + if h.Store != nil && h.nodeHasPublication(ctx) && h.Aggregator != nil { + if all, err := h.Store.List(ctx); err == nil { + if peer := findOtherPeer(all, h.LocalID); peer != nil { + results := h.Aggregator.FanOut(ctx, + []models.HANode{*peer}, repairAgentPath+"/status", h.LocalID) if len(results) == 1 && results[0].OK && len(results[0].Data) > 0 { c.Data(200, "application/json", wrapEnvelope(results[0].Data)) return } - // Peer nicht erreichbar → idle zurückgeben statt Fehler, - // damit das UI-Polling nicht hart abbricht. + // Peer nicht erreichbar → idle statt Fehler, damit das + // UI-Polling nicht hart abbricht. response.OK(c, repairStatusResponse{Phase: "idle", Log: []string{}}) return } @@ -295,7 +327,7 @@ func localRepairStatus() repairStatusResponse { return out } -// findNode / findByPGRole: kleine Helfer über die ha_nodes-Liste. +// findNode liefert die ha_nodes-Row mit der gegebenen ID. func findNode(nodes []models.HANode, id string) *models.HANode { for i := range nodes { if nodes[i].ID == id { @@ -305,32 +337,13 @@ func findNode(nodes []models.HANode, id string) *models.HANode { return nil } -// isPrimaryNode: ein Node gilt als Primary (Publication-Quelle), wenn -// role ODER pg_role "primary" ist. pg_role bleibt nach cluster-setup- -// standby auf "standalone" (nur `promote` setzt es), daher ist role das -// verlässliche Signal — analog zur keepalived-Logik. -func isPrimaryNode(n *models.HANode) bool { - return n != nil && (n.Role == "primary" || n.PGRole == "primary") -} - -// findPrimary liefert den Primary-Node (Resync-Quelle). -func findPrimary(nodes []models.HANode) *models.HANode { - for i := range nodes { - if isPrimaryNode(&nodes[i]) { - return &nodes[i] - } - } - return nil -} - -// findSubscriberPeer liefert den Resync-Ziel-Peer: ein anderer Node, der -// NICHT der Primary ist (in einem 2-Node-Cluster der Standby/Subscriber). +// findOtherPeer liefert den (einen) anderen Node im 2-Node-Cluster. // Bevorzugt einen online erreichbaren Peer. -func findSubscriberPeer(nodes []models.HANode, localID string) *models.HANode { +func findOtherPeer(nodes []models.HANode, localID string) *models.HANode { var fallback *models.HANode for i := range nodes { n := &nodes[i] - if n.ID == localID || isPrimaryNode(n) { + if n.ID == localID { continue } if n.Status == "online" { @@ -343,10 +356,13 @@ func findSubscriberPeer(nodes []models.HANode, localID string) *models.HANode { return fallback } -// pickPrimaryHost wählt die beste erreichbare Adresse des Primary: +// pickPrimaryHost wählt die beste erreichbare Adresse eines Node: // Mgmt-IP → Internal-IP → Public-IP → FQDN. Strippt eine etwaige // CIDR-Maske (inet-Spalten können "10.0.0.5/32" liefern). func pickPrimaryHost(n *models.HANode) string { + if n == nil { + return "" + } for _, cand := range []*string{n.MgmtIP, n.InternalIP, n.PublicIP} { if cand != nil { if h := strings.TrimSpace(strings.SplitN(*cand, "/", 2)[0]); h != "" { diff --git a/management-ui/src/pages/Cluster/index.tsx b/management-ui/src/pages/Cluster/index.tsx index c940c67..8e0d65b 100644 --- a/management-ui/src/pages/Cluster/index.tsx +++ b/management-ui/src/pages/Cluster/index.tsx @@ -392,17 +392,13 @@ export default function ClusterPage() { const primaryFqdn = data?.local_node?.fqdn ?? window.location.hostname - // Repair-Button: sichtbar bei Drift, für Admins, wenn ein Resync-Ziel - // existiert — auf dem Standby (lokal) oder auf dem Primary (delegiert - // an den Subscriber-Peer). Primary = role ODER pg_role 'primary' - // (pg_role bleibt nach setup-standby 'standalone', role ist verlässlich). - const isPrimaryNode = (n?: HANode | null) => !!n && (n.pg_role === 'primary' || n.role === 'primary') - const localIsPrimary = isPrimaryNode(data?.local_node) - const hasSubscriberPeer = data?.peers?.some(p => !isPrimaryNode(p)) ?? false - const hasPrimary = localIsPrimary || (data?.peers?.some(isPrimaryNode) ?? false) + // Repair-Button: sichtbar bei Drift, für Admins, sobald ein Peer + // existiert. Welche Node Primary (Publication-Quelle) bzw. Subscriber + // ist, entscheidet das Backend zur Laufzeit über pg_publication — die + // UI muss das nicht raten (ha_nodes.role ist je Node lokal/unzuverlässig). const canRepair = !isViewer && !!data?.drift_found - && (localIsPrimary ? hasSubscriberPeer : hasPrimary) + && ((data?.peers?.length ?? 0) > 0) const peerColumns: ColumnsType = [ { @@ -540,13 +536,7 @@ export default function ClusterPage() { banner className="mb-16" message={t('cluster.driftBanner')} - description={ - <> - {t('cluster.driftBannerDesc')} - {localIsPrimary && !hasSubscriberPeer - && {t('cluster.repair.noStandbyHint')}} - - } + description={t('cluster.driftBannerDesc')} action={ canRepair ? (