fix(cluster): periodischer Peer-Heartbeat (30s) + Rolling-Update candidate-aware — v1.2.96
Fix 1 — Peer zeigt fälschlich 'offline': runPrimaryPush (Secondary→Primary, einziger periodischer Cross-Node-ha_nodes-Refresh) tickte mit 5 min, SweepStaleNodes-Threshold ist aber 2 min → Secondary war 2 min online, dann 3 min offline, im 5-min-Takt. Tick auf 30s (4× Marge unter Threshold). Receiver lädt nftables nur bei IP-Änderung → kein Reload-Sturm. Fix 2 — Rolling-Update konnte nie fertig werden wenn der Secondary die Zielversion schon hatte (baseline==target → Warten auf unmöglichen Flip → 10-min-Timeout). runRollingUpdate ist jetzt candidate-aware: ermittelt apt-Candidate, überspringt den Secondary-Schritt wenn dieser schon aktuell ist, erkennt den Flip via 'erreicht candidate ODER bewegt sich von baseline', und schließt direkt mit 'done' wenn auch der Primary schon aktuell ist. FinishRollingUpdateIfPending setzt hängende updating/waiting-secondary-Phasen beim Boot auf idle zurück (tote Orchestrierungs-Goroutine nach Restart). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Debian
@@ -826,12 +826,20 @@ func runSecondaryConfigRender(ctx context.Context, pool *pgxpoolPool, box *secre
|
||||
}
|
||||
|
||||
// runPrimaryPush periodically pushes this secondary node's config_hash to the
|
||||
// primary via mTLS. The primary's ha_nodes view only gets config_hash written
|
||||
// during join-time autoRegister — after that the primary never hears about
|
||||
// hash changes unless we push. Without this, the drift banner shows stale
|
||||
// hashes from join-time forever.
|
||||
// primary via mTLS. The primary's ha_nodes view only gets config_hash + last_seen
|
||||
// written during join-time autoRegister — after that the primary never hears about
|
||||
// the secondary unless we push. Without this, the drift banner shows stale hashes
|
||||
// from join-time forever AND the secondary's last_seen freezes → SweepStaleNodes
|
||||
// marks it offline.
|
||||
//
|
||||
// WICHTIG: tick MUSS deutlich unter dem Stale-Threshold (4× 30s = 2 min, siehe
|
||||
// scheduler.staleThreshold / cluster.SweepStaleNodes) liegen. Sonst flippt der
|
||||
// Secondary zwischen den Pushes zwangsläufig auf "offline" (bei 5-min-Tick:
|
||||
// 2 min online, 3 min offline). 30s = 4 Pushes pro Stale-Fenster → ein
|
||||
// verpasster Push (Netz-Glitch) ist unkritisch. Der Receiver (AgentRegisterPeer)
|
||||
// lädt nftables nur bei IP-Änderung neu → kein Reload-Sturm durch häufige Pushes.
|
||||
func runPrimaryPush(ctx context.Context, pool *pgxpoolPool, nodeID, fqdn, version, primaryURL string) {
|
||||
const tick = 5 * time.Minute
|
||||
const tick = 30 * time.Second
|
||||
t := time.NewTicker(tick)
|
||||
defer t.Stop()
|
||||
push := func() {
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"git.netcell-it.de/projekte/edgeguard-native/internal/configgen"
|
||||
"git.netcell-it.de/projekte/edgeguard-native/internal/handlers/response"
|
||||
"git.netcell-it.de/projekte/edgeguard-native/internal/models"
|
||||
aptsvc "git.netcell-it.de/projekte/edgeguard-native/internal/services/apt"
|
||||
)
|
||||
|
||||
// ruStateMu serialisiert Lesen/Schreiben der Rolling-Update-State-Datei
|
||||
@@ -32,17 +33,24 @@ const (
|
||||
phaseFailed = "failed"
|
||||
)
|
||||
|
||||
// FinishRollingUpdateIfPending wird beim API-Start aufgerufen. Wenn die
|
||||
// State-Datei "updating-primary" enthält, bedeutet das dass der Primary
|
||||
// gerade erfolgreich neugestartet ist → Update abgeschlossen → "done" schreiben.
|
||||
// FinishRollingUpdateIfPending wird beim API-Start aufgerufen.
|
||||
// - "updating-primary": der Primary ist gerade erfolgreich neugestartet →
|
||||
// Update abgeschlossen → "done".
|
||||
// - "updating-secondary"/"waiting-secondary": die orchestrierende Goroutine
|
||||
// lief in DIESEM (jetzt neu gestarteten) Prozess und ist mit ihm gestorben.
|
||||
// Die Phase kann nicht weiterlaufen → auf "idle" zurücksetzen, sonst zeigt
|
||||
// die UI ewig "Rolling Update läuft". (Vorher blieb so ein Stand hängen.)
|
||||
func FinishRollingUpdateIfPending() {
|
||||
st := readRollingUpdateState()
|
||||
if st.Phase == phaseUpdatingPrimary {
|
||||
switch st.Phase {
|
||||
case phaseUpdatingPrimary:
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseDone,
|
||||
SecondaryID: st.SecondaryID,
|
||||
SecondaryFQDN: st.SecondaryFQDN,
|
||||
})
|
||||
case phaseUpdatingSecondary, phaseWaitingSecondary:
|
||||
writeRollingUpdateState(RollingUpdateState{Phase: phaseIdle})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -154,72 +162,99 @@ func (h *ClusterHandler) RollingUpdateStatus(c *gin.Context) {
|
||||
func (h *ClusterHandler) runRollingUpdate(secondary *models.HANode) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Vor dem Upgrade die aktuelle Secondary-Version als Baseline merken —
|
||||
// der Flip wird gegen DIESEN Wert geprüft (nicht gegen die Primary-
|
||||
// Version, die fälschlich sofort/nie „flippen" konnte).
|
||||
// Zielversion = das verfügbare apt-Candidate (worauf wir hochziehen) und
|
||||
// die aktuelle Secondary-Version als Baseline. Beides steuert, ob der
|
||||
// Secondary überhaupt etwas zu tun hat.
|
||||
candidate := rollingCandidateVersion(ctx)
|
||||
baseline := secondaryVersion(ctx, h, secondary)
|
||||
target := baseline
|
||||
if target == "" {
|
||||
target = h.Version // Fallback, falls Baseline nicht abrufbar
|
||||
}
|
||||
|
||||
// 1. Secondary triggern
|
||||
slog.Info("rolling-update: posting trigger-update to secondary", "fqdn", secondary.FQDN)
|
||||
result := h.Aggregator.PostPeer(ctx, *secondary, "/agent/cluster/trigger-update")
|
||||
if !result.OK {
|
||||
// Ist der Secondary bereits auf der Zielversion, gibt es nichts
|
||||
// hochzuziehen — KEIN Trigger, KEIN Warten. Sonst würde auf einen
|
||||
// Version-Flip gewartet, der nie kommt → 10-min-Timeout (der frühere Bug,
|
||||
// wenn beide Nodes schon aktuell waren).
|
||||
secondaryUpToDate := candidate != "" && baseline != "" && baseline == candidate
|
||||
if secondaryUpToDate {
|
||||
slog.Info("rolling-update: secondary already at target — skipping secondary step",
|
||||
"version", candidate)
|
||||
} else {
|
||||
// 1. Secondary triggern
|
||||
slog.Info("rolling-update: posting trigger-update to secondary", "fqdn", secondary.FQDN)
|
||||
result := h.Aggregator.PostPeer(ctx, *secondary, "/agent/cluster/trigger-update")
|
||||
if !result.OK {
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseFailed,
|
||||
SecondaryID: secondary.ID,
|
||||
SecondaryFQDN: secondary.FQDN,
|
||||
Error: "trigger-update failed: " + result.Err,
|
||||
})
|
||||
slog.Warn("rolling-update: secondary trigger failed", "error", result.Err)
|
||||
return
|
||||
}
|
||||
|
||||
// 2. Secondary-Version pollen — der Secondary restartet nach dem
|
||||
// Upgrade, danach zeigt /agent/cluster/version eine neue Version.
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseFailed,
|
||||
Phase: phaseWaitingSecondary,
|
||||
SecondaryID: secondary.ID,
|
||||
SecondaryFQDN: secondary.FQDN,
|
||||
Error: "trigger-update failed: " + result.Err,
|
||||
})
|
||||
slog.Warn("rolling-update: secondary trigger failed", "error", result.Err)
|
||||
return
|
||||
}
|
||||
slog.Info("rolling-update: waiting for secondary version flip",
|
||||
"baseline", baseline, "candidate", candidate)
|
||||
|
||||
// 2. Secondary-Version pollen — der Secondary restartet nach dem
|
||||
// Upgrade, danach zeigt /agent/cluster/version eine neue Version.
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseWaitingSecondary,
|
||||
SecondaryID: secondary.ID,
|
||||
SecondaryFQDN: secondary.FQDN,
|
||||
})
|
||||
slog.Info("rolling-update: waiting for secondary version flip")
|
||||
// Kurze Wartezeit damit apt auf dem Secondary erst losläuft
|
||||
time.Sleep(20 * time.Second)
|
||||
|
||||
// Kurze Wartezeit damit apt auf dem Secondary erst losläuft
|
||||
time.Sleep(20 * time.Second)
|
||||
|
||||
deadline := time.Now().Add(10 * time.Minute)
|
||||
versionFlipped := false
|
||||
for time.Now().Before(deadline) {
|
||||
results := h.Aggregator.FanOut(ctx, []models.HANode{*secondary}, "/agent/cluster/version", h.LocalID)
|
||||
if len(results) > 0 && results[0].OK {
|
||||
var ver struct {
|
||||
Version string `json:"version"`
|
||||
}
|
||||
if err := json.Unmarshal(results[0].Data, &ver); err == nil {
|
||||
slog.Info("rolling-update: secondary version", "version", ver.Version, "baseline", target)
|
||||
if ver.Version != "" && ver.Version != target {
|
||||
versionFlipped = true
|
||||
break
|
||||
deadline := time.Now().Add(10 * time.Minute)
|
||||
versionFlipped := false
|
||||
for time.Now().Before(deadline) {
|
||||
results := h.Aggregator.FanOut(ctx, []models.HANode{*secondary}, "/agent/cluster/version", h.LocalID)
|
||||
if len(results) > 0 && results[0].OK {
|
||||
var ver struct {
|
||||
Version string `json:"version"`
|
||||
}
|
||||
if err := json.Unmarshal(results[0].Data, &ver); err == nil {
|
||||
slog.Info("rolling-update: secondary version", "version", ver.Version,
|
||||
"baseline", baseline, "candidate", candidate)
|
||||
// Erfolg = Secondary hat die Zielversion erreicht (candidate)
|
||||
// ODER hat sich gegenüber der Baseline überhaupt bewegt
|
||||
// (Fallback, wenn candidate nicht ermittelbar war).
|
||||
if ver.Version != "" &&
|
||||
((candidate != "" && ver.Version == candidate) || ver.Version != baseline) {
|
||||
versionFlipped = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
time.Sleep(10 * time.Second)
|
||||
}
|
||||
|
||||
if !versionFlipped {
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseFailed,
|
||||
SecondaryID: secondary.ID,
|
||||
SecondaryFQDN: secondary.FQDN,
|
||||
Error: "timeout (10 min) waiting for secondary version flip",
|
||||
})
|
||||
slog.Warn("rolling-update: secondary version flip timeout")
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Second)
|
||||
}
|
||||
|
||||
if !versionFlipped {
|
||||
// 3. Primary (uns selbst) aktualisieren — identisch zu /system/upgrade.
|
||||
// Ist der Primary bereits auf der Zielversion (z. B. beide Nodes schon
|
||||
// aktuell), gibt es nichts zu tun → direkt "done". Sonst liefe ein
|
||||
// apt-Lauf ohne Paket-Wechsel → kein Restart → Phase hinge ewig in
|
||||
// "updating-primary".
|
||||
if candidate != "" && h.Version == candidate {
|
||||
slog.Info("rolling-update: primary already at target — nothing to upgrade", "version", candidate)
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseFailed,
|
||||
Phase: phaseDone,
|
||||
SecondaryID: secondary.ID,
|
||||
SecondaryFQDN: secondary.FQDN,
|
||||
Error: "timeout (10 min) waiting for secondary version flip",
|
||||
})
|
||||
slog.Warn("rolling-update: secondary version flip timeout")
|
||||
return
|
||||
}
|
||||
|
||||
// 3. Primary (uns selbst) aktualisieren — identisch zu /system/upgrade
|
||||
writeRollingUpdateState(RollingUpdateState{
|
||||
Phase: phaseUpdatingPrimary,
|
||||
SecondaryID: secondary.ID,
|
||||
@@ -280,6 +315,15 @@ rm -f /var/lib/edgeguard/upgrade.sh
|
||||
slog.Info("rolling-update: primary upgrade dispatched, process will restart")
|
||||
}
|
||||
|
||||
// rollingCandidateVersion liefert best-effort die verfügbare apt-Candidate-
|
||||
// Version des Meta-Pakets "edgeguard" — also die Version, auf die das Rolling-
|
||||
// Update hochzieht. Leerer String, wenn apt sie nicht ermitteln kann (dann
|
||||
// fällt runRollingUpdate auf reine Baseline-Flip-Erkennung zurück).
|
||||
func rollingCandidateVersion(ctx context.Context) string {
|
||||
vers := aptsvc.PackageVersions(ctx, false)
|
||||
return vers["edgeguard_available"]
|
||||
}
|
||||
|
||||
// secondaryVersion holt best-effort die laufende Version des Peers via mTLS.
|
||||
func secondaryVersion(ctx context.Context, h *ClusterHandler, secondary *models.HANode) string {
|
||||
results := h.Aggregator.FanOut(ctx, []models.HANode{*secondary}, "/agent/cluster/version", h.LocalID)
|
||||
|
||||
Reference in New Issue
Block a user