[Closed] Over-engineered invalid-rule quarantine

.github/alert-rule-quarantine-patch/TRIGGER

@@ -0,0 +1 @@
+trigger

.github/alert-rule-quarantine-patch/part-00

@@ -0,0 +1,91 @@
+diff --git a/cmd/server/schema_optional.go b/cmd/server/schema_optional.go
+index e9e9a0d..877dcc6 100644
+--- a/cmd/server/schema_optional.go
++++ b/cmd/server/schema_optional.go
+@@ -44,7 +44,9 @@ func ensureRuntimeSchema(db *sql.DB) error {
+ 		  ADD COLUMN IF NOT EXISTS target_role VARCHAR(16) NOT NULL DEFAULT 'all',
+ 		  ADD COLUMN IF NOT EXISTS require_alertable BOOLEAN NOT NULL DEFAULT TRUE,
+ 		  ADD COLUMN IF NOT EXISTS severity VARCHAR(20) NOT NULL DEFAULT 'auto',
+-		  ADD COLUMN IF NOT EXISTS notify_recovery BOOLEAN NOT NULL DEFAULT TRUE`,
++		  ADD COLUMN IF NOT EXISTS notify_recovery BOOLEAN NOT NULL DEFAULT TRUE,
++		  ADD COLUMN IF NOT EXISTS quarantine_reason TEXT,
++		  ADD COLUMN IF NOT EXISTS quarantined_at TIMESTAMPTZ`,
+ 		`ALTER TABLE alerts
+ 		  ADD COLUMN IF NOT EXISTS notified_at TIMESTAMPTZ,
+ 		  ADD COLUMN IF NOT EXISTS notify_error TEXT,
+@@ -165,7 +167,7 @@ func validateRuntimeSchema(db *sql.DB) error {
+ 		"sites":                      {"id", "name", "tower_h_m"},
+ 		"devices":                    {"id", "mac", "ip_address", "role", "managed", "alertable", "alert_silenced_until", "username", "password", "status"},
+ 		"scheduled_jobs":             {"id", "status", "progress", "total_devices", "completed_devices", "error_message"},
+-		"alert_rules":                {"id", "enabled", "scope", "scope_id", "target_role", "require_alertable", "metric", "operator", "threshold", "severity", "notify_channels", "notify_recovery"},
++		"alert_rules":                {"id", "enabled", "scope", "scope_id", "target_role", "require_alertable", "metric", "operator", "threshold", "severity", "notify_channels", "notify_recovery", "quarantine_reason", "quarantined_at"},
+ 		"alerts":                     {"id", "rule_id", "device_id", "status", "triggered_at", "resolved_at", "notified_at", "notify_error", "recovery_notified_at", "recovery_notify_error"},
+ 		"alert_states":               {"rule_id", "device_id", "first_triggered_at", "last_value", "last_checked_at", "notified"},
+ 		"alert_notification_outbox":  {"id", "alert_id", "channel", "event", "payload", "status", "attempts", "next_attempt_at", "last_error", "updated_at", "sent_at"},
+diff --git a/internal/alerting/manager.go b/internal/alerting/manager.go
+index fcdbfe6..aaaaecf 100644
+--- a/internal/alerting/manager.go
++++ b/internal/alerting/manager.go
+@@ -50,25 +50,27 @@ const (
+ var ErrNotFound = errors.New("not found")
+
+ type Rule struct {
+-	ID               int       `json:"id"`
+-	Name             string    `json:"name"`
+-	Enabled          bool      `json:"enabled"`
+-	Scope            string    `json:"scope"`
+-	ScopeID          *int      `json:"scope_id,omitempty"`
+-	TargetRole       string    `json:"target_role"`
+-	RequireAlertable bool      `json:"require_alertable"`
+-	Metric           string    `json:"metric"`
+-	Operator         string    `json:"operator"`
+-	Threshold        float64   `json:"threshold"`
+-	DurationSeconds  int       `json:"duration_seconds"`
+-	Severity         string    `json:"severity"`
+-	NotifyChannels   []string  `json:"notify_channels"`
+-	NotifyEmails     []string  `json:"notify_emails,omitempty"`
+-	WebhookURL       string    `json:"webhook_url,omitempty"`
+-	NotifyRecovery   bool      `json:"notify_recovery"`
+-	CooldownSeconds  int       `json:"cooldown_seconds"`
+-	CreatedAt        time.Time `json:"created_at"`
+-	CreatedBy        int       `json:"created_by,omitempty"`
++	ID               int        `json:"id"`
++	Name             string     `json:"name"`
++	Enabled          bool       `json:"enabled"`
++	Scope            string     `json:"scope"`
++	ScopeID          *int       `json:"scope_id,omitempty"`
++	TargetRole       string     `json:"target_role"`
++	RequireAlertable bool       `json:"require_alertable"`
++	Metric           string     `json:"metric"`
++	Operator         string     `json:"operator"`
++	Threshold        float64    `json:"threshold"`
++	DurationSeconds  int        `json:"duration_seconds"`
++	Severity         string     `json:"severity"`
++	NotifyChannels   []string   `json:"notify_channels"`
++	NotifyEmails     []string   `json:"notify_emails,omitempty"`
++	WebhookURL       string     `json:"webhook_url,omitempty"`
++	NotifyRecovery   bool       `json:"notify_recovery"`
++	CooldownSeconds  int        `json:"cooldown_seconds"`
++	CreatedAt        time.Time  `json:"created_at"`
++	CreatedBy        int        `json:"created_by,omitempty"`
++	QuarantineReason string     `json:"quarantine_reason,omitempty"`
++	QuarantinedAt    *time.Time `json:"quarantined_at,omitempty"`
+ }
+
+ type Alert struct {
+@@ -167,18 +169,25 @@ func (m *Manager) Start(ctx context.Context) {
+ 	go m.sysmonClient.Run(ctx)
+ }
+
++type invalidEnabledRule struct {
++	rule Rule
++	err  error
++}
++
+ func (m *Manager) loadRules() ([]Rule, error) {
+ 	rows, err := m.db.Query(`
+-		SELECT id, name, enabled, scope, scope_id, target_role, require_alertable, metric, operator, threshold,
+-		       duration_seconds, severity, notify_channels, notify_emails, webhook_url, notify_recovery, cooldown_seconds
++		SELECT id, name, enabled, COALESCE(scope, 'all'), scope_id, target_role, require_alertable, metric, operator, threshold,
++		       COALESCE(duration_seconds, 0), severity, COALESCE(notify_channels, ARRAY[]::TEXT[]),
++		    

.github/alert-rule-quarantine-patch/part-01

@@ -0,0 +1,109 @@
+   COALESCE(notify_emails, ARRAY[]::TEXT[]), webhook_url, notify_recovery, COALESCE(cooldown_seconds, 0)
+ 		FROM alert_rules WHERE enabled = true
++		ORDER BY id
+ 	`)
+ 	if err != nil {
+ 		return nil, err
+ 	}
+-	defer rows.Close()
+
+ 	var rules []Rule
++	var invalidRules []invalidEnabledRule
+ 	for rows.Next() {
+ 		var r Rule
+ 		var scopeID sql.NullInt64
+@@ -187,6 +196,7 @@ func (m *Manager) loadRules() ([]Rule, error) {
+ 		if err := rows.Scan(&r.ID, &r.Name, &r.Enabled, &r.Scope, &scopeID, &r.TargetRole, &r.RequireAlertable,
+ 			&r.Metric, &r.Operator, &r.Threshold, &r.DurationSeconds, &r.Severity, &channels, &emails, &webhookURL,
+ 			&r.NotifyRecovery, &r.CooldownSeconds); err != nil {
++			rows.Close()
+ 			return nil, err
+ 		}
+ 		if scopeID.Valid {
+@@ -200,14 +210,84 @@ func (m *Manager) loadRules() ([]Rule, error) {
+ 		}
+ 		normalizeRule(&r)
+ 		if err := ValidateRule(&r); err != nil {
+-			return nil, fmt.Errorf("enabled alert rule %d is invalid: %w", r.ID, err)
++			invalidRules = append(invalidRules, invalidEnabledRule{rule: r, err: err})
++			continue
+ 		}
+ 		rules = append(rules, r)
+ 	}
+-	return rules, rows.Err()
++	if err := rows.Err(); err != nil {
++		rows.Close()
++		return nil, err
++	}
++	if err := rows.Close(); err != nil {
++		return nil, err
++	}
++
++	wakeNotifications := false
++	for _, invalid := range invalidRules {
+		enqueued, err := m.quarantineInvalidRule(context.Background(), invalid.rule, invalid.err)
+		if err != nil {
+			// Bad user data must not take the monitoring daemon down. Keep the
+			// rule out of memory and make the persistence failure unmissable in
+			// the service log so an operator can repair the database manually.
+			log.Printf("alert rule %d (%q) is invalid and was skipped, but could not be quarantined: %v (validation error: %v)",
+				invalid.rule.ID, invalid.rule.Name, err, invalid.err)
+			continue
+		}
+		wakeNotifications = wakeNotifications || enqueued
+		log.Printf("alert rule %d (%q) was automatically disabled: %v", invalid.rule.ID, invalid.rule.Name, invalid.err)
+	}
+	if wakeNotifications {
+		m.wakeNotificationWorker()
+	}
+
+	return rules, nil
+}
+
+func (m *Manager) quarantineInvalidRule(ctx context.Context, rule Rule, validationErr error) (bool, error) {
+	if rule.ID <= 0 {
+		return false, fmt.Errorf("invalid rule id %d", rule.ID)
+	}
+	reason := truncateError(fmt.Sprintf("Automatically disabled by WaveControl because the rule is invalid: %v", validationErr), 2048)
+	tx, err := m.db.BeginTx(ctx, nil)
+	if err != nil {
+		return false, err
+	}
+	defer tx.Rollback()
+
+	result, err := tx.ExecContext(ctx, `
+		UPDATE alert_rules
+		SET enabled=false, quarantine_reason=$2, quarantined_at=NOW(), updated_at=NOW()
+		WHERE id=$1 AND enabled=true
+	`, rule.ID, reason)
+	if err != nil {
+		return false, err
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	if affected == 0 {
++		return false, tx.Commit()
++	}
++
++	// Close any occurrence created by the old rule through the normal durable
++	// resolution path. This cancels undelivered triggers and sends a matching
++	// recovery to channels that may already have received the trigger.
++	enqueued, err := m.resolveRuleAlertsTx(ctx, tx, rule, reason)
++	if err != nil {
++		return false, err
++	}
++	if _, err := tx.ExecContext(ctx, `DDELETE FROM alert_states WHERE rule_id=$1`, rule.ID); err != nil {
++		return false, err
++	}
++	if err := tx.Commit(); err != nil {
++		return false, err
++	}
++	return enqueued, nil
+ }
+
+-func (m *Manager) loadStates() (map[string]*AlertState, error) {
++func (m *Manager) loadStates(activeRuleIDs map[int]struct{}) (map[string]*AlertState, error) {
+ 	rows, err := m.db.Query(`
+		SELECT s.

.github/alert-rule-quarantine-patch/part-02

@@ -0,0 +1,129 @@
+oldown state: %w", err)
+ 	}
+@@ -1023,13 +1113,16 @@ func (m *Manager) GetRule(id int) (Rule, error) {
+ 	var r Rule
+ 	var scopeID sql.NullInt64
+ 	var channels, emails pq.StringArray
+-	var webhook sql.NullString
++	var webhook, quarantineReason sql.NullString
++	var quarantinedAt sql.NullTime
+ 	err := m.db.QueryRow(`
+ 		SELECT id,name,enabled,scope,scope_id,target_role,require_alertable,metric,operator,threshold,
+-		 duration_seconds,severity,notify_channels,notify_emails,webhook_url,notify_recovery,cooldown_seconds,created_at,COALESCE(created_by,0)
++		 duration_seconds,severity,notify_channels,notify_emails,webhook_url,notify_recovery,cooldown_seconds,created_at,
++		 COALESCE(created_by,0),quarantine_reason,quarantined_at
+ 		FROM alert_rules WHERE id=$1
+ 	`, id).Scan(&r.ID, &r.Name, &r.Enabled, &r.Scope, &scopeID, &r.TargetRole, &r.RequireAlertable, &r.Metric,
+-		&r.Operator, &r.Threshold, &r.DurationSeconds, &r.Severity, &channels, &emails, &webhook, &r.NotifyRecovery, &r.CooldownSeconds, &r.CreatedAt, &r.CreatedBy)
++		&r.Operator, &r.Threshold, &r.DurationSeconds, &r.Severity, &channels, &emails, &webhook, &r.NotifyRecovery,
++		&r.CooldownSeconds, &r.CreatedAt, &r.CreatedBy, &quarantineReason, &quarantinedAt)
+ 	if errors.Is(err, sql.ErrNoRows) {
+ 		return r, ErrNotFound
+ 	}
+@@ -1045,6 +1138,13 @@ func (m *Manager) GetRule(id int) (Rule, error) {
+ 	if webhook.Valid {
+ 		r.WebhookURL = webhook.String
+ 	}
++	if quarantineReason.Valid {
++		r.QuarantineReason = quarantineReason.String
++	}
++	if quarantinedAt.Valid {
++		t := quarantinedAt.Time
++		r.QuarantinedAt = &t
++	}
+ 	normalizeRule(&r)
+ 	return r, nil
+ }
+@@ -1076,7 +1176,8 @@ func (m *Manager) UpdateRule(id int, rule *Rule) error {
+ 	if _, err := tx.Exec(`
+ 		UPDATE alert_rules SET name=$1,enabled=$2,scope=$3,scope_id=$4,target_role=$5,require_alertable=$6,
+ 		 metric=$7,operator=$8,threshold=$9,duration_seconds=$10,severity=$11,notify_channels=$12,notify_emails=$13,
+-		 webhook_url=NULLIF($14,''),notify_recovery=$15,cooldown_seconds=$16,updated_at=NOW() WHERE id=$17
++		 webhook_url=NULLIF($14,''),notify_recovery=$15,cooldown_seconds=$16,
++		 quarantine_reason=NULL,quarantined_at=NULL,updated_at=NOW() WHERE id=$17
+ 	`, rule.Name, rule.Enabled, rule.Scope, rule.ScopeID, rule.TargetRole, rule.RequireAlertable, rule.Metric, rule.Operator,
+ 		rule.Threshold, rule.DurationSeconds, rule.Severity, pq.Array(rule.NotifyChannels), pq.Array(rule.NotifyEmails), rule.WebhookURL,
+ 		rule.NotifyRecovery, rule.CooldownSeconds, id); err != nil {
+@@ -1128,7 +1229,8 @@ func (m *Manager) DeleteRule(id int) error {
+ func (m *Manager) ListRules() ([]Rule, error) {
+ 	rows, err := m.db.Query(`
+ 		SELECT id,name,enabled,scope,scope_id,target_role,require_alertable,metric,operator,threshold,
+-		 duration_seconds,severity,notify_channels,notify_emails,webhook_url,notify_recovery,cooldown_seconds,created_at
++		 duration_seconds,severity,notify_channels,notify_emails,webhook_url,notify_recovery,cooldown_seconds,created_at,
++		 quarantine_reason,quarantined_at
+ 		FROM alert_rules ORDER BY name
+ 	`)
+ 	if err != nil {
+@@ -1140,10 +1242,11 @@ func (m *Manager) ListRules() ([]Rule, error) {
+ 		var r Rule
+ 		var scopeID sql.NullInt64
+ 		var channels, emails pq.StringArray
+-		var webhook sql.NullString
++		var webhook, quarantineReason sql.NullString
++		var quarantinedAt sql.NullTime
+ 		if err := rows.Scan(&r.ID, &r.Name, &r.Enabled, &r.Scope, &scopeID, &r.TargetRole, &r.RequireAlertable,
+ 			&r.Metric, &r.Operator, &r.Threshold, &r.DurationSeconds, &r.Severity, &channels, &emails, &webhook,
+-			&r.NotifyRecovery, &r.CooldownSeconds, &r.CreatedAt); err != nil {
++			&r.NotifyRecovery, &r.CooldownSeconds, &r.CreatedAt, &quarantineReason, &quarantinedAt); err != nil {
+ 			return nil, err
+ 		}
+ 		if scopeID.Valid {
+@@ -1155,6 +1258,13 @@ func (m *Manager) ListRules() ([]Rule, error) {
+ 		if webhook.Valid {
+ 			r.WebhookURL = webhook.String
+ 		}
++		if quarantineReason.Valid {
++			r.QuarantineReason = quarantineReason.String
++		}
++		if quarantinedAt.Valid {
++			t := quarantinedAt.Time
++			r.QuarantinedAt = &t
++		}
+ 		normalizeRule(&r)
+ 		rules = append(rules, r)
+ 	}
+diff --git a/internal/alerting/quarantine_test.go b/internal/alerting/quarantine_test.go
+new file mode 100644
+index 0000000..d227141
+--- /dev/null
++++ b/internal/alerting/quarantine_test.go
+@@ -0,0 +1,163 @@
++package alerting
++
++import (
++	"context"
++	"database/sql"
++	"database/sql/driver"
++	"errors"
++	"io"
++	"strings"
++	"sync"
++	"testing"
++)
++
++type quarantineTestConnector struct {
++	conn *quarantineTestConn
++}
++
++func (c quarantineTestConnector) Connect(context.Context) (driver.Conn, error) { return c.conn, nil }
++func (c quarantineTestConnector) Driver() driver.Driver                        { return quarantineTestDriver{} }
++
++type quarantineTestDriver struct{}
++
++func (quarantineTestDriver) Open(string) (driver.Conn, error) {
++	return nil, errors.New("quarantine test driver must be opened through its connector")
++}
++
++type quarantineTestConn struct {
++	mu        sync.Mutex
++	execs     []quarantineTestExec
++	committed bool
++	failExec  bool
++}
++
++type quarantineTestExec struct {
++	query string
++	args  []driver.NamedValue
++}
++
++func (c *quarantineTestConn) Prepare(string) (driver.Stmt,