From c8eb54d64a403a27085ba1c4432b82842fecec33 Mon Sep 17 00:00:00 2001 From: Arthur Silva Sens Date: Tue, 2 Jun 2026 22:22:43 -0300 Subject: [PATCH] Refactor stat_replication_slots into standalone collector (#1307) * Refactor stat_replication_slots into standalone collector Signed-off-by: Arthur Silva Sens * Do not expose metrics for NULL results Signed-off-by: Arthur Silva Sens * Merge slot and slots collectors Signed-off-by: Arthur Silva Sens * Remove unused queries map Signed-off-by: Arthur Silva Sens * Add changelog entry for breaking changes Signed-off-by: Arthur Silva Sens --------- Signed-off-by: Arthur Silva Sens --- CHANGELOG.md | 25 ++ README.md | 4 +- collector/pg_replication_slot.go | 204 --------------- collector/pg_replication_slot_test.go | 192 -------------- collector/pg_replication_slots.go | 342 +++++++++++++++++++++++++ collector/pg_replication_slots_test.go | 282 ++++++++++++++++++++ exporter/postgres_exporter.go | 22 +- exporter/queries.go | 56 ---- 8 files changed, 653 insertions(+), 474 deletions(-) delete mode 100644 collector/pg_replication_slot.go delete mode 100644 collector/pg_replication_slot_test.go create mode 100644 collector/pg_replication_slots.go create mode 100644 collector/pg_replication_slots_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cc3b9c4..c6f346bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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] ... diff --git a/README.md b/README.md index a3577c15..8f269299 100644 --- a/README.md +++ b/README.md @@ -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). diff --git a/collector/pg_replication_slot.go b/collector/pg_replication_slot.go deleted file mode 100644 index e6c9773e..00000000 --- a/collector/pg_replication_slot.go +++ /dev/null @@ -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() -} diff --git a/collector/pg_replication_slot_test.go b/collector/pg_replication_slot_test.go deleted file mode 100644 index 981b5db6..00000000 --- a/collector/pg_replication_slot_test.go +++ /dev/null @@ -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) - } -} diff --git a/collector/pg_replication_slots.go b/collector/pg_replication_slots.go new file mode 100644 index 00000000..088fb9a3 --- /dev/null +++ b/collector/pg_replication_slots.go @@ -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" +} diff --git a/collector/pg_replication_slots_test.go b/collector/pg_replication_slots_test.go new file mode 100644 index 00000000..ceb602ba --- /dev/null +++ b/collector/pg_replication_slots_test.go @@ -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) + } +} diff --git a/exporter/postgres_exporter.go b/exporter/postgres_exporter.go index 4c918d82..3f49182a 100644 --- a/exporter/postgres_exporter.go +++ b/exporter/postgres_exporter.go @@ -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) diff --git a/exporter/queries.go b/exporter/queries.go index cb24f0b1..db8b6dac 100644 --- a/exporter/queries.go +++ b/exporter/queries.go @@ -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