From 300750001c2957386ee3c3af3b35ef3fe79393b6 Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Tue, 24 Jan 2023 14:45:31 -0500 Subject: [PATCH 1/6] adding codified functionality for logical replication metrics Signed-off-by: Zachary Caldarola --- collector/pg_replication_slots.go | 72 +++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 collector/pg_replication_slots.go diff --git a/collector/pg_replication_slots.go b/collector/pg_replication_slots.go new file mode 100644 index 000000000..1da841e3c --- /dev/null +++ b/collector/pg_replication_slots.go @@ -0,0 +1,72 @@ +// Copyright 2022 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/go-kit/log" + "github.com/prometheus/client_golang/prometheus" +) + +func init() { + registerCollector("replication_slot", defaultEnabled, NewPGReplicationSlotCollector) +} + +type PGReplicationSlotCollector struct { + log log.Logger +} + +func NewPGReplicationSlotCollector(logger log.Logger) (Collector, error) { + return &PGReplicationSlotCollector{log: logger}, nil +} + +var pgReplicationSlot = map[string]*prometheus.Desc{ + "lsn_distance": prometheus.NewDesc( + "pg_replication_slot_lsn_distance", + "Disk space used by the database", + []string{"slot_name"}, nil, + ), +} + +func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch chan<- prometheus.Metric) error { + rows, err := db.QueryContext(ctx, + `SELECT + slot_name, + (pg_current_wal_lsn() - confirmed_flush_lsn) AS lsn_distance + FROM + pg_replication_slots;`) + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + var slot_name string + var size int64 + if err := rows.Scan(&slot_name, &size); err != nil { + return err + } + + ch <- prometheus.MustNewConstMetric( + pgReplicationSlot["size_bytes"], + prometheus.GaugeValue, float64(size), slot_name, + ) + } + if err := rows.Err(); err != nil { + return err + } + return nil +} From 9b13780d142ba5b6abf1dca84504170137acb970 Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Wed, 25 Jan 2023 11:15:24 -0500 Subject: [PATCH 2/6] addressing comments Signed-off-by: Zachary Caldarola --- collector/pg_replication_slots.go | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/collector/pg_replication_slots.go b/collector/pg_replication_slots.go index 1da841e3c..c21d7f411 100644 --- a/collector/pg_replication_slots.go +++ b/collector/pg_replication_slots.go @@ -1,4 +1,4 @@ -// Copyright 2022 The Prometheus Authors +// 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 @@ -34,9 +34,14 @@ func NewPGReplicationSlotCollector(logger log.Logger) (Collector, error) { } var pgReplicationSlot = map[string]*prometheus.Desc{ - "lsn_distance": prometheus.NewDesc( - "pg_replication_slot_lsn_distance", - "Disk space used by the database", + "current_wal_lsn": prometheus.NewDesc( + "pg_replication_slot_current_wal_lsn", + "current wal lsn value", + []string{"slot_name"}, nil, + ), + "confirmed_flush_lsn": prometheus.NewDesc( + "pg_replication_slot_confirmed_flush_lsn", + "last lsn confirmed flushed to the replication slot", []string{"slot_name"}, nil, ), } @@ -45,7 +50,8 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha rows, err := db.QueryContext(ctx, `SELECT slot_name, - (pg_current_wal_lsn() - confirmed_flush_lsn) AS lsn_distance + pg_current_wal_lsn() AS current_wal_lsn, + confirmed_flush_lsn FROM pg_replication_slots;`) if err != nil { @@ -55,14 +61,19 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha for rows.Next() { var slot_name string - var size int64 - if err := rows.Scan(&slot_name, &size); err != nil { + var wal_lsn int64 + var flush_lsn int64 + if err := rows.Scan(&slot_name, &wal_lsn, &flush_lsn); err != nil { return err } ch <- prometheus.MustNewConstMetric( - pgReplicationSlot["size_bytes"], - prometheus.GaugeValue, float64(size), slot_name, + pgReplicationSlot["current_wal_lsn"], + prometheus.GaugeValue, float64(wal_lsn), slot_name, + ) + ch <- prometheus.MustNewConstMetric( + pgReplicationSlot["confirmed_flush_lsn"], + prometheus.GaugeValue, float64(flush_lsn), slot_name, ) } if err := rows.Err(); err != nil { From 600ad185cf47aa1180dbb073d55c22b6caf3587c Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Sat, 28 Jan 2023 17:05:45 -0500 Subject: [PATCH 3/6] more comments Signed-off-by: Zachary Caldarola --- ...lication_slots.go => replication_slots.go} | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) rename collector/{pg_replication_slots.go => replication_slots.go} (74%) diff --git a/collector/pg_replication_slots.go b/collector/replication_slots.go similarity index 74% rename from collector/pg_replication_slots.go rename to collector/replication_slots.go index c21d7f411..e1c2f0a8d 100644 --- a/collector/pg_replication_slots.go +++ b/collector/replication_slots.go @@ -44,16 +44,22 @@ var pgReplicationSlot = map[string]*prometheus.Desc{ "last lsn confirmed flushed to the replication slot", []string{"slot_name"}, nil, ), + "is_active": prometheus.NewDesc( + "pg_replication_slot_is_active", + "last lsn confirmed flushed to the replication slot", + []string{"slot_name"}, nil, + ), } func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch chan<- prometheus.Metric) error { rows, err := db.QueryContext(ctx, `SELECT slot_name, - pg_current_wal_lsn() AS current_wal_lsn, - confirmed_flush_lsn - FROM - pg_replication_slots;`) + pg_current_wal_lsn() - '0/0' AS current_wal_lsn, + coalesce(confirmed_flush_lsn, '0/0') - '0/0', + active + FROM + pg_replication_slots;`) if err != nil { return err } @@ -63,7 +69,8 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha var slot_name string var wal_lsn int64 var flush_lsn int64 - if err := rows.Scan(&slot_name, &wal_lsn, &flush_lsn); err != nil { + var is_active int + if err := rows.Scan(&slot_name, &wal_lsn, &flush_lsn, &is_active); err != nil { return err } @@ -71,9 +78,15 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha pgReplicationSlot["current_wal_lsn"], prometheus.GaugeValue, float64(wal_lsn), slot_name, ) + if (is_active == 1) { + ch <- prometheus.MustNewConstMetric( + pgReplicationSlot["confirmed_flush_lsn"], + prometheus.GaugeValue, float64(flush_lsn), slot_name, + ) + } ch <- prometheus.MustNewConstMetric( - pgReplicationSlot["confirmed_flush_lsn"], - prometheus.GaugeValue, float64(flush_lsn), slot_name, + pgReplicationSlot["is_active"], + prometheus.GaugeValue, int(flush_lsn), slot_name, ) } if err := rows.Err(); err != nil { From 1ec69c9168c042314b8ef67360ba8e18f0e4fd84 Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Sat, 28 Jan 2023 17:37:41 -0500 Subject: [PATCH 4/6] fmt Signed-off-by: Zachary Caldarola --- collector/replication_slots.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/collector/replication_slots.go b/collector/replication_slots.go index e1c2f0a8d..8705f69c1 100644 --- a/collector/replication_slots.go +++ b/collector/replication_slots.go @@ -78,7 +78,7 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha pgReplicationSlot["current_wal_lsn"], prometheus.GaugeValue, float64(wal_lsn), slot_name, ) - if (is_active == 1) { + if is_active == 1 { ch <- prometheus.MustNewConstMetric( pgReplicationSlot["confirmed_flush_lsn"], prometheus.GaugeValue, float64(flush_lsn), slot_name, From f3952f941b833f8626330d64ea885f3086b1f3a1 Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Sat, 28 Jan 2023 17:40:46 -0500 Subject: [PATCH 5/6] typing Signed-off-by: Zachary Caldarola --- collector/replication_slots.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/collector/replication_slots.go b/collector/replication_slots.go index 8705f69c1..129b2efd5 100644 --- a/collector/replication_slots.go +++ b/collector/replication_slots.go @@ -86,7 +86,7 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha } ch <- prometheus.MustNewConstMetric( pgReplicationSlot["is_active"], - prometheus.GaugeValue, int(flush_lsn), slot_name, + prometheus.GaugeValue, float64(flush_lsn), slot_name, ) } if err := rows.Err(); err != nil { From 076eb2c3f54bf6d95ad59fcb63a96f5a29e31636 Mon Sep 17 00:00:00 2001 From: Zachary Caldarola Date: Sat, 28 Jan 2023 18:45:48 -0500 Subject: [PATCH 6/6] fmt Signed-off-by: Zachary Caldarola --- collector/replication_slots.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/collector/replication_slots.go b/collector/replication_slots.go index 129b2efd5..224db3ccf 100644 --- a/collector/replication_slots.go +++ b/collector/replication_slots.go @@ -69,7 +69,7 @@ func (PGReplicationSlotCollector) Update(ctx context.Context, db *sql.DB, ch cha var slot_name string var wal_lsn int64 var flush_lsn int64 - var is_active int + var is_active int64 if err := rows.Scan(&slot_name, &wal_lsn, &flush_lsn, &is_active); err != nil { return err }