hub v0.41.0: OffsiteChecker (fill 90/95 + staleness >48h) + operator freeze lever (SLICE 4)
Sibling checker over the controller report's offsite object: quota-fill warn/crit + the silently-stuck staleness detector (escrowed-only, red-proofed; nil-safe on pre-v0.109 reports; same-second tie-guard). SetOffsiteFrozen flips ONLY readonly on the exactly-1 labelled sub-account (SSH preserved); Freeze/Unfreeze buttons — manual only, never automatic. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
This commit is contained in:
@@ -0,0 +1,244 @@
|
||||
package monitor
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-hub/internal/store"
|
||||
)
|
||||
|
||||
// OffsiteChecker (SLICE 4) watches each customer's offsite backup health from the controller report's
|
||||
// `offsite` status object. Two independent signals, one checker (sibling of StorageFillChecker — same
|
||||
// born/persistent, escalation-only emit, recovery re-arm shape; NOT bolted onto the disk checkers —
|
||||
// different data source, different remedy text):
|
||||
//
|
||||
// - FILL: repo_size_bytes vs the shared-model soft quota (quota_gb>0) at warn 90% / crit 95% — the
|
||||
// operator's early warning before the controller's own 100% run-refusal bites the customer.
|
||||
// - STALENESS: enabled + escrowed but no run in >48h (or never) — the silently-STUCK detector. A
|
||||
// RECENTLY-failing offsite is NOT stale (backup_failed already alerts it); staleness is the
|
||||
// complement: nothing is even trying. Pending/disabled targets are normal onboarding, never stale.
|
||||
//
|
||||
// Reports without an `offsite` object (pre-v0.109 controllers, offbox not enabled) are skipped nil-safe.
|
||||
type OffsiteChecker struct {
|
||||
store *store.Store
|
||||
logger *log.Logger
|
||||
onEvent EventNotifyFunc
|
||||
staleAfter time.Duration
|
||||
now func() time.Time // injectable clock (tests)
|
||||
|
||||
mu sync.Mutex
|
||||
fillStates map[string]string // customerID → fill band
|
||||
staleStates map[string]string // customerID → "ok" | "stale"
|
||||
}
|
||||
|
||||
const defaultOffsiteStaleAfter = 48 * time.Hour
|
||||
|
||||
// offsiteReport mirrors the controller report's `offsite` object (v0.109.0).
|
||||
type offsiteReport struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
EscrowState string `json:"escrow_state"`
|
||||
LastRun string `json:"last_run"`
|
||||
LastStatus string `json:"last_status"`
|
||||
SnapshotCount int `json:"snapshot_count"`
|
||||
RepoSizeBytes int64 `json:"repo_size_bytes"`
|
||||
QuotaGB int `json:"quota_gb"`
|
||||
}
|
||||
|
||||
// NewOffsiteChecker builds the checker. Same seeding philosophy as StorageFillChecker: already-breached
|
||||
// customers are left UNSEEDED so their first Check emits (born/persistent); the dispatcher's cooldown
|
||||
// dedups a hub restart.
|
||||
func NewOffsiteChecker(s *store.Store, staleAfter time.Duration, onEvent EventNotifyFunc, logger *log.Logger) *OffsiteChecker {
|
||||
if staleAfter <= 0 {
|
||||
staleAfter = defaultOffsiteStaleAfter
|
||||
}
|
||||
oc := &OffsiteChecker{
|
||||
store: s, logger: logger, onEvent: onEvent, staleAfter: staleAfter, now: time.Now,
|
||||
fillStates: make(map[string]string), staleStates: make(map[string]string),
|
||||
}
|
||||
customers, err := s.GetCustomers()
|
||||
if err != nil {
|
||||
logger.Printf("[WARN] Offsite checker: failed to seed states: %v", err)
|
||||
return oc
|
||||
}
|
||||
var seeded int
|
||||
for _, c := range customers {
|
||||
off := parseOffsite(c.ReportJSON)
|
||||
if off == nil || s.IsCustomerBlocked(c.CustomerID) {
|
||||
continue
|
||||
}
|
||||
if band := oc.fillBand(off); band == bandOK {
|
||||
oc.fillStates[c.CustomerID] = bandOK
|
||||
seeded++
|
||||
}
|
||||
if !oc.isStale(off) {
|
||||
oc.staleStates[c.CustomerID] = "ok"
|
||||
}
|
||||
}
|
||||
logger.Printf("[INFO] Offsite checker initialized: fill warn=90%% crit=95%%, stale after %s, %d ok-seeded", staleAfter, seeded)
|
||||
return oc
|
||||
}
|
||||
|
||||
func parseOffsite(reportJSON string) *offsiteReport {
|
||||
var r struct {
|
||||
Offsite *offsiteReport `json:"offsite"`
|
||||
}
|
||||
if json.Unmarshal([]byte(reportJSON), &r) != nil {
|
||||
return nil
|
||||
}
|
||||
return r.Offsite // nil when absent (old controller / offbox not enabled) — the caller skips
|
||||
}
|
||||
|
||||
// fillBand maps the quota usage to a band. quota<=0 (dedicated/unset) never alerts.
|
||||
func (oc *OffsiteChecker) fillBand(off *offsiteReport) string {
|
||||
if off.QuotaGB <= 0 || off.RepoSizeBytes <= 0 {
|
||||
return bandOK
|
||||
}
|
||||
pct := float64(off.RepoSizeBytes) * 100 / float64(int64(off.QuotaGB)<<30)
|
||||
return bandForPercent(pct, 90, 95)
|
||||
}
|
||||
|
||||
// isStale: enabled + ESCROWED (the only state where runs are expected) with no run in >staleAfter (or
|
||||
// never). Pending/disabled = normal onboarding, never stale. A recent-but-failing run is NOT stale
|
||||
// (backup_failed owns that signal).
|
||||
func (oc *OffsiteChecker) isStale(off *offsiteReport) bool {
|
||||
if !off.Enabled || off.EscrowState != "escrowed" {
|
||||
return false
|
||||
}
|
||||
if off.LastRun == "" {
|
||||
return true
|
||||
}
|
||||
t, err := time.Parse(time.RFC3339, off.LastRun)
|
||||
if err != nil {
|
||||
return true // unparseable = unknown-old — fail toward visibility
|
||||
}
|
||||
return oc.now().Sub(t) > oc.staleAfter
|
||||
}
|
||||
|
||||
// Check evaluates every customer's latest report. Escalation-only emits; recovery re-arms silently.
|
||||
func (oc *OffsiteChecker) Check() {
|
||||
customers, err := oc.store.GetCustomers()
|
||||
if err != nil {
|
||||
oc.logger.Printf("[WARN] Offsite check failed: %v", err)
|
||||
return
|
||||
}
|
||||
oc.mu.Lock()
|
||||
defer oc.mu.Unlock()
|
||||
|
||||
seen := make(map[string]bool, len(customers))
|
||||
for _, c := range customers {
|
||||
// GetCustomers can return the same customer twice when two reports tie on received_at
|
||||
// (second-resolution timestamps) — process each customer once per sweep.
|
||||
if seen[c.CustomerID] {
|
||||
continue
|
||||
}
|
||||
seen[c.CustomerID] = true
|
||||
off := parseOffsite(c.ReportJSON)
|
||||
if off == nil {
|
||||
delete(oc.fillStates, c.CustomerID) // vanished object (disabled / downgraded) → re-arm
|
||||
delete(oc.staleStates, c.CustomerID)
|
||||
continue
|
||||
}
|
||||
if oc.store.IsCustomerBlocked(c.CustomerID) {
|
||||
delete(oc.fillStates, c.CustomerID)
|
||||
delete(oc.staleStates, c.CustomerID)
|
||||
continue
|
||||
}
|
||||
|
||||
// FILL (quota>0 only)
|
||||
newBand := oc.fillBand(off)
|
||||
if bandRank(newBand) > bandRank(oc.fillStates[c.CustomerID]) {
|
||||
oc.emitFill(c.CustomerID, off, newBand)
|
||||
}
|
||||
oc.fillStates[c.CustomerID] = newBand
|
||||
|
||||
// STALENESS (binary, warn-severity)
|
||||
newStale := "ok"
|
||||
if oc.isStale(off) {
|
||||
newStale = "stale"
|
||||
}
|
||||
if newStale == "stale" && oc.staleStates[c.CustomerID] != "stale" {
|
||||
oc.emitStale(c.CustomerID, off)
|
||||
}
|
||||
oc.staleStates[c.CustomerID] = newStale
|
||||
}
|
||||
for k := range oc.fillStates {
|
||||
if !seen[k] {
|
||||
delete(oc.fillStates, k)
|
||||
}
|
||||
}
|
||||
for k := range oc.staleStates {
|
||||
if !seen[k] {
|
||||
delete(oc.staleStates, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GetFillState / GetStaleState expose current states for tests.
|
||||
func (oc *OffsiteChecker) GetFillState(customerID string) string {
|
||||
oc.mu.Lock()
|
||||
defer oc.mu.Unlock()
|
||||
if s := oc.fillStates[customerID]; s != "" {
|
||||
return s
|
||||
}
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
func (oc *OffsiteChecker) GetStaleState(customerID string) string {
|
||||
oc.mu.Lock()
|
||||
defer oc.mu.Unlock()
|
||||
if s := oc.staleStates[customerID]; s != "" {
|
||||
return s
|
||||
}
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
func (oc *OffsiteChecker) emitFill(customerID string, off *offsiteReport, band string) {
|
||||
usedGB := off.RepoSizeBytes >> 30
|
||||
pct := float64(off.RepoSizeBytes) * 100 / float64(int64(off.QuotaGB)<<30)
|
||||
var eventType, severity, message string
|
||||
switch band {
|
||||
case bandCritical:
|
||||
eventType, severity = "offsite_fill_critical", "critical"
|
||||
message = fmt.Sprintf("Customer %s: offsite backup at %.0f%% of its %d GB quota (%d GB used) — at 100%% new offsite runs are refused; consider the freeze lever or a bigger quota", customerID, pct, off.QuotaGB, usedGB)
|
||||
case bandWarning:
|
||||
eventType, severity = "offsite_fill_warning", "warning"
|
||||
message = fmt.Sprintf("Customer %s: offsite backup at %.0f%% of its %d GB quota (%d GB used)", customerID, pct, off.QuotaGB, usedGB)
|
||||
default:
|
||||
return
|
||||
}
|
||||
details, _ := json.Marshal(map[string]any{
|
||||
"customer_id": customerID, "quota_gb": off.QuotaGB, "repo_size_bytes": off.RepoSizeBytes, "percent": pct,
|
||||
})
|
||||
oc.logger.Printf("[INFO] Offsite fill: %s %.0f%% (%s)", customerID, pct, eventType)
|
||||
if _, err := oc.store.SaveEvent(customerID, eventType, severity, message, string(details), "hub"); err != nil {
|
||||
oc.logger.Printf("[WARN] Failed to save offsite fill event for %s: %v", customerID, err)
|
||||
return
|
||||
}
|
||||
if oc.onEvent != nil {
|
||||
oc.onEvent(customerID, eventType, severity, message, string(details), "hub")
|
||||
}
|
||||
}
|
||||
|
||||
func (oc *OffsiteChecker) emitStale(customerID string, off *offsiteReport) {
|
||||
age := "never ran"
|
||||
if off.LastRun != "" {
|
||||
if t, err := time.Parse(time.RFC3339, off.LastRun); err == nil {
|
||||
age = fmt.Sprintf("last run %s ago", oc.now().Sub(t).Round(time.Hour))
|
||||
}
|
||||
}
|
||||
message := fmt.Sprintf("Customer %s: offsite backup is STALE — enabled + escrowed but %s (threshold %s). The offsite leg is silently not running; check the controller/schedule", customerID, age, oc.staleAfter)
|
||||
details, _ := json.Marshal(map[string]any{
|
||||
"customer_id": customerID, "last_run": off.LastRun, "last_status": off.LastStatus, "stale_after": oc.staleAfter.String(),
|
||||
})
|
||||
oc.logger.Printf("[INFO] Offsite staleness: %s (%s)", customerID, age)
|
||||
if _, err := oc.store.SaveEvent(customerID, "offsite_stale", "warning", message, string(details), "hub"); err != nil {
|
||||
oc.logger.Printf("[WARN] Failed to save offsite staleness event for %s: %v", customerID, err)
|
||||
return
|
||||
}
|
||||
if oc.onEvent != nil {
|
||||
oc.onEvent(customerID, "offsite_stale", "warning", message, string(details), "hub")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
package monitor
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// saveOffsiteReport records a customer report whose report_json carries the given offsite object
|
||||
// (or none, when raw==""). The 1.1s sleep before a RE-save of the same customer makes received_at
|
||||
// strictly increasing (second resolution) so GetCustomers' MAX(received_at) picks the new row
|
||||
// deterministically.
|
||||
var offsiteSaved = map[string]bool{}
|
||||
|
||||
func saveOffsiteReport(t *testing.T, st interface {
|
||||
SaveReport(string, []byte) error
|
||||
}, customerID, raw string) {
|
||||
t.Helper()
|
||||
if offsiteSaved[customerID] {
|
||||
time.Sleep(1100 * time.Millisecond)
|
||||
}
|
||||
offsiteSaved[customerID] = true
|
||||
body := `{"customer_id":"` + customerID + `"`
|
||||
if raw != "" {
|
||||
body += `,"offsite":` + raw
|
||||
}
|
||||
body += `}`
|
||||
if err := st.SaveReport(customerID, []byte(body)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func offsiteJSON(enabled bool, escrow, lastRun, lastStatus string, sizeBytes int64, quotaGB int) string {
|
||||
return fmt.Sprintf(`{"enabled":%v,"escrow_state":%q,"last_run":%q,"last_status":%q,"snapshot_count":3,"repo_size_bytes":%d,"quota_gb":%d}`,
|
||||
enabled, escrow, lastRun, lastStatus, sizeBytes, quotaGB)
|
||||
}
|
||||
|
||||
// Scenario D — staleness: enabled+escrowed with no run >48h (or never) alerts once (deduped across
|
||||
// sweeps); a pending customer NEVER alerts (normal onboarding — the red-proofed filter); recovery re-arms.
|
||||
func TestOffsite_StalenessAlert(t *testing.T) {
|
||||
st := newDiskStore(t)
|
||||
now := time.Now().UTC()
|
||||
old := now.Add(-72 * time.Hour).Format(time.RFC3339)
|
||||
fresh := now.Add(-1 * time.Hour).Format(time.RFC3339)
|
||||
|
||||
// stale-c: escrowed, last run 72h ago → stale. pend-c: PENDING with the same old run → silent.
|
||||
saveOffsiteReport(t, st, "stale-c", offsiteJSON(true, "escrowed", old, "ok", 1<<30, 50))
|
||||
saveOffsiteReport(t, st, "pend-c", offsiteJSON(true, "pending", old, "", 1<<30, 50))
|
||||
|
||||
var types []string
|
||||
oc := NewOffsiteChecker(st, 48*time.Hour, func(_, et, _, _, _, _ string) { types = append(types, et) }, quietLog())
|
||||
oc.Check()
|
||||
if got := count(types, "offsite_stale"); got != 1 {
|
||||
t.Fatalf("want exactly 1 offsite_stale (stale-c only; pending must NEVER alert), got %d (%v)", got, types)
|
||||
}
|
||||
if oc.GetStaleState("pend-c") != "ok" {
|
||||
t.Fatalf("a pending customer must not be stale, got %s", oc.GetStaleState("pend-c"))
|
||||
}
|
||||
// repeated sweep with the same data → NO re-page (dedupe)
|
||||
oc.Check()
|
||||
if got := count(types, "offsite_stale"); got != 1 {
|
||||
t.Fatalf("staleness must not re-page every sweep, got %d", got)
|
||||
}
|
||||
// recovery: a fresh run clears + re-arms; going stale again re-alerts
|
||||
saveOffsiteReport(t, st, "stale-c", offsiteJSON(true, "escrowed", fresh, "ok", 1<<30, 50))
|
||||
oc.Check()
|
||||
if oc.GetStaleState("stale-c") != "ok" {
|
||||
t.Fatal("recovery must clear the stale state")
|
||||
}
|
||||
saveOffsiteReport(t, st, "stale-c", offsiteJSON(true, "escrowed", old, "ok", 1<<30, 50))
|
||||
oc.Check()
|
||||
if got := count(types, "offsite_stale"); got != 2 {
|
||||
t.Fatalf("re-staleness after recovery must alert again, got %d", got)
|
||||
}
|
||||
// never-ran escrowed customer is stale too
|
||||
saveOffsiteReport(t, st, "never-c", offsiteJSON(true, "escrowed", "", "", 0, 50))
|
||||
oc.Check()
|
||||
if got := count(types, "offsite_stale"); got != 3 {
|
||||
t.Fatalf("a never-ran escrowed target must be stale, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Scenario E (fill) — 90/95 of quota_gb; quota 0 never alerts; escalation-only; recovery re-arms.
|
||||
func TestOffsite_FillAlert(t *testing.T) {
|
||||
st := newDiskStore(t)
|
||||
fresh := time.Now().UTC().Format(time.RFC3339)
|
||||
gb := int64(1) << 30
|
||||
|
||||
var types, sevs []string
|
||||
onEvent := func(_, et, sev, _, _, _ string) { types = append(types, et); sevs = append(sevs, sev) }
|
||||
saveOffsiteReport(t, st, "fill-c", offsiteJSON(true, "escrowed", fresh, "ok", 40*gb, 50)) // 80% — under warn
|
||||
saveOffsiteReport(t, st, "noq-c", offsiteJSON(true, "escrowed", fresh, "ok", 900*gb, 0)) // dedicated: quota 0
|
||||
oc := NewOffsiteChecker(st, 48*time.Hour, onEvent, quietLog())
|
||||
oc.Check()
|
||||
if len(types) != 0 {
|
||||
t.Fatalf("80%% and quota-0 must not alert, got %v", types)
|
||||
}
|
||||
|
||||
saveOffsiteReport(t, st, "fill-c", offsiteJSON(true, "escrowed", fresh, "ok", 46*gb, 50)) // 92% warn
|
||||
oc.Check()
|
||||
if count(types, "offsite_fill_warning") != 1 {
|
||||
t.Fatalf("92%% must warn once, got %v", types)
|
||||
}
|
||||
oc.Check() // same data — no re-page
|
||||
if len(types) != 1 {
|
||||
t.Fatalf("no re-page on an unchanged band, got %v", types)
|
||||
}
|
||||
saveOffsiteReport(t, st, "fill-c", offsiteJSON(true, "escrowed", fresh, "ok", 48*gb, 50)) // 96% crit
|
||||
oc.Check()
|
||||
if count(types, "offsite_fill_critical") != 1 || sevs[len(sevs)-1] != "critical" {
|
||||
t.Fatalf("96%% must escalate to critical, got types=%v sevs=%v", types, sevs)
|
||||
}
|
||||
// recovery re-arms
|
||||
saveOffsiteReport(t, st, "fill-c", offsiteJSON(true, "escrowed", fresh, "ok", 10*gb, 50))
|
||||
oc.Check()
|
||||
if oc.GetFillState("fill-c") != bandOK {
|
||||
t.Fatal("recovery must re-arm the fill state")
|
||||
}
|
||||
}
|
||||
|
||||
// Nil-safety — reports without an offsite object (pre-v0.109 controllers / offbox disabled) are skipped:
|
||||
// no alert, no state.
|
||||
func TestOffsite_NilSafeOnOldReports(t *testing.T) {
|
||||
st := newDiskStore(t)
|
||||
saveOffsiteReport(t, st, "old-c", "") // no offsite key at all
|
||||
var types []string
|
||||
oc := NewOffsiteChecker(st, 48*time.Hour, func(_, et, _, _, _, _ string) { types = append(types, et) }, quietLog())
|
||||
oc.Check()
|
||||
if len(types) != 0 {
|
||||
t.Fatalf("a report without an offsite object must never alert, got %v", types)
|
||||
}
|
||||
if oc.GetStaleState("old-c") != "unknown" || oc.GetFillState("old-c") != "unknown" {
|
||||
t.Fatal("no state may exist for a customer without an offsite object")
|
||||
}
|
||||
}
|
||||
|
||||
func count(list []string, want string) int {
|
||||
n := 0
|
||||
for _, s := range list {
|
||||
if s == want {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
var _ = strings.Contains // keep strings import if unused by future edits
|
||||
Reference in New Issue
Block a user