From 097180e8d78ca114ad5e22cc701f6b4acb5a3328 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Martin=20Stenr=C3=B6se?= <22470448+martinstenrose@users.noreply.github.com> Date: Wed, 2 Sep 2026 02:41:48 +0200 Subject: [PATCH] feat(alerts): add alert for failed systemd services (#2173) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a user-configurable "Failed Services" alert that notifies when any tracked systemd service enters the failed state, and again when all services recover. --------- Signed-off-by: Martin Stenröse Co-authored-by: henrygd --- agent/agent.go | 6 + internal/alerts/alerts.go | 3 + internal/alerts/alerts_system.go | 19 +- internal/alerts/alerts_systemd.go | 183 +++++++++ internal/alerts/alerts_systemd_test.go | 383 ++++++++++++++++++ internal/alerts/alerts_test_helpers.go | 4 + internal/entities/system/system.go | 3 + internal/entities/system/system_test.go | 22 + internal/hub/systems/system.go | 22 +- internal/hub/systems/systemd_records_test.go | 126 ++++++ internal/hub/systems/systems_test_helpers.go | 5 + .../migrations/1_add_systemd_failed_alert.go | 44 ++ .../site/src/components/active-alerts.tsx | 4 +- .../src/components/alerts-history-columns.tsx | 6 +- .../src/components/alerts/alerts-sheet.tsx | 98 +++-- internal/site/src/lib/alerts.ts | 11 +- internal/site/src/types.d.ts | 4 + 17 files changed, 891 insertions(+), 52 deletions(-) create mode 100644 internal/alerts/alerts_systemd.go create mode 100644 internal/alerts/alerts_systemd_test.go create mode 100644 internal/hub/systems/systemd_records_test.go create mode 100644 internal/migrations/1_add_systemd_failed_alert.go diff --git a/agent/agent.go b/agent/agent.go index b3598f4e..dcabdb92 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -201,6 +201,12 @@ func (a *Agent) gatherStats(options common.DataRequestOptions) *system.CombinedD } if a.systemdManager.hasFreshStats { data.SystemdServices = a.systemdManager.getServiceStats(nil, false) + data.SystemdServicesUpdated = true + // Preserve an explicit zero count so the hub can distinguish a fresh + // empty snapshot from a response that omitted systemd data. + if totalCount == 0 { + data.Info.Services = []uint16{0, 0} + } } } diff --git a/internal/alerts/alerts.go b/internal/alerts/alerts.go index bbf6452e..35739702 100644 --- a/internal/alerts/alerts.go +++ b/internal/alerts/alerts.go @@ -129,6 +129,9 @@ func (am *AlertManager) bindEvents() { if err := resolveStatusAlerts(e.App); err != nil { e.App.Logger().Error("Failed to resolve stale status alerts", "err", err) } + if err := resolveSystemdAlerts(e.App); err != nil { + e.App.Logger().Error("Failed to resolve stale systemd alerts", "err", err) + } if err := am.restorePendingStatusAlerts(); err != nil { e.App.Logger().Error("Failed to restore pending status alerts", "err", err) } diff --git a/internal/alerts/alerts_system.go b/internal/alerts/alerts_system.go index 3277dd53..11226b8c 100644 --- a/internal/alerts/alerts_system.go +++ b/internal/alerts/alerts_system.go @@ -37,7 +37,24 @@ func cpuStateAlertValue(name string, breakdown []float64) (float64, bool) { } func (am *AlertManager) HandleSystemAlerts(systemRecord *core.Record, data *system.CombinedData) error { - alerts := am.alertsCache.GetAlertsExcludingNames(systemRecord.Id, "Status") + // Systemd alerts are binary state, not numeric thresholds, so they're handled + // separately. They read their own state from the database and don't use data. + // Read the confirmed empty state from the record being saved instead of data: + // dashboard polling can replace the system's in-memory payload concurrently. + var currentInfo system.Info + confirmedEmptySnapshot := false + if err := systemRecord.UnmarshalJSONField("info", ¤tInfo); err == nil { + confirmedEmptySnapshot = len(currentInfo.Services) > 0 && currentInfo.Services[0] == 0 + } + if err := am.HandleSystemdAlerts(systemRecord, confirmedEmptySnapshot); err != nil { + am.hub.Logger().Error("Error handling systemd alerts", "err", err) + } + + if data == nil { + return nil + } + + alerts := am.alertsCache.GetAlertsExcludingNames(systemRecord.Id, "Status", alertNameSystemdFailed) if len(alerts) == 0 { return nil } diff --git a/internal/alerts/alerts_systemd.go b/internal/alerts/alerts_systemd.go new file mode 100644 index 00000000..567cbf4b --- /dev/null +++ b/internal/alerts/alerts_systemd.go @@ -0,0 +1,183 @@ +package alerts + +import ( + "fmt" + "strings" + + "github.com/henrygd/beszel/internal/entities/systemd" + + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" +) + +// alertNameSystemdFailed is the alerts.name value for the failed systemd services alert. +const alertNameSystemdFailed = "SystemdFailed" + +// maxListedServices caps how many service names are listed in a notification body. +const maxListedServices = 10 + +// HandleSystemdAlerts manages alerts for systemd services in the failed state. +// +// This is a binary state alert and fires on the first observation of a failed +// service rather than using a delay. The agent only refreshes systemd state every +// 10 minutes, so a shorter delay could never observe new data before expiring, and +// that poll interval already hides services that fail and restart quickly. +func (am *AlertManager) HandleSystemdAlerts(systemRecord *core.Record, confirmedEmptySnapshot bool) error { + alerts := am.alertsCache.GetAlertsByName(systemRecord.Id, alertNameSystemdFailed) + if len(alerts) == 0 { + return nil + } + + // State is read from the systemd_services snapshot rather than the update payload. + // The payload is not a reliable source here: realtime dashboard subscriptions fetch + // from the agent with a shorter cache time, and the agent omits systemd services from + // those responses, overwriting the cached payload roughly once a second while a system + // is being viewed. The snapshot table is only written by the full update cycle. + total, failed, err := am.queryServiceStates(systemRecord.Id) + if err != nil { + return err + } + // No rows normally means no systemd data for this system (agent without systemd, + // or not yet reported), which must not be treated as a recovery. A fresh snapshot + // marker disambiguates that case from an agent explicitly reporting zero services. + if total == 0 && !confirmedEmptySnapshot { + return nil + } + + systemName := systemRecord.GetString("name") + + for _, alertData := range alerts { + triggered := len(failed) > 0 + // Only notify on a change of state, so a service that stays failed across + // cycles doesn't re-notify every update. + if triggered == alertData.Triggered { + continue + } + if err := am.sendSystemdAlert(triggered, systemName, alertData, failed); err != nil { + am.hub.Logger().Error("Failed to send alert", "err", err) + } + } + return nil +} + +// queryServiceStates returns the number of services reported in the most recent update +// for a system, and the names of those in the failed state. +// +// Rows are restricted to the latest update because systemd_services is upserted, never +// pruned on change: a service that no longer exists on the host stops being reported and +// its row keeps its last known state until the retention sweep removes it. Every row +// written in one cycle shares a single updated timestamp, so the newest timestamp +// identifies exactly the services the agent last reported. +func (am *AlertManager) queryServiceStates(systemID string) (total int, failed []string, err error) { + var rows []struct { + Name string `db:"name"` + State systemd.ServiceState `db:"state"` + } + err = am.hub.DB(). + Select("name", "state"). + From("systemd_services"). + Where(dbx.NewExp( + "system={:system} AND updated=(SELECT MAX(updated) FROM systemd_services WHERE system={:system})", + dbx.Params{"system": systemID}, + )). + OrderBy("name"). + All(&rows) + if err != nil { + return 0, nil, err + } + for _, row := range rows { + if row.State == systemd.StatusFailed { + failed = append(failed, row.Name) + } + } + return len(rows), failed, nil +} + +// sendSystemdAlert sends a failed or recovered systemd services alert to the alert's user. +func (am *AlertManager) sendSystemdAlert(triggered bool, systemName string, alertData CachedAlertData, failed []string) error { + // Update trigger state for alert record before sending alert + if err := am.setAlertTriggered(alertData, triggered); err != nil { + return err + } + + var title, message string + if triggered { + title = fmt.Sprintf("Failed services on %s %v", systemName, "\U0001F534") // Red alert emoji + message = fmt.Sprintf("%s on %s: %s", pluralizeServices(len(failed)), systemName, formatServiceList(failed)) + } else { + title = fmt.Sprintf("Services recovered on %s %v", systemName, "✅") // Green checkmark emoji + message = fmt.Sprintf("No services are in the failed state on %s.", systemName) + } + + systemID := alertData.SystemID + + return am.SendAlert(AlertMessageData{ + UserID: alertData.UserID, + SystemID: systemID, + Title: title, + Message: message, + Link: am.hub.MakeLink("system", systemID), + LinkText: "View " + systemName, + }) +} + +// pluralizeServices returns a count label like "1 failed service" or "3 failed services". +func pluralizeServices(count int) string { + if count == 1 { + return "1 failed service" + } + return fmt.Sprintf("%d failed services", count) +} + +// formatServiceList joins service names, truncating long lists. +func formatServiceList(names []string) string { + if len(names) <= maxListedServices { + return strings.Join(names, ", ") + } + remaining := len(names) - maxListedServices + return fmt.Sprintf("%s and %d more", strings.Join(names[:maxListedServices], ", "), remaining) +} + +// resolveSystemdAlerts resolves triggered systemd alerts for systems that no longer +// have any failed services. This clears stale state left by a hub restart. +func resolveSystemdAlerts(app core.App) error { + db := app.DB() + var alertIds []string + err := db.NewQuery(` + SELECT a.id + FROM alerts a + JOIN systems sys ON sys.id = a.system + WHERE a.name = {:name} + AND a.triggered = true + AND ( + EXISTS ( + SELECT 1 FROM systemd_services cur + WHERE cur.system = a.system + AND cur.updated = (SELECT MAX(updated) FROM systemd_services WHERE system = a.system) + ) + OR json_extract(sys.info, '$.sv[0]') = 0 + ) + AND NOT EXISTS ( + SELECT 1 FROM systemd_services s + WHERE s.system = a.system AND s.state = {:state} + AND s.updated = (SELECT MAX(updated) FROM systemd_services WHERE system = a.system) + ) + `).Bind(dbx.Params{ + "name": alertNameSystemdFailed, + "state": systemd.StatusFailed, + }).Column(&alertIds) + if err != nil { + return err + } + for _, alertId := range alertIds { + alert, err := app.FindRecordById("alerts", alertId) + if err != nil { + return err + } + alert.Set("triggered", false) + if err := app.Save(alert); err != nil { + return err + } + } + return nil +} diff --git a/internal/alerts/alerts_systemd_test.go b/internal/alerts/alerts_systemd_test.go new file mode 100644 index 00000000..b6e6ee28 --- /dev/null +++ b/internal/alerts/alerts_systemd_test.go @@ -0,0 +1,383 @@ +//go:build testing + +package alerts_test + +import ( + "testing" + "time" + + "github.com/henrygd/beszel/internal/alerts" + systemEntity "github.com/henrygd/beszel/internal/entities/system" + "github.com/henrygd/beszel/internal/entities/systemd" + beszelTests "github.com/henrygd/beszel/internal/tests" + "github.com/pocketbase/dbx" + "github.com/pocketbase/pocketbase/core" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// setSystemdServiceState upserts a systemd_services row mirroring the raw SQL write +// path used by the hub (createSystemdStatsRecords), which bypasses record hooks. +func setSystemdServiceState(t *testing.T, hub core.App, systemID, name string, state systemd.ServiceState, updated int64) { + t.Helper() + + _, err := hub.DB().NewQuery( + "INSERT INTO systemd_services (id, system, name, state, sub, cpu, cpuPeak, memory, memPeak, updated) " + + "VALUES ({:id}, {:system}, {:name}, {:state}, 0, 0, 0, 0, 0, {:updated}) " + + "ON CONFLICT(id) DO UPDATE SET state = excluded.state, updated = excluded.updated", + ).Bind(dbx.Params{ + "id": systemID + "-" + name, + "system": systemID, + "name": name, + "state": state, + "updated": updated, + }).Execute() + require.NoError(t, err) +} + +// seedServices writes a set of services into the systemd_services snapshot, which is the +// source HandleSystemdAlerts reads from. All rows share one updated timestamp, matching +// how the hub writes a batch in createSystemdStatsRecords. +func seedServices(t *testing.T, hub core.App, systemID string, states ...systemd.ServiceState) { + t.Helper() + seedServicesAt(t, hub, systemID, time.Now().UTC().UnixMilli(), states...) +} + +// seedServicesAt writes services with an explicit batch timestamp. +func seedServicesAt(t *testing.T, hub core.App, systemID string, updated int64, states ...systemd.ServiceState) { + t.Helper() + for i, state := range states { + setSystemdServiceState(t, hub, systemID, serviceName(i), state, updated) + } +} + +func serviceName(i int) string { + return string(rune('a'+i)) + ".service" +} + +// systemdTestSetup creates a user with an email, a system, and a SystemdFailed alert. +func systemdTestSetup(t *testing.T, triggered bool) (*beszelTests.TestHub, *core.Record, *core.Record) { + t.Helper() + + hub, user := beszelTests.GetHubWithUser(t) + + userSettings, err := hub.FindFirstRecordByFilter("user_settings", "user={:user}", map[string]any{"user": user.Id}) + require.NoError(t, err) + userSettings.Set("settings", `{"emails":["test@example.com"],"webhooks":[]}`) + require.NoError(t, hub.Save(userSettings)) + + // "paused" avoids spawning a background updater goroutine that would outlive + // the test hub; these tests drive HandleSystemdAlerts directly. + systems, err := beszelTests.CreateSystems(hub, 1, user.Id, "paused") + require.NoError(t, err) + system := systems[0] + + alert, err := beszelTests.CreateRecord(hub, "alerts", map[string]any{ + "name": "SystemdFailed", + "system": system.Id, + "user": user.Id, + "triggered": triggered, + }) + require.NoError(t, err) + + return hub, system, alert +} + +func TestSystemdAlertFiresImmediately(t *testing.T) { + hub, system, alert := systemdTestSetup(t, false) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + seedServices(t, hub, system.Id, systemd.StatusFailed, systemd.StatusActive) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend(), "failed service should notify on first observation") + + messages := hub.TestMailer.Messages() + require.NotEmpty(t, messages) + last := messages[len(messages)-1] + assert.Contains(t, last.Subject, "Failed services") + assert.Contains(t, last.Text, "a.service", "notification should name the failed service") + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.True(t, alertRecord.GetBool("triggered"), "alert should be marked triggered") + + // history record should be created via the alerts update hook + historyCount, err := hub.CountRecords("alerts_history", dbx.HashExp{"resolved": ""}) + require.NoError(t, err) + assert.EqualValues(t, 1, historyCount, "should have one unresolved alert history record") +} + +func TestSystemdAlertFullCycle(t *testing.T) { + hub, system, alert := systemdTestSetup(t, false) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + // Fail, then recover. + seedServices(t, hub, system.Id, systemd.StatusFailed) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + seedServices(t, hub, system.Id, systemd.StatusActive) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + assert.Equal(t, initialEmailCount+2, hub.TestMailer.TotalSend(), "should send a failure and a recovery notification") + + messages := hub.TestMailer.Messages() + require.Len(t, messages, 2) + assert.Contains(t, messages[0].Subject, "Failed services") + assert.Contains(t, messages[1].Subject, "Services recovered") + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "alert should be cleared after recovery") + + // history record should be resolved + historyCount, err := hub.CountRecords("alerts_history", dbx.HashExp{"resolved": ""}) + require.NoError(t, err) + assert.Zero(t, historyCount, "alert history record should be resolved") +} + +func TestSystemdAlertSendsRecoveryWhenTriggered(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + seedServices(t, hub, system.Id, systemd.StatusActive, systemd.StatusInactive) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend(), "recovery notification should be sent") + messages := hub.TestMailer.Messages() + require.NotEmpty(t, messages) + assert.Contains(t, messages[len(messages)-1].Subject, "Services recovered") + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "alert should be cleared after recovery") +} + +func TestSystemdAlertDoesNotResendWhileTriggered(t *testing.T) { + hub, system, _ := systemdTestSetup(t, true) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + // Still failing across several cycles — should not re-notify. + for range 3 { + seedServices(t, hub, system.Id, systemd.StatusFailed) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + } + + assert.Equal(t, initialEmailCount, hub.TestMailer.TotalSend(), "should not re-notify while still triggered") +} + +func TestSystemdAlertRepeatedFailureNotifiesOnce(t *testing.T) { + hub, system, _ := systemdTestSetup(t, false) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + for range 3 { + seedServices(t, hub, system.Id, systemd.StatusFailed) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + } + + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend(), "repeated failures should only notify once") +} + +// A service that no longer exists on the host stops being reported, but its row stays +// in systemd_services with its last known state until the retention sweep. That stale +// row must not keep the alert triggered. +func TestSystemdAlertIgnoresServicesNoLongerReported(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + now := time.Now().UTC().UnixMilli() + // Older batch still holding a failed service that has since been removed. + setSystemdServiceState(t, hub, system.Id, "gone.service", systemd.StatusFailed, now-60_000) + // Current batch reports only healthy services. + seedServicesAt(t, hub, system.Id, now, systemd.StatusActive, systemd.StatusActive) + + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend(), "stale failed row should not block recovery") + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "alert should resolve once the service stops being reported") +} + +func TestResolveSystemdAlertsIgnoresStaleFailedRows(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + now := time.Now().UTC().UnixMilli() + setSystemdServiceState(t, hub, system.Id, "gone.service", systemd.StatusFailed, now-60_000) + seedServicesAt(t, hub, system.Id, now, systemd.StatusActive) + + require.NoError(t, alerts.ResolveSystemdAlerts(hub)) + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "stale failed row should not keep the alert triggered") +} + +func TestSystemdAlertNoSystemdDataIsIgnored(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + // A system with no systemd_services rows (agent without systemd, or nothing + // reported yet) must not be treated as a recovery. + require.NoError(t, am.HandleSystemdAlerts(system, false)) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + assert.Equal(t, initialEmailCount, hub.TestMailer.TotalSend(), "missing systemd data should not send a recovery") + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.True(t, alertRecord.GetBool("triggered"), "triggered state should be preserved when data is absent") +} + +func TestSystemdAlertFreshEmptySnapshotResolves(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + // An explicit zero service count on the saved system record distinguishes a + // confirmed empty snapshot from an agent response that omitted systemd data. + system.Set("info", systemEntity.Info{Services: []uint16{0, 0}}) + require.NoError(t, am.HandleSystemAlerts(system, nil)) + + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend(), "fresh empty snapshot should send a recovery") + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "fresh empty snapshot should resolve the alert") +} + +func TestSystemdAlertNoAlertRecord(t *testing.T) { + hub, user := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + systems, err := beszelTests.CreateSystems(hub, 1, user.Id, "paused") + require.NoError(t, err) + system := systems[0] + + initialEmailCount := hub.TestMailer.TotalSend() + am := alerts.NewTestAlertManagerWithoutWorker(hub) + + seedServices(t, hub, system.Id, systemd.StatusFailed) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + assert.Equal(t, initialEmailCount, hub.TestMailer.TotalSend(), "no email when no alert record exists") +} + +func TestResolveSystemdAlertsClearsStaleTriggered(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + // No failed services in the snapshot, but the alert is still marked triggered + // (e.g. the hub restarted while the alert was active). + setSystemdServiceState(t, hub, system.Id, "a.service", systemd.StatusActive, time.Now().UTC().UnixMilli()) + + require.NoError(t, alerts.ResolveSystemdAlerts(hub)) + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "stale triggered flag should be cleared") +} + +func TestResolveSystemdAlertsKeepsTriggeredWithoutSystemdData(t *testing.T) { + hub, _, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + // Missing rows do not prove recovery. This can happen when a system is offline + // and its last service snapshot has been removed by retention. + require.NoError(t, alerts.ResolveSystemdAlerts(hub)) + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.True(t, alertRecord.GetBool("triggered"), "missing systemd data should preserve triggered state") +} + +func TestResolveSystemdAlertsClearsConfirmedEmptySnapshot(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + // Update the persisted snapshot directly so record hooks don't alter alert state + // before the startup resolver is exercised. + _, err := hub.DB().NewQuery( + "UPDATE systems SET info = {:info} WHERE id = {:id}", + ).Bind(dbx.Params{"info": `{"sv":[0,0]}`, "id": system.Id}).Execute() + require.NoError(t, err) + require.NoError(t, alerts.ResolveSystemdAlerts(hub)) + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alertRecord.GetBool("triggered"), "confirmed empty snapshot should clear triggered state") +} + +func TestResolveSystemdAlertsKeepsStillFailing(t *testing.T) { + hub, system, alert := systemdTestSetup(t, true) + defer hub.Cleanup() + + setSystemdServiceState(t, hub, system.Id, "a.service", systemd.StatusFailed, time.Now().UTC().UnixMilli()) + + require.NoError(t, alerts.ResolveSystemdAlerts(hub)) + + alertRecord, err := hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.True(t, alertRecord.GetBool("triggered"), "alert should stay triggered while a service is still failed") +} + +func TestSystemdAlertMultipleUsersRespectOwnAlerts(t *testing.T) { + hub, user1 := beszelTests.GetHubWithUser(t) + defer hub.Cleanup() + + setStatusAlertEmail(t, hub, user1.Id, "user1@example.com") + + user2, err := beszelTests.CreateUser(hub, "user2@example.com", "password") + require.NoError(t, err) + _, err = beszelTests.CreateRecord(hub, "user_settings", map[string]any{ + "user": user2.Id, + "settings": map[string]any{ + "emails": []string{"user2@example.com"}, + "webhooks": []string{}, + }, + }) + require.NoError(t, err) + + system, err := beszelTests.CreateRecord(hub, "systems", map[string]any{ + "name": "shared-system", + "users": []string{user1.Id, user2.Id}, + "host": "127.0.0.1", + }) + require.NoError(t, err) + + for _, user := range []*core.Record{user1, user2} { + _, err = beszelTests.CreateRecord(hub, "alerts", map[string]any{ + "name": "SystemdFailed", + "system": system.Id, + "user": user.Id, + }) + require.NoError(t, err) + } + + am := alerts.NewTestAlertManagerWithoutWorker(hub) + seedServices(t, hub, system.Id, systemd.StatusFailed) + require.NoError(t, am.HandleSystemdAlerts(system, false)) + + messages := hub.TestMailer.Messages() + require.Len(t, messages, 2, "each user should receive their own alert") +} diff --git a/internal/alerts/alerts_test_helpers.go b/internal/alerts/alerts_test_helpers.go index 6567f76a..017f3edb 100644 --- a/internal/alerts/alerts_test_helpers.go +++ b/internal/alerts/alerts_test_helpers.go @@ -88,6 +88,10 @@ func ResolveStatusAlerts(app core.App) error { return resolveStatusAlerts(app) } +func ResolveSystemdAlerts(app core.App) error { + return resolveSystemdAlerts(app) +} + func (am *AlertManager) RestorePendingStatusAlerts() error { return am.restorePendingStatusAlerts() } diff --git a/internal/entities/system/system.go b/internal/entities/system/system.go index 66858397..3ccaea32 100644 --- a/internal/entities/system/system.go +++ b/internal/entities/system/system.go @@ -204,4 +204,7 @@ type CombinedData struct { Containers []*container.Stats `json:"container" cbor:"2,keyasint"` SystemdServices []*systemd.Service `json:"systemd,omitempty" cbor:"3,keyasint,omitempty"` Details *Details `cbor:"4,keyasint,omitempty"` + // SystemdServicesUpdated distinguishes a fresh empty snapshot from a response + // that omitted systemd data (for example, a short-cache dashboard request). + SystemdServicesUpdated bool `json:"systemdUpdated,omitempty" cbor:"5,keyasint,omitempty"` } diff --git a/internal/entities/system/system_test.go b/internal/entities/system/system_test.go index 665019cb..f75fecd6 100644 --- a/internal/entities/system/system_test.go +++ b/internal/entities/system/system_test.go @@ -63,3 +63,25 @@ func TestStatsLegacyBatteryPayload(t *testing.T) { assert.Contains(t, payload, "bat") assert.NotContains(t, payload, "bats") } + +func TestCombinedDataSystemdUpdateMarkerTransport(t *testing.T) { + data := CombinedData{SystemdServicesUpdated: true} + + jsonData, err := json.Marshal(data) + require.NoError(t, err) + var decodedJSON CombinedData + require.NoError(t, json.Unmarshal(jsonData, &decodedJSON)) + assert.True(t, decodedJSON.SystemdServicesUpdated) + assert.Empty(t, decodedJSON.SystemdServices) + + cborData, err := cbor.Marshal(data) + require.NoError(t, err) + var decodedCBOR CombinedData + require.NoError(t, cbor.Unmarshal(cborData, &decodedCBOR)) + assert.True(t, decodedCBOR.SystemdServicesUpdated) + assert.Empty(t, decodedCBOR.SystemdServices) + + var legacy CombinedData + require.NoError(t, json.Unmarshal([]byte(`{"stats":{},"info":{},"container":[]}`), &legacy)) + assert.False(t, legacy.SystemdServicesUpdated) +} diff --git a/internal/hub/systems/system.go b/internal/hub/systems/system.go index 43955b52..1bcdb8bf 100644 --- a/internal/hub/systems/system.go +++ b/internal/hub/systems/system.go @@ -250,8 +250,10 @@ func (sys *System) createRecords(data *system.CombinedData) (*core.Record, error } } - // add new systemd_stats record - if len(data.SystemdServices) > 0 { + // Update systemd service records when the agent reports a fresh snapshot. + // The length check keeps snapshots from older agents working, while the + // explicit marker lets newer agents report that a fresh snapshot is empty. + if data.SystemdServicesUpdated || len(data.SystemdServices) > 0 { if err := createSystemdStatsRecords(txApp, data.SystemdServices, sys.Id); err != nil { return err } @@ -307,7 +309,10 @@ func createSystemDetailsRecord(app core.App, data *system.Details, systemId stri func createSystemdStatsRecords(app core.App, data []*systemd.Service, systemId string) error { if len(data) == 0 { - return nil + _, err := app.DB().NewQuery( + "DELETE FROM systemd_services WHERE system = {:system}", + ).Bind(dbx.Params{"system": systemId}).Execute() + return err } // shared params for all records params := dbx.Params{ @@ -332,7 +337,16 @@ func createSystemdStatsRecords(app core.App, data []*systemd.Service, systemId s "INSERT INTO systemd_services (id, system, name, state, sub, cpu, cpuPeak, memory, memPeak, updated) VALUES %s ON CONFLICT(id) DO UPDATE SET system = excluded.system, name = excluded.name, state = excluded.state, sub = excluded.sub, cpu = excluded.cpu, cpuPeak = excluded.cpuPeak, memory = excluded.memory, memPeak = excluded.memPeak, updated = excluded.updated", strings.Join(valueStrings, ","), ) - _, err := app.DB().NewQuery(queryString).Bind(params).Execute() + if _, err := app.DB().NewQuery(queryString).Bind(params).Execute(); err != nil { + return err + } + // Remove services the agent no longer reports. Every row in this batch shares the + // same updated timestamp, so anything older no longer exists on the host. Left in + // place these rows survive until the retention sweep and surface inconsistently + // across the dashboard, the services table, and alerts. + _, err := app.DB().NewQuery( + "DELETE FROM systemd_services WHERE system = {:system} AND updated < {:updated}", + ).Bind(dbx.Params{"system": systemId, "updated": params["updated"]}).Execute() return err } diff --git a/internal/hub/systems/systemd_records_test.go b/internal/hub/systems/systemd_records_test.go new file mode 100644 index 00000000..91af6f48 --- /dev/null +++ b/internal/hub/systems/systemd_records_test.go @@ -0,0 +1,126 @@ +//go:build testing + +package systems_test + +import ( + "testing" + "time" + + "github.com/henrygd/beszel/internal/entities/system" + "github.com/henrygd/beszel/internal/entities/systemd" + "github.com/henrygd/beszel/internal/hub/systems" + "github.com/henrygd/beszel/internal/tests" + "github.com/pocketbase/dbx" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCreateRecordsHandlesSystemdAlertLifecycle(t *testing.T) { + hub, user := tests.GetHubWithUser(t) + defer hub.Cleanup() + + settings, err := hub.FindFirstRecordByFilter("user_settings", "user={:user}", dbx.Params{"user": user.Id}) + require.NoError(t, err) + settings.Set("settings", `{"emails":["test@example.com"],"webhooks":[]}`) + require.NoError(t, hub.Save(settings)) + + systemRecords, err := tests.CreateSystems(hub, 1, user.Id, "paused") + require.NoError(t, err) + systemRecord := systemRecords[0] + alert, err := tests.CreateRecord(hub, "alerts", map[string]any{ + "name": "SystemdFailed", + "system": systemRecord.Id, + "user": user.Id, + }) + require.NoError(t, err) + + monitoredSystem, err := hub.GetSystemManager().GetSystem(systemRecord.Id) + require.NoError(t, err) + initialEmailCount := hub.TestMailer.TotalSend() + + // Exercise the production path: persist the snapshot transactionally, save the + // system record, and let its update hook evaluate and deliver the alert. + _, err = monitoredSystem.CreateRecords(&system.CombinedData{ + Info: system.Info{Services: []uint16{1, 1}}, + SystemdServicesUpdated: true, + SystemdServices: []*systemd.Service{ + {Name: "failed.service", State: systemd.StatusFailed}, + }, + }) + require.NoError(t, err) + + alert, err = hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.True(t, alert.GetBool("triggered")) + assert.Equal(t, initialEmailCount+1, hub.TestMailer.TotalSend()) + serviceCount, err := hub.CountRecords("systemd_services", dbx.HashExp{"system": systemRecord.Id}) + require.NoError(t, err) + assert.EqualValues(t, 1, serviceCount) + unresolvedCount, err := hub.CountRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id, "resolved": ""}) + require.NoError(t, err) + assert.EqualValues(t, 1, unresolvedCount) + + // A fresh empty snapshot must delete the old failed row and resolve the alert. + _, err = monitoredSystem.CreateRecords(&system.CombinedData{ + Info: system.Info{Services: []uint16{0, 0}}, + SystemdServicesUpdated: true, + }) + require.NoError(t, err) + + alert, err = hub.FindRecordById("alerts", alert.Id) + require.NoError(t, err) + assert.False(t, alert.GetBool("triggered")) + assert.Equal(t, initialEmailCount+2, hub.TestMailer.TotalSend()) + serviceCount, err = hub.CountRecords("systemd_services", dbx.HashExp{"system": systemRecord.Id}) + require.NoError(t, err) + assert.Zero(t, serviceCount) + unresolvedCount, err = hub.CountRecords("alerts_history", dbx.HashExp{"alert_id": alert.Id, "resolved": ""}) + require.NoError(t, err) + assert.Zero(t, unresolvedCount) +} + +// createSystemdStatsRecords upserts the reported services and must drop rows for +// services the agent has stopped reporting, so a unit removed from the host doesn't +// linger with its last known state until the retention sweep. +func TestCreateSystemdStatsRecordsRemovesUnreportedServices(t *testing.T) { + hub, err := tests.NewTestHub(t.TempDir()) + require.NoError(t, err) + defer hub.Cleanup() + + user, err := tests.CreateUser(hub, "test@example.com", "password") + require.NoError(t, err) + system, err := tests.CreateRecord(hub, "systems", map[string]any{ + "name": "test-system", + "host": "127.0.0.1", + "users": []string{user.Id}, + }) + require.NoError(t, err) + + serviceNames := func() []string { + var out []string + require.NoError(t, hub.DB().Select("name").From("systemd_services"). + Where(dbx.NewExp("system={:s}", dbx.Params{"s": system.Id})). + OrderBy("name").Column(&out)) + return out + } + + require.NoError(t, systems.CreateSystemdStatsRecords(hub, []*systemd.Service{ + {Name: "a.service", State: systemd.StatusActive}, + {Name: "gone.service", State: systemd.StatusFailed}, + }, system.Id)) + assert.Equal(t, []string{"a.service", "gone.service"}, serviceNames()) + + // Batches are stamped with millisecond precision and update cycles are a minute + // apart in practice; ensure the next batch gets a distinct timestamp. + time.Sleep(2 * time.Millisecond) + + // gone.service is no longer reported, so its row must not survive. + require.NoError(t, systems.CreateSystemdStatsRecords(hub, []*systemd.Service{ + {Name: "a.service", State: systemd.StatusActive}, + }, system.Id)) + assert.Equal(t, []string{"a.service"}, serviceNames()) + + // A fresh empty snapshot means the agent no longer reports any services. + require.NoError(t, systems.CreateSystemdStatsRecords(hub, nil, system.Id)) + assert.Empty(t, serviceNames()) +} diff --git a/internal/hub/systems/systems_test_helpers.go b/internal/hub/systems/systems_test_helpers.go index 5d69ac12..ca61895f 100644 --- a/internal/hub/systems/systems_test_helpers.go +++ b/internal/hub/systems/systems_test_helpers.go @@ -7,6 +7,7 @@ import ( "fmt" entities "github.com/henrygd/beszel/internal/entities/system" + "github.com/henrygd/beszel/internal/entities/systemd" "github.com/pocketbase/pocketbase/core" ) @@ -134,3 +135,7 @@ func (s *System) CreateRecords(data *entities.CombinedData) (*core.Record, error s.data = data return s.createRecords(data) } + +func CreateSystemdStatsRecords(app core.App, data []*systemd.Service, systemId string) error { + return createSystemdStatsRecords(app, data, systemId) +} diff --git a/internal/migrations/1_add_systemd_failed_alert.go b/internal/migrations/1_add_systemd_failed_alert.go new file mode 100644 index 00000000..ca8f7810 --- /dev/null +++ b/internal/migrations/1_add_systemd_failed_alert.go @@ -0,0 +1,44 @@ +package migrations + +import ( + "errors" + "slices" + + "github.com/pocketbase/pocketbase/core" + m "github.com/pocketbase/pocketbase/migrations" +) + +// alertNameSystemdFailed is the alerts.name select value for the +// "failed systemd services" alert type. +const alertNameSystemdFailed = "SystemdFailed" + +func init() { + m.Register(func(app core.App) error { + return updateAlertNameValues(app, func(values []string) []string { + if slices.Contains(values, alertNameSystemdFailed) { + return values + } + return append(values, alertNameSystemdFailed) + }) + }, func(app core.App) error { + return updateAlertNameValues(app, func(values []string) []string { + return slices.DeleteFunc(values, func(value string) bool { + return value == alertNameSystemdFailed + }) + }) + }) +} + +// updateAlertNameValues applies fn to the values of the alerts.name select field and saves the collection. +func updateAlertNameValues(app core.App, fn func([]string) []string) error { + collection, err := app.FindCollectionByNameOrId("alerts") + if err != nil { + return err + } + field, ok := collection.Fields.GetByName("name").(*core.SelectField) + if !ok { + return errors.New("alerts.name is not a select field") + } + field.Values = fn(field.Values) + return app.Save(collection) +} diff --git a/internal/site/src/components/active-alerts.tsx b/internal/site/src/components/active-alerts.tsx index 8a04bcc1..2c5019d9 100644 --- a/internal/site/src/components/active-alerts.tsx +++ b/internal/site/src/components/active-alerts.tsx @@ -59,7 +59,9 @@ export const ActiveAlerts = () => { {systems[alert.system]?.name} {info.name()} - {alert.name === "Status" ? ( + {info.triggeredDesc ? ( + info.triggeredDesc() + ) : alert.name === "Status" ? ( Connection is down ) : info.invert ? ( diff --git a/internal/site/src/components/alerts-history-columns.tsx b/internal/site/src/components/alerts-history-columns.tsx index b3162e23..887ae0b2 100644 --- a/internal/site/src/components/alerts-history-columns.tsx +++ b/internal/site/src/components/alerts-history-columns.tsx @@ -60,11 +60,15 @@ export const alertsHistoryColumns: ColumnDef[] = [ ), cell({ row, getValue }) { const name = row.original.name + const info = alertInfo[name] + if (info?.triggeredDesc) { + return {info.triggeredDesc()} + } if (name === "Status") { return {t`Down`} } const value = getValue() as number - const unit = alertInfo[name]?.unit + const unit = info?.unit return ( {toFixedFloat(value, value < 10 ? 2 : 1)} diff --git a/internal/site/src/components/alerts/alerts-sheet.tsx b/internal/site/src/components/alerts/alerts-sheet.tsx index c7550175..8e73c6fa 100644 --- a/internal/site/src/components/alerts/alerts-sheet.tsx +++ b/internal/site/src/components/alerts/alerts-sheet.tsx @@ -236,10 +236,16 @@ export function AlertContent({ const { name } = alertData const singleDescription = alertData.singleDesc?.() + /** Alerts that fire on first observation have no duration to configure */ + const noDuration = alertData.noDuration === true + /** Binary alerts have no threshold to configure */ + const noThreshold = !!singleDescription || noDuration + /** Whether enabling the alert reveals anything to configure */ + const hasControls = !(noThreshold && noDuration) const [checked, setChecked] = useState(global ? false : !!alert) - const [min, setMin] = useState(alert?.min || 10) - const [value, setValue] = useState(alert?.value || (singleDescription ? 0 : (alertData.start ?? 80))) + const [min, setMin] = useState(alert?.min || (noDuration ? 0 : 10)) + const [value, setValue] = useState(alert?.value || (noThreshold ? 0 : (alertData.start ?? 80))) const Icon = alertData.icon @@ -277,14 +283,16 @@ export function AlertContent({ - {checked && ( + {checked && hasControls && (
}> - {!singleDescription && ( + {!noThreshold && (

{alertData.invert ? ( @@ -361,45 +369,47 @@ export function AlertContent({

)} -
-

- {singleDescription && ( - <> - {singleDescription} - {` `} - - )} - - For {min}{" "} - - -

-
- sendUpsert(val[0], value)} - onValueChange={(val) => setMin(val[0])} - min={1} - max={60} - /> - { - let val = parseInt(e.target.value, 10) - if (!Number.isNaN(val)) { - val = Math.max(1, Math.min(val, 60)) - setMin(val) - sendUpsert(val, value) - } - }} - min={1} - max={60} - className="w-16 h-8 text-center px-1" - /> + {!noDuration && ( +
+

+ {singleDescription && ( + <> + {singleDescription} + {` `} + + )} + + For {min}{" "} + + +

+
+ sendUpsert(val[0], value)} + onValueChange={(val) => setMin(val[0])} + min={1} + max={60} + /> + { + let val = parseInt(e.target.value, 10) + if (!Number.isNaN(val)) { + val = Math.max(1, Math.min(val, 60)) + setMin(val) + sendUpsert(val, value) + } + }} + min={1} + max={60} + className="w-16 h-8 text-center px-1" + /> +
-
+ )}
)} diff --git a/internal/site/src/lib/alerts.ts b/internal/site/src/lib/alerts.ts index 736cec4e..b193bfb8 100644 --- a/internal/site/src/lib/alerts.ts +++ b/internal/site/src/lib/alerts.ts @@ -1,5 +1,5 @@ import { t } from "@lingui/core/macro" -import { CpuIcon, HardDriveIcon, MemoryStickIcon, ServerIcon } from "lucide-react" +import { CpuIcon, HardDriveIcon, MemoryStickIcon, ServerCrashIcon, ServerIcon } from "lucide-react" import type { RecordSubscription } from "pocketbase" import { EthernetIcon, GpuIcon } from "@/components/ui/icons" import { $alerts } from "@/lib/stores" @@ -104,6 +104,15 @@ export const alertInfo: Record = { start: 20, invert: true, }, + SystemdFailed: { + name: () => t`Failed Services`, + unit: "", + icon: ServerCrashIcon, + desc: () => t`Triggers when any systemd service enters the failed state`, + triggeredDesc: () => t`One or more services are in the failed state`, + /** Fires on first observation - the agent only polls systemd every 10 minutes */ + noDuration: true, + }, } as const /** Helper to manage user alerts */ diff --git a/internal/site/src/types.d.ts b/internal/site/src/types.d.ts index 3ef5293c..60d38266 100644 --- a/internal/site/src/types.d.ts +++ b/internal/site/src/types.d.ts @@ -396,6 +396,10 @@ export interface AlertInfo { start?: number /** Single value description (when there's only one value, like status) */ singleDesc?: () => string + /** Hides the duration slider for alerts that fire on first observation */ + noDuration?: boolean + /** Description shown instead of numeric threshold and duration values */ + triggeredDesc?: () => string invert?: boolean }