Export ntfy_db_up and health-check the primary like the replicas

A node cut off from its database only showed in the logs. The health check loop now always
runs and pings the primary too; ntfy_db_up (1/0, per pool, next to ntfy_db_pool_*) is the
result of the last check, so an alert can fire on ntfy_db_up{role="primary"} == 0.

db/db.go

@@ -10,10 +10,10 @@ import (
 )
 
 const (
-	tag                            = "db"
-	replicaHealthCheckInitialDelay = 5 * time.Second
-	replicaHealthCheckInterval     = 30 * time.Second
-	replicaHealthCheckTimeout      = 10 * time.Second
+	tag                     = "db"
+	healthCheckInitialDelay = 5 * time.Second
+	healthCheckInterval     = 30 * time.Second
+	healthCheckTimeout      = 10 * time.Second
 )
 
 // DB wraps a primary *sql.DB and optional read replicas. All standard query/exec methods
@@ -28,7 +28,8 @@ type DB struct {
 
 // New creates a new DB that wraps the given primary and optional replica connections.
 // If replicas is nil or empty, ReadOnly() simply returns the primary.
-// Replicas start unhealthy and are checked immediately by a background goroutine.
+// Replicas start unhealthy, the primary healthy (it was just opened); a background goroutine
+// checks all of them, so the metrics show a node that is cut off from its database.
 func New(primary *Host, replicas []*Host) *DB {
 	ctx, cancel := context.WithCancel(context.Background())
 	d := &DB{
@@ -36,9 +37,8 @@ func New(primary *Host, replicas []*Host) *DB {
 		replicas: replicas,
 		cancel:   cancel,
 	}
-	if len(d.replicas) > 0 {
-		go d.healthCheckLoop(ctx)
-	}
+	primary.healthy.Store(true)
+	go d.healthCheckLoop(ctx)
 	return d
 }
 
@@ -87,16 +87,18 @@ func (d *DB) Primary() *sql.DB {
 func (d *DB) Stats() []*PoolStats {
 	stats := []*PoolStats{
 		{
-			Role:  PoolRolePrimary,
-			Addr:  d.primary.Addr,
-			Stats: d.primary.DB.Stats(),
+			Role:    PoolRolePrimary,
+			Addr:    d.primary.Addr,
+			Healthy: d.primary.healthy.Load(),
+			Stats:   d.primary.DB.Stats(),
 		},
 	}
 	for _, r := range d.replicas {
 		stats = append(stats, &PoolStats{
-			Role:  PoolRoleReplica,
-			Addr:  r.Addr,
-			Stats: r.DB.Stats(),
+			Role:    PoolRoleReplica,
+			Addr:    r.Addr,
+			Healthy: r.healthy.Load(),
+			Stats:   r.DB.Stats(),
 		})
 	}
 	return stats
@@ -137,39 +139,46 @@ func (d *DB) Close() error {
 	return d.primary.DB.Close()
 }
 
-// healthCheckLoop checks replicas immediately, then periodically on a ticker.
+// healthCheckLoop checks the primary and the replicas shortly after start, then periodically.
 func (d *DB) healthCheckLoop(ctx context.Context) {
 	select {
 	case <-ctx.Done():
 		return
-	case <-time.After(replicaHealthCheckInitialDelay):
-		d.checkReplicas(ctx)
+	case <-time.After(healthCheckInitialDelay):
+		d.checkHealth(ctx)
 	}
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-time.After(replicaHealthCheckInterval):
-			d.checkReplicas(ctx)
+		case <-time.After(healthCheckInterval):
+			d.checkHealth(ctx)
 		}
 	}
 }
 
-// checkReplicas pings each replica with a timeout and updates its health status.
-func (d *DB) checkReplicas(ctx context.Context) {
+// checkHealth pings the primary and each replica with a timeout and updates their health status.
+func (d *DB) checkHealth(ctx context.Context) {
+	checkHost(ctx, d.primary, "primary")
 	for _, r := range d.replicas {
-		wasHealthy := r.healthy.Load()
-		pingCtx, cancel := context.WithTimeout(ctx, replicaHealthCheckTimeout)
-		err := r.DB.PingContext(pingCtx)
-		cancel()
-		if err != nil {
-			r.healthy.Store(false)
-			log.Tag(tag).Error("Database replica %s is unhealthy: %s", r.Addr, err)
-		} else {
-			r.healthy.Store(true)
-			if !wasHealthy {
-				log.Tag(tag).Info("Database replica %s is healthy", r.Addr)
-			}
+		checkHost(ctx, r, "replica")
+	}
+}
+
+// checkHost pings the host and updates its health status, logging when it changes (and every
+// failure, so a lasting outage stays visible in the logs)
+func checkHost(ctx context.Context, h *Host, kind string) {
+	wasHealthy := h.healthy.Load()
+	pingCtx, cancel := context.WithTimeout(ctx, healthCheckTimeout)
+	err := h.DB.PingContext(pingCtx)
+	cancel()
+	if err != nil {
+		h.healthy.Store(false)
+		log.Tag(tag).Error("Database %s %s is unhealthy: %s", kind, h.Addr, err)
+	} else {
+		h.healthy.Store(true)
+		if !wasHealthy {
+			log.Tag(tag).Info("Database %s %s is healthy", kind, h.Addr)
 		}
 	}
 }

db/db_test.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"path/filepath"
 	"testing"
+	"time"
 
 	_ "github.com/mattn/go-sqlite3"
 	"github.com/stretchr/testify/require"
@@ -74,3 +75,21 @@ func TestDB_Stats(t *testing.T) {
 	require.Equal(t, 3, stats[1].Stats.MaxOpenConnections)
 	require.Equal(t, 0, stats[1].Stats.InUse)
 }
+
+func TestDB_HealthCheck(t *testing.T) {
+	// The primary is checked like the replicas, so a node cut off from it shows as unhealthy
+	// (ntfy_db_up), and healthy again once it is back
+	path := filepath.Join(t.TempDir(), "primary.db")
+	primary, err := sql.Open("sqlite3", path)
+	require.Nil(t, err)
+	d := db.New(&db.Host{
+		Addr: "primary:5432",
+		DB:   primary,
+	}, nil)
+	defer d.Close()
+	require.True(t, d.Stats()[0].Healthy) // Just opened: healthy until a check says otherwise
+	require.Nil(t, primary.Close())
+	require.Eventually(t, func() bool {
+		return !d.Stats()[0].Healthy
+	}, 15*time.Second, 50*time.Millisecond)
+}

db/types.go

@@ -32,7 +32,8 @@ const (
 
 // PoolStats is one connection pool's statistics: the primary's or a replica's
 type PoolStats struct {
-	Role  string // PoolRolePrimary or PoolRoleReplica
-	Addr  string // "host:port"
-	Stats sql.DBStats
+	Role    string // PoolRolePrimary or PoolRoleReplica
+	Addr    string // "host:port"
+	Healthy bool   // The last health check (a ping) succeeded
+	Stats   sql.DBStats
 }

docs/config.md

@@ -500,6 +500,10 @@ To size the pools, watch the `ntfy_db_pool_*` [metrics](#monitoring), one series
 and `ntfy_db_pool_wait_seconds_total` count queries that had to wait for a free connection. The open connection count
 says little, since the pool rotates through all of them. Waits that keep going up mean the pool is too small.
 
+`ntfy_db_up` is 1 if the server reached the pool's database at its last health check (a ping every 30 seconds), and 0
+if not, one series per pool like the above. Alert on `ntfy_db_up{role="primary"} == 0`: a server that cannot reach its
+primary cannot publish or authenticate, and enforces the daily limits on its own until it is back.
+
 
 ## Message cache
 If desired, ntfy can temporarily keep notifications in an in-memory or an on-disk cache. Caching messages for a short period

docs/releases.md

@@ -2294,6 +2294,7 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 
 * Local admin socket (`listen-admin-unix`, default `/run/ntfy/admin.sock`) that serves the admin API without credentials to whoever can open it (the ntfy user and group, and root); the shipped systemd unit now sets `RuntimeDirectory=ntfy` so the server can create it. Custom unit files without it log a warning at startup and keep working as before (see [admin socket](config.md#admin-socket))
 * Admin API (internal, undocumented): manage tokens of other users, tiers and password reset links
+* Export `ntfy_db_up` (1 if the last health check of a PostgreSQL pool's database succeeded, per pool) and health-check the primary like the replicas, to alert on a server that is cut off from its database (see [PostgreSQL](config.md#postgresql))
 
 **Bug fixes + maintenance:**
 

metrics/metrics.go

@@ -167,10 +167,11 @@ func init() {
 type DBPool struct {
 	Role  string // db.PoolRolePrimary or db.PoolRoleReplica
 	Addr  string // "host:port"
+	Up    bool   // The last health check of the pool's database succeeded
 	Stats sql.DBStats
 }
 
-// SetDBPoolsFunc sets the function that the ntfy_db_pool_* metrics read their values from on
+// SetDBPoolsFunc sets the function that the ntfy_db_* metrics read their values from on
 // every scrape; nil (the default) reports no pools
 func SetDBPoolsFunc(fn func() []*DBPool) {
 	dbPools.fn.Store(&fn)
@@ -184,39 +185,45 @@ type dbPoolCollector struct {
 	metrics []*dbPoolMetric
 }
 
-// dbPoolMetric is one ntfy_db_pool_* metric: its description, and how to read it from a pool
+// dbPoolMetric is one ntfy_db_* metric: its description, and how to read it from a pool
 type dbPoolMetric struct {
 	desc      *prometheus.Desc
 	valueType prometheus.ValueType
-	value     func(stats sql.DBStats) float64
+	value     func(pool *DBPool) float64
 }
 
 func newDBPoolCollector() *dbPoolCollector {
 	return &dbPoolCollector{
 		metrics: []*dbPoolMetric{
-			newDBPoolMetric("ntfy_db_pool_connections_max", "Maximum open database connections (pool_max_conns), per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
-				return float64(stats.MaxOpenConnections)
+			newDBPoolMetric("ntfy_db_up", "1 if the last health check of the pool's database (a ping every 30s) succeeded, 0 if not, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
+				if pool.Up {
+					return 1
+				}
+				return 0
 			}),
-			newDBPoolMetric("ntfy_db_pool_connections_open", "Open database connections, in use or idle, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
-				return float64(stats.OpenConnections)
+			newDBPoolMetric("ntfy_db_pool_connections_max", "Maximum open database connections (pool_max_conns), per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
+				return float64(pool.Stats.MaxOpenConnections)
 			}),
-			newDBPoolMetric("ntfy_db_pool_connections_in_use", "Database connections currently in use, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
-				return float64(stats.InUse)
+			newDBPoolMetric("ntfy_db_pool_connections_open", "Open database connections, in use or idle, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
+				return float64(pool.Stats.OpenConnections)
 			}),
-			newDBPoolMetric("ntfy_db_pool_connections_idle", "Idle database connections, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
-				return float64(stats.Idle)
+			newDBPoolMetric("ntfy_db_pool_connections_in_use", "Database connections currently in use, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
+				return float64(pool.Stats.InUse)
 			}),
-			newDBPoolMetric("ntfy_db_pool_waits_total", "Times a query had to wait for a free database connection, per pool", prometheus.CounterValue, func(stats sql.DBStats) float64 {
-				return float64(stats.WaitCount)
+			newDBPoolMetric("ntfy_db_pool_connections_idle", "Idle database connections, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
+				return float64(pool.Stats.Idle)
 			}),
-			newDBPoolMetric("ntfy_db_pool_wait_seconds_total", "Total time spent waiting for a free database connection, per pool", prometheus.CounterValue, func(stats sql.DBStats) float64 {
-				return stats.WaitDuration.Seconds()
+			newDBPoolMetric("ntfy_db_pool_waits_total", "Times a query had to wait for a free database connection, per pool", prometheus.CounterValue, func(pool *DBPool) float64 {
+				return float64(pool.Stats.WaitCount)
+			}),
+			newDBPoolMetric("ntfy_db_pool_wait_seconds_total", "Total time spent waiting for a free database connection, per pool", prometheus.CounterValue, func(pool *DBPool) float64 {
+				return pool.Stats.WaitDuration.Seconds()
 			}),
 		},
 	}
 }
 
-func newDBPoolMetric(name, help string, valueType prometheus.ValueType, value func(stats sql.DBStats) float64) *dbPoolMetric {
+func newDBPoolMetric(name, help string, valueType prometheus.ValueType, value func(pool *DBPool) float64) *dbPoolMetric {
 	return &dbPoolMetric{
 		desc:      prometheus.NewDesc(name, help, []string{"role", "host"}, nil),
 		valueType: valueType,
@@ -239,7 +246,7 @@ func (c *dbPoolCollector) Collect(ch chan<- prometheus.Metric) {
 	}
 	for _, pool := range (*fn)() {
 		for _, m := range c.metrics {
-			ch <- prometheus.MustNewConstMetric(m.desc, m.valueType, m.value(pool.Stats), pool.Role, pool.Addr)
+			ch <- prometheus.MustNewConstMetric(m.desc, m.valueType, m.value(pool), pool.Role, pool.Addr)
 		}
 	}
 }

metrics/metrics_test.go

@@ -36,6 +36,7 @@ var expectedMetricNames = []string{
 	"ntfy_db_pool_connections_open",
 	"ntfy_db_pool_wait_seconds_total",
 	"ntfy_db_pool_waits_total",
+	"ntfy_db_up",
 	"ntfy_emails_received_failure",
 	"ntfy_emails_received_success",
 	"ntfy_emails_sent_failure",
@@ -101,6 +102,7 @@ func TestDBPoolMetrics(t *testing.T) {
 			{
 				Role: db.PoolRolePrimary,
 				Addr: "db:5432",
+				Up:   true,
 				Stats: sql.DBStats{
 					MaxOpenConnections: 30,
 					OpenConnections:    12,
@@ -140,9 +142,13 @@ ntfy_db_pool_wait_seconds_total{host="replica:5432",role="%[2]s"} 0
 # TYPE ntfy_db_pool_waits_total counter
 ntfy_db_pool_waits_total{host="db:5432",role="%[1]s"} 3
 ntfy_db_pool_waits_total{host="replica:5432",role="%[2]s"} 0
+# HELP ntfy_db_up 1 if the last health check of the pool's database (a ping every 30s) succeeded, 0 if not, per pool
+# TYPE ntfy_db_up gauge
+ntfy_db_up{host="db:5432",role="%[1]s"} 1
+ntfy_db_up{host="replica:5432",role="%[2]s"} 0
 `, db.PoolRolePrimary, db.PoolRoleReplica)
 	require.Nil(t, testutil.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expected),
-		"ntfy_db_pool_connections_in_use", "ntfy_db_pool_connections_max", "ntfy_db_pool_waits_total", "ntfy_db_pool_wait_seconds_total"))
+		"ntfy_db_pool_connections_in_use", "ntfy_db_pool_connections_max", "ntfy_db_pool_waits_total", "ntfy_db_pool_wait_seconds_total", "ntfy_db_up"))
 	open, err := testutil.GatherAndCount(prometheus.DefaultGatherer, "ntfy_db_pool_connections_open", "ntfy_db_pool_connections_idle")
 	require.Nil(t, err)
 	require.Equal(t, 4, open)

server/server.go

@@ -383,13 +383,14 @@ func New(conf *Config) (*Server, error) {
 	return s, nil
 }
 
-// dbPoolStats reports the database connection pools for the ntfy_db_pool_* metrics
+// dbPoolStats reports the database connection pools for the ntfy_db_* metrics
 func (s *Server) dbPoolStats() []*metrics.DBPool {
 	pools := make([]*metrics.DBPool, 0)
 	for _, p := range s.db.Stats() {
 		pools = append(pools, &metrics.DBPool{
 			Role:  p.Role,
 			Addr:  p.Addr,
+			Up:    p.Healthy,
 			Stats: p.Stats,
 		})
 	}