pangolin_exporter

1
// Copyright 2023 The Prometheus Authors
2
// Licensed under the Apache License, Version 2.0 (the "License");
3
// you may not use this file except in compliance with the License.
4
// You may obtain a copy of the License at
5
//
6
// http://www.apache.org/licenses/LICENSE-2.0
7
//
8
// Unless required by applicable law or agreed to in writing, software
9
// distributed under the License is distributed on an "AS IS" BASIS,
10
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11
// See the License for the specific language governing permissions and
12
// limitations under the License.
13

14
package collector
15

16
import (
17
	"context"
18
	"database/sql"
19

20
	"github.com/go-kit/log"
21
	"github.com/prometheus/client_golang/prometheus"
22
)
23

24
const replicationSlotSubsystem = "replication_slot"
25

26
func init() {
27
	registerCollector(replicationSlotSubsystem, defaultEnabled, NewPGReplicationSlotCollector)
28
}
29

30
type PGReplicationSlotCollector struct {
31
	log log.Logger
32
}
33

34
func NewPGReplicationSlotCollector(config collectorConfig) (Collector, error) {
35
	return &PGReplicationSlotCollector{log: config.logger}, nil
36
}
37

38
var (
39
	pgReplicationSlotCurrentWalDesc = prometheus.NewDesc(
40
		prometheus.BuildFQName(
41
			namespace,
42
			replicationSlotSubsystem,
43
			"slot_current_wal_lsn",
44
		),
45
		"current wal lsn value",
46
		[]string{"slot_name", "slot_type"}, nil,
47
	)
48
	pgReplicationSlotCurrentFlushDesc = prometheus.NewDesc(
49
		prometheus.BuildFQName(
50
			namespace,
51
			replicationSlotSubsystem,
52
			"slot_confirmed_flush_lsn",
53
		),
54
		"last lsn confirmed flushed to the replication slot",
55
		[]string{"slot_name", "slot_type"}, nil,
56
	)
57
	pgReplicationSlotIsActiveDesc = prometheus.NewDesc(
58
		prometheus.BuildFQName(
59
			namespace,
60
			replicationSlotSubsystem,
61
			"slot_is_active",
62
		),
63
		"whether the replication slot is active or not",
64
		[]string{"slot_name", "slot_type"}, nil,
65
	)
66

67
	pgReplicationSlotQuery = `SELECT
68
		slot_name,
69
		slot_type,
70
		CASE WHEN pg_is_in_recovery() THEN
71
		    pg_last_wal_receive_lsn() - '0/0'
72
		ELSE
73
		    pg_current_wal_lsn() - '0/0'
74
		END AS current_wal_lsn,
75
		COALESCE(confirmed_flush_lsn, '0/0') - '0/0' AS confirmed_flush_lsn,
76
		active
77
	FROM pg_replication_slots;`
78
)
79

80
func (PGReplicationSlotCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
81
	db := instance.getDB()
82
	rows, err := db.QueryContext(ctx,
83
		pgReplicationSlotQuery)
84
	if err != nil {
85
		return err
86
	}
87
	defer rows.Close()
88

89
	for rows.Next() {
90
		var slotName sql.NullString
91
		var slotType sql.NullString
92
		var walLSN sql.NullFloat64
93
		var flushLSN sql.NullFloat64
94
		var isActive sql.NullBool
95
		if err := rows.Scan(&slotName, &slotType, &walLSN, &flushLSN, &isActive); err != nil {
96
			return err
97
		}
98

99
		isActiveValue := 0.0
100
		if isActive.Valid && isActive.Bool {
101
			isActiveValue = 1.0
102
		}
103
		slotNameLabel := "unknown"
104
		if slotName.Valid {
105
			slotNameLabel = slotName.String
106
		}
107
		slotTypeLabel := "unknown"
108
		if slotType.Valid {
109
			slotTypeLabel = slotType.String
110
		}
111

112
		var walLSNMetric float64
113
		if walLSN.Valid {
114
			walLSNMetric = walLSN.Float64
115
		}
116
		ch <- prometheus.MustNewConstMetric(
117
			pgReplicationSlotCurrentWalDesc,
118
			prometheus.GaugeValue, walLSNMetric, slotNameLabel, slotTypeLabel,
119
		)
120
		if isActive.Valid && isActive.Bool {
121
			var flushLSNMetric float64
122
			if flushLSN.Valid {
123
				flushLSNMetric = flushLSN.Float64
124
			}
125
			ch <- prometheus.MustNewConstMetric(
126
				pgReplicationSlotCurrentFlushDesc,
127
				prometheus.GaugeValue, flushLSNMetric, slotNameLabel, slotTypeLabel,
128
			)
129
		}
130
		ch <- prometheus.MustNewConstMetric(
131
			pgReplicationSlotIsActiveDesc,
132
			prometheus.GaugeValue, isActiveValue, slotNameLabel, slotTypeLabel,
133
		)
134
	}
135
	return rows.Err()
136
}
137