diff --git a/internal/api/v1/handlers.go b/internal/api/v1/handlers.go index 5bd5e94..de4ddb2 100644 --- a/internal/api/v1/handlers.go +++ b/internal/api/v1/handlers.go @@ -215,9 +215,51 @@ func (api *API) StartCentralStatusBroadcaster(startTime time.Time) { } // 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) { r.Use(RequestLoggingMiddleware) + // Health + r.HandleFunc("/api/v1/health/reconcile", api.ReconcileHealth).Methods("GET") + // Resource-oriented settings (persisted in state.yaml) r.HandleFunc("/api/v1/operator", api.GetOperator).Methods("GET") r.HandleFunc("/api/v1/operator", api.UpdateOperator).Methods("PUT") diff --git a/internal/dnsmasq/config.go b/internal/dnsmasq/config.go index 8252e54..660fc78 100644 --- a/internal/dnsmasq/config.go +++ b/internal/dnsmasq/config.go @@ -300,6 +300,62 @@ log-dhcp 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 func (g *Manager) ValidateConfig(configPath string) error { cmd := exec.Command("dnsmasq", "--test", "-C", configPath) diff --git a/internal/haproxy/config.go b/internal/haproxy/config.go index 54e2eea..e56e00c 100644 --- a/internal/haproxy/config.go +++ b/internal/haproxy/config.go @@ -13,6 +13,7 @@ import ( "time" "github.com/wild-cloud/wild-central/internal/domains" + "github.com/wild-cloud/wild-central/internal/storage" ) // L4Route represents an L4 TCP passthrough route. @@ -550,6 +551,51 @@ func (m *Manager) WriteConfig(content string) error { 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) func (m *Manager) ReloadService() error { cmd := exec.Command("systemctl", "reload-or-restart", "haproxy.service") diff --git a/internal/nftables/manager.go b/internal/nftables/manager.go index 6bbb6bd..0f1cfeb 100644 --- a/internal/nftables/manager.go +++ b/internal/nftables/manager.go @@ -131,6 +131,62 @@ func (m *Manager) ValidateRules(rulesPath string) error { 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. // Validation uses a temp file in /tmp (world-writable) to avoid needing // write permission on the /etc/nftables.d/ directory itself. diff --git a/internal/reconcile/reconciler.go b/internal/reconcile/reconciler.go index 696f750..1cad5d3 100644 --- a/internal/reconcile/reconciler.go +++ b/internal/reconcile/reconciler.go @@ -26,13 +26,15 @@ import ( // HAProxyManager is the subset of haproxy.Manager used by reconciliation. type HAProxyManager interface { GenerateWithOpts(instances []haproxy.L4Route, custom []haproxy.CustomRoute, opts haproxy.GenerateOpts) string + SafeApply(content string) error WriteConfig(content string) error ReloadService() error } // DNSManager is the subset of dnsmasq.Manager used by reconciliation. 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) GetStatus() (*dnsmasq.Status, error) } @@ -68,9 +70,24 @@ type EventBroadcaster interface { // that the reconciler doesn't own. 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. type Reconciler struct { mu sync.Mutex // serializes concurrent Reconcile() calls + health Health Domains DomainManager HAProxy HAProxyManager DNS DNSManager @@ -82,6 +99,13 @@ type Reconciler struct { 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 // and HAProxy routes to match. Serialized by mutex — concurrent calls wait // rather than racing on config files. @@ -119,6 +143,8 @@ func (r *Reconciler) Reconcile() { } } + previousHealth := r.health + r.writeHAProxyConfig(l4Routes, activeHTTPRoutes, globalCfg) // Build and apply DNS config @@ -141,9 +167,13 @@ func (r *Reconciler) Reconcile() { 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) + r.health.DNS = SubsystemHealth{Status: "error", Message: err.Error()} + r.broadcastEvent("dnsmasq:error", err.Error()) } else { + r.health.DNS = SubsystemHealth{Status: "ok"} r.broadcastDNSEvent("dnsmasq:config", "DNS config regenerated from registered domains") } @@ -153,10 +183,22 @@ func (r *Reconciler) Reconcile() { 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", "domains", len(doms), "l4Routes", len(l4Routes), "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 } -// writeHAProxyConfig generates, validates, and writes the HAProxy config. -// On validation failure, it identifies broken domains and retries without them. +// writeHAProxyConfig generates and applies the HAProxy config via SafeApply. +// On validation failure, identifies broken domains and retries without them. func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes []haproxy.HTTPRoute, globalCfg *config.State) { genOpts := haproxy.GenerateOpts{ HTTPRoutes: httpRoutes, @@ -254,7 +296,8 @@ func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes [ 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())) if len(brokenDomains) > 0 { slog.Error("excluding broken domains and retrying", @@ -280,19 +323,23 @@ func (r *Reconciler) writeHAProxyConfig(l4Routes []haproxy.L4Route, httpRoutes [ genOpts.HTTPRoutes = filteredHTTP 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) - } else if err := r.HAProxy.ReloadService(); err != nil { - slog.Warn("failed to reload HAProxy", "component", "reconcile", "error", err) + r.health.HAProxy = SubsystemHealth{Status: "error", Message: err.Error()} + r.broadcastEvent("haproxy:error", err.Error()) } 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)") } } 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 { + r.health.HAProxy = SubsystemHealth{Status: "ok"} + r.health.ExcludedDomains = nil r.broadcastEvent("haproxy:config", "HAProxy config regenerated from registered domains") } } diff --git a/internal/reconcile/reconciler_test.go b/internal/reconcile/reconciler_test.go index bbb05be..d194fbe 100644 --- a/internal/reconcile/reconciler_test.go +++ b/internal/reconcile/reconciler_test.go @@ -36,22 +36,26 @@ func (s *stubHAProxy) GenerateWithOpts(l4 []haproxy.L4Route, _ []haproxy.CustomR s.lastL4Routes = l4 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 } type stubDNS struct { - entries []dnsmasq.DNSEntry - filterPath string - updateCalled bool + entries []dnsmasq.DNSEntry + filterPath string + 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.updateCalled = true + return "generated-dns-config" +} +func (s *stubDNS) SafeApply(_ string) error { + s.applyCalled = true return nil } -func (s *stubDNS) SetFilterConfPath(p string) { s.filterPath = p } -func (s *stubDNS) GetStatus() (*dnsmasq.Status, error) { return nil, nil } +func (s *stubDNS) SetFilterConfPath(p string) { s.filterPath = p } +func (s *stubDNS) GetStatus() (*dnsmasq.Status, error) { return nil, nil } type stubAuth struct { userCount int @@ -188,7 +192,7 @@ func TestReconcile_EmptyDomains(t *testing.T) { r, hp, dns, ddns := newTestReconciler(t, nil) r.Reconcile() - if !dns.updateCalled { + if !dns.applyCalled { t.Error("expected dnsmasq update") } if len(dns.entries) != 0 { diff --git a/main.go b/main.go index b8ca161..358ce89 100644 --- a/main.go +++ b/main.go @@ -160,9 +160,14 @@ func main() { // Tell the API what port it's running on, register Central as a domain // (if a domain is configured), and reconcile all networking. api.SetPort(port) + api.CheckPrerequisites() api.EnsureCentralDomain() 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) slog.Info("wild-central started", "addr", addr, "version", Version)