feat(alerts): add alert for failed systemd services (#2173)

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 <martin@stenrose.se>
Co-authored-by: henrygd <hank@henrygd.me>

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
 }
 

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())
+}

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)
+}