From 053b38e46c0e804f6afd4ae5b8a81a5c4910cc62 Mon Sep 17 00:00:00 2001 From: Debian Date: Sat, 6 Jun 2026 11:21:50 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20AlertWriter=20graceful=20flush=20(#15)?= =?UTF-8?q?=20+=20Rolling-Update=20Robustheit=20(#19)=20=E2=80=94=20v1.2.9?= =?UTF-8?q?5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #15 waf/alerts.go: AlertWriter.Close() flusht gepufferte Alerts + stoppt die Goroutine (stop/done-Channels, sync.Once, atomic closed; Kanal wird NIE geschlossen → Send racet ohne Panic). Wiring in cmd/edgeguard-waf nach ListenAndServe (graceful shutdown). -race-Test alerts_test.go. #19 handlers/cluster_rollingupdate.go: (a) RollingUpdateStatus mutiert State nicht mehr beim GET — terminale Zustände altern in readRollingUpdateState nach 10 min aus (kein verlorenes 'done' bei parallelen Pollern). (b) State-File via sync.Mutex + configgen.AtomicWrite (kein partieller Read / Race zwischen Handler & Goroutine). (c) Version-Flip wird gegen die VORHER erfasste Secondary-Baseline geprüft statt gegen die Primary-Version (verhindert sofort-/nie-Flip). Bewusst belassen: geteilter upgrade.sh-Pfad ist deterministischer Inhalt + an exakte sudoers-Zeile gebunden → Überschreib-Race benign; MST-Timestamp-Parse locale (Server laufen C-Locale). Co-Authored-By: Claude Opus 4.8 (1M context) --- VERSION | 2 +- cmd/edgeguard-waf/main.go | 2 + internal/handlers/cluster_rollingupdate.go | 57 ++++++++++++++++---- internal/waf/alerts.go | 48 ++++++++++++++--- internal/waf/alerts_test.go | 61 ++++++++++++++++++++++ 5 files changed, 153 insertions(+), 17 deletions(-) create mode 100644 internal/waf/alerts_test.go diff --git a/VERSION b/VERSION index 60a3294..dbbd6b1 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.2.94 \ No newline at end of file +1.2.95 \ No newline at end of file diff --git a/cmd/edgeguard-waf/main.go b/cmd/edgeguard-waf/main.go index d917a86..4f7f121 100644 --- a/cmd/edgeguard-waf/main.go +++ b/cmd/edgeguard-waf/main.go @@ -83,6 +83,8 @@ func main() { slog.Error("waf: SPOE agent stopped", "error", err) os.Exit(1) } + // Graceful shutdown (ctx cancelled): gepufferte Alerts flushen. + alertWriter.Close() } // reload fetches all domain+waf_config pairs from DB and rebuilds engines. diff --git a/internal/handlers/cluster_rollingupdate.go b/internal/handlers/cluster_rollingupdate.go index 64f63af..c86fc05 100644 --- a/internal/handlers/cluster_rollingupdate.go +++ b/internal/handlers/cluster_rollingupdate.go @@ -7,14 +7,20 @@ import ( "net/http" "os" "os/exec" + "sync" "time" "github.com/gin-gonic/gin" + "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" ) +// ruStateMu serialisiert Lesen/Schreiben der Rolling-Update-State-Datei +// (HTTP-Handler + Hintergrund-Goroutine greifen gleichzeitig zu). +var ruStateMu sync.Mutex + const rollingUpdateStateFile = "/var/lib/edgeguard/rolling-update-state.json" const ( @@ -53,6 +59,8 @@ type RollingUpdateState struct { } func readRollingUpdateState() RollingUpdateState { + ruStateMu.Lock() + defer ruStateMu.Unlock() data, err := os.ReadFile(rollingUpdateStateFile) if err != nil { return RollingUpdateState{Phase: phaseIdle, UpdatedAt: time.Now()} @@ -61,6 +69,13 @@ func readRollingUpdateState() RollingUpdateState { if err := json.Unmarshal(data, &s); err != nil { return RollingUpdateState{Phase: phaseIdle, UpdatedAt: time.Now()} } + // Terminale Zustände altern aus (statt Mutation-on-GET): nach 10 min + // gilt done/failed als idle — so verliert kein paralleler Poller das + // Ergebnis und ein alter Stand bleibt nicht hängen. + if (s.Phase == phaseDone || s.Phase == phaseFailed) && !s.UpdatedAt.IsZero() && + time.Since(s.UpdatedAt) > 10*time.Minute { + return RollingUpdateState{Phase: phaseIdle, UpdatedAt: time.Now()} + } return s } @@ -71,7 +86,10 @@ func writeRollingUpdateState(s RollingUpdateState) { slog.Warn("rolling-update: failed to marshal state", "error", err) return } - if err := os.WriteFile(rollingUpdateStateFile, data, 0o600); err != nil { + ruStateMu.Lock() + defer ruStateMu.Unlock() + // AtomicWrite (temp+rename) → Leser sehen nie einen partiellen Stand. + if err := configgen.AtomicWrite(rollingUpdateStateFile, data, 0o600); err != nil { slog.Warn("rolling-update: failed to write state file", "error", err) } } @@ -127,19 +145,24 @@ func (h *ClusterHandler) RollingUpdate(c *gin.Context) { } // RollingUpdateStatus gibt den aktuellen Rolling-Update-State zurück. -// Bei phase == "done" wird nach Auslieferung sofort auf idle zurückgesetzt -// damit der nächste Pageload keinen Stale-done vorfindet. +// Read-only — terminale Zustände altern in readRollingUpdateState aus +// (kein Reset-on-GET mehr, das parallelen Pollern das "done" wegnahm). func (h *ClusterHandler) RollingUpdateStatus(c *gin.Context) { - st := readRollingUpdateState() - response.OK(c, st) - if st.Phase == phaseDone { - writeRollingUpdateState(RollingUpdateState{Phase: phaseIdle}) - } + response.OK(c, readRollingUpdateState()) } 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). + 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") @@ -175,8 +198,8 @@ func (h *ClusterHandler) runRollingUpdate(secondary *models.HANode) { Version string `json:"version"` } if err := json.Unmarshal(results[0].Data, &ver); err == nil { - slog.Info("rolling-update: secondary version", "version", ver.Version, "primary", h.Version) - if ver.Version != h.Version { + slog.Info("rolling-update: secondary version", "version", ver.Version, "baseline", target) + if ver.Version != "" && ver.Version != target { versionFlipped = true break } @@ -256,3 +279,17 @@ rm -f /var/lib/edgeguard/upgrade.sh // UI erkennt Version-Flip via /system/health und schließt den Flow. slog.Info("rolling-update: primary upgrade dispatched, process will restart") } + +// 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) + if len(results) > 0 && results[0].OK { + var ver struct { + Version string `json:"version"` + } + if json.Unmarshal(results[0].Data, &ver) == nil { + return ver.Version + } + } + return "" +} diff --git a/internal/waf/alerts.go b/internal/waf/alerts.go index 227bf01..9a90da9 100644 --- a/internal/waf/alerts.go +++ b/internal/waf/alerts.go @@ -3,6 +3,8 @@ package waf import ( "context" "log/slog" + "sync" + "sync/atomic" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -26,8 +28,12 @@ type Alert struct { // AlertWriter accepts Alert values via a buffered channel and writes // them to PostgreSQL asynchronously so SPOE handling stays low-latency. type AlertWriter struct { - pool *pgxpool.Pool - ch chan Alert + pool *pgxpool.Pool + ch chan Alert + stop chan struct{} + done chan struct{} + closeOnce sync.Once + closed atomic.Bool } // NewAlertWriter creates an AlertWriter and starts its background goroutine. @@ -36,14 +42,19 @@ func NewAlertWriter(pool *pgxpool.Pool, bufSize int) *AlertWriter { aw := &AlertWriter{ pool: pool, ch: make(chan Alert, bufSize), + stop: make(chan struct{}), + done: make(chan struct{}), } go aw.run() return aw } -// Send enqueues an alert. Drops silently if the channel is full to -// avoid slowing down SPOE request handling. +// Send enqueues an alert. Drops silently if the channel is full (or the +// writer is closing) to avoid slowing down / panicking SPOE handling. func (aw *AlertWriter) Send(a Alert) { + if aw.closed.Load() { + return + } select { case aw.ch <- a: default: @@ -51,9 +62,34 @@ func (aw *AlertWriter) Send(a Alert) { } } +// Close stops the writer and flushes buffered alerts (best-effort). +// Safe to call multiple times. The channel is never closed → Send never +// panics even if it races with Close. +func (aw *AlertWriter) Close() { + aw.closeOnce.Do(func() { + aw.closed.Store(true) + close(aw.stop) + }) + <-aw.done +} + func (aw *AlertWriter) run() { - for a := range aw.ch { - aw.write(a) + defer close(aw.done) + for { + select { + case a := <-aw.ch: + aw.write(a) + case <-aw.stop: + // Restliche gepufferte Alerts noch wegschreiben, dann Ende. + for { + select { + case a := <-aw.ch: + aw.write(a) + default: + return + } + } + } } } diff --git a/internal/waf/alerts_test.go b/internal/waf/alerts_test.go new file mode 100644 index 0000000..267b8a8 --- /dev/null +++ b/internal/waf/alerts_test.go @@ -0,0 +1,61 @@ +package waf + +import ( + "context" + "os" + "sync" + "testing" + "time" + + "git.netcell-it.de/projekte/edgeguard-native/internal/database" +) + +// Beweist Fix #15: AlertWriter.Close() flusht, ist idempotent, und Send/Close +// racen ohne Panic (Kanal wird nie geschlossen). Guarded per EG_FWTEST_DSN. +func TestAlertWriter_CloseFlush(t *testing.T) { + dsn := os.Getenv("EG_FWTEST_DSN") + if dsn == "" { + t.Skip("set EG_FWTEST_DSN to run the alert-writer test") + } + ctx := context.Background() + var mErr error + for i := 0; i < 3; i++ { + if mErr = database.Migrate(ctx, dsn); mErr == nil { + break + } + time.Sleep(700 * time.Millisecond) + } + if mErr != nil { + t.Fatalf("migrate: %v", mErr) + } + pool, err := database.Open(ctx, dsn) + if err != nil { + t.Fatalf("open: %v", err) + } + defer pool.Close() + + aw := NewAlertWriter(pool, 64) + for i := 0; i < 20; i++ { + aw.Send(Alert{Hostname: "t.local", ClientIP: "203.0.113.1", Method: "GET", URI: "/", Action: "detected"}) + } + + // Send parallel zu Close → darf nicht paniken. + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { defer wg.Done(); aw.Send(Alert{Hostname: "t.local", Action: "detected"}) }() + } + + done := make(chan struct{}) + go func() { aw.Close(); close(done) }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Close() did not return (flush hung)") + } + wg.Wait() + + // Idempotent + Send nach Close ist No-op (kein Panic). + aw.Close() + aw.Send(Alert{Hostname: "after.local", Action: "detected"}) +}