feat: Add safe_wal_size to replication_slot

Author: MarcWort

collector/pg_replication_slot.go

@@ -63,6 +63,15 @@ var (
 		"whether the replication slot is active or not",
 		[]string{"slot_name", "slot_type"}, nil,
 	)
+	pgReplicationSlotSafeWal = prometheus.NewDesc(
+		prometheus.BuildFQName(
+			namespace,
+			replicationSlotSubsystem,
+			"safe_wal_size",
+		),
+		"number of bytes that can be written to WAL such that this slot is not in danger of getting in state lost",
+		[]string{"slot_name", "slot_type"}, nil,
+	)
 
 	pgReplicationSlotQuery = `SELECT
 		slot_name,
@@ -73,7 +82,8 @@ var (
 		    pg_current_wal_lsn() - '0/0'
 		END AS current_wal_lsn,
 		COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
-		active
+		active,
+		safe_wal_size
 	FROM pg_replication_slots;`
 )
 
@@ -92,7 +102,8 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance
 		var walLSN sql.NullFloat64
 		var flushLSN sql.NullFloat64
 		var isActive sql.NullBool
-		if err := rows.Scan(&slotName, &slotType, &walLSN, &flushLSN, &isActive); err != nil {
+		var safeWalSize sql.NullInt64
+		if err := rows.Scan(&slotName, &slotType, &walLSN, &flushLSN, &isActive, &safeWalSize); err != nil {
 			return err
 		}
 
@@ -131,6 +142,13 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance
 			pgReplicationSlotIsActiveDesc,
 			prometheus.GaugeValue, isActiveValue, slotNameLabel, slotTypeLabel,
 		)
+
+		if safeWalSize.Valid {
+			ch <- prometheus.MustNewConstMetric(
+				pgReplicationSlotSafeWal,
+				prometheus.GaugeValue, float64(safeWalSize.Int64), slotNameLabel, slotTypeLabel,
+			)
+		}
 	}
 	return rows.Err()
 }

collector/pg_replication_slot_test.go

@@ -31,9 +31,9 @@ func TestPgReplicationSlotCollectorActive(t *testing.T) {
 
 	inst := &instance{db: db}
 
-	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
 	rows := sqlmock.NewRows(columns).
-		AddRow("test_slot", "physical", 5, 3, true)
+		AddRow("test_slot", "physical", 5, 3, true, 323906992)
 	mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
 
 	ch := make(chan prometheus.Metric)
@@ -50,6 +50,7 @@ func TestPgReplicationSlotCollectorActive(t *testing.T) {
 		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 5, metricType: dto.MetricType_GAUGE},
 		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 3, metricType: dto.MetricType_GAUGE},
 		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 1, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 323906992, metricType: dto.MetricType_GAUGE},
 	}
 
 	convey.Convey("Metrics comparison", t, func() {
@@ -72,9 +73,9 @@ func TestPgReplicationSlotCollectorInActive(t *testing.T) {
 
 	inst := &instance{db: db}
 
-	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
 	rows := sqlmock.NewRows(columns).
-		AddRow("test_slot", "physical", 6, 12, false)
+		AddRow("test_slot", "physical", 6, 12, false, -4000)
 	mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
 
 	ch := make(chan prometheus.Metric)
@@ -90,6 +91,7 @@ func TestPgReplicationSlotCollectorInActive(t *testing.T) {
 	expected := []MetricResult{
 		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 6, metricType: dto.MetricType_GAUGE},
 		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: 0, metricType: dto.MetricType_GAUGE},
+		{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical"}, value: -4000, metricType: dto.MetricType_GAUGE},
 	}
 
 	convey.Convey("Metrics comparison", t, func() {
@@ -113,9 +115,9 @@ func TestPgReplicationSlotCollectorActiveNil(t *testing.T) {
 
 	inst := &instance{db: db}
 
-	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
 	rows := sqlmock.NewRows(columns).
-		AddRow("test_slot", "physical", 6, 12, nil)
+		AddRow("test_slot", "physical", 6, 12, nil, nil)
 	mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
 
 	ch := make(chan prometheus.Metric)
@@ -153,9 +155,9 @@ func TestPgReplicationSlotCollectorTestNilValues(t *testing.T) {
 
 	inst := &instance{db: db}
 
-	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active"}
+	columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
 	rows := sqlmock.NewRows(columns).
-		AddRow(nil, nil, nil, nil, true)
+		AddRow(nil, nil, nil, nil, true, nil)
 	mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)
 
 	ch := make(chan prometheus.Metric)