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