Skip to content

Commit

Permalink
feat: Add wal_status to replication_slot
Browse files Browse the repository at this point in the history 
Signed-off-by: MarcWort <[email protected]>
  • Loading branch information
@MarcWort
MarcWort committed May 8, 2024
1 parent b81d26c commit 7525c2a
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 10 deletions.
22 changes: 20 additions & 2 deletions collector/pg_replication_slot.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,15 @@ var (
"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,
)
pgReplicationSlotWalStatus = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotSubsystem,
"wal_status",
),
"availability of WAL files claimed by this slot",
[]string{"slot_name", "slot_type", "wal_status"}, nil,
)

pgReplicationSlotQuery = `SELECT
slot_name,
Expand All @@ -83,7 +92,8 @@ var (
END AS current_wal_lsn,
COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
active,
safe_wal_size
safe_wal_size,
wal_status
FROM pg_replication_slots;`
)

Expand All @@ -103,7 +113,8 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance
var flushLSN sql.NullFloat64
var isActive sql.NullBool
var safeWalSize sql.NullInt64
if err := rows.Scan(&slotName, &slotType, &walLSN, &flushLSN, &isActive, &safeWalSize); err != nil {
var walStatus sql.NullString
if err := rows.Scan(&slotName, &slotType, &walLSN, &flushLSN, &isActive, &safeWalSize, &walStatus); err != nil {
return err
}

Expand Down Expand Up @@ -149,6 +160,13 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance
prometheus.GaugeValue, float64(safeWalSize.Int64), slotNameLabel, slotTypeLabel,
)
}

if walStatus.Valid {
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotWalStatus,
prometheus.GaugeValue, 1, slotNameLabel, slotTypeLabel, walStatus.String,
)
}
}
return rows.Err()
}
19 changes: 11 additions & 8 deletions collector/pg_replication_slot_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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", "safe_wal_size"}
columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}
rows := sqlmock.NewRows(columns).
AddRow("test_slot", "physical", 5, 3, true, 323906992)
AddRow("test_slot", "physical", 5, 3, true, 323906992, "reserved")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)

ch := make(chan prometheus.Metric)
Expand All @@ -51,6 +51,7 @@ func TestPgReplicationSlotCollectorActive(t *testing.T) {
{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},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical", "wal_status": "reserved"}, value: 1, metricType: dto.MetricType_GAUGE},
}

convey.Convey("Metrics comparison", t, func() {
Expand All @@ -73,9 +74,9 @@ func TestPgReplicationSlotCollectorInActive(t *testing.T) {

inst := &instance{db: db}

columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}
rows := sqlmock.NewRows(columns).
AddRow("test_slot", "physical", 6, 12, false, -4000)
AddRow("test_slot", "physical", 6, 12, false, -4000, "extended")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)

ch := make(chan prometheus.Metric)
Expand All @@ -92,6 +93,7 @@ func TestPgReplicationSlotCollectorInActive(t *testing.T) {
{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},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical", "wal_status": "extended"}, value: 1, metricType: dto.MetricType_GAUGE},
}

convey.Convey("Metrics comparison", t, func() {
Expand All @@ -115,9 +117,9 @@ func TestPgReplicationSlotCollectorActiveNil(t *testing.T) {

inst := &instance{db: db}

columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}
rows := sqlmock.NewRows(columns).
AddRow("test_slot", "physical", 6, 12, nil, nil)
AddRow("test_slot", "physical", 6, 12, nil, nil, "lost")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)

ch := make(chan prometheus.Metric)
Expand All @@ -133,6 +135,7 @@ func TestPgReplicationSlotCollectorActiveNil(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", "wal_status": "lost"}, value: 1, metricType: dto.MetricType_GAUGE},
}

convey.Convey("Metrics comparison", t, func() {
Expand All @@ -155,9 +158,9 @@ func TestPgReplicationSlotCollectorTestNilValues(t *testing.T) {

inst := &instance{db: db}

columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size"}
columns := []string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}
rows := sqlmock.NewRows(columns).
AddRow(nil, nil, nil, nil, true, nil)
AddRow(nil, nil, nil, nil, true, nil, nil)
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotQuery)).WillReturnRows(rows)

ch := make(chan prometheus.Metric)
Expand Down

0 comments on commit 7525c2a

Please sign in to comment.