diff --git a/collector/pg_stat_checkpointer.go b/collector/pg_stat_checkpointer.go new file mode 100644 index 000000000..224af27e5 --- /dev/null +++ b/collector/pg_stat_checkpointer.go @@ -0,0 +1,219 @@ +// Copyright 2021 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/prometheus/client_golang/prometheus" +) + +const statCheckpointerSubsystem = "stat_checkpointer" + +func init() { + registerCollector(statCheckpointerSubsystem, defaultEnabled, NewPGStatCheckpointerCollector) +} + +type PGStatCheckpointerCollector struct { +} + +func NewPGStatCheckpointerCollector(collectorConfig) (Collector, error) { + return &PGStatCheckpointerCollector{}, nil +} + +var ( + statCheckpointerNumTimedDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "num_timed_total"), + "Number of scheduled checkpoints due to timeout", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerNumRequestedDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "num_requested_total"), + "Number of requested checkpoints that have been performed", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerRestartpointsTimedDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "restartpoints_timed_total"), + "Number of scheduled restartpoints due to timeout or after a failed attempt to perform it", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerRestartpointsReqDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "restartpoints_req_total"), + "Number of requested restartpoints", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerRestartpointsDoneDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "restartpoints_done_total"), + "Number of restartpoints that have been performed", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerWriteTimeDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "write_time_total"), + "Total amount of time that has been spent in the portion of processing checkpoints and restartpoints where files are written to disk, in milliseconds", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerSyncTimeDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "sync_time_total"), + "Total amount of time that has been spent in the portion of processing checkpoints and restartpoints where files are synchronized to disk, in milliseconds", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerBuffersWrittenDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "buffers_written_total"), + "Number of buffers written during checkpoints and restartpoints", + []string{}, + prometheus.Labels{}, + ) + statCheckpointerStatsResetDesc = prometheus.NewDesc( + prometheus.BuildFQName(namespace, statCheckpointerSubsystem, "stats_reset_total"), + "Time at which these statistics were last reset", + []string{}, + prometheus.Labels{}, + ) + + statCheckpointerQuery = `SELECT + num_timed + ,num_requested + ,restartpoints_timed + ,restartpoints_req + ,restartpoints_done + ,write_time + ,sync_time + ,buffers_written + ,stats_reset + FROM pg_stat_checkpointer;` +) + +func (PGStatCheckpointerCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error { + db := instance.getDB() + row := db.QueryRowContext(ctx, statCheckpointerQuery) + + // num_timed = nt = bigint + // num_requested = nr = bigint + // restartpoints_timed = rpt = bigint + // restartpoints_req = rpr = bigint + // restartpoints_done = rpd = bigint + // write_time = wt = double precision + // sync_time = st = double precision + // buffers_written = bw = bigint + // stats_reset = sr = timestamp + + var nt, nr, rpt, rpr, rpd, bw sql.NullInt64 + var wt, st sql.NullFloat64 + var sr sql.NullTime + + err := row.Scan(&nt, &nr, &rpt, &rpr, &rpd, &wt, &st, &bw, &sr) + if err != nil { + return err + } + + ntMetric := 0.0 + if nt.Valid { + ntMetric = float64(nt.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerNumTimedDesc, + prometheus.CounterValue, + ntMetric, + ) + + nrMetric := 0.0 + if nr.Valid { + nrMetric = float64(nr.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerNumRequestedDesc, + prometheus.CounterValue, + nrMetric, + ) + + rptMetric := 0.0 + if rpt.Valid { + rptMetric = float64(rpt.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerRestartpointsTimedDesc, + prometheus.CounterValue, + rptMetric, + ) + + rprMetric := 0.0 + if rpr.Valid { + rprMetric = float64(rpr.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerRestartpointsReqDesc, + prometheus.CounterValue, + rprMetric, + ) + + rpdMetric := 0.0 + if rpd.Valid { + rpdMetric = float64(rpd.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerRestartpointsDoneDesc, + prometheus.CounterValue, + rpdMetric, + ) + + wtMetric := 0.0 + if wt.Valid { + wtMetric = float64(wt.Float64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerWriteTimeDesc, + prometheus.CounterValue, + wtMetric, + ) + + stMetric := 0.0 + if st.Valid { + stMetric = float64(st.Float64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerSyncTimeDesc, + prometheus.CounterValue, + stMetric, + ) + + bwMetric := 0.0 + if bw.Valid { + bwMetric = float64(bw.Int64) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerBuffersWrittenDesc, + prometheus.CounterValue, + bwMetric, + ) + + srMetric := 0.0 + if sr.Valid { + srMetric = float64(sr.Time.Unix()) + } + ch <- prometheus.MustNewConstMetric( + statCheckpointerStatsResetDesc, + prometheus.CounterValue, + srMetric, + ) + + return nil +} diff --git a/collector/pg_stat_checkpointer_test.go b/collector/pg_stat_checkpointer_test.go new file mode 100644 index 000000000..a2d2837f9 --- /dev/null +++ b/collector/pg_stat_checkpointer_test.go @@ -0,0 +1,143 @@ +// 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" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/smartystreets/goconvey/convey" +) + +func TestPGStatCheckpointerCollector(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} + + columns := []string{ + "num_timed", + "num_requested", + "restartpoints_timed", + "restartpoints_req", + "restartpoints_done", + "write_time", + "sync_time", + "buffers_written", + "stats_reset"} + + srT, err := time.Parse("2006-01-02 15:04:05.00000-07", "2023-05-25 17:10:42.81132-07") + if err != nil { + t.Fatalf("Error parsing time: %s", err) + } + + rows := sqlmock.NewRows(columns). + AddRow(354, 4945, 289097744, 1242257, 3275602074, 89320867, 450139, 2034563757, srT) + mock.ExpectQuery(sanitizeQuery(statCheckpointerQuery)).WillReturnRows(rows) + + ch := make(chan prometheus.Metric) + go func() { + defer close(ch) + c := PGStatCheckpointerCollector{} + + if err := c.Update(context.Background(), inst, ch); err != nil { + t.Errorf("Error calling PGStatCheckpointerCollector.Update: %s", err) + } + }() + + expected := []MetricResult{ + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 354}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 4945}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 289097744}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 1242257}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 3275602074}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 89320867}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 450139}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 2034563757}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 1685059842}, + } + + 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 TestPGStatCheckpointerCollectorNullValues(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} + + columns := []string{ + "num_timed", + "num_requested", + "restartpoints_timed", + "restartpoints_req", + "restartpoints_done", + "write_time", + "sync_time", + "buffers_written", + "stats_reset"} + + rows := sqlmock.NewRows(columns). + AddRow(nil, nil, nil, nil, nil, nil, nil, nil, nil) + mock.ExpectQuery(sanitizeQuery(statCheckpointerQuery)).WillReturnRows(rows) + + ch := make(chan prometheus.Metric) + go func() { + defer close(ch) + c := PGStatCheckpointerCollector{} + + if err := c.Update(context.Background(), inst, ch); err != nil { + t.Errorf("Error calling PGStatCheckpointerCollector.Update: %s", err) + } + }() + + expected := []MetricResult{ + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + {labels: labelMap{}, metricType: dto.MetricType_COUNTER, value: 0}, + } + + 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) + } +}