prometheus/postgres_exporter
1
0
Fork 0
mirror of https://github.com/prometheus-community/postgres_exporter.git synced 2026-10-23 12:08:57 +03:00
Code Activity

stat_replication: add pid label to disambiguate replication connections (#1353)

Browse Source 
* 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 <github@joeadams.io>

* Add PR# in CHANGELOG

Signed-off-by: Joe Adams <github@joeadams.io>

---------

Signed-off-by: Joe Adams <github@joeadams.io>
This commit is contained in:
Joe Adams
2026-07-26 10:16:37 -04:00
committed by GitHub
parent a6bc2f6ab6
commit eaa67f0255
4 changed files with 104 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@@ -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

31
collector/pg_stat_replication.go
View File

@@ -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, &currentWalLSNBytes, &walLSNDiff); err != nil {
if err := rows.Scan(&applicationName, &clientAddr, &state, &slotName, &pid, &currentWalLSNBytes, &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 ""
}

83
collector/pg_stat_replication_test.go
View File

@@ -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 {

1
go.mod
View File

@@ -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