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

Refactor stat_replication_slots into standalone collector (#1307)

Browse Source 
* Refactor stat_replication_slots into standalone collector

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>

* Do not expose metrics for NULL results

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>

* Merge slot and slots collectors

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>

* Remove unused queries map

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>

* Add changelog entry for breaking changes

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>

---------

Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com>
This commit is contained in:
Arthur Silva Sens
2026-06-02 22:22:43 -03:00
committed by GitHub
parent 6c3e586d90
commit c8eb54d64a
8 changed files with 653 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
CHANGELOG.md
View File

@@ -1,5 +1,30 @@
## main / (unreleased)
BREAKING CHANGES:
The `replication_slot` collector has been renamed to `replication_slots`.
Users explicitly enabling or disabling this collector must update their flags from
`--collector.replication_slot` / `--no-collector.replication_slot` to
`--collector.replication_slots` / `--no-collector.replication_slots`.
The metrics previously exposed by the `replication_slot` collector have also been
renamed to use the `pg_replication_slots_` prefix. Update alerts, recording
rules, and dashboards that reference these metrics:
* `pg_replication_slot_slot_current_wal_lsn` -> `pg_replication_slots_slot_current_wal_lsn`
* `pg_replication_slot_slot_confirmed_flush_lsn` -> `pg_replication_slots_slot_confirmed_flush_lsn`
* `pg_replication_slot_slot_is_active` -> `pg_replication_slots_slot_is_active`
* `pg_replication_slot_safe_wal_size_bytes` -> `pg_replication_slots_safe_wal_size_bytes`
* `pg_replication_slot_wal_status` -> `pg_replication_slots_wal_status`
Some metrics previously emitted through the legacy default metric map are now
emitted by the standalone `replication_slots` collector. This affects
`pg_replication_slots_active`, `pg_replication_slots_pg_wal_lsn_diff`, and
`pg_replication_slots_pg_xlog_location_diff`. These metrics no longer honor
`--metric-prefix`, no longer include deprecated `constantLabels`, and are no
longer disabled by `--disable-default-metrics`. To stop collecting them, disable
the `replication_slots` collector explicitly with `--no-collector.replication_slots`.
* [CHANGE] ...
* [FEATURE] ...
* [ENHANCEMENT] ...

4
README.md
View File

@@ -133,8 +133,8 @@ This will build the docker image as `prometheuscommunity/postgres_exporter:${bra
* `[no-]collector.replication`
Enable the `replication` collector (default: enabled).
* `[no-]collector.replication_slot`
Enable the `replication_slot` collector (default: enabled).
* `[no-]collector.replication_slots`
Enable the `replication_slots` collector (default: enabled).
* `[no-]collector.settings`
Enable the `settings` collector (default: enabled).

204
collector/pg_replication_slot.go
View File

@@ -1,204 +0,0 @@
// Copyright 2023 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package collector
import (
"context"
"database/sql"
"log/slog"
"github.com/blang/semver/v4"
"github.com/prometheus/client_golang/prometheus"
)
const replicationSlotSubsystem = "replication_slot"
func init() {
registerCollector(replicationSlotSubsystem, defaultEnabled, NewPGReplicationSlotCollector)
}
type PGReplicationSlotCollector struct {
log *slog.Logger
}
func NewPGReplicationSlotCollector(config collectorConfig) (Collector, error) {
return &PGReplicationSlotCollector{log: config.logger}, nil
}
var (
pgReplicationSlotCurrentWalDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotSubsystem,
"slot_current_wal_lsn",
),
"current wal lsn value",
[]string{"slot_name", "slot_type"}, nil,
)
pgReplicationSlotCurrentFlushDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotSubsystem,
"slot_confirmed_flush_lsn",
),
"last lsn confirmed flushed to the replication slot",
[]string{"slot_name", "slot_type"}, nil,
)
pgReplicationSlotIsActiveDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotSubsystem,
"slot_is_active",
),
"whether the replication slot is active or not",
[]string{"slot_name", "slot_type"}, nil,
)
pgReplicationSlotSafeWal = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotSubsystem,
"safe_wal_size_bytes",
),
"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,
slot_type,
CASE WHEN pg_is_in_recovery() THEN
pg_last_wal_receive_lsn() - '0/0'
ELSE
pg_current_wal_lsn() - '0/0'
END AS current_wal_lsn,
COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
active
FROM pg_replication_slots;`
pgReplicationSlotNewQuery = `SELECT
slot_name,
slot_type,
CASE WHEN pg_is_in_recovery() THEN
pg_last_wal_receive_lsn() - '0/0'
ELSE
pg_current_wal_lsn() - '0/0'
END AS current_wal_lsn,
COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
active,
safe_wal_size,
wal_status
FROM pg_replication_slots;`
)
func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
query := pgReplicationSlotQuery
abovePG13 := instance.version.GTE(semver.MustParse("13.0.0"))
if abovePG13 {
query = pgReplicationSlotNewQuery
}
db := instance.getDB()
rows, err := db.QueryContext(ctx,
query)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var slotName sql.NullString
var slotType sql.NullString
var walLSN sql.NullFloat64
var flushLSN sql.NullFloat64
var isActive sql.NullBool
var safeWalSize sql.NullInt64
var walStatus sql.NullString
r := []any{
&slotName,
&slotType,
&walLSN,
&flushLSN,
&isActive,
}
if abovePG13 {
r = append(r, &safeWalSize)
r = append(r, &walStatus)
}
err := rows.Scan(r...)
if err != nil {
return err
}
isActiveValue := 0.0
if isActive.Valid && isActive.Bool {
isActiveValue = 1.0
}
slotNameLabel := "unknown"
if slotName.Valid {
slotNameLabel = slotName.String
}
slotTypeLabel := "unknown"
if slotType.Valid {
slotTypeLabel = slotType.String
}
var walLSNMetric float64
if walLSN.Valid {
walLSNMetric = walLSN.Float64
}
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotCurrentWalDesc,
prometheus.GaugeValue, walLSNMetric, slotNameLabel, slotTypeLabel,
)
if isActive.Valid && isActive.Bool {
var flushLSNMetric float64
if flushLSN.Valid {
flushLSNMetric = flushLSN.Float64
}
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotCurrentFlushDesc,
prometheus.GaugeValue, flushLSNMetric, slotNameLabel, slotTypeLabel,
)
}
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotIsActiveDesc,
prometheus.GaugeValue, isActiveValue, slotNameLabel, slotTypeLabel,
)
if safeWalSize.Valid {
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotSafeWal,
prometheus.GaugeValue, float64(safeWalSize.Int64), slotNameLabel, slotTypeLabel,
)
}
if walStatus.Valid {
ch <- prometheus.MustNewConstMetric(
pgReplicationSlotWalStatus,
prometheus.GaugeValue, 1, slotNameLabel, slotTypeLabel, walStatus.String,
)
}
}
return rows.Err()
}

192
collector/pg_replication_slot_test.go
View File

@@ -1,192 +0,0 @@
// Copyright 2023 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package collector
import (
"context"
"testing"
"github.com/DATA-DOG/go-sqlmock"
"github.com/blang/semver/v4"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
"github.com/smartystreets/goconvey/convey"
)
func TestPgReplicationSlotCollectorActive(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("13.3.7")}
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, "reserved")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotNewQuery)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGPostmasterCollector.Update: %s", err)
}
}()
expected := []MetricResult{
{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},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical", "wal_status": "reserved"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPgReplicationSlotCollectorInActive(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("13.3.7")}
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, "extended")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotNewQuery)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotCollector.Update: %s", err)
}
}()
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},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical", "wal_status": "extended"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPgReplicationSlotCollectorActiveNil(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("13.3.7")}
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, "lost")
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotNewQuery)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotCollector.Update: %s", err)
}
}()
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() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPgReplicationSlotCollectorTestNilValues(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("13.3.7")}
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, nil)
mock.ExpectQuery(sanitizeQuery(pgReplicationSlotNewQuery)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotCollector.Update: %s", err)
}
}()
expected := []MetricResult{
{labels: labelMap{"slot_name": "unknown", "slot_type": "unknown"}, value: 0, metricType: dto.MetricType_GAUGE},
{labels: labelMap{"slot_name": "unknown", "slot_type": "unknown"}, value: 0, metricType: dto.MetricType_GAUGE},
{labels: labelMap{"slot_name": "unknown", "slot_type": "unknown"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}

342
collector/pg_replication_slots.go Normal file
View File

@@ -0,0 +1,342 @@
// Copyright The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package collector
import (
"context"
"database/sql"
"github.com/blang/semver/v4"
"github.com/prometheus/client_golang/prometheus"
)
const replicationSlotsSubsystem = "replication_slots"
func init() {
registerCollector(replicationSlotsSubsystem, defaultEnabled, NewPGReplicationSlotsCollector)
}
type PGReplicationSlotsCollector struct{}
func NewPGReplicationSlotsCollector(collectorConfig) (Collector, error) {
return &PGReplicationSlotsCollector{}, nil
}
var (
replicationSlotsLabels = []string{"slot_name", "database"}
replicationSlotsSlotLabels = []string{"slot_name", "slot_type"}
replicationSlotsActiveDesc = prometheus.NewDesc(
prometheus.BuildFQName(namespace, replicationSlotsSubsystem, "active"),
"Flag indicating if the slot is active",
replicationSlotsLabels,
prometheus.Labels{},
)
replicationSlotsWalLSNDiffDesc = prometheus.NewDesc(
prometheus.BuildFQName(namespace, replicationSlotsSubsystem, "pg_wal_lsn_diff"),
"Replication lag in bytes",
replicationSlotsLabels,
prometheus.Labels{},
)
replicationSlotsXlogLocationDiffDesc = prometheus.NewDesc(
prometheus.BuildFQName(namespace, replicationSlotsSubsystem, "pg_xlog_location_diff"),
"Unknown metric from pg_replication_slots",
replicationSlotsLabels,
prometheus.Labels{},
)
replicationSlotsCurrentWalDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotsSubsystem,
"slot_current_wal_lsn",
),
"current wal lsn value",
replicationSlotsSlotLabels, nil,
)
replicationSlotsCurrentFlushDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotsSubsystem,
"slot_confirmed_flush_lsn",
),
"last lsn confirmed flushed to the replication slot",
replicationSlotsSlotLabels, nil,
)
replicationSlotsIsActiveDesc = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotsSubsystem,
"slot_is_active",
),
"whether the replication slot is active or not",
replicationSlotsSlotLabels, nil,
)
replicationSlotsSafeWal = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotsSubsystem,
"safe_wal_size_bytes",
),
"number of bytes that can be written to WAL such that this slot is not in danger of getting in state lost",
replicationSlotsSlotLabels, nil,
)
replicationSlotsWalStatus = prometheus.NewDesc(
prometheus.BuildFQName(
namespace,
replicationSlotsSubsystem,
"wal_status",
),
"availability of WAL files claimed by this slot",
[]string{"slot_name", "slot_type", "wal_status"}, nil,
)
replicationSlotsQuery = `
SELECT slot_name, database, active,
(case pg_is_in_recovery() when 't' then pg_wal_lsn_diff(pg_last_wal_receive_lsn(), restart_lsn) else pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) end) as pg_wal_lsn_diff
FROM pg_replication_slots
`
replicationSlotsQueryBefore10 = `
SELECT slot_name, database, active,
(case pg_is_in_recovery() when 't' then pg_xlog_location_diff(pg_last_xlog_receive_location(), restart_lsn) else pg_xlog_location_diff(pg_current_xlog_location(), restart_lsn) end) as pg_xlog_location_diff
FROM pg_replication_slots
`
replicationSlotsSlotQuery = `SELECT
slot_name,
slot_type,
CASE WHEN pg_is_in_recovery() THEN
pg_last_wal_receive_lsn() - '0/0'
ELSE
pg_current_wal_lsn() - '0/0'
END AS current_wal_lsn,
COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
active
FROM pg_replication_slots;`
replicationSlotsSlotNewQuery = `SELECT
slot_name,
slot_type,
CASE WHEN pg_is_in_recovery() THEN
pg_last_wal_receive_lsn() - '0/0'
ELSE
pg_current_wal_lsn() - '0/0'
END AS current_wal_lsn,
COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
active,
safe_wal_size,
wal_status
FROM pg_replication_slots;`
)
func (PGReplicationSlotsCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
switch {
case instance.version.GTE(semver.MustParse("10.0.0")):
if err := updateReplicationSlots(ctx, instance, ch); err != nil {
return err
}
return updateReplicationSlotsSlotMetrics(ctx, instance, ch)
case instance.version.GTE(semver.MustParse("9.4.0")):
return updateReplicationSlotsBefore10(ctx, instance, ch)
default:
return nil
}
}
func updateReplicationSlots(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
db := instance.getDB()
rows, err := db.QueryContext(ctx, replicationSlotsQuery)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var slotName, database sql.NullString
var active sql.NullBool
var walLSNDiff sql.NullFloat64
if err := rows.Scan(&slotName, &database, &active, &walLSNDiff); err != nil {
return err
}
labels := replicationSlotsLabelValues(slotName, database)
emitReplicationSlotsActive(ch, active, labels)
if walLSNDiff.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsWalLSNDiffDesc,
prometheus.GaugeValue,
walLSNDiff.Float64,
labels...,
)
}
}
return rows.Err()
}
func updateReplicationSlotsBefore10(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
db := instance.getDB()
rows, err := db.QueryContext(ctx, replicationSlotsQueryBefore10)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var slotName, database sql.NullString
var active sql.NullBool
var xlogLocationDiff sql.NullFloat64
if err := rows.Scan(&slotName, &database, &active, &xlogLocationDiff); err != nil {
return err
}
labels := replicationSlotsLabelValues(slotName, database)
emitReplicationSlotsActive(ch, active, labels)
if xlogLocationDiff.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsXlogLocationDiffDesc,
prometheus.UntypedValue,
xlogLocationDiff.Float64,
labels...,
)
}
}
return rows.Err()
}
func updateReplicationSlotsSlotMetrics(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
query := replicationSlotsSlotQuery
abovePG13 := instance.version.GTE(semver.MustParse("13.0.0"))
if abovePG13 {
query = replicationSlotsSlotNewQuery
}
db := instance.getDB()
rows, err := db.QueryContext(ctx, query)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var slotName sql.NullString
var slotType sql.NullString
var walLSN sql.NullFloat64
var flushLSN sql.NullFloat64
var isActive sql.NullBool
var safeWalSize sql.NullInt64
var walStatus sql.NullString
r := []any{
&slotName,
&slotType,
&walLSN,
&flushLSN,
&isActive,
}
if abovePG13 {
r = append(r, &safeWalSize)
r = append(r, &walStatus)
}
if err := rows.Scan(r...); err != nil {
return err
}
slotLabels := replicationSlotsSlotLabelValues(slotName, slotType)
if walLSN.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsCurrentWalDesc,
prometheus.GaugeValue, walLSN.Float64, slotLabels...,
)
}
if isActive.Valid && isActive.Bool && flushLSN.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsCurrentFlushDesc,
prometheus.GaugeValue, flushLSN.Float64, slotLabels...,
)
}
emitReplicationSlotsSlotIsActive(ch, isActive, slotLabels)
if safeWalSize.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsSafeWal,
prometheus.GaugeValue, float64(safeWalSize.Int64), slotLabels...,
)
}
if walStatus.Valid {
ch <- prometheus.MustNewConstMetric(
replicationSlotsWalStatus,
prometheus.GaugeValue, 1, slotLabels[0], slotLabels[1], walStatus.String,
)
}
}
return rows.Err()
}
func emitReplicationSlotsActive(ch chan<- prometheus.Metric, active sql.NullBool, labels []string) {
if !active.Valid {
return
}
activeValue := 0.0
if active.Bool {
activeValue = 1.0
}
ch <- prometheus.MustNewConstMetric(
replicationSlotsActiveDesc,
prometheus.GaugeValue,
activeValue,
labels...,
)
}
func emitReplicationSlotsSlotIsActive(ch chan<- prometheus.Metric, active sql.NullBool, labels []string) {
if !active.Valid {
return
}
activeValue := 0.0
if active.Bool {
activeValue = 1.0
}
ch <- prometheus.MustNewConstMetric(
replicationSlotsIsActiveDesc,
prometheus.GaugeValue,
activeValue,
labels...,
)
}
func replicationSlotsLabelValues(slotName, database sql.NullString) []string {
return []string{replicationSlotsNullStringValue(slotName), replicationSlotsNullStringValue(database)}
}
func replicationSlotsNullStringValue(s sql.NullString) string {
if s.Valid {
return s.String
}
return ""
}
func replicationSlotsSlotLabelValues(slotName, slotType sql.NullString) []string {
return []string{unknownStringValue(slotName), unknownStringValue(slotType)}
}
func unknownStringValue(s sql.NullString) string {
if s.Valid {
return s.String
}
return "unknown"
}

282
collector/pg_replication_slots_test.go Normal file
View File

@@ -0,0 +1,282 @@
// Copyright The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package collector
import (
"context"
"testing"
"github.com/DATA-DOG/go-sqlmock"
"github.com/blang/semver/v4"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
"github.com/smartystreets/goconvey/convey"
)
func TestPGReplicationSlotsCollector(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{"slot_name", "database", "active", "pg_wal_lsn_diff"}).
AddRow("slot_a", "postgres", true, 64.0)
mock.ExpectQuery(sanitizeQuery(replicationSlotsQuery)).WillReturnRows(rows)
slotRows := sqlmock.NewRows([]string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}).
AddRow("slot_a", "physical", 5, 3, true, 323906992, "reserved")
mock.ExpectQuery(sanitizeQuery(replicationSlotsSlotNewQuery)).WillReturnRows(slotRows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
expectedLabels := labelMap{"slot_name": "slot_a", "database": "postgres"}
expectedSlotLabels := labelMap{"slot_name": "slot_a", "slot_type": "physical"}
expected := []MetricResult{
{labels: expectedLabels, value: 1, metricType: dto.MetricType_GAUGE},
{labels: expectedLabels, value: 64, metricType: dto.MetricType_GAUGE},
{labels: expectedSlotLabels, value: 5, metricType: dto.MetricType_GAUGE},
{labels: expectedSlotLabels, value: 3, metricType: dto.MetricType_GAUGE},
{labels: expectedSlotLabels, value: 1, metricType: dto.MetricType_GAUGE},
{labels: expectedSlotLabels, value: 323906992, metricType: dto.MetricType_GAUGE},
{labels: labelMap{"slot_name": "slot_a", "slot_type": "physical", "wal_status": "reserved"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorBefore10(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("9.6.0")}
rows := sqlmock.NewRows([]string{"slot_name", "database", "active", "pg_xlog_location_diff"}).
AddRow("slot_a", "postgres", false, 32.0)
mock.ExpectQuery(sanitizeQuery(replicationSlotsQueryBefore10)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
expectedLabels := labelMap{"slot_name": "slot_a", "database": "postgres"}
expected := []MetricResult{
{labels: expectedLabels, value: 0, metricType: dto.MetricType_GAUGE},
{labels: expectedLabels, value: 32, metricType: dto.MetricType_UNTYPED},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorSlotInactive(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("13.3.7")}
rows := sqlmock.NewRows([]string{"slot_name", "database", "active", "pg_wal_lsn_diff"})
mock.ExpectQuery(sanitizeQuery(replicationSlotsQuery)).WillReturnRows(rows)
slotRows := sqlmock.NewRows([]string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}).
AddRow("test_slot", "physical", 6, 12, false, -4000, "extended")
mock.ExpectQuery(sanitizeQuery(replicationSlotsSlotNewQuery)).WillReturnRows(slotRows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
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},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "physical", "wal_status": "extended"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorSlotBefore13(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("12.0.0")}
rows := sqlmock.NewRows([]string{"slot_name", "database", "active", "pg_wal_lsn_diff"})
mock.ExpectQuery(sanitizeQuery(replicationSlotsQuery)).WillReturnRows(rows)
slotRows := sqlmock.NewRows([]string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active"}).
AddRow("test_slot", "logical", 10, 8, true)
mock.ExpectQuery(sanitizeQuery(replicationSlotsSlotQuery)).WillReturnRows(slotRows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
expected := []MetricResult{
{labels: labelMap{"slot_name": "test_slot", "slot_type": "logical"}, value: 10, metricType: dto.MetricType_GAUGE},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "logical"}, value: 8, metricType: dto.MetricType_GAUGE},
{labels: labelMap{"slot_name": "test_slot", "slot_type": "logical"}, value: 1, metricType: dto.MetricType_GAUGE},
}
convey.Convey("Metrics comparison", t, func() {
for _, expect := range expected {
m := readMetric(<-ch)
convey.So(expect, convey.ShouldResemble, m)
}
})
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorNullValues(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{"slot_name", "database", "active", "pg_wal_lsn_diff"}).
AddRow(nil, nil, nil, nil)
mock.ExpectQuery(sanitizeQuery(replicationSlotsQuery)).WillReturnRows(rows)
slotRows := sqlmock.NewRows([]string{"slot_name", "slot_type", "current_wal_lsn", "confirmed_flush_lsn", "active", "safe_wal_size", "wal_status"}).
AddRow(nil, nil, nil, nil, nil, nil, nil)
mock.ExpectQuery(sanitizeQuery(replicationSlotsSlotNewQuery)).WillReturnRows(slotRows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
if metric, ok := <-ch; ok {
t.Fatalf("unexpected metric emitted for NULL replication_slots value: %s", metric.Desc())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorBefore10NullValues(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("9.6.0")}
rows := sqlmock.NewRows([]string{"slot_name", "database", "active", "pg_xlog_location_diff"}).
AddRow(nil, nil, nil, nil)
mock.ExpectQuery(sanitizeQuery(replicationSlotsQueryBefore10)).WillReturnRows(rows)
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
if metric, ok := <-ch; ok {
t.Fatalf("unexpected metric emitted for NULL replication_slots value: %s", metric.Desc())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}
func TestPGReplicationSlotsCollectorBefore94(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("9.3.0")}
ch := make(chan prometheus.Metric)
go func() {
defer close(ch)
c := PGReplicationSlotsCollector{}
if err := c.Update(context.Background(), inst, ch); err != nil {
t.Errorf("Error calling PGReplicationSlotsCollector.Update: %s", err)
}
}()
if metric, ok := <-ch; ok {
t.Fatalf("unexpected metric emitted for PostgreSQL 9.3: %s", metric.Desc())
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("there were unfulfilled exceptions: %s", err)
}
}

22
exporter/postgres_exporter.go
View File

@@ -158,15 +158,7 @@ func (e *ErrorConnectToServer) Error() string {
// TODO: revisit this with the semver system
func DumpMaps() {
// TODO: make this function part of the exporter
for name, cmap := range builtinMetricMaps {
query, ok := queryOverrides[name]
if !ok {
fmt.Println(name)
} else {
for _, queryOverride := range query {
fmt.Println(name, queryOverride.versionRange, queryOverride.query)
}
}
for _, cmap := range builtinMetricMaps {
for column, details := range cmap.columnMappings {
fmt.Printf(" %-40s %v\n", column, details)
@@ -189,16 +181,6 @@ var builtinMetricMaps = map[string]intermediateMetricMap{
true,
0,
},
"pg_replication_slots": {
map[string]ColumnMapping{
"slot_name": {LABEL, "Name of the replication slot", nil, nil},
"database": {LABEL, "Name of the database", nil, nil},
"active": {GAUGE, "Flag indicating if the slot is active", nil, nil},
"pg_wal_lsn_diff": {GAUGE, "Replication lag in bytes", nil, nil},
},
true,
0,
},
}
// Turn the MetricMap column mapping into a prometheus descriptor mapping.
@@ -558,7 +540,7 @@ func (e *Exporter) checkMapVersions(ch chan<- prometheus.Metric, server *Server)
// Get Default Metrics only for master database
if !e.disableDefaultMetrics && server.master {
server.metricMap = makeDescMap(semanticVersion, server.labels, e.builtinMetricMaps, server.logger, e.metricPrefix)
server.queryOverrides = makeQueryOverrideMap(semanticVersion, queryOverrides, server.logger)
server.queryOverrides = make(map[string]string)
} else {
server.metricMap = make(map[string]MetricMapNamespace)
server.queryOverrides = make(map[string]string)

56
exporter/queries.go
View File

@@ -34,62 +34,6 @@ type UserQuery struct {
// UserQueries represents a set of UserQuery objects
type UserQueries map[string]UserQuery
// OverrideQuery 's are run in-place of simple namespace look ups, and provide
// advanced functionality. But they have a tendency to postgres version specific.
// There aren't too many versions, so we simply store customized versions using
// the semver matching we do for columns.
type OverrideQuery struct {
versionRange semver.Range
query string
}
// Overriding queries for namespaces above.
// TODO: validate this is a closed set in tests, and there are no overlaps
var queryOverrides = map[string][]OverrideQuery{
"pg_replication_slots": {
{
semver.MustParseRange(">=9.4.0 <10.0.0"),
`
SELECT slot_name, database, active,
(case pg_is_in_recovery() when 't' then pg_xlog_location_diff(pg_last_xlog_receive_location(), restart_lsn) else pg_xlog_location_diff(pg_current_xlog_location(), restart_lsn) end) as pg_xlog_location_diff
FROM pg_replication_slots
`,
},
{
semver.MustParseRange(">=10.0.0"),
`
SELECT slot_name, database, active,
(case pg_is_in_recovery() when 't' then pg_wal_lsn_diff(pg_last_wal_receive_lsn(), restart_lsn) else pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) end) as pg_wal_lsn_diff
FROM pg_replication_slots
`,
},
},
}
// Convert the query override file to the version-specific query override file
// for the exporter.
func makeQueryOverrideMap(pgVersion semver.Version, queryOverrides map[string][]OverrideQuery, logger *slog.Logger) map[string]string {
resultMap := make(map[string]string)
for name, overrideDef := range queryOverrides {
// Find a matching semver. We make it an error to have overlapping
// ranges at test-time, so only 1 should ever match.
matched := false
for _, queryDef := range overrideDef {
if queryDef.versionRange(pgVersion) {
resultMap[name] = queryDef.query
matched = true
break
}
}
if !matched {
logger.Warn("No query matched override, disabling metric space", "name", name)
resultMap[name] = ""
}
}
return resultMap
}
func parseUserQueries(content []byte, logger *slog.Logger) (map[string]intermediateMetricMap, map[string]string, error) {
var userQueries UserQueries