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
package collector
14

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

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

25
func init() {
26
	registerCollector(statWalReceiverSubsystem, defaultDisabled, NewPGStatWalReceiverCollector)
27
}
28

29
type PGStatWalReceiverCollector struct {
30
	log log.Logger
31
}
32

33
const statWalReceiverSubsystem = "stat_wal_receiver"
34

35
func NewPGStatWalReceiverCollector(config collectorConfig) (Collector, error) {
36
	return &PGStatWalReceiverCollector{log: config.logger}, nil
37
}
38

39
var (
40
	labelCats                      = []string{"upstream_host", "slot_name", "status"}
41
	statWalReceiverReceiveStartLsn = prometheus.NewDesc(
42
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "receive_start_lsn"),
43
		"First write-ahead log location used when WAL receiver is started represented as a decimal",
44
		labelCats,
45
		prometheus.Labels{},
46
	)
47
	statWalReceiverReceiveStartTli = prometheus.NewDesc(
48
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "receive_start_tli"),
49
		"First timeline number used when WAL receiver is started",
50
		labelCats,
51
		prometheus.Labels{},
52
	)
53
	statWalReceiverFlushedLSN = prometheus.NewDesc(
54
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "flushed_lsn"),
55
		"Last write-ahead log location already received and flushed to disk, the initial value of this field being the first log location used when WAL receiver is started represented as a decimal",
56
		labelCats,
57
		prometheus.Labels{},
58
	)
59
	statWalReceiverReceivedTli = prometheus.NewDesc(
60
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "received_tli"),
61
		"Timeline number of last write-ahead log location received and flushed to disk",
62
		labelCats,
63
		prometheus.Labels{},
64
	)
65
	statWalReceiverLastMsgSendTime = prometheus.NewDesc(
66
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "last_msg_send_time"),
67
		"Send time of last message received from origin WAL sender",
68
		labelCats,
69
		prometheus.Labels{},
70
	)
71
	statWalReceiverLastMsgReceiptTime = prometheus.NewDesc(
72
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "last_msg_receipt_time"),
73
		"Send time of last message received from origin WAL sender",
74
		labelCats,
75
		prometheus.Labels{},
76
	)
77
	statWalReceiverLatestEndLsn = prometheus.NewDesc(
78
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "latest_end_lsn"),
79
		"Last write-ahead log location reported to origin WAL sender as integer",
80
		labelCats,
81
		prometheus.Labels{},
82
	)
83
	statWalReceiverLatestEndTime = prometheus.NewDesc(
84
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "latest_end_time"),
85
		"Time of last write-ahead log location reported to origin WAL sender",
86
		labelCats,
87
		prometheus.Labels{},
88
	)
89
	statWalReceiverUpstreamNode = prometheus.NewDesc(
90
		prometheus.BuildFQName(namespace, statWalReceiverSubsystem, "upstream_node"),
91
		"Node ID of the upstream node",
92
		labelCats,
93
		prometheus.Labels{},
94
	)
95

96
	pgStatWalColumnQuery = `
97
	SELECT
98
		column_name
99
	FROM information_schema.columns
100
	WHERE
101
		table_name = 'pg_stat_wal_receiver' and
102
		column_name = 'flushed_lsn'
103
	`
104

105
	pgStatWalReceiverQueryTemplate = `
106
	SELECT
107
		trim(both '''' from substring(conninfo from 'host=([^ ]*)')) as upstream_host,
108
		slot_name,
109
		status,
110
		(receive_start_lsn- '0/0') %% (2^52)::bigint as receive_start_lsn,
111
		%s
112
receive_start_tli,
113
		received_tli,
114
		extract(epoch from last_msg_send_time) as last_msg_send_time,
115
		extract(epoch from last_msg_receipt_time) as last_msg_receipt_time,
116
		(latest_end_lsn - '0/0') %% (2^52)::bigint as latest_end_lsn,
117
		extract(epoch from latest_end_time) as latest_end_time,
118
		substring(slot_name from 'repmgr_slot_([0-9]*)') as upstream_node
119
	FROM pg_catalog.pg_stat_wal_receiver
120
	`
121
)
122

123
func (c *PGStatWalReceiverCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
124
	db := instance.getDB()
125
	hasFlushedLSNRows, err := db.QueryContext(ctx, pgStatWalColumnQuery)
126
	if err != nil {
127
		return err
128
	}
129

130
	hasFlushedLSN := hasFlushedLSNRows.Next()
131
	var query string
132
	if hasFlushedLSN {
133
		query = fmt.Sprintf(pgStatWalReceiverQueryTemplate, "(flushed_lsn - '0/0') % (2^52)::bigint as flushed_lsn,\n")
134
	} else {
135
		query = fmt.Sprintf(pgStatWalReceiverQueryTemplate, "")
136
	}
137

138
	hasFlushedLSNRows.Close()
139

140
	rows, err := db.QueryContext(ctx, query)
141
	if err != nil {
142
		return err
143
	}
144
	defer rows.Close()
145
	for rows.Next() {
146
		var upstreamHost, slotName, status sql.NullString
147
		var receiveStartLsn, receiveStartTli, flushedLsn, receivedTli, latestEndLsn, upstreamNode sql.NullInt64
148
		var lastMsgSendTime, lastMsgReceiptTime, latestEndTime sql.NullFloat64
149

150
		if hasFlushedLSN {
151
			if err := rows.Scan(&upstreamHost, &slotName, &status, &receiveStartLsn, &receiveStartTli, &flushedLsn, &receivedTli, &lastMsgSendTime, &lastMsgReceiptTime, &latestEndLsn, &latestEndTime, &upstreamNode); err != nil {
152
				return err
153
			}
154
		} else {
155
			if err := rows.Scan(&upstreamHost, &slotName, &status, &receiveStartLsn, &receiveStartTli, &receivedTli, &lastMsgSendTime, &lastMsgReceiptTime, &latestEndLsn, &latestEndTime, &upstreamNode); err != nil {
156
				return err
157
			}
158
		}
159
		if !upstreamHost.Valid {
160
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because upstream host is null")
161
			continue
162
		}
163

164
		if !slotName.Valid {
165
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because slotname host is null")
166
			continue
167
		}
168

169
		if !status.Valid {
170
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because status is null")
171
			continue
172
		}
173
		labels := []string{upstreamHost.String, slotName.String, status.String}
174

175
		if !receiveStartLsn.Valid {
176
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because receive_start_lsn is null")
177
			continue
178
		}
179
		if !receiveStartTli.Valid {
180
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because receive_start_tli is null")
181
			continue
182
		}
183
		if hasFlushedLSN && !flushedLsn.Valid {
184
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because flushed_lsn is null")
185
			continue
186
		}
187
		if !receivedTli.Valid {
188
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because received_tli is null")
189
			continue
190
		}
191
		if !lastMsgSendTime.Valid {
192
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because last_msg_send_time is null")
193
			continue
194
		}
195
		if !lastMsgReceiptTime.Valid {
196
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because last_msg_receipt_time is null")
197
			continue
198
		}
199
		if !latestEndLsn.Valid {
200
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because latest_end_lsn is null")
201
			continue
202
		}
203
		if !latestEndTime.Valid {
204
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because latest_end_time is null")
205
			continue
206
		}
207
		if !upstreamNode.Valid {
208
			level.Debug(c.log).Log("msg", "Skipping wal receiver stats because upstream_node is null")
209
			continue
210
		}
211
		ch <- prometheus.MustNewConstMetric(
212
			statWalReceiverReceiveStartLsn,
213
			prometheus.CounterValue,
214
			float64(receiveStartLsn.Int64),
215
			labels...)
216

217
		ch <- prometheus.MustNewConstMetric(
218
			statWalReceiverReceiveStartTli,
219
			prometheus.GaugeValue,
220
			float64(receiveStartTli.Int64),
221
			labels...)
222

223
		if hasFlushedLSN {
224
			ch <- prometheus.MustNewConstMetric(
225
				statWalReceiverFlushedLSN,
226
				prometheus.CounterValue,
227
				float64(flushedLsn.Int64),
228
				labels...)
229
		}
230

231
		ch <- prometheus.MustNewConstMetric(
232
			statWalReceiverReceivedTli,
233
			prometheus.GaugeValue,
234
			float64(receivedTli.Int64),
235
			labels...)
236

237
		ch <- prometheus.MustNewConstMetric(
238
			statWalReceiverLastMsgSendTime,
239
			prometheus.CounterValue,
240
			float64(lastMsgSendTime.Float64),
241
			labels...)
242

243
		ch <- prometheus.MustNewConstMetric(
244
			statWalReceiverLastMsgReceiptTime,
245
			prometheus.CounterValue,
246
			float64(lastMsgReceiptTime.Float64),
247
			labels...)
248

249
		ch <- prometheus.MustNewConstMetric(
250
			statWalReceiverLatestEndLsn,
251
			prometheus.CounterValue,
252
			float64(latestEndLsn.Int64),
253
			labels...)
254

255
		ch <- prometheus.MustNewConstMetric(
256
			statWalReceiverLatestEndTime,
257
			prometheus.CounterValue,
258
			latestEndTime.Float64,
259
			labels...)
260

261
		ch <- prometheus.MustNewConstMetric(
262
			statWalReceiverUpstreamNode,
263
			prometheus.GaugeValue,
264
			float64(upstreamNode.Int64),
265
			labels...)
266
	}
267
	if err := rows.Err(); err != nil {
268
		return err
269
	}
270
	return nil
271
}
272