feat(cluster): pull-basierte Peer-Registrierung als autoRegister-Fallback

Wenn autoRegister (Push utm-2→utm-1) fehlschlägt, bleibt der Peer
dauerhaft im 'joining'-Status. Neuer Self-Healing-Mechanismus:

- GET /agent/cluster/identity auf dem Agent-Listener: gibt die eigene
  ha_nodes-Row zurück (echte Node-ID, FQDN, Version)
- ClusterHandler.Status() startet für jeden 'joining'/'pending'-Peer
  einen Background-Reconcile via Aggregator: holt Identity, upsertet
  mit echter ID (public_ip aus Placeholder übernommen für nftables),
  löscht Placeholder
- autoRegister-Fehler werden jetzt als Warn-Log sichtbar statt silent
  verworfen

Damit reicht ein Cluster-Page-Aufruf nach Update beider Nodes um den
Status von 'joining' auf 'online' zu bringen.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Debian
2026-05-29 19:06:19 +02:00
parent 6eb6d6d6a4
commit ffb579879d
6 changed files with 113 additions and 7 deletions

View File

@@ -138,6 +138,27 @@ func (h *ClusterHandler) RegisterPublic(rg *gin.RouterGroup) {
func (h *ClusterHandler) RegisterAgent(rg *gin.RouterGroup) {
g := rg.Group("/agent/cluster")
g.POST("/peers", h.AgentRegisterPeer)
g.GET("/identity", h.AgentIdentity)
}
// AgentIdentity gibt die eigene ha_nodes-Row zurück. Wird vom Primary
// genutzt um joining-Peers aktiv zu reconcilen wenn autoRegister (Push)
// fehlgeschlagen ist — Pull-Fallback.
func (h *ClusterHandler) AgentIdentity(c *gin.Context) {
if h.Store == nil || h.LocalID == "" {
response.NotFound(c, simpleError("node not registered"))
return
}
node, err := h.Store.Get(c.Request.Context(), h.LocalID)
if err != nil {
if err == cluster.ErrNotFound {
response.NotFound(c, simpleError("local node not in ha_nodes yet"))
return
}
response.Internal(c, err)
return
}
response.OK(c, node)
}
// PeerReloader: optionale Funktion die nach einem Auto-Register
@@ -212,6 +233,23 @@ func (h *ClusterHandler) Status(c *gin.Context) {
if len(out.Peers) > 0 {
out.Mode = "cluster"
}
// Pull-Reconcile für joining-Peers: wenn ein Peer via Aggregator
// erreichbar ist aber noch mit Placeholder-ID in ha_nodes steht,
// holen wir seine echte Identity aktiv ab. Best-effort goroutine —
// blockiert die Status-Response nicht.
if h.Aggregator != nil && h.Store != nil {
var joining []models.HANode
for _, p := range out.Peers {
if p.Status == "joining" || p.Status == "pending" {
joining = append(joining, p)
}
}
if len(joining) > 0 {
go h.reconcileJoiningPeers(joining)
}
}
// Drift-Detection: jeder peer mit anderem config_hash als unser
// lokaler → Banner-Trigger im UI.
if localHash != nil && *localHash != "" {
@@ -458,6 +496,62 @@ func cnFromCSR(csrPEM string) string {
return csr.Subject.CommonName
}
// reconcileJoiningPeers versucht für jeden Peer im Status "joining" oder
// "pending" die echte Node-ID via /agent/cluster/identity zu holen und
// ihn in ha_nodes mit der richtigen ID einzutragen. Self-Healing-
// Fallback wenn autoRegister (Push von joining-Peer zu Primary) wegen
// eines temporären Netzwerkproblems fehlgeschlagen ist.
//
// Die public_ip des Placeholder-Rows wird in der neuen Row übernommen
// damit @peer_ipv4 (nftables) korrekt bleibt.
func (h *ClusterHandler) reconcileJoiningPeers(placeholders []models.HANode) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
results := h.Aggregator.FanOut(ctx, placeholders, "/agent/cluster/identity", h.LocalID)
changed := false
for i, res := range results {
if !res.OK || len(res.Data) == 0 {
continue
}
var identity models.HANode
if err := json.Unmarshal(res.Data, &identity); err != nil || identity.ID == "" {
continue
}
placeholder := placeholders[i]
if identity.ID == placeholder.ID {
continue // ID bereits korrekt
}
// Echte ID gefunden — row mit realer ID anlegen, public_ip aus
// dem Placeholder-Row übernehmen damit nftables korrekt bleibt.
n := identity
n.Status = "online"
if n.PublicIP == nil {
n.PublicIP = placeholder.PublicIP
}
if n.InternalIP == nil {
n.InternalIP = placeholder.InternalIP
}
out, err := h.Store.UpsertSelf(ctx, n)
if err != nil {
slog.Warn("cluster: reconcile joining peer: upsert failed",
"fqdn", n.FQDN, "real_id", n.ID, "error", err)
continue
}
_ = h.Store.DeletePlaceholdersByFQDN(ctx, n.FQDN, n.ID)
changed = true
slog.Info("cluster: joining peer reconciled via identity pull",
"id", out.ID, "fqdn", out.FQDN, "placeholder_id", placeholder.ID)
}
if changed && h.PeerReloader != nil {
rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := h.PeerReloader(rctx); err != nil {
slog.Warn("cluster: PeerReloader failed after reconcile", "error", err)
}
}
}
var errInvalidJoinRequest = simpleError("missing token or csr")
type simpleError string