beszel-ipv6/internal/alerts/alerts_network_monitors_test.go

//go:build testing

package alerts_test

import (
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/henrygd/beszel/internal/alerts"
	"github.com/henrygd/beszel/internal/entities/monitor"
	beszelTests "github.com/henrygd/beszel/internal/tests"
	"github.com/pocketbase/dbx"
	"github.com/pocketbase/pocketbase/core"
	pbTests "github.com/pocketbase/pocketbase/tests"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func networkAlertSetup(t *testing.T) (*beszelTests.TestHub, *core.Record, *core.Record, []*core.Record) {
	t.Helper()
	hub, system, alert := systemdTestSetup(t, false)
	t.Cleanup(hub.Cleanup)
	alert.Set("name", "NetworkMonitorLoss")
	alert.Set("value", 5)
	require.NoError(t, hub.Save(alert))
	var monitors []*core.Record
	for _, name := range []string{"gateway", "website"} {
		record, err := beszelTests.CreateRecord(hub, "network_monitors", map[string]any{
			"system": system.Id, "target": name + ".example.com", "protocol": "icmp", "interval": 60, "enabled": true,
		})
		require.NoError(t, err)
		monitors = append(monitors, record)
	}
	// Avoid starting a system update worker in tests.
	_, err := hub.DB().Update("systems", dbx.Params{"status": "up"}, dbx.HashExp{"id": system.Id}).Execute()
	require.NoError(t, err)
	system.Set("status", "up")
	return hub, system, alert, monitors
}

func monitorResult(loss float64) monitor.Result {
	return monitor.Result{LastProbeAt: time.Now().UnixMilli(), SampleCount: 60, PacketLoss1h: loss}
}

func TestNetworkMonitorAlertIndependentIncidents(t *testing.T) {
	hub, system, alert, monitors := networkAlertSetup(t)
	am := alerts.NewTestAlertManagerWithoutWorker(hub)
	count := hub.TestMailer.TotalSend()
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(10), monitors[1].Id: monitorResult(0)}
	check := func(active bool, open, sent int) {
		t.Helper()
		record, err := hub.FindRecordById("alerts", alert.Id)
		require.NoError(t, err)
		assert.Equal(t, active, record.GetBool("triggered"))
		total, err := hub.CountRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id, "resolved": ""})
		require.NoError(t, err)
		assert.EqualValues(t, open, total)
		assert.Equal(t, count+sent, hub.TestMailer.TotalSend())
	}
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	check(true, 1, 1)
	message := hub.TestMailer.Messages()[count]
	assert.Contains(t, message.Text, "gateway.example.com")
	assert.Contains(t, message.Text, "10.00%")
	assert.Contains(t, message.Text, "5.00%")
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	check(true, 1, 1)
	// Persisted monitor state prevents duplicate notifications after a hub restart.
	am = alerts.NewTestAlertManagerWithoutWorker(hub)
	results[monitors[1].Id] = monitorResult(20)
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	check(true, 2, 2)
	results[monitors[0].Id] = monitorResult(5) // Equality is a recovery.
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	check(true, 1, 3)
	results[monitors[1].Id] = monitorResult(0)
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	check(false, 0, 4)
	histories, err := hub.FindAllRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id})
	require.NoError(t, err)
	require.Len(t, histories, 2)
	for _, history := range histories {
		assert.NotEmpty(t, history.GetString("monitor_name"))
	}
}

func TestNetworkMonitorAlertTargetLabel(t *testing.T) {
	for _, tc := range []struct {
		protocol, target, label string
		port                    int
	}{
		{"icmp", "gateway.example.com", "gateway.example.com", 0},
		{"http", "https://example.com/health", "https://example.com/health", 0},
		{"tcp", "example.com", "example.com:8443", 8443},
		{"tcp", "2001:db8::1", "[2001:db8::1]:443", 443},
	} {
		t.Run(tc.label, func(t *testing.T) {
			hub, system, alert, monitors := networkAlertSetup(t)
			m := monitors[0]
			m.Set("protocol", tc.protocol)
			m.Set("target", tc.target)
			m.Set("port", tc.port)
			require.NoError(t, hub.Save(m))
			am := alerts.NewTestAlertManagerWithoutWorker(hub)
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, map[string]monitor.Result{m.Id: monitorResult(10)}))
			histories, err := hub.FindAllRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id})
			require.NoError(t, err)
			require.Len(t, histories, 1)
			assert.Equal(t, tc.label, histories[0].GetString("monitor_name"))
			assert.Contains(t, hub.TestMailer.Messages()[hub.TestMailer.TotalSend()-1].Text, tc.label)
		})
	}
}

func TestNetworkMonitorAlertIgnoresUnknownResults(t *testing.T) {
	for _, scenario := range []string{"missing", "stale", "warmup", "no probes", "down", "paused", "future", "expired hourly window"} {
		t.Run(scenario, func(t *testing.T) {
			hub, system, alert, monitors := networkAlertSetup(t)
			am := alerts.NewTestAlertManagerWithoutWorker(hub)
			apply := func(loss float64) {
				result := monitorResult(loss)
				results := map[string]monitor.Result{monitors[0].Id: result}
				switch scenario {
				case "missing":
					results = nil
				case "stale":
					result.LastProbeAt = time.Now().Add(-10 * time.Minute).UnixMilli()
					results[monitors[0].Id] = result
				case "expired hourly window":
					monitors[0].Set("interval", 3600)
					require.NoError(t, hub.Save(monitors[0]))
					result.LastProbeAt = time.Now().Add(-2 * time.Hour).UnixMilli()
					results[monitors[0].Id] = result
				case "future":
					result.LastProbeAt = time.Now().Add(time.Hour).UnixMilli()
					results[monitors[0].Id] = result
				case "warmup":
					result.SampleCount = 2
					results[monitors[0].Id] = result
				case "no probes":
					result.SampleCount = 0
					results[monitors[0].Id] = result
				case "down":
					_, err := hub.DB().Update("systems", dbx.Params{"status": scenario}, dbx.HashExp{"id": system.Id}).Execute()
					require.NoError(t, err)
				case "paused":
					record, err := hub.FindRecordById("systems", system.Id)
					require.NoError(t, err)
					record.Set("status", "paused")
					require.NoError(t, hub.Save(record))
				}
				require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
			}
			count := hub.TestMailer.TotalSend()
			apply(100)
			assert.Equal(t, count, hub.TestMailer.TotalSend())
			_, err := hub.DB().Update("systems", dbx.Params{"status": "up"}, dbx.HashExp{"id": system.Id}).Execute()
			require.NoError(t, err)
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, map[string]monitor.Result{monitors[0].Id: monitorResult(10)}))
			apply(0)
			assert.Equal(t, count+1, hub.TestMailer.TotalSend(), "unknown data must not recover an incident")
			record, err := hub.FindRecordById("alerts", alert.Id)
			require.NoError(t, err)
			assert.True(t, record.GetBool("triggered"))
		})
	}
}

func TestNetworkMonitorAlertCleanup(t *testing.T) {
	for _, scenario := range []string{"disable monitor", "delete monitor", "disable alert", "purge history", "delete system"} {
		t.Run(scenario, func(t *testing.T) {
			hub, system, alert, monitors := networkAlertSetup(t)
			am := alerts.NewTestAlertManagerWithoutWorker(hub)
			results := map[string]monitor.Result{monitors[0].Id: monitorResult(10), monitors[1].Id: monitorResult(20)}
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
			count := hub.TestMailer.TotalSend()
			switch scenario {
			case "disable monitor":
				monitors[0].Set("enabled", false)
				require.NoError(t, hub.Save(monitors[0]))
			case "delete monitor":
				require.NoError(t, hub.Delete(monitors[0]))
			case "disable alert":
				require.NoError(t, hub.Delete(alert))
			case "delete system":
				require.NoError(t, hub.Delete(system))
			case "purge history":
				history, err := hub.FindAllRecords("alerts_history")
				require.NoError(t, err)
				for _, record := range history {
					require.NoError(t, hub.Delete(record))
				}
			}
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
			assert.Equal(t, count, hub.TestMailer.TotalSend())
			open, err := hub.CountRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id, "resolved": ""})
			require.NoError(t, err)
			if scenario == "disable monitor" || scenario == "delete monitor" {
				assert.EqualValues(t, 1, open)
				record, err := hub.FindRecordById("alerts", alert.Id)
				require.NoError(t, err)
				assert.True(t, record.GetBool("triggered"))
				require.NoError(t, hub.Delete(monitors[1]))
				record, err = hub.FindRecordById("alerts", alert.Id)
				require.NoError(t, err)
				assert.False(t, record.GetBool("triggered"))
			} else {
				assert.Zero(t, open)
			}
		})
	}
}

func TestNetworkMonitorAlertPerUserThresholds(t *testing.T) {
	hub, system, alert, monitors := networkAlertSetup(t)
	user, err := beszelTests.CreateUser(hub, "monitor2@example.com", "password")
	require.NoError(t, err)
	other, err := beszelTests.CreateRecord(hub, "alerts", map[string]any{"name": "NetworkMonitorLoss", "system": system.Id, "user": user.Id, "value": 20})
	require.NoError(t, err)
	am := alerts.NewTestAlertManagerWithoutWorker(hub)
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(10)}
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	other, err = hub.FindRecordById("alerts", other.Id)
	require.NoError(t, err)
	assert.False(t, other.GetBool("triggered"))
	// Editing the threshold re-evaluates on the next batch, without losing state.
	alert, err = hub.FindRecordById("alerts", alert.Id)
	require.NoError(t, err)
	alert.Set("value", 15)
	require.NoError(t, hub.Save(alert))
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	alert, err = hub.FindRecordById("alerts", alert.Id)
	require.NoError(t, err)
	assert.False(t, alert.GetBool("triggered"))
}

func TestNetworkMonitorAlertAPI(t *testing.T) {
	for _, tc := range []struct {
		name                  string
		value                 float64
		direct, denied, patch bool
		status                int
	}{
		{name: "zero threshold", value: 0, status: 200},
		{name: "fractional threshold", value: 5.5, status: 200},
		{name: "negative threshold", value: -1, status: 400},
		{name: "unreachable threshold", value: 100, status: 400},
		{name: "bulk inaccessible system", value: 5, denied: true, status: 200},
		{name: "direct inaccessible system", value: 5, direct: true, denied: true, status: 403},
		{name: "direct invalid threshold", value: -1, direct: true, status: 400},
		{name: "direct private state", value: 5, direct: true, status: 200},
		{name: "patch preserves state", value: 10, direct: true, patch: true, status: 200},
	} {
		t.Run(tc.name, func(t *testing.T) {
			hub, user := beszelTests.GetHubWithUser(t)
			defer hub.Cleanup()
			owner := user.Id
			if tc.denied {
				other, err := beszelTests.CreateUser(hub, "other@example.com", "password")
				require.NoError(t, err)
				owner = other.Id
			}
			systems, err := beszelTests.CreateSystems(hub, 1, owner, "paused")
			require.NoError(t, err)
			token, err := user.NewAuthToken()
			require.NoError(t, err)
			body := map[string]any{"name": "NetworkMonitorLoss", "value": tc.value, "min": 60, "systems": []string{systems[0].Id}, "overwrite": true}
			url, method := "/api/beszel/user-alerts", "POST"
			if tc.direct {
				url = "/api/collections/alerts/records"
				body["system"], body["user"] = systems[0].Id, user.Id
				body["state"], body["triggered"] = map[string]any{"monitors": map[string]string{"fake": "fake"}}, true
			}
			if tc.patch {
				alert, err := beszelTests.CreateRecord(hub, "alerts", map[string]any{"name": "NetworkMonitorLoss", "system": systems[0].Id, "user": user.Id, "value": 5, "triggered": true, "state": map[string]any{"monitors": map[string]string{"real": "history"}}})
				require.NoError(t, err)
				url += "/" + alert.Id
				method = "PATCH"
				body["triggered"] = false
			}
			content := `"success":true`
			if tc.direct {
				content = `"name":"NetworkMonitorLoss"`
			}
			if tc.status == 400 {
				content = `"status":400`
			}
			if tc.status == 403 {
				content = `"status":403`
			}
			scenario := beszelTests.ApiScenario{
				Name: tc.name, Method: method, URL: url, Body: jsonReader(body),
				Headers: map[string]string{"Authorization": token}, ExpectedStatus: tc.status, ExpectedContent: []string{content},
				TestAppFactory: func(testing.TB) *pbTests.TestApp { return hub.TestApp },
			}
			scenario.Test(t)
			records, err := hub.FindAllRecords("alerts")
			require.NoError(t, err)
			if tc.status != 200 || tc.denied {
				assert.Empty(t, records)
				return
			}
			require.Len(t, records, 1)
			assert.Equal(t, tc.value, records[0].GetFloat("value"))
			assert.Zero(t, records[0].GetInt("min"))
			state := struct {
				Monitors map[string]string `json:"monitors"`
			}{}
			require.NoError(t, records[0].UnmarshalJSONField("state", &state))
			states := state.Monitors
			if tc.patch {
				assert.Equal(t, map[string]string{"real": "history"}, states)
				assert.True(t, records[0].GetBool("triggered"))
			} else {
				assert.Empty(t, states)
				assert.False(t, records[0].GetBool("triggered"))
			}
		})
	}
}

type monitorCountingHub struct {
	*beszelTests.TestHub
	transactions      atomic.Int64
	beforeTransaction func()
}

func (h *monitorCountingHub) RunInTransaction(fn func(core.App) error) error {
	h.transactions.Add(1)
	if h.beforeTransaction != nil {
		h.beforeTransaction()
	}
	return h.App.RunInTransaction(fn)
}

// Count actual SQL on both DB connections, including queries through record APIs.
func monitorSQLCounter(t *testing.T, app core.App) *atomic.Int64 {
	t.Helper()
	count := &atomic.Int64{}
	for _, builder := range []dbx.Builder{app.ConcurrentDB(), app.NonconcurrentDB()} {
		db := builder.(*dbx.DB)
		old := db.LogFunc
		db.LogFunc = func(string, ...any) { count.Add(1) }
		t.Cleanup(func() { db.LogFunc = old })
	}
	return count
}

func TestNetworkMonitorAlertSteadyStateNoDatabaseWork(t *testing.T) {
	hub, system, alert, monitors := networkAlertSetup(t)
	counted := &monitorCountingHub{TestHub: hub}
	am := alerts.NewTestAlertManagerWithoutWorker(counted)
	sql := monitorSQLCounter(t, hub)
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(0)}
	evaluate := func() { t.Helper(); require.NoError(t, am.HandleNetworkMonitorAlerts(system, results)) }
	noWork := func() {
		t.Helper()
		sql.Store(0)
		counted.transactions.Store(0)
		for range 100 {
			evaluate()
		}
		assert.Zero(t, sql.Load(), "steady state must not issue SQL")
		assert.Zero(t, counted.transactions.Load(), "steady state must not open transactions")
	}
	// One-time lazy loads are permitted, including on hub restart.
	evaluate()
	assert.Positive(t, sql.Load())
	noWork()
	// Realtime metric saves invoke record hooks but must not invalidate config.
	fresh, err := hub.FindRecordById("network_monitors", monitors[0].Id)
	require.NoError(t, err)
	monitors[0] = fresh
	monitors[0].Set("loss1h", 0)
	monitors[0].Set("res", 100)
	require.NoError(t, hub.Save(monitors[0]))
	noWork()
	results[monitors[0].Id] = monitorResult(10)
	evaluate()
	assert.Positive(t, sql.Load(), "transitions must still be persisted")
	assert.EqualValues(t, 1, counted.transactions.Load())
	noWork()
	// Missing and stale observations must not enter the transaction either.
	results = nil
	noWork()
	results = map[string]monitor.Result{monitors[0].Id: {SampleCount: 60, LastProbeAt: time.Now().Add(-10 * time.Minute).UnixMilli()}}
	noWork()
	results[monitors[0].Id] = monitorResult(0)
	evaluate()
	noWork()
	require.NoError(t, hub.Delete(alert))
	noWork()
}

func TestNetworkMonitorAlertConfigCacheInvalidation(t *testing.T) {
	hub, system, alert, monitors := networkAlertSetup(t)
	am := alerts.NewTestAlertManagerWithoutWorker(hub)
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(0)}
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	count := hub.TestMailer.TotalSend()
	// Widening the interval makes this observation fresh. A stale interval cache
	// would miss the failure indefinitely, even though results keep arriving.
	result := monitorResult(10)
	result.LastProbeAt = time.Now().Add(-4 * time.Minute).UnixMilli()
	results[monitors[0].Id] = result
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count, hub.TestMailer.TotalSend())
	monitors[0].Set("interval", 120)
	require.NoError(t, hub.Save(monitors[0]))
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count+1, hub.TestMailer.TotalSend())
	// Disable, then re-enable the same ID: its new failure must be detected.
	monitors[0].Set("enabled", false)
	require.NoError(t, hub.Save(monitors[0]))
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	monitors[0].Set("enabled", true)
	require.NoError(t, hub.Save(monitors[0]))
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count+2, hub.TestMailer.TotalSend())
	// A new monitor must also become eligible without restarting the hub.
	created, err := beszelTests.CreateRecord(hub, "network_monitors", map[string]any{
		"system": system.Id, "name": "new", "target": "new.example.com", "protocol": "icmp", "interval": 60, "enabled": true,
	})
	require.NoError(t, err)
	results[created.Id] = monitorResult(10)
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count+3, hub.TestMailer.TotalSend())
	// Threshold changes refresh cached config and preserve the active incidents.
	alert, err = hub.FindRecordById("alerts", alert.Id)
	require.NoError(t, err)
	alert.Set("value", 15)
	require.NoError(t, hub.Save(alert))
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count+5, hub.TestMailer.TotalSend())
}

func TestNetworkMonitorAlertRevalidatesCandidate(t *testing.T) {
	for _, change := range []string{"threshold", "disable alert", "disable monitor", "down"} {
		t.Run(change, func(t *testing.T) {
			hub, system, alert, monitors := networkAlertSetup(t)
			counted := &monitorCountingHub{TestHub: hub}
			am := alerts.NewTestAlertManagerWithoutWorker(counted)
			results := map[string]monitor.Result{monitors[0].Id: monitorResult(0)}
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
			count := hub.TestMailer.TotalSend()
			// Change the DB after the cache predicts a transition, before its transaction.
			counted.beforeTransaction = func() {
				counted.beforeTransaction = nil
				switch change {
				case "threshold":
					alert.Set("value", 20)
					require.NoError(t, hub.Save(alert))
				case "disable alert":
					require.NoError(t, hub.Delete(alert))
				case "disable monitor":
					monitors[0].Set("enabled", false)
					require.NoError(t, hub.Save(monitors[0]))
				case "down":
					_, err := hub.DB().Update("systems", dbx.Params{"status": "down"}, dbx.HashExp{"id": system.Id}).Execute()
					require.NoError(t, err)
				}
			}
			results[monitors[0].Id] = monitorResult(10)
			require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
			assert.EqualValues(t, 1, counted.transactions.Load())
			assert.Equal(t, count, hub.TestMailer.TotalSend())
			histories, err := hub.CountRecords("alerts_history")
			require.NoError(t, err)
			assert.Zero(t, histories)
		})
	}
}

func TestNetworkMonitorAlertConcurrentEvaluations(t *testing.T) {
	hub, system, _, monitors := networkAlertSetup(t)
	am := alerts.NewTestAlertManagerWithoutWorker(hub)
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(10)}
	count := hub.TestMailer.TotalSend()
	var wg sync.WaitGroup
	errs := make(chan error, 8)
	for range 8 {
		wg.Go(func() { errs <- am.HandleNetworkMonitorAlerts(system, results) })
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		require.NoError(t, err)
	}
	assert.Equal(t, count+1, hub.TestMailer.TotalSend())
}

func TestNetworkMonitorAlertCacheAfterRollback(t *testing.T) {
	hub, system, _, monitors := networkAlertSetup(t)
	am := alerts.NewTestAlertManagerWithoutWorker(hub)
	results := map[string]monitor.Result{monitors[0].Id: monitorResult(0)}
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	_, err := hub.DB().NewQuery(`CREATE TRIGGER fail_alert BEFORE UPDATE ON alerts BEGIN SELECT RAISE(ABORT, 'test rollback'); END`).Execute()
	require.NoError(t, err)
	count := hub.TestMailer.TotalSend()
	results[monitors[0].Id] = monitorResult(10)
	require.Error(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count, hub.TestMailer.TotalSend())
	histories, err := hub.CountRecords("alerts_history")
	require.NoError(t, err)
	assert.Zero(t, histories)
	_, err = hub.DB().NewQuery("DROP TRIGGER fail_alert").Execute()
	require.NoError(t, err)
	// A failed transition must not be published to the cache and mask the retry.
	require.NoError(t, am.HandleNetworkMonitorAlerts(system, results))
	assert.Equal(t, count+1, hub.TestMailer.TotalSend())
}