Paul Payne
6aa1e7d438
When haproxy -c fails, parse ALERT line numbers, map to service domains via # service: comments in generated config, exclude broken services, and regenerate. One retry — prevents a single bad registration from blocking all config updates.
380 lines
11 KiB
Go
380 lines
11 KiB
Go
package v1
|
|
|
|
import (
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/wild-cloud/wild-central/internal/certbot"
|
|
"github.com/wild-cloud/wild-central/internal/config"
|
|
"github.com/wild-cloud/wild-central/internal/haproxy"
|
|
"github.com/wild-cloud/wild-central/internal/network"
|
|
"github.com/wild-cloud/wild-central/internal/services"
|
|
"github.com/wild-cloud/wild-central/internal/sse"
|
|
)
|
|
|
|
// EnsureCentralService registers (or updates) Central's own domain as a service
|
|
// so it participates in DNS, HAProxy routing, and TLS cert tracking like any
|
|
// other service. Called on startup and when global config changes.
|
|
func (api *API) EnsureCentralService() {
|
|
centralDomain := api.getCentralDomain()
|
|
port := api.getRunningPort()
|
|
backend := fmt.Sprintf("127.0.0.1:%d", port)
|
|
|
|
// Check if the current registration already matches — skip if so.
|
|
if centralDomain != "" {
|
|
if existing, _ := api.services.Get(centralDomain); existing != nil &&
|
|
existing.Source == "central" &&
|
|
existing.Backend.Address == backend {
|
|
return
|
|
}
|
|
}
|
|
|
|
// Clean up stale Central registrations (e.g. domain changed).
|
|
svcs, _ := api.services.List()
|
|
for _, svc := range svcs {
|
|
if svc.Source == "central" && svc.Domain != centralDomain {
|
|
_ = api.services.Deregister(svc.Domain)
|
|
}
|
|
}
|
|
|
|
if centralDomain == "" {
|
|
return
|
|
}
|
|
|
|
_ = api.services.Register(services.Service{
|
|
Domain: centralDomain,
|
|
Source: "central",
|
|
Backend: services.Backend{
|
|
Address: backend,
|
|
Type: services.BackendHTTP,
|
|
},
|
|
// Public defaults to false (private)
|
|
TLS: services.TLSTerminate,
|
|
})
|
|
}
|
|
|
|
// Reconcile runs networking reconciliation immediately. Called on startup
|
|
// and automatically whenever a service is registered/updated/deregistered.
|
|
func (api *API) Reconcile() {
|
|
api.reconcileNetworking()
|
|
}
|
|
|
|
// reconcileNetworking reads all registered services and regenerates
|
|
// dnsmasq DNS entries and HAProxy routes to match.
|
|
func (api *API) reconcileNetworking() {
|
|
svcs, err := api.services.List()
|
|
if err != nil {
|
|
slog.Error("reconcile: failed to list services", "error", err)
|
|
return
|
|
}
|
|
|
|
globalCfg, err := config.LoadState(api.statePath())
|
|
if err != nil {
|
|
slog.Warn("reconcile: failed to load state, using empty", "error", err)
|
|
globalCfg = &config.GlobalConfig{}
|
|
}
|
|
|
|
// Build HAProxy routes from registered services
|
|
var instanceRoutes []haproxy.InstanceRoute
|
|
var httpRoutes []haproxy.HTTPRoute
|
|
|
|
for _, svc := range svcs {
|
|
switch svc.EffectiveBackendType() {
|
|
case services.BackendTCPPassthrough:
|
|
instanceRoutes = append(instanceRoutes, haproxy.InstanceRoute{
|
|
Name: svc.Domain,
|
|
Domain: svc.Domain,
|
|
BackendIP: extractHost(svc.EffectiveBackendAddress()),
|
|
Subdomains: svc.Subdomains,
|
|
})
|
|
case services.BackendHTTP:
|
|
var routeBackends []haproxy.HTTPRouteBackend
|
|
for _, r := range svc.EffectiveRoutes() {
|
|
routeBackends = append(routeBackends, haproxy.HTTPRouteBackend{
|
|
Paths: r.Paths,
|
|
Backend: r.Backend.Address,
|
|
HealthPath: r.Backend.Health,
|
|
Headers: r.Headers,
|
|
IPAllow: r.IPAllow,
|
|
})
|
|
}
|
|
httpRoutes = append(httpRoutes, haproxy.HTTPRoute{
|
|
Name: svc.Domain,
|
|
Domain: svc.Domain,
|
|
Subdomains: svc.Subdomains,
|
|
Routes: routeBackends,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Only include L7 HTTP routes for services that have a cert.
|
|
certsDir := "/etc/haproxy/certs/"
|
|
var activeHTTPRoutes []haproxy.HTTPRoute
|
|
for _, route := range httpRoutes {
|
|
if hasCertForDomain(certsDir, route.Domain) {
|
|
activeHTTPRoutes = append(activeHTTPRoutes, route)
|
|
} else {
|
|
slog.Warn("reconcile: skipping L7 route (no cert)", "domain", route.Domain)
|
|
}
|
|
}
|
|
|
|
// Generate and write HAProxy config
|
|
genOpts := haproxy.GenerateOpts{
|
|
HTTPRoutes: activeHTTPRoutes,
|
|
CertsDir: certsDir,
|
|
}
|
|
haproxyCfg := api.haproxy.GenerateWithOpts(instanceRoutes, nil, genOpts)
|
|
|
|
if err := api.haproxy.WriteConfig(haproxyCfg); err != nil {
|
|
// Validation failed — identify and exclude broken services, then retry
|
|
brokenDomains := haproxy.FindBrokenServices(haproxyCfg, haproxy.ParseValidationErrors(err.Error()))
|
|
if len(brokenDomains) > 0 {
|
|
slog.Error("reconcile: excluding broken services and retrying",
|
|
"broken", brokenDomains, "error", err)
|
|
|
|
exclude := map[string]bool{}
|
|
for _, d := range brokenDomains {
|
|
exclude[d] = true
|
|
}
|
|
|
|
var filteredInstances []haproxy.InstanceRoute
|
|
for _, r := range instanceRoutes {
|
|
if !exclude[r.Domain] {
|
|
filteredInstances = append(filteredInstances, r)
|
|
}
|
|
}
|
|
var filteredHTTP []haproxy.HTTPRoute
|
|
for _, r := range activeHTTPRoutes {
|
|
if !exclude[r.Domain] {
|
|
filteredHTTP = append(filteredHTTP, r)
|
|
}
|
|
}
|
|
|
|
genOpts.HTTPRoutes = filteredHTTP
|
|
haproxyCfg = api.haproxy.GenerateWithOpts(filteredInstances, nil, genOpts)
|
|
if err := api.haproxy.WriteConfig(haproxyCfg); err != nil {
|
|
slog.Error("reconcile: retry also failed", "error", err)
|
|
} else {
|
|
if err := api.haproxy.ReloadService(); err != nil {
|
|
slog.Warn("reconcile: failed to reload HAProxy", "error", err)
|
|
} else {
|
|
api.broadcastHaproxyEvent("haproxy:config", "HAProxy config regenerated (excluded broken services)")
|
|
}
|
|
}
|
|
} else {
|
|
slog.Error("reconcile: failed to write HAProxy config (no broken services identified)", "error", err)
|
|
}
|
|
} else {
|
|
if err := api.haproxy.ReloadService(); err != nil {
|
|
slog.Warn("reconcile: failed to reload HAProxy", "error", err)
|
|
} else {
|
|
api.broadcastHaproxyEvent("haproxy:config", "HAProxy config regenerated from registered services")
|
|
}
|
|
}
|
|
|
|
// Generate dnsmasq DNS entries from registered services.
|
|
// DNS target IP depends on service type:
|
|
// tcp-passthrough (k8s): DNS → k8s LB IP (LAN traffic goes direct to k8s)
|
|
// http/static: DNS → Central IP (HAProxy terminates TLS)
|
|
// Use the configured dnsmasq IP (eth0) as Central's IP, not auto-detected
|
|
// (auto-detect can pick wlan0 if that's the default route).
|
|
centralIP := globalCfg.Cloud.Dnsmasq.IP
|
|
if centralIP == "" {
|
|
centralIP, _ = network.GetWildCentralIP()
|
|
}
|
|
if centralIP == "" {
|
|
centralIP = "127.0.0.1"
|
|
}
|
|
|
|
var instanceConfigs []config.InstanceConfig
|
|
for _, svc := range svcs {
|
|
if svc.Domain == "" {
|
|
continue
|
|
}
|
|
|
|
// Choose the DNS target IP based on service type
|
|
dnsIP := centralIP
|
|
if svc.EffectiveBackendType() == services.BackendTCPPassthrough {
|
|
// k8s instances: LAN clients connect directly to the k8s LB
|
|
dnsIP = extractHost(svc.EffectiveBackendAddress())
|
|
}
|
|
|
|
ic := config.InstanceConfig{}
|
|
ic.Cluster.LoadBalancerIp = dnsIP
|
|
|
|
if !svc.Public {
|
|
// Internal-only: local=/ prevents upstream DNS forwarding
|
|
ic.Cloud.InternalDomain = svc.Domain
|
|
} else {
|
|
ic.Cloud.Domain = svc.Domain
|
|
}
|
|
|
|
instanceConfigs = append(instanceConfigs, ic)
|
|
}
|
|
|
|
if err := api.dnsmasq.UpdateConfig(globalCfg, instanceConfigs, true); err != nil {
|
|
slog.Error("reconcile: failed to update dnsmasq", "error", err)
|
|
} else {
|
|
api.broadcastDnsmasqEvent("dnsmasq:config", "DNS config regenerated from registered services")
|
|
}
|
|
|
|
// Ensure TLS certs exist for HTTP services (where Central terminates TLS).
|
|
// For services under the gateway domain, a wildcard cert covers them all.
|
|
// For others, provision individual certs.
|
|
if len(httpRoutes) > 0 {
|
|
api.ensureTLSCerts(globalCfg, svcs)
|
|
}
|
|
|
|
slog.Info("reconcile: networking updated",
|
|
"services", len(svcs),
|
|
"l4Routes", len(instanceRoutes),
|
|
"l7Routes", len(httpRoutes),
|
|
)
|
|
}
|
|
|
|
// ensureTLSCerts logs which certificates are missing for registered services.
|
|
// Does NOT auto-provision — cert provisioning is user-initiated via the
|
|
// Certificates page. This just logs warnings so the operator knows what's needed.
|
|
func (api *API) ensureTLSCerts(globalCfg *config.GlobalConfig, svcs []services.Service) {
|
|
gatewayDomain := ""
|
|
if parts := strings.SplitN(globalCfg.Cloud.Central.Domain, ".", 2); len(parts) == 2 {
|
|
gatewayDomain = parts[1]
|
|
}
|
|
|
|
for _, svc := range svcs {
|
|
if svc.TLS != services.TLSTerminate || svc.Domain == "" {
|
|
continue
|
|
}
|
|
|
|
// Check if covered by wildcard
|
|
if gatewayDomain != "" && strings.HasSuffix(svc.Domain, "."+gatewayDomain) {
|
|
wildcardCert := certbot.HAProxyCertPath(gatewayDomain)
|
|
if _, err := os.Stat(wildcardCert); err != nil {
|
|
slog.Warn("reconcile/tls: no wildcard cert for gateway domain — provision via Certificates page",
|
|
"domain", svc.Domain, "needed", "*."+gatewayDomain)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Check individual cert
|
|
if _, err := os.Stat(certbot.HAProxyCertPath(svc.Domain)); err != nil {
|
|
slog.Warn("reconcile/tls: no cert for service — provision via Certificates page",
|
|
"domain", svc.Domain)
|
|
}
|
|
}
|
|
}
|
|
|
|
// hasCertForDomain checks if a cert exists for a domain — either an individual
|
|
// cert (<domain>.pem) or a wildcard cert that covers it.
|
|
func hasCertForDomain(certsDir, domain string) bool {
|
|
// Check individual cert
|
|
if _, err := os.Stat(certbot.HAProxyCertPath(domain)); err == nil {
|
|
return true
|
|
}
|
|
// Check wildcard certs — a *.example.com cert covers foo.example.com
|
|
parts := strings.SplitN(domain, ".", 2)
|
|
if len(parts) == 2 {
|
|
// Wildcard cert might be stored as the base domain PEM
|
|
wildcardBase := parts[1]
|
|
if _, err := os.Stat(certbot.HAProxyCertPath(wildcardBase)); err == nil {
|
|
return true
|
|
}
|
|
}
|
|
// Check if the certs directory has ANY .pem files (HAProxy needs at least one)
|
|
entries, err := os.ReadDir(certsDir)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
for _, e := range entries {
|
|
if strings.HasSuffix(e.Name(), ".pem") {
|
|
// There's at least one cert — HAProxy can start with the dir bind.
|
|
// The specific domain may not have its own cert but HAProxy won't crash.
|
|
// It will serve whichever cert matches best (or the first one as default).
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// extractHost gets the host part from a host:port string
|
|
func extractHost(addr string) string {
|
|
for i := len(addr) - 1; i >= 0; i-- {
|
|
if addr[i] == ':' {
|
|
return addr[:i]
|
|
}
|
|
}
|
|
return addr
|
|
}
|
|
|
|
// broadcastDnsmasqEvent broadcasts SSE events for dnsmasq status changes
|
|
func (api *API) broadcastDnsmasqEvent(eventType string, message string) {
|
|
if api.sseManager == nil {
|
|
return
|
|
}
|
|
|
|
// Get current dnsmasq status
|
|
status, err := api.dnsmasq.GetStatus()
|
|
if err != nil {
|
|
status = nil
|
|
}
|
|
|
|
event := &sse.Event{
|
|
ID: fmt.Sprintf("dnsmasq-%d", time.Now().UnixNano()),
|
|
Type: eventType,
|
|
InstanceName: "global", // dnsmasq is global/central-level, not instance-specific
|
|
Timestamp: time.Now(),
|
|
Data: map[string]any{
|
|
"message": message,
|
|
"status": status,
|
|
},
|
|
}
|
|
|
|
api.sseManager.Broadcast(event)
|
|
}
|
|
|
|
// broadcastHaproxyEvent broadcasts SSE events for HAProxy status changes
|
|
func (api *API) broadcastHaproxyEvent(eventType string, message string) {
|
|
if api.sseManager == nil {
|
|
return
|
|
}
|
|
|
|
event := &sse.Event{
|
|
ID: fmt.Sprintf("haproxy-%d", time.Now().UnixNano()),
|
|
Type: eventType,
|
|
InstanceName: "global",
|
|
Timestamp: time.Now(),
|
|
Data: map[string]any{
|
|
"message": message,
|
|
},
|
|
}
|
|
|
|
api.sseManager.Broadcast(event)
|
|
}
|
|
|
|
// broadcastCentralStatusEvent broadcasts SSE events for central status changes
|
|
func (api *API) broadcastCentralStatusEvent(startTime time.Time) {
|
|
if api.sseManager == nil {
|
|
return
|
|
}
|
|
|
|
uptime := time.Since(startTime)
|
|
|
|
event := &sse.Event{
|
|
ID: fmt.Sprintf("central-%d", time.Now().UnixNano()),
|
|
Type: "central:status",
|
|
InstanceName: "global",
|
|
Timestamp: time.Now(),
|
|
Data: map[string]any{
|
|
"status": "running",
|
|
"version": api.version,
|
|
"uptime": uptime.String(),
|
|
"uptimeSeconds": int(uptime.Seconds()),
|
|
},
|
|
}
|
|
|
|
api.sseManager.Broadcast(event)
|
|
}
|