From eaa67f02559c071cbe41b3111e930c154e8ec315 Mon Sep 17 00:00:00 2001 From: Joe Adams Date: Sun, 26 Jul 2026 10:16:37 -0400 Subject: [PATCH] stat_replication: add pid label to disambiguate replication connections (#1353) * stat_replication: add pid label to disambiguate replication connections application_name, client_addr, state, and slot_name are not a unique key for pg_stat_replication rows. Multiple standbys with the same application_name, client_addr, state, and slot_name (empty) were causing duplicate metrics which would get rejected by the registry. This regressed visibly in 0.20.1: 0.20.0 had the stat_replication query erroring on every scrape ("column slot_name does not exist", #1310), so the collector never actually emitted rows. #1313 fixed that query error, which let real data flow again and exposed this pre-existing label collision for the first time. One downside is that the additional label could increase series churn because every time the walsender reconnects, the pid will change and therefore create a new series. Fixes #1352 Signed-off-by: Joe Adams * Add PR# in CHANGELOG Signed-off-by: Joe Adams --------- Signed-off-by: Joe Adams --- CHANGELOG.md | 2 + collector/pg_stat_replication.go | 31 +++++++--- collector/pg_stat_replication_test.go | 83 +++++++++++++++++++++++++-- go.mod | 1 + 4 files changed, 104 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b8a06d4..ab02ec18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## main / (unreleased) +* [CHANGE] stat_replication: add `pid` label to disambiguate replication connections that otherwise share identical labels by @sysadmind in https://github.com/prometheus-community/postgres_exporter/pull/1353 + ## 0.20.1 / 2026-07-07 * [BUGFIX] stat_replication: fix "slot_name does not exist" by joining `pg_replication_slots` by @megative in https://github.com/prometheus-community/postgres_exporter/pull/1313 diff --git a/collector/pg_stat_replication.go b/collector/pg_stat_replication.go index 6f10c555..fdb328a1 100644 --- a/collector/pg_stat_replication.go +++ b/collector/pg_stat_replication.go @@ -16,6 +16,7 @@ package collector import ( "context" "database/sql" + "strconv" "github.com/blang/semver/v4" "github.com/prometheus/client_golang/prometheus" @@ -34,7 +35,7 @@ func NewPGStatReplicationCollector(collectorConfig) (Collector, error) { } var ( - statReplicationLabels = []string{"application_name", "client_addr", "state", "slot_name"} + statReplicationLabels = []string{"application_name", "client_addr", "state", "slot_name", "pid"} statReplicationCurrentWalLSNBytesDesc = prometheus.NewDesc( prometheus.BuildFQName(namespace, statReplicationSubsystem, "pg_current_wal_lsn_bytes"), @@ -65,6 +66,7 @@ var ( client_addr::text, state, s.slot_name, + pg_stat_replication.pid, (case pg_is_in_recovery() when 't' then pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_lsn('0/0'))::float else pg_wal_lsn_diff(pg_current_wal_lsn(), pg_lsn('0/0'))::float end) AS pg_current_wal_lsn_bytes, (case pg_is_in_recovery() when 't' then pg_wal_lsn_diff(pg_last_wal_receive_lsn(), replay_lsn)::float else pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)::float end) AS pg_wal_lsn_diff FROM pg_stat_replication @@ -77,6 +79,7 @@ var ( client_addr::text, state, s.slot_name, + pg_stat_replication.pid, (case pg_is_in_recovery() when 't' then pg_xlog_location_diff(pg_last_xlog_receive_location(), replay_location)::float else pg_xlog_location_diff(pg_current_xlog_location(), replay_location)::float end) AS pg_xlog_location_diff FROM pg_stat_replication LEFT JOIN pg_replication_slots s ON s.active_pid = pg_stat_replication.pid @@ -90,6 +93,7 @@ var ( application_name, client_addr::text, state, + pid, (case pg_is_in_recovery() when 't' then pg_xlog_location_diff(pg_last_xlog_receive_location(), replay_location)::float else pg_xlog_location_diff(pg_current_xlog_location(), replay_location)::float end) AS pg_xlog_location_diff FROM pg_stat_replication ` @@ -118,11 +122,12 @@ func updateStatReplication(ctx context.Context, instance *instance, ch chan<- pr for rows.Next() { var applicationName, clientAddr, state, slotName sql.NullString + var pid sql.NullInt64 var currentWalLSNBytes, walLSNDiff sql.NullFloat64 - if err := rows.Scan(&applicationName, &clientAddr, &state, &slotName, ¤tWalLSNBytes, &walLSNDiff); err != nil { + if err := rows.Scan(&applicationName, &clientAddr, &state, &slotName, &pid, ¤tWalLSNBytes, &walLSNDiff); err != nil { return err } - labels := statReplicationLabelValues(applicationName, clientAddr, state, slotName) + labels := statReplicationLabelValues(applicationName, clientAddr, state, slotName, pid) if currentWalLSNBytes.Valid { ch <- prometheus.MustNewConstMetric( @@ -155,8 +160,9 @@ func updateStatReplicationBefore10(ctx context.Context, instance *instance, ch c for rows.Next() { var applicationName, clientAddr, state, slotName sql.NullString + var pid sql.NullInt64 var xlogLocationDiff sql.NullFloat64 - if err := rows.Scan(&applicationName, &clientAddr, &state, &slotName, &xlogLocationDiff); err != nil { + if err := rows.Scan(&applicationName, &clientAddr, &state, &slotName, &pid, &xlogLocationDiff); err != nil { return err } @@ -165,7 +171,7 @@ func updateStatReplicationBefore10(ctx context.Context, instance *instance, ch c statReplicationXlogLocationDiffDesc, prometheus.GaugeValue, xlogLocationDiff.Float64, - statReplicationLabelValues(applicationName, clientAddr, state, slotName)..., + statReplicationLabelValues(applicationName, clientAddr, state, slotName, pid)..., ) } } @@ -183,8 +189,9 @@ func updateStatReplicationBefore95(ctx context.Context, instance *instance, ch c for rows.Next() { var applicationName, clientAddr, state sql.NullString + var pid sql.NullInt64 var xlogLocationDiff sql.NullFloat64 - if err := rows.Scan(&applicationName, &clientAddr, &state, &xlogLocationDiff); err != nil { + if err := rows.Scan(&applicationName, &clientAddr, &state, &pid, &xlogLocationDiff); err != nil { return err } @@ -193,7 +200,7 @@ func updateStatReplicationBefore95(ctx context.Context, instance *instance, ch c statReplicationXlogLocationDiffDesc, prometheus.GaugeValue, xlogLocationDiff.Float64, - statReplicationLabelValues(applicationName, clientAddr, state, sql.NullString{})..., + statReplicationLabelValues(applicationName, clientAddr, state, sql.NullString{}, pid)..., ) } } @@ -201,12 +208,13 @@ func updateStatReplicationBefore95(ctx context.Context, instance *instance, ch c return rows.Err() } -func statReplicationLabelValues(applicationName, clientAddr, state, slotName sql.NullString) []string { +func statReplicationLabelValues(applicationName, clientAddr, state, slotName sql.NullString, pid sql.NullInt64) []string { return []string{ nullStringValue(applicationName), nullStringValue(clientAddr), nullStringValue(state), nullStringValue(slotName), + nullInt64Value(pid), } } @@ -216,3 +224,10 @@ func nullStringValue(s sql.NullString) string { } return "" } + +func nullInt64Value(i sql.NullInt64) string { + if i.Valid { + return strconv.FormatInt(i.Int64, 10) + } + return "" +} diff --git a/collector/pg_stat_replication_test.go b/collector/pg_stat_replication_test.go index 0816a0ca..a26de2f1 100644 --- a/collector/pg_stat_replication_test.go +++ b/collector/pg_stat_replication_test.go @@ -15,11 +15,13 @@ package collector import ( "context" + "strings" "testing" "github.com/DATA-DOG/go-sqlmock" "github.com/blang/semver/v4" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" dto "github.com/prometheus/client_model/go" "github.com/smartystreets/goconvey/convey" ) @@ -38,9 +40,10 @@ func TestPGStatReplicationCollector(t *testing.T) { "client_addr", "state", "slot_name", + "pid", "pg_current_wal_lsn_bytes", "pg_wal_lsn_diff", - }).AddRow("standby", "10.0.0.1", "streaming", "slot_a", 1024.0, 64.0) + }).AddRow("standby", "10.0.0.1", "streaming", "slot_a", 123, 1024.0, 64.0) mock.ExpectQuery(sanitizeQuery(statReplicationQuery)).WillReturnRows(rows) ch := make(chan prometheus.Metric) @@ -57,6 +60,7 @@ func TestPGStatReplicationCollector(t *testing.T) { "client_addr": "10.0.0.1", "state": "streaming", "slot_name": "slot_a", + "pid": "123", } expected := []MetricResult{ {labels: expectedLabels, value: 1024, metricType: dto.MetricType_GAUGE}, @@ -87,8 +91,9 @@ func TestPGStatReplicationCollectorBefore10(t *testing.T) { "client_addr", "state", "slot_name", + "pid", "pg_xlog_location_diff", - }).AddRow("standby", "10.0.0.1", "streaming", "slot_a", 32.0) + }).AddRow("standby", "10.0.0.1", "streaming", "slot_a", 123, 32.0) mock.ExpectQuery(sanitizeQuery(statReplicationQueryBefore10)).WillReturnRows(rows) ch := make(chan prometheus.Metric) @@ -107,6 +112,7 @@ func TestPGStatReplicationCollectorBefore10(t *testing.T) { "client_addr": "10.0.0.1", "state": "streaming", "slot_name": "slot_a", + "pid": "123", }, value: 32, metricType: dto.MetricType_GAUGE, @@ -136,8 +142,9 @@ func TestPGStatReplicationCollectorBefore95(t *testing.T) { "application_name", "client_addr", "state", + "pid", "pg_xlog_location_diff", - }).AddRow("standby", "10.0.0.1", "streaming", 32.0) + }).AddRow("standby", "10.0.0.1", "streaming", 123, 32.0) mock.ExpectQuery(sanitizeQuery(statReplicationQueryBefore95)).WillReturnRows(rows) ch := make(chan prometheus.Metric) @@ -156,6 +163,7 @@ func TestPGStatReplicationCollectorBefore95(t *testing.T) { "client_addr": "10.0.0.1", "state": "streaming", "slot_name": "", + "pid": "123", }, value: 32, metricType: dto.MetricType_GAUGE, @@ -186,9 +194,10 @@ func TestPGStatReplicationCollectorNullValues(t *testing.T) { "client_addr", "state", "slot_name", + "pid", "pg_current_wal_lsn_bytes", "pg_wal_lsn_diff", - }).AddRow(nil, nil, nil, nil, nil, nil) + }).AddRow(nil, nil, nil, nil, nil, nil, nil) mock.ExpectQuery(sanitizeQuery(statReplicationQuery)).WillReturnRows(rows) ch := make(chan prometheus.Metric) @@ -222,8 +231,9 @@ func TestPGStatReplicationCollectorBefore10NullValues(t *testing.T) { "client_addr", "state", "slot_name", + "pid", "pg_xlog_location_diff", - }).AddRow(nil, nil, nil, nil, nil) + }).AddRow(nil, nil, nil, nil, nil, nil) mock.ExpectQuery(sanitizeQuery(statReplicationQueryBefore10)).WillReturnRows(rows) ch := make(chan prometheus.Metric) @@ -243,6 +253,69 @@ func TestPGStatReplicationCollectorBefore10NullValues(t *testing.T) { } } +// collectFunc adapts a plain function to prometheus.Collector so it can be +// exercised through testutil.CollectAndCompare. +type collectFunc func(ch chan<- prometheus.Metric) + +func (f collectFunc) Describe(chan<- *prometheus.Desc) {} +func (f collectFunc) Collect(ch chan<- prometheus.Metric) { f(ch) } + +// TestPGStatReplicationCollectorDuplicateConnections is a regression test +// for https://github.com/prometheus-community/postgres_exporter/issues/1352. +// +// application_name, client_addr, state and slot_name are not a unique key +// for pg_stat_replication. Only pid actually distinguishes the rows. +func TestPGStatReplicationCollectorDuplicateConnections(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("Error opening a stub db connection: %s", err) + } + defer db.Close() + + inst := &instance{db: db, version: semver.MustParse("16.0.0")} + + rows := sqlmock.NewRows([]string{ + "application_name", + "client_addr", + "state", + "slot_name", + "pid", + "pg_current_wal_lsn_bytes", + "pg_wal_lsn_diff", + }). + AddRow("walreceiver", "172.18.0.254", "streaming", "", 111, 1024.0, 64.0). + AddRow("walreceiver", "172.18.0.254", "streaming", "", 222, 2048.0, 128.0) + mock.ExpectQuery(sanitizeQuery(statReplicationQuery)).WillReturnRows(rows) + + collector := collectFunc(func(ch chan<- prometheus.Metric) { + c := PGStatReplicationCollector{} + if err := c.Update(context.Background(), inst, ch); err != nil { + t.Errorf("Error calling PGStatReplicationCollector.Update: %s", err) + } + }) + + expected := strings.NewReader(` +# HELP pg_stat_replication_pg_current_wal_lsn_bytes WAL position in bytes +# TYPE pg_stat_replication_pg_current_wal_lsn_bytes gauge +pg_stat_replication_pg_current_wal_lsn_bytes{application_name="walreceiver",client_addr="172.18.0.254",pid="111",slot_name="",state="streaming"} 1024 +pg_stat_replication_pg_current_wal_lsn_bytes{application_name="walreceiver",client_addr="172.18.0.254",pid="222",slot_name="",state="streaming"} 2048 +# HELP pg_stat_replication_pg_wal_lsn_diff Lag in bytes between master and slave +# TYPE pg_stat_replication_pg_wal_lsn_diff gauge +pg_stat_replication_pg_wal_lsn_diff{application_name="walreceiver",client_addr="172.18.0.254",pid="111",slot_name="",state="streaming"} 64 +pg_stat_replication_pg_wal_lsn_diff{application_name="walreceiver",client_addr="172.18.0.254",pid="222",slot_name="",state="streaming"} 128 +`) + if err := testutil.CollectAndCompare(collector, expected, + "pg_stat_replication_pg_current_wal_lsn_bytes", + "pg_stat_replication_pg_wal_lsn_diff", + ); err != nil { + t.Errorf("unexpected collecting result for two connections with identical application_name/client_addr/state/slot_name:\n%s", err) + } + + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("there were unfulfilled exceptions: %s", err) + } +} + func TestPGStatReplicationCollectorBefore92(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { diff --git a/go.mod b/go.mod index 77bc7c9b..6c001bdd 100644 --- a/go.mod +++ b/go.mod @@ -29,6 +29,7 @@ require ( github.com/jtolds/gls v4.20.0+incompatible // indirect github.com/kr/pretty v0.3.1 // indirect github.com/kr/text v0.2.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/mdlayher/socket v0.6.0 // indirect github.com/mdlayher/vsock v1.3.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect