fix(collect): stop false fleet-wide offline alerts

collect/.env.example

@@ -79,8 +79,21 @@ REDIS_URL=redis://localhost:6379
 # Notification configuration
 #NOTIFICATION_RETRY_ATTEMPTS=3
 
-# Heartbeat monitoring configuration
-#HEARTBEAT_TIMEOUT_MINUTES=20
+# Heartbeat monitoring configuration.
+# Clients beat on wall-clock-aligned waves every ~10 min, so this timeout is a
+# budget of missed beats: 30 min tolerates 3. Lower it and a single lost wave
+# can mark the whole fleet inactive at once.
+#HEARTBEAT_TIMEOUT_MINUTES=30
+# How often the monitor re-evaluates system liveness.
+#HEARTBEAT_CHECK_INTERVAL_SECONDS=300
+
+# LinkFailed lost-wave guard. Because heartbeats arrive in synchronized waves, a
+# gap in our own recording leaves every affected system with the same
+# last_heartbeat. When this many inactive systems share a last_heartbeat inside
+# the window below, collect treats it as its own ingest fault and suppresses
+# their alerts instead of paging every customer. Set the count to 0 to disable.
+#LINKFAILED_LOST_WAVE_MIN_SYSTEMS=20
+#LINKFAILED_LOST_WAVE_WINDOW_SECONDS=180
 
 # Backup storage (DigitalOcean Spaces — shared S3 account with Mimir)
 # Endpoint, access key, and secret key are the same values configured for

collect/README.md

@@ -45,7 +45,7 @@ REDIS_PASSWORD=
 ```bash
 LISTEN_ADDRESS=127.0.0.1:8081
 API_MAX_REQUEST_SIZE=10MB
-HEARTBEAT_TIMEOUT_MINUTES=20
+HEARTBEAT_TIMEOUT_MINUTES=30
 LOG_LEVEL=info
 ```
 
@@ -84,11 +84,11 @@ LOG_LEVEL=info
   - `inactive` → `active` when a fresh heartbeat arrives (within `HEARTBEAT_TIMEOUT_MINUTES`); this recovery path also resolves the firing `LinkFailed` alert
   - `active` → `inactive` when the heartbeat is stale (older than `HEARTBEAT_TIMEOUT_MINUTES`)
   - `unknown` → `active` as a safety net for the inline flip
-- Configurable timeout via `HEARTBEAT_TIMEOUT_MINUTES` (default: 20 minutes)
+- Configurable timeout via `HEARTBEAT_TIMEOUT_MINUTES` (default: 30 minutes)
 
 **7. LinkFailed Synchronization**
 - **LinkFailed Monitor Cron** runs every 5 minutes
-- Fires the internal `LinkFailed` alert for inactive, non-deleted, non-suspended systems after `HEARTBEAT_TIMEOUT_MINUTES` (20 minutes by default); systems under a suspended organization are excluded too
+- Fires the internal `LinkFailed` alert for inactive, non-deleted, non-suspended systems after `HEARTBEAT_TIMEOUT_MINUTES` (30 minutes by default); systems under a suspended organization are excluded too
 - On recovery (`inactive` → `active`) the heartbeat monitor posts an explicit resolve with the same fingerprint, so Alertmanager clears the alert immediately instead of waiting for the TTL; the 10 minute TTL from the last refresh remains a backstop if a resolve is ever missed
 - Reuses the same server-side label enrichment as the Mimir proxy so internal alerts carry the same authoritative system and organization labels
 

collect/alerting/mimir.go

@@ -362,7 +362,18 @@ func PostAlerts(orgID string, alerts []models.AlertmanagerPostAlert) error {
 // alert for a recovered system. It reuses the exact firing label set — the same
 // three base labels plus EnrichAlerts(systemContext) — so Alertmanager recomputes
 // the identical fingerprint and clears the firing alert instead of opening a new
-// one. Annotations are omitted: they don't affect the fingerprint.
+// one.
+//
+// Annotations do not affect the fingerprint, but they must still be set:
+// Alertmanager's Alert.Merge copies the younger alert wholesale and only
+// special-cases StartsAt, so a resolve with no annotations blanks them on the
+// alert it clears. That left every normally-resolved LinkFailed with empty
+// annotations in both the resolved notification and alert_history.
+//
+// The text describes the recovery rather than replaying the firing alert's
+// "no heartbeat since X" line: system_heartbeats keeps only the latest beat, so
+// by the time we resolve, the timestamp that line referred to is already
+// overwritten by the beat that triggered the recovery.
 func BuildResolvedLinkFailedAlert(systemContext *SystemAlertContext) (models.AlertmanagerPostAlert, error) {
 	now := time.Now().UTC()
 	enriched, err := EnrichAlerts([]models.AlertmanagerPostAlert{
@@ -372,6 +383,12 @@ func BuildResolvedLinkFailedAlert(systemContext *SystemAlertContext) (models.Ale
 				"severity":     "critical",
 				ManagedByLabel: ManagedByCollect,
 			},
+			Annotations: map[string]string{
+				"summary_en":     "System is communicating again",
+				"summary_it":     "Il sistema comunica di nuovo",
+				"description_en": "The system has resumed sending heartbeats to My Nethesis.",
+				"description_it": "Il sistema ha ripreso a inviare heartbeat a My Nethesis.",
+			},
 			StartsAt: now.Add(-time.Minute),
 			EndsAt:   now,
 		},

collect/alerting/mimir_test.go

@@ -178,6 +178,27 @@ func TestEnrichAlerts(t *testing.T) {
 	assert.Equal(t, endsAt, alerts[0].EndsAt)
 }
 
+// Alertmanager's Alert.Merge copies the younger alert wholesale and only
+// special-cases StartsAt, so an annotation-free resolve wipes the annotations off
+// the alert it clears, leaving both the resolved notification and the
+// alert_history row without a description. The resolve must therefore carry its
+// own annotations.
+func TestBuildResolvedLinkFailedAlert_CarriesAnnotations(t *testing.T) {
+	systemContext := BuildSystemAlertContext(SystemAlertMetadata{
+		SystemID:       "sys-1",
+		OrganizationID: "org-1",
+		SystemKey:      "SYS-001",
+		SystemName:     "web-01",
+	})
+
+	alert, err := BuildResolvedLinkFailedAlert(systemContext)
+	require.NoError(t, err)
+
+	for _, key := range []string{"summary_en", "summary_it", "description_en", "description_it"} {
+		assert.NotEmpty(t, alert.Annotations[key], "resolve must set %s or Alertmanager blanks it", key)
+	}
+}
+
 func TestBuildResolvedLinkFailedAlert(t *testing.T) {
 	// SystemFQDN/IPv4/VAT left empty on purpose: EnrichAlerts must strip them so
 	// the resolve fingerprint matches the firing alert (which also strips empties).

collect/configuration/configuration.go

@@ -90,6 +90,10 @@ type Configuration struct {
 	HeartbeatTimeoutMinutes       int `json:"heartbeat_timeout_minutes"`
 	HeartbeatCheckIntervalSeconds int `json:"heartbeat_check_interval_seconds"`
 
+	// LinkFailed lost-wave guard
+	LinkFailedLostWaveMinSystems    int `json:"linkfailed_lost_wave_min_systems"`
+	LinkFailedLostWaveWindowSeconds int `json:"linkfailed_lost_wave_window_seconds"`
+
 	// Mimir configuration
 	MimirURL string `json:"mimir_url"`
 
@@ -215,13 +219,26 @@ func Init() {
 	Config.NotificationRetryAttempts = parseIntWithDefault("NOTIFICATION_RETRY_ATTEMPTS", 3)
 
 	// Heartbeat monitoring configuration.
-	// Timeout is 2x the ~10min client send interval: at 10min (== the interval)
-	// the synchronized heartbeat waves straddle the cutoff and the fleet flaps
-	// active<->inactive en masse, churning LinkFailed fire/resolve. 20min absorbs
-	// a late/missed beat; genuinely-down systems still alert within 20min.
-	Config.HeartbeatTimeoutMinutes = parseIntWithDefault("HEARTBEAT_TIMEOUT_MINUTES", 20)
+	// Clients send on wall-clock-aligned waves at a ~10min interval, so the
+	// timeout is really a budget of missed beats: at 10min (== the interval) the
+	// waves straddle the cutoff and the fleet flaps active<->inactive en masse.
+	// 20min allowed only 2 beats, which is thin enough that two consecutive lost
+	// waves can mark the whole fleet inactive at once. 30min gives 3 beats, so one
+	// lost wave plus a late one is absorbed; genuinely-down systems still alert
+	// within 30min.
+	// The remaining fleet-wide risk is handled by the lost-wave guard below.
+	Config.HeartbeatTimeoutMinutes = parseIntWithDefault("HEARTBEAT_TIMEOUT_MINUTES", 30)
 	Config.HeartbeatCheckIntervalSeconds = parseIntWithDefault("HEARTBEAT_CHECK_INTERVAL_SECONDS", 300)
 
+	// LinkFailed lost-wave guard. Because heartbeats arrive in synchronized
+	// waves, a gap in our own recording leaves every affected system with the
+	// same last_heartbeat to within seconds. Machines that are really dead stop
+	// beating at arbitrary moments, so a tight cluster this large is our fault,
+	// not theirs, and must not page customers. The window is wider than an
+	// observed wave (~60s) to absorb clock skew and ingest lag.
+	Config.LinkFailedLostWaveMinSystems = parseIntWithDefault("LINKFAILED_LOST_WAVE_MIN_SYSTEMS", 20)
+	Config.LinkFailedLostWaveWindowSeconds = parseIntWithDefault("LINKFAILED_LOST_WAVE_WINDOW_SECONDS", 180)
+
 	// Mimir configuration
 	if mimirURL := os.Getenv("MIMIR_URL"); mimirURL != "" {
 		Config.MimirURL = mimirURL

collect/cron/linkfailed_monitor.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"sort"
 	"time"
 
 	collectalerting "github.com/nethesis/my/collect/alerting"
@@ -39,15 +40,20 @@ type LinkFailedMonitor struct {
 	timeoutMinutes int
 	syncInterval   time.Duration
 	postAlerts     postAlertsFunc
+	// Lost-wave guard: see suppressLostWave.
+	lostWaveMinSystems int
+	lostWaveWindow     time.Duration
 }
 
 // NewLinkFailedMonitor creates a new LinkFailed monitor instance.
 func NewLinkFailedMonitor() *LinkFailedMonitor {
 	return &LinkFailedMonitor{
-		db:             database.DB,
-		timeoutMinutes: configuration.Config.HeartbeatTimeoutMinutes,
-		syncInterval:   linkFailedSyncInterval,
-		postAlerts:     collectalerting.PostAlerts,
+		db:                 database.DB,
+		timeoutMinutes:     configuration.Config.HeartbeatTimeoutMinutes,
+		syncInterval:       linkFailedSyncInterval,
+		postAlerts:         collectalerting.PostAlerts,
+		lostWaveMinSystems: configuration.Config.LinkFailedLostWaveMinSystems,
+		lostWaveWindow:     time.Duration(configuration.Config.LinkFailedLostWaveWindowSeconds) * time.Second,
 	}
 }
 
@@ -81,6 +87,18 @@ func (m *LinkFailedMonitor) sync(ctx context.Context) {
 		return
 	}
 
+	if suppressed, clusters := suppressLostWave(desiredByOrg, m.lostWaveMinSystems, m.lostWaveWindow); suppressed > 0 {
+		clusterStarts := make([]string, 0, len(clusters))
+		for _, at := range clusters {
+			clusterStarts = append(clusterStarts, at.UTC().Format(time.RFC3339))
+		}
+		logger.Error().
+			Int("systems_suppressed", suppressed).
+			Strs("heartbeat_cluster_starts", clusterStarts).
+			Dur("cluster_window", m.lostWaveWindow).
+			Msg("LinkFailed monitor: suppressed a synchronized heartbeat gap — heartbeat recording broke on our side, these systems are not down")
+	}
+
 	for tenantOrgID, systems := range desiredByOrg {
 		if err := m.syncOrganization(tenantOrgID, systems); err != nil {
 			logger.Error().Err(err).Str("tenant_org_id", tenantOrgID).Msg("LinkFailed monitor: sync failed")
@@ -185,6 +203,93 @@ func (m *LinkFailedMonitor) loadInactiveSystems(ctx context.Context) (map[string
 	return systemsByOrg, nil
 }
 
+// lostWaveEntry locates one system inside the desired-alert map alongside its
+// heartbeat time, so a detected cluster can be removed in place.
+type lostWaveEntry struct {
+	tenantOrgID string
+	systemKey   string
+	at          time.Time
+}
+
+// suppressLostWave drops inactive systems whose last_heartbeat values all fall
+// inside a single narrow window.
+//
+// Clients send heartbeats on wall-clock-aligned waves, so when the platform
+// briefly stops recording them every affected system is left with the same
+// last_heartbeat to within a few seconds. Hundreds of machines do not genuinely
+// lose connectivity in the same second: a tight cluster means our own ingest
+// gapped, and firing on it pages every customer at once for a fault on our side.
+// A machine that is really dead stopped beating at an arbitrary moment, so it
+// never joins the cluster and still alerts normally.
+//
+// The scan deliberately runs across tenants: an ingest gap hits every tenant at
+// once, which is what separates it from one reseller losing an uplink.
+//
+// One ingest gap usually spans several consecutive waves, so every qualifying
+// cluster is removed, not just the largest. Once two waves have been lost the
+// inactive set holds both, and suppressing only the biggest would still page the
+// systems belonging to the smaller one.
+//
+// Known trade-off: a wave can straddle the staleness cutoff and so reach the
+// inactive set across two ticks. If the first slice lands below minSystems it
+// alerts, and when the rest arrives the now-qualifying cluster is suppressed, so
+// those few alerts stop being refreshed and expire into a spurious "resolved"
+// notification one TTL later. Bounded by minSystems and self-correcting, so it is
+// accepted rather than tracked with per-alert state.
+//
+// Returns how many systems were dropped and the start of each offending window,
+// oldest first. A minSystems or window of zero or less disables the guard.
+func suppressLostWave(byOrg map[string]map[string]linkFailedSystem, minSystems int, window time.Duration) (int, []time.Time) {
+	if minSystems <= 0 || window <= 0 {
+		return 0, nil
+	}
+
+	entries := make([]lostWaveEntry, 0, len(byOrg))
+	for tenantOrgID, systems := range byOrg {
+		for systemKey, system := range systems {
+			entries = append(entries, lostWaveEntry{
+				tenantOrgID: tenantOrgID,
+				systemKey:   systemKey,
+				at:          system.LastHeartbeat,
+			})
+		}
+	}
+	if len(entries) < minSystems {
+		return 0, nil
+	}
+
+	sort.Slice(entries, func(i, j int) bool { return entries[i].at.Before(entries[j].at) })
+
+	// Sweep once, carving out every maximal run that fits inside one window and
+	// meets the floor. Advancing past a suppressed run rather than restarting
+	// keeps this linear and stops a run from being counted twice.
+	var (
+		suppressed int
+		clusters   []time.Time
+		start      int
+	)
+	for start < len(entries) {
+		end := start
+		for end+1 < len(entries) && entries[end+1].at.Sub(entries[start].at) <= window {
+			end++
+		}
+
+		if count := end - start + 1; count >= minSystems {
+			for _, entry := range entries[start : end+1] {
+				delete(byOrg[entry.tenantOrgID], entry.systemKey)
+			}
+			suppressed += count
+			clusters = append(clusters, entries[start].at)
+			start = end + 1
+			continue
+		}
+
+		start++
+	}
+
+	return suppressed, clusters
+}
+
 // syncOrganization pushes the firing alerts for one Mimir tenant. tenantOrgID
 // is the reseller/managing org (X-Scope-OrgID); the systems it carries may
 // belong to several customer orgs, each identified by the organization_id