Skip to content

Commit 3d7edd4

Browse files
SuperQSuperQ
authored
Merge pull request #613 from sysadmind/pg_database
Add pg_database collector
2 parents 8d6d304 + 9dad33c commit 3d7edd4

File tree

6 files changed

+395
-12
lines changed

6 files changed

+395
-12
lines changed

Diff for: ‎CHANGELOG.md

+4
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,7 @@
1+
## master / unreleased
2+
3+
* [ENHANCEMENT] Add pg_database_size_bytes metric #613
4+
15
## 0.10.1 / 2022-01-14
26

37
* [BUGFIX] Fix broken log-level for values other than debug. #560

Diff for: ‎cmd/postgres_exporter/main.go

+8
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919

2020
"github.com/go-kit/log"
2121
"github.com/go-kit/log/level"
22+
"github.com/prometheus-community/postgres_exporter/collector"
2223
"github.com/prometheus/client_golang/prometheus"
2324
"github.com/prometheus/client_golang/prometheus/promhttp"
2425
"github.com/prometheus/common/promlog"
@@ -114,6 +115,13 @@ func main() {
114115

115116
prometheus.MustRegister(exporter)
116117

118+
pe, err := collector.NewPostgresCollector(logger, dsn)
119+
if err != nil {
120+
level.Error(logger).Log("msg", "Failed to create PostgresCollector", "err", err.Error())
121+
os.Exit(1)
122+
}
123+
prometheus.MustRegister(pe)
124+
117125
http.Handle(*metricPath, promhttp.Handler())
118126
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
119127
w.Header().Set("Content-Type", "text/html; charset=UTF-8") // nolint: errcheck

Diff for: ‎collector/collector.go

+210
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,210 @@
1+
// Copyright 2022 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+
"errors"
19+
"fmt"
20+
"sync"
21+
"time"
22+
23+
"github.com/go-kit/log"
24+
"github.com/go-kit/log/level"
25+
"github.com/prometheus/client_golang/prometheus"
26+
"gopkg.in/alecthomas/kingpin.v2"
27+
)
28+
29+
var (
30+
factories = make(map[string]func(logger log.Logger) (Collector, error))
31+
initiatedCollectorsMtx = sync.Mutex{}
32+
initiatedCollectors = make(map[string]Collector)
33+
collectorState = make(map[string]*bool)
34+
forcedCollectors = map[string]bool{} // collectors which have been explicitly enabled or disabled
35+
)
36+
37+
const (
38+
// Namespace for all metrics.
39+
namespace = "pg"
40+
41+
defaultEnabled = true
42+
// defaultDisabled = false
43+
)
44+
45+
var (
46+
scrapeDurationDesc = prometheus.NewDesc(
47+
prometheus.BuildFQName(namespace, "scrape", "collector_duration_seconds"),
48+
"postgres_exporter: Duration of a collector scrape.",
49+
[]string{"collector"},
50+
nil,
51+
)
52+
scrapeSuccessDesc = prometheus.NewDesc(
53+
prometheus.BuildFQName(namespace, "scrape", "collector_success"),
54+
"postgres_exporter: Whether a collector succeeded.",
55+
[]string{"collector"},
56+
nil,
57+
)
58+
)
59+
60+
type Collector interface {
61+
Update(ctx context.Context, server *server, ch chan<- prometheus.Metric) error
62+
}
63+
64+
func registerCollector(name string, isDefaultEnabled bool, createFunc func(logger log.Logger) (Collector, error)) {
65+
var helpDefaultState string
66+
if isDefaultEnabled {
67+
helpDefaultState = "enabled"
68+
} else {
69+
helpDefaultState = "disabled"
70+
}
71+
72+
// Create flag for this collector
73+
flagName := fmt.Sprintf("collector.%s", name)
74+
flagHelp := fmt.Sprintf("Enable the %s collector (default: %s).", name, helpDefaultState)
75+
defaultValue := fmt.Sprintf("%v", isDefaultEnabled)
76+
77+
flag := kingpin.Flag(flagName, flagHelp).Default(defaultValue).Action(collectorFlagAction(name)).Bool()
78+
collectorState[name] = flag
79+
80+
// Register the create function for this collector
81+
factories[name] = createFunc
82+
}
83+
84+
// PostgresCollector implements the prometheus.Collector interface.
85+
type PostgresCollector struct {
86+
Collectors map[string]Collector
87+
logger log.Logger
88+
89+
servers map[string]*server
90+
}
91+
92+
// NewPostgresCollector creates a new PostgresCollector.
93+
func NewPostgresCollector(logger log.Logger, dsns []string, filters ...string) (*PostgresCollector, error) {
94+
f := make(map[string]bool)
95+
for _, filter := range filters {
96+
enabled, exist := collectorState[filter]
97+
if !exist {
98+
return nil, fmt.Errorf("missing collector: %s", filter)
99+
}
100+
if !*enabled {
101+
return nil, fmt.Errorf("disabled collector: %s", filter)
102+
}
103+
f[filter] = true
104+
}
105+
collectors := make(map[string]Collector)
106+
initiatedCollectorsMtx.Lock()
107+
defer initiatedCollectorsMtx.Unlock()
108+
for key, enabled := range collectorState {
109+
if !*enabled || (len(f) > 0 && !f[key]) {
110+
continue
111+
}
112+
if collector, ok := initiatedCollectors[key]; ok {
113+
collectors[key] = collector
114+
} else {
115+
collector, err := factories[key](log.With(logger, "collector", key))
116+
if err != nil {
117+
return nil, err
118+
}
119+
collectors[key] = collector
120+
initiatedCollectors[key] = collector
121+
}
122+
}
123+
124+
servers := make(map[string]*server)
125+
for _, dsn := range dsns {
126+
s, err := makeServer(dsn)
127+
if err != nil {
128+
return nil, err
129+
}
130+
servers[dsn] = s
131+
}
132+
133+
return &PostgresCollector{
134+
Collectors: collectors,
135+
logger: logger,
136+
servers: servers,
137+
}, nil
138+
}
139+
140+
// Describe implements the prometheus.Collector interface.
141+
func (n PostgresCollector) Describe(ch chan<- *prometheus.Desc) {
142+
ch <- scrapeDurationDesc
143+
ch <- scrapeSuccessDesc
144+
}
145+
146+
// Collect implements the prometheus.Collector interface.
147+
func (n PostgresCollector) Collect(ch chan<- prometheus.Metric) {
148+
ctx := context.TODO()
149+
wg := sync.WaitGroup{}
150+
wg.Add(len(n.servers))
151+
for _, s := range n.servers {
152+
go func(s *server) {
153+
n.subCollect(ctx, s, ch)
154+
wg.Done()
155+
}(s)
156+
}
157+
wg.Wait()
158+
}
159+
160+
func (n PostgresCollector) subCollect(ctx context.Context, server *server, ch chan<- prometheus.Metric) {
161+
wg := sync.WaitGroup{}
162+
wg.Add(len(n.Collectors))
163+
for name, c := range n.Collectors {
164+
go func(name string, c Collector) {
165+
execute(ctx, name, c, server, ch, n.logger)
166+
wg.Done()
167+
}(name, c)
168+
}
169+
wg.Wait()
170+
}
171+
172+
func execute(ctx context.Context, name string, c Collector, s *server, ch chan<- prometheus.Metric, logger log.Logger) {
173+
begin := time.Now()
174+
err := c.Update(ctx, s, ch)
175+
duration := time.Since(begin)
176+
var success float64
177+
178+
if err != nil {
179+
if IsNoDataError(err) {
180+
level.Debug(logger).Log("msg", "collector returned no data", "name", name, "duration_seconds", duration.Seconds(), "err", err)
181+
} else {
182+
level.Error(logger).Log("msg", "collector failed", "name", name, "duration_seconds", duration.Seconds(), "err", err)
183+
}
184+
success = 0
185+
} else {
186+
level.Debug(logger).Log("msg", "collector succeeded", "name", name, "duration_seconds", duration.Seconds())
187+
success = 1
188+
}
189+
ch <- prometheus.MustNewConstMetric(scrapeDurationDesc, prometheus.GaugeValue, duration.Seconds(), name)
190+
ch <- prometheus.MustNewConstMetric(scrapeSuccessDesc, prometheus.GaugeValue, success, name)
191+
}
192+
193+
// collectorFlagAction generates a new action function for the given collector
194+
// to track whether it has been explicitly enabled or disabled from the command line.
195+
// A new action function is needed for each collector flag because the ParseContext
196+
// does not contain information about which flag called the action.
197+
// See: https://github.com/alecthomas/kingpin/issues/294
198+
func collectorFlagAction(collector string) func(ctx *kingpin.ParseContext) error {
199+
return func(ctx *kingpin.ParseContext) error {
200+
forcedCollectors[collector] = true
201+
return nil
202+
}
203+
}
204+
205+
// ErrNoData indicates the collector found no data to collect, but had no other error.
206+
var ErrNoData = errors.New("collector returned no data")
207+
208+
func IsNoDataError(err error) bool {
209+
return err == ErrNoData
210+
}

Diff for: ‎collector/pg_database.go

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
// Copyright 2022 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+
19+
"github.com/go-kit/log"
20+
"github.com/prometheus/client_golang/prometheus"
21+
)
22+
23+
func init() {
24+
registerCollector("database", defaultEnabled, NewPGDatabaseCollector)
25+
}
26+
27+
type PGDatabaseCollector struct {
28+
log log.Logger
29+
}
30+
31+
func NewPGDatabaseCollector(logger log.Logger) (Collector, error) {
32+
return &PGDatabaseCollector{log: logger}, nil
33+
}
34+
35+
var pgDatabase = map[string]*prometheus.Desc{
36+
"size_bytes": prometheus.NewDesc(
37+
"pg_database_size_bytes",
38+
"Disk space used by the database",
39+
[]string{"datname", "server"}, nil,
40+
),
41+
}
42+
43+
func (PGDatabaseCollector) Update(ctx context.Context, server *server, ch chan<- prometheus.Metric) error {
44+
db, err := server.GetDB()
45+
if err != nil {
46+
return err
47+
}
48+
rows, err := db.QueryContext(ctx,
49+
`SELECT pg_database.datname
50+
,pg_database_size(pg_database.datname)
51+
FROM pg_database;`)
52+
if err != nil {
53+
return err
54+
}
55+
defer rows.Close()
56+
57+
for rows.Next() {
58+
var datname string
59+
var size int64
60+
if err := rows.Scan(&datname, &size); err != nil {
61+
return err
62+
}
63+
64+
ch <- prometheus.MustNewConstMetric(
65+
pgDatabase["size_bytes"],
66+
prometheus.GaugeValue, float64(size), datname, server.GetName(),
67+
)
68+
}
69+
if err := rows.Err(); err != nil {
70+
return err
71+
}
72+
return nil
73+
}

0 commit comments

Comments
 (0)