Add a Collector for pg_stat_io.

Docs: https://www.postgresql.org/docs/current/monitoring-stats.html#MONITORING-PG-STAT-IO-VIEW

We use this in production. One problem is that the stats for a vacuum
are reported when the vacuum ends, not continuously.

Signed-off-by: Peter Nuttall <[email protected]>

Author: sfc-gh-pnuttall

collector/pg_stat_io.go

@@ -0,0 +1,325 @@
+// 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 statIOSubsystem = "stat_io"
+
+func init() {
+	registerCollector(statIOSubsystem, defaultDisabled, NewStatIOCollector)
+}
+
+type StatIOCollector struct {
+	log *slog.Logger
+}
+
+func NewStatIOCollector(config collectorConfig) (Collector, error) {
+	return &StatIOCollector{
+		log: config.logger,
+	}, nil
+}
+
+var (
+	statIOReadsTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "reads_total"),
+		"Number of read operations, each of the size specified in op_bytes.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOReadTimeTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "read_time_total"),
+		"Time spent in read operations in milliseconds (if track_io_timing is enabled, otherwise zero)",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOWritesTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "writes_total"),
+		"Number of write operations, each of the size specified in op_bytes.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOWriteTimeTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "writes_time_total"),
+		"Time spent in write operations in milliseconds (if track_io_timing is enabled, otherwise zero)",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOWriteBackTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "write_back_total"),
+		"Number of units of size op_bytes which the process requested the kernel write out to permanent storage.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOWriteBackTimeTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "write_back_time_total"),
+		"Time spent in writeback operations in milliseconds (if track_io_timing is enabled, otherwise zero). This includes the time spent queueing write-out requests and, potentially, the time spent to write out the dirty data.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOExtendsTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "extends_total"),
+		"Number of relation extend operations, each of the size specified in op_bytes.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOExtendsTimeTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "extends_time_total"),
+		"Time spent in extend operations in milliseconds (if track_io_timing is enabled, otherwise zero)",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOHitsTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "hits_total"),
+		"The number of times a desired block was found in a shared buffer.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOEvictionsTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "evictions_total"),
+		"Number of times a block has been written out from a shared or local buffer in order to make it available for another use.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOReusesTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "reuses_total"),
+		"The number of times an existing buffer in a size-limited ring buffer outside of shared buffers was reused as part of an I/O operation in the bulkread, bulkwrite, or vacuum contexts.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+
+	statIOFsyncsTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "fsync_total"),
+		"Number of fsync calls. These are only tracked in context normal.",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOFsyncTimeTotal = prometheus.NewDesc(
+		prometheus.BuildFQName(namespace, statIOSubsystem, "fsync_time_total"),
+		"Time spent in fsync operations in milliseconds (if track_io_timing is enabled, otherwise zero)",
+		[]string{"backend_type", "object", "context"},
+		prometheus.Labels{},
+	)
+	statIOQuery = `
+		SELECT
+		backend_type,
+		object,
+		context,
+		reads,
+		read_time,
+		writes,
+		write_time,
+		writebacks,
+		writeback_time,
+		extends,
+		extend_time,
+		hits,
+		evictions,
+		reuses,
+		fsyncs,
+		fsync_time
+
+		FROM
+		pg_stat_io
+		`
+)
+
+// Update implements Collector and exposes database locks.
+// It is called by the Prometheus registry when collecting metrics.
+func (c StatIOCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error {
+	// pg_stat_io is only in v16, and we don't need support for earlier currently.
+	if !instance.version.GE(semver.MustParse("16.0.0")) {
+		return nil
+	}
+	db := instance.getDB()
+	// Query the list of databases
+	rows, err := db.QueryContext(ctx, statIOQuery)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	var backendType, object, PGContext sql.NullString
+	var reads, writes, writeBacks, extends, hits, evictions, reuses, fsyncs sql.NullInt64
+	var readTime, writeTime, writeBackTime, extendsTime, fsyncTime sql.NullFloat64
+
+	for rows.Next() {
+		if err := rows.Scan(
+			&backendType, &object, &PGContext,
+			&reads,
+			&readTime,
+			&writes,
+			&writeTime,
+			&writeBacks,
+			&writeBackTime,
+			&extends,
+			&extendsTime,
+			&hits,
+			&evictions,
+			&reuses,
+			&fsyncs,
+			&fsyncTime); err != nil {
+			return err
+		}
+
+		if !backendType.Valid || !object.Valid || !PGContext.Valid {
+			continue
+		}
+
+		readsMetric := 0.0
+		if reads.Valid {
+			readsMetric = float64(reads.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOReadsTotal,
+			prometheus.CounterValue,
+			readsMetric,
+			backendType.String, object.String, PGContext.String)
+
+		readTimeMetric := 0.0
+		if readTime.Valid {
+			readTimeMetric = readTime.Float64
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOReadTimeTotal,
+			prometheus.CounterValue,
+			readTimeMetric,
+			backendType.String, object.String, PGContext.String)
+
+		writesMetric := 0.0
+		if writes.Valid {
+			writesMetric = float64(writes.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOWritesTotal,
+			prometheus.CounterValue,
+			writesMetric,
+			backendType.String, object.String, PGContext.String)
+
+		writeTimeMetric := 0.0
+		if writeTime.Valid {
+			writeTimeMetric = writeTime.Float64
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOWriteTimeTotal,
+			prometheus.CounterValue,
+			writeTimeMetric,
+			backendType.String, object.String, PGContext.String)
+
+		writeBackMetric := 0.0
+		if writeBacks.Valid {
+			writeBackMetric = float64(writeBacks.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOWriteBackTotal,
+			prometheus.CounterValue,
+			writeBackMetric,
+			backendType.String, object.String, PGContext.String)
+
+		writeBackTimeMetric := 0.0
+		if writeBackTime.Valid {
+			writeBackTimeMetric = writeBackTime.Float64
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOWriteBackTimeTotal,
+			prometheus.CounterValue,
+			writeBackTimeMetric,
+			backendType.String, object.String, PGContext.String)
+
+		extendsMetric := 0.0
+		if extends.Valid {
+			extendsMetric = float64(extends.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOExtendsTotal,
+			prometheus.CounterValue,
+			extendsMetric,
+			backendType.String, object.String, PGContext.String)
+
+		extendsTimeMetric := 0.0
+		if extendsTime.Valid {
+			extendsTimeMetric = extendsTime.Float64
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOExtendsTimeTotal,
+			prometheus.CounterValue,
+			extendsTimeMetric,
+			backendType.String, object.String, PGContext.String)
+
+		hitsMetric := 0.0
+		if hits.Valid {
+			hitsMetric = float64(hits.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOHitsTotal,
+			prometheus.CounterValue,
+			hitsMetric,
+			backendType.String, object.String, PGContext.String)
+
+		evictionsMetric := 0.0
+		if evictions.Valid {
+			evictionsMetric = float64(evictions.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOEvictionsTotal,
+			prometheus.CounterValue,
+			evictionsMetric,
+			backendType.String, object.String, PGContext.String)
+
+		reusesMetric := 0.0
+		if reuses.Valid {
+			reusesMetric = float64(reuses.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOReusesTotal,
+			prometheus.CounterValue,
+			reusesMetric,
+			backendType.String, object.String, PGContext.String)
+
+		fsyncsMetric := 0.0
+		if fsyncs.Valid {
+			fsyncsMetric = float64(fsyncs.Int64)
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOFsyncsTotal,
+			prometheus.CounterValue,
+			fsyncsMetric,
+			backendType.String, object.String, PGContext.String)
+
+		fsyncTimeMetric := 0.0
+		if fsyncTime.Valid {
+			fsyncTimeMetric = fsyncTime.Float64
+		}
+		ch <- prometheus.MustNewConstMetric(
+			statIOFsyncTimeTotal,
+			prometheus.CounterValue,
+			fsyncTimeMetric,
+			backendType.String, object.String, PGContext.String)
+
+	}
+
+	return rows.Err()
+}