diff --git a/collector/pg_blocked.go b/collector/pg_blocked.go new file mode 100644 index 00000000..f215ed8c --- /dev/null +++ b/collector/pg_blocked.go @@ -0,0 +1,101 @@ +// 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" + + "github.com/go-kit/log" + "github.com/prometheus/client_golang/prometheus" +) + +const blockedSubsystem = "blocked" + +func init() { + registerCollector(blockedSubsystem, defaultDisabled, NewPGBlockedCollector) +} + +type PGBlockedCollector struct { + log log.Logger +} + +func NewPGBlockedCollector(config collectorConfig) (Collector, error) { + return &PGBlockedCollector{log: config.logger}, nil +} + +var ( + blockedQueries = prometheus.NewDesc( + prometheus.BuildFQName(namespace, blockedSubsystem, "queries"), + "The current number of blocked queries", + []string{"table"}, + prometheus.Labels{}, + ) + + blockedQuery = ` + SELECT + count(blocked.transactionid) AS queries, + '__transaction__' AS table + FROM pg_catalog.pg_locks blocked + WHERE NOT blocked.granted AND locktype = 'transactionid' + GROUP BY locktype + UNION + SELECT + count(blocked.relation) AS queries, + blocked.relation::regclass::text AS table + FROM pg_catalog.pg_locks blocked + WHERE NOT blocked.granted AND locktype != 'transactionid' + GROUP BY relation + ` +) + +func (PGBlockedCollector) Update(ctx context.Context, instance *instance, ch chan<- prometheus.Metric) error { + db := instance.getDB() + rows, err := db.QueryContext(ctx, + blockedQuery) + + if err != nil { + return err + } + defer rows.Close() + + for rows.Next() { + var table sql.NullString + var queries sql.NullFloat64 + + if err := rows.Scan(&queries, &table); err != nil { + return err + } + + tableLabel := "unknown" + if table.Valid { + tableLabel = table.String + } + + queriesMetric := 0.0 + if queries.Valid { + queriesMetric = queries.Float64 + } + ch <- prometheus.MustNewConstMetric( + blockedQueries, + prometheus.GaugeValue, + queriesMetric, + tableLabel, + ) + } + if err := rows.Err(); err != nil { + return err + } + return nil +} diff --git a/collector/pg_blocked_test.go b/collector/pg_blocked_test.go new file mode 100644 index 00000000..281edc82 --- /dev/null +++ b/collector/pg_blocked_test.go @@ -0,0 +1,101 @@ +// 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" + + "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 TestPgBlockedCollector(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{ + "queries", + "table", + } + rows := sqlmock.NewRows(columns). + AddRow(1000, "pgbouncer") + + mock.ExpectQuery(sanitizeQuery(blockedQuery)).WillReturnRows(rows) + + ch := make(chan prometheus.Metric) + go func() { + defer close(ch) + c := PGBlockedCollector{} + + if err := c.Update(context.Background(), inst, ch); err != nil { + t.Errorf("Error calling PGBlockedCollector.Update: %s", err) + } + }() + expected := []MetricResult{ + {labels: labelMap{"table": "pgbouncer"}, value: 1000, metricType: dto.MetricType_GAUGE}, + } + 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 TestPgBlockedCollectorNull(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{ + "queries", + "table", + } + rows := sqlmock.NewRows(columns). + AddRow(nil, nil) + + mock.ExpectQuery(sanitizeQuery(blockedQuery)).WillReturnRows(rows) + + ch := make(chan prometheus.Metric) + go func() { + defer close(ch) + c := PGBlockedCollector{} + + if err := c.Update(context.Background(), inst, ch); err != nil { + t.Errorf("Error calling PGBlockedCollector.Update: %s", err) + } + }() + expected := []MetricResult{ + {labels: labelMap{"table": "unknown"}, value: 0, metricType: dto.MetricType_GAUGE}, + } + 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) + } +}