Add SafeApply lifecycle, health tracking, convergence loop, and startup checks

SafeApply pattern (validate → backup → write → reload → verify → rollback):
- HAProxy: SafeApply wraps existing validate+write+reload with backup and
  post-reload health check; rolls back to .bak on failure
- dnsmasq: SafeApply validates via dnsmasq --test, backs up, atomic writes,
  restarts, verifies daemon is active; rolls back on failure
- nftables: SafeApply validates via nft -c, backs up, atomic writes, applies
  to kernel, verifies table loaded; rolls back on failure

Health tracking:
- Add SubsystemHealth and Health structs to reconciler
- Track per-subsystem status (ok/degraded/error) after each reconcile
- Detect recovery: previous error → current ok broadcasts recovery event
- GET /api/v1/health/reconcile endpoint exposes health state
- HAProxy tracks excluded domains as "degraded" state

SSE error events:
- Broadcast haproxy:error, dnsmasq:error on SafeApply failure
- Broadcast haproxy:recovered, dnsmasq:recovered on recovery from error

Convergence loop:
- 5-minute periodic reconcile drives system toward desired state
- Catches config drift, daemon crashes, transient failures
- Serialized by reconcile mutex — no race with event-driven reconciles

Startup:
- CheckPrerequisites verifies required (dnsmasq, haproxy) and optional
  (wg, authelia, cscli, cloudflared, nft) binaries before first reconcile
This commit is contained in:
2026-07-14 11:44:44 +00:00
parent 88acd437a1
commit a0ab8faa1c
7 changed files with 276 additions and 20 deletions

View File

@@ -215,9 +215,51 @@ func (api *API) StartCentralStatusBroadcaster(startTime time.Time) {
} }
// RegisterRoutes registers all Central API routes // RegisterRoutes registers all Central API routes
// ReconcileHealth returns the health state from the last reconciliation.
func (api *API) ReconcileHealth(w http.ResponseWriter, r *http.Request) {
respondJSON(w, http.StatusOK, api.reconciler.GetHealth())
}
// CheckPrerequisites verifies that required daemons are available on the system.
// Non-fatal — logs errors for required and info for optional missing binaries.
func (api *API) CheckPrerequisites() {
for _, bin := range []string{"dnsmasq", "haproxy"} {
if _, err := exec.LookPath(bin); err != nil {
slog.Error("required daemon not found", "component", "startup", "binary", bin)
}
}
for _, bin := range []string{"wg", "authelia", "cscli", "cloudflared", "nft"} {
if _, err := exec.LookPath(bin); err != nil {
slog.Info("optional daemon not found (some features unavailable)", "component", "startup", "binary", bin)
}
}
}
// StartConvergenceLoop starts a periodic reconciliation loop that continuously
// drives the system toward desired state. Catches config drift, daemon crashes,
// and transient failures.
func (api *API) StartConvergenceLoop(ctx gocontext.Context, interval time.Duration) {
go func() {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
api.reconciler.Reconcile()
}
}
}()
slog.Info("convergence loop started", "component", "reconcile", "interval", interval)
}
func (api *API) RegisterRoutes(r *mux.Router) { func (api *API) RegisterRoutes(r *mux.Router) {
r.Use(RequestLoggingMiddleware) r.Use(RequestLoggingMiddleware)
// Health
r.HandleFunc("/api/v1/health/reconcile", api.ReconcileHealth).Methods("GET")
// Resource-oriented settings (persisted in state.yaml) // Resource-oriented settings (persisted in state.yaml)
r.HandleFunc("/api/v1/operator", api.GetOperator).Methods("GET") r.HandleFunc("/api/v1/operator", api.GetOperator).Methods("GET")
r.HandleFunc("/api/v1/operator", api.UpdateOperator).Methods("PUT") r.HandleFunc("/api/v1/operator", api.UpdateOperator).Methods("PUT")

View File

@@ -300,6 +300,62 @@ log-dhcp
return sb.String() return sb.String()
} }
// SafeApply validates, backs up, writes, restarts, and verifies the dnsmasq config.
// On restart or verification failure, rolls back to the previous config.
func (g *Manager) SafeApply(content string) error {
// Validate via temp file
tmpFile := g.configPath + ".validate"
if err := os.WriteFile(tmpFile, []byte(content), 0644); err != nil {
return fmt.Errorf("writing validation file: %w", err)
}
defer os.Remove(tmpFile)
if err := g.ValidateConfig(tmpFile); err != nil {
return fmt.Errorf("validation failed: %w", err)
}
// Backup current config
if storage.FileExists(g.configPath) {
_ = storage.CopyFile(g.configPath, g.configPath+".bak")
}
// Write atomically
if err := storage.WriteFileAtomic(g.configPath, []byte(content), 0644); err != nil {
return fmt.Errorf("write failed: %w", err)
}
// Restart daemon
if err := g.RestartService(); err != nil {
g.rollback()
return fmt.Errorf("restart failed (rolled back): %w", err)
}
// Verify daemon is running
if err := g.verify(); err != nil {
g.rollback()
return fmt.Errorf("health check failed (rolled back): %w", err)
}
return nil
}
func (g *Manager) rollback() {
bakPath := g.configPath + ".bak"
if storage.FileExists(bakPath) {
_ = os.Rename(bakPath, g.configPath)
_ = g.RestartService()
slog.Warn("rolled back to previous config", "component", "dnsmasq", "path", g.configPath)
}
}
func (g *Manager) verify() error {
cmd := exec.Command("systemctl", "is-active", "--quiet", "dnsmasq.service")
if err := cmd.Run(); err != nil {
return fmt.Errorf("dnsmasq not active after restart")
}
return nil
}
// ValidateConfig tests a dnsmasq configuration file for syntax errors // ValidateConfig tests a dnsmasq configuration file for syntax errors
func (g *Manager) ValidateConfig(configPath string) error { func (g *Manager) ValidateConfig(configPath string) error {
cmd := exec.Command("dnsmasq", "--test", "-C", configPath) cmd := exec.Command("dnsmasq", "--test", "-C", configPath)

View File

@@ -13,6 +13,7 @@ import (
"time" "time"
"github.com/wild-cloud/wild-central/internal/domains" "github.com/wild-cloud/wild-central/internal/domains"
"github.com/wild-cloud/wild-central/internal/storage"
) )
// L4Route represents an L4 TCP passthrough route. // L4Route represents an L4 TCP passthrough route.
@@ -550,6 +551,51 @@ func (m *Manager) WriteConfig(content string) error {
return nil return nil
} }
// SafeApply validates, backs up, writes, reloads, and verifies the HAProxy config.
// On reload or verification failure, rolls back to the previous config.
func (m *Manager) SafeApply(content string) error {
// Backup current config
if storage.FileExists(m.configPath) {
_ = storage.CopyFile(m.configPath, m.configPath+".bak")
}
// Validate + atomic write
if err := m.WriteConfig(content); err != nil {
return fmt.Errorf("validation failed: %w", err)
}
// Reload daemon
if err := m.ReloadService(); err != nil {
m.rollback()
return fmt.Errorf("reload failed (rolled back): %w", err)
}
// Verify daemon is running
if err := m.verify(); err != nil {
m.rollback()
return fmt.Errorf("health check failed (rolled back): %w", err)
}
return nil
}
func (m *Manager) rollback() {
bakPath := m.configPath + ".bak"
if storage.FileExists(bakPath) {
_ = os.Rename(bakPath, m.configPath)
_ = m.ReloadService()
slog.Warn("rolled back to previous config", "component", "haproxy", "path", m.configPath)
}
}
func (m *Manager) verify() error {
cmd := exec.Command("systemctl", "is-active", "--quiet", "haproxy.service")
if err := cmd.Run(); err != nil {
return fmt.Errorf("haproxy not active after reload")
}
return nil
}
// ReloadService gracefully reloads HAProxy (starts it if not running) // ReloadService gracefully reloads HAProxy (starts it if not running)
func (m *Manager) ReloadService() error { func (m *Manager) ReloadService() error {
cmd := exec.Command("systemctl", "reload-or-restart", "haproxy.service") cmd := exec.Command("systemctl", "reload-or-restart", "haproxy.service")

View File

@@ -131,6 +131,62 @@ func (m *Manager) ValidateRules(rulesPath string) error {
return nil return nil
} }
// SafeApply validates, backs up, writes, and applies nftables rules.
// On apply failure, rolls back to the previous rules.
func (m *Manager) SafeApply(content string) error {
// Validate syntax
tmpFile := "/tmp/wild-cloud-nft-safeapply.tmp"
if err := os.WriteFile(tmpFile, []byte(content), 0644); err != nil {
return fmt.Errorf("writing validation file: %w", err)
}
defer os.Remove(tmpFile)
if err := m.ValidateRules(tmpFile); err != nil {
return fmt.Errorf("validation failed: %w", err)
}
// Backup current rules
if storage.FileExists(m.rulesPath) {
_ = storage.CopyFile(m.rulesPath, m.rulesPath+".bak")
}
// Write atomically
if err := storage.WriteFileAtomic(m.rulesPath, []byte(content), 0644); err != nil {
return fmt.Errorf("write failed: %w", err)
}
// Apply to kernel
if err := m.ApplyRules(); err != nil {
m.rollback()
return fmt.Errorf("apply failed (rolled back): %w", err)
}
// Verify rules loaded
if err := m.verify(); err != nil {
m.rollback()
return fmt.Errorf("verification failed (rolled back): %w", err)
}
return nil
}
func (m *Manager) rollback() {
bakPath := m.rulesPath + ".bak"
if storage.FileExists(bakPath) {
_ = os.Rename(bakPath, m.rulesPath)
_ = m.ApplyRules()
slog.Warn("rolled back to previous rules", "component", "nftables", "path", m.rulesPath)
}
}
func (m *Manager) verify() error {
status, err := m.GetStatus()
if err != nil || status == "" {
return fmt.Errorf("nftables table not loaded after apply")
}
return nil
}
// WriteRules validates the content then writes the nftables rules file. // WriteRules validates the content then writes the nftables rules file.
// Validation uses a temp file in /tmp (world-writable) to avoid needing // Validation uses a temp file in /tmp (world-writable) to avoid needing
// write permission on the /etc/nftables.d/ directory itself. // write permission on the /etc/nftables.d/ directory itself.

View File

@@ -26,13 +26,15 @@ import (
// HAProxyManager is the subset of haproxy.Manager used by reconciliation. // HAProxyManager is the subset of haproxy.Manager used by reconciliation.
type HAProxyManager interface { type HAProxyManager interface {
GenerateWithOpts(instances []haproxy.L4Route, custom []haproxy.CustomRoute, opts haproxy.GenerateOpts) string GenerateWithOpts(instances []haproxy.L4Route, custom []haproxy.CustomRoute, opts haproxy.GenerateOpts) string
SafeApply(content string) error
WriteConfig(content string) error WriteConfig(content string) error
ReloadService() error ReloadService() error
} }
// DNSManager is the subset of dnsmasq.Manager used by reconciliation. // DNSManager is the subset of dnsmasq.Manager used by reconciliation.
type DNSManager interface { type DNSManager interface {
UpdateConfig(cfg *config.State, entries []dnsmasq.DNSEntry, restart bool) error Generate(cfg *config.State, entries []dnsmasq.DNSEntry) string
SafeApply(content string) error
SetFilterConfPath(path string) SetFilterConfPath(path string)
GetStatus() (*dnsmasq.Status, error) GetStatus() (*dnsmasq.Status, error)
} }
@@ -68,9 +70,24 @@ type EventBroadcaster interface {
// that the reconciler doesn't own. // that the reconciler doesn't own.
type GenerateAutheliaConfigFn func(state *config.State) error type GenerateAutheliaConfigFn func(state *config.State) error
// SubsystemHealth records the outcome of the last reconciliation for one subsystem.
type SubsystemHealth struct {
Status string `json:"status"` // "ok", "degraded", "error"
Message string `json:"message,omitempty"`
}
// Health records the outcome of the last reconciliation across all subsystems.
type Health struct {
HAProxy SubsystemHealth `json:"haproxy"`
DNS SubsystemHealth `json:"dns"`
UpdatedAt time.Time `json:"updatedAt"`
ExcludedDomains []string `json:"excludedDomains,omitempty"`
}
// Reconciler orchestrates config regeneration when domains change. // Reconciler orchestrates config regeneration when domains change.
type Reconciler struct { type Reconciler struct {
mu sync.Mutex // serializes concurrent Reconcile() calls mu sync.Mutex // serializes concurrent Reconcile() calls
health Health
Domains DomainManager Domains DomainManager
HAProxy HAProxyManager HAProxy HAProxyManager
DNS DNSManager DNS DNSManager
@@ -82,6 +99,13 @@ type Reconciler struct {
GenerateAutheliaConfig GenerateAutheliaConfigFn GenerateAutheliaConfig GenerateAutheliaConfigFn
} }
// GetHealth returns a snapshot of the last reconciliation health state.
func (r *Reconciler) GetHealth() Health {
r.mu.Lock()
defer r.mu.Unlock()
return r.health
}
// Reconcile reads all registered domains and regenerates dnsmasq DNS entries // Reconcile reads all registered domains and regenerates dnsmasq DNS entries
// and HAProxy routes to match. Serialized by mutex — concurrent calls wait // and HAProxy routes to match. Serialized by mutex — concurrent calls wait
// rather than racing on config files. // rather than racing on config files.
@@ -119,6 +143,8 @@ func (r *Reconciler) Reconcile() {
} }
} }
previousHealth := r.health
r.writeHAProxyConfig(l4Routes, activeHTTPRoutes, globalCfg) r.writeHAProxyConfig(l4Routes, activeHTTPRoutes, globalCfg)
// Build and apply DNS config // Build and apply DNS config
@@ -141,9 +167,13 @@ func (r *Reconciler) Reconcile() {
r.DNS.SetFilterConfPath("") r.DNS.SetFilterConfPath("")
} }
if err := r.DNS.UpdateConfig(globalCfg, dnsEntries, true); err != nil { dnsConfig := r.DNS.Generate(globalCfg, dnsEntries)
if err := r.DNS.SafeApply(dnsConfig); err != nil {
slog.Error("failed to update dnsmasq", "component", "reconcile", "error", err) slog.Error("failed to update dnsmasq", "component", "reconcile", "error", err)
r.health.DNS = SubsystemHealth{Status: "error", Message: err.Error()}
r.broadcastEvent("dnsmasq:error", err.Error())
} else { } else {
r.health.DNS = SubsystemHealth{Status: "ok"}
r.broadcastDNSEvent("dnsmasq:config", "DNS config regenerated from registered domains") r.broadcastDNSEvent("dnsmasq:config", "DNS config regenerated from registered domains")
} }
@@ -153,10 +183,22 @@ func (r *Reconciler) Reconcile() {
r.DDNS.Trigger() r.DDNS.Trigger()
r.health.UpdatedAt = time.Now()
// Detect recoveries
if previousHealth.HAProxy.Status == "error" && r.health.HAProxy.Status == "ok" {
r.broadcastEvent("haproxy:recovered", "HAProxy recovered")
}
if previousHealth.DNS.Status == "error" && r.health.DNS.Status == "ok" {
r.broadcastEvent("dnsmasq:recovered", "DNS recovered")
}
slog.Info("networking updated", "component", "reconcile", slog.Info("networking updated", "component", "reconcile",
"domains", len(doms), "domains", len(doms),
"l4Routes", len(l4Routes), "l4Routes", len(l4Routes),
"l7Routes", len(httpRoutes), "l7Routes", len(httpRoutes),
"haproxy", r.health.HAProxy.Status,
"dns", r.health.DNS.Status,
) )
} }
@@ -229,8 +271,8 @@ func buildDNSEntries(doms []domains.Domain, centralIP string) []dnsmasq.DNSEntry
return entries return entries
} }
// writeHAProxyConfig generates, validates, and writes the HAProxy config. // writeHAProxyConfig generates and applies the HAProxy config via SafeApply.
// On validation failure, it identifies broken domains and retries without them. // On validation failure, identifies broken domains and retries without them.
func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes []haproxy.HTTPRoute, globalCfg *config.State) { func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes []haproxy.HTTPRoute, globalCfg *config.State) {
genOpts := haproxy.GenerateOpts{ genOpts := haproxy.GenerateOpts{
HTTPRoutes: httpRoutes, HTTPRoutes: httpRoutes,
@@ -254,7 +296,8 @@ func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes [
cfg := r.HAProxy.GenerateWithOpts(l4Routes, nil, genOpts) cfg := r.HAProxy.GenerateWithOpts(l4Routes, nil, genOpts)
if err := r.HAProxy.WriteConfig(cfg); err != nil { if err := r.HAProxy.SafeApply(cfg); err != nil {
// SafeApply handles rollback internally. Try to identify broken domains and retry.
brokenDomains := haproxy.FindBrokenServices(cfg, haproxy.ParseValidationErrors(err.Error())) brokenDomains := haproxy.FindBrokenServices(cfg, haproxy.ParseValidationErrors(err.Error()))
if len(brokenDomains) > 0 { if len(brokenDomains) > 0 {
slog.Error("excluding broken domains and retrying", slog.Error("excluding broken domains and retrying",
@@ -280,19 +323,23 @@ func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes [
genOpts.HTTPRoutes = filteredHTTP genOpts.HTTPRoutes = filteredHTTP
cfg = r.HAProxy.GenerateWithOpts(filteredL4, nil, genOpts) cfg = r.HAProxy.GenerateWithOpts(filteredL4, nil, genOpts)
if err := r.HAProxy.WriteConfig(cfg); err != nil { if err := r.HAProxy.SafeApply(cfg); err != nil {
slog.Error("retry also failed", "component", "reconcile", "error", err) slog.Error("retry also failed", "component", "reconcile", "error", err)
} else if err := r.HAProxy.ReloadService(); err != nil { r.health.HAProxy = SubsystemHealth{Status: "error", Message: err.Error()}
slog.Warn("failed to reload HAProxy", "component", "reconcile", "error", err) r.broadcastEvent("haproxy:error", err.Error())
} else { } else {
r.health.HAProxy = SubsystemHealth{Status: "degraded", Message: fmt.Sprintf("excluded domains: %v", brokenDomains)}
r.health.ExcludedDomains = brokenDomains
r.broadcastEvent("haproxy:config", "HAProxy config regenerated (excluded broken domains)") r.broadcastEvent("haproxy:config", "HAProxy config regenerated (excluded broken domains)")
} }
} else { } else {
slog.Error("failed to write HAProxy config (no broken domains identified)", "component", "reconcile", "error", err) slog.Error("failed to apply HAProxy config (no broken domains identified)", "component", "reconcile", "error", err)
r.health.HAProxy = SubsystemHealth{Status: "error", Message: err.Error()}
r.broadcastEvent("haproxy:error", err.Error())
} }
} else if err := r.HAProxy.ReloadService(); err != nil {
slog.Warn("failed to reload HAProxy", "component", "reconcile", "error", err)
} else { } else {
r.health.HAProxy = SubsystemHealth{Status: "ok"}
r.health.ExcludedDomains = nil
r.broadcastEvent("haproxy:config", "HAProxy config regenerated from registered domains") r.broadcastEvent("haproxy:config", "HAProxy config regenerated from registered domains")
} }
} }

View File

@@ -36,22 +36,26 @@ func (s *stubHAProxy) GenerateWithOpts(l4 []haproxy.L4Route, _ []haproxy.CustomR
s.lastL4Routes = l4 s.lastL4Routes = l4
return "generated-config" return "generated-config"
} }
func (s *stubHAProxy) WriteConfig(c string) error { s.writtenConfig = c; return s.writeErr } func (s *stubHAProxy) SafeApply(c string) error { s.writtenConfig = c; return s.writeErr }
func (s *stubHAProxy) WriteConfig(c string) error { s.writtenConfig = c; return s.writeErr }
func (s *stubHAProxy) ReloadService() error { s.reloadCalled = true; return nil } func (s *stubHAProxy) ReloadService() error { s.reloadCalled = true; return nil }
type stubDNS struct { type stubDNS struct {
entries []dnsmasq.DNSEntry entries []dnsmasq.DNSEntry
filterPath string filterPath string
updateCalled bool applyCalled bool
} }
func (s *stubDNS) UpdateConfig(_ *config.State, entries []dnsmasq.DNSEntry, _ bool) error { func (s *stubDNS) Generate(_ *config.State, entries []dnsmasq.DNSEntry) string {
s.entries = entries s.entries = entries
s.updateCalled = true return "generated-dns-config"
}
func (s *stubDNS) SafeApply(_ string) error {
s.applyCalled = true
return nil return nil
} }
func (s *stubDNS) SetFilterConfPath(p string) { s.filterPath = p } func (s *stubDNS) SetFilterConfPath(p string) { s.filterPath = p }
func (s *stubDNS) GetStatus() (*dnsmasq.Status, error) { return nil, nil } func (s *stubDNS) GetStatus() (*dnsmasq.Status, error) { return nil, nil }
type stubAuth struct { type stubAuth struct {
userCount int userCount int
@@ -188,7 +192,7 @@ func TestReconcile_EmptyDomains(t *testing.T) {
r, hp, dns, ddns := newTestReconciler(t, nil) r, hp, dns, ddns := newTestReconciler(t, nil)
r.Reconcile() r.Reconcile()
if !dns.updateCalled { if !dns.applyCalled {
t.Error("expected dnsmasq update") t.Error("expected dnsmasq update")
} }
if len(dns.entries) != 0 { if len(dns.entries) != 0 {

View File

@@ -160,9 +160,14 @@ func main() {
// Tell the API what port it's running on, register Central as a domain // Tell the API what port it's running on, register Central as a domain
// (if a domain is configured), and reconcile all networking. // (if a domain is configured), and reconcile all networking.
api.SetPort(port) api.SetPort(port)
api.CheckPrerequisites()
api.EnsureCentralDomain() api.EnsureCentralDomain()
api.Reconcile() api.Reconcile()
// Start periodic convergence loop — continuously drives system toward
// desired state, recovering from daemon crashes and config drift.
api.StartConvergenceLoop(ctx, 5*time.Minute)
addr := fmt.Sprintf("%s:%d", host, port) addr := fmt.Sprintf("%s:%d", host, port)
slog.Info("wild-central started", "addr", addr, "version", Version) slog.Info("wild-central started", "addr", addr, "version", Version)