Skip to content

Commit c3b0206

Browse files
sysadmindsysadmind
committed
Add collector interface
Uses node_exporter style collector registration Signed-off-by: Joe Adams <[email protected]>
1 parent 1981623 commit c3b0206

File tree

5 files changed

+342
-28
lines changed

5 files changed

+342
-28
lines changed

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: ‎cmd/postgres_exporter/server.go

-14
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,13 @@
1414
package main
1515

1616
import (
17-
"context"
1817
"database/sql"
1918
"fmt"
20-
"log"
2119
"sync"
2220
"time"
2321

2422
"github.com/blang/semver"
2523
"github.com/go-kit/log/level"
26-
"github.com/prometheus-community/postgres_exporter/collector"
2724
"github.com/prometheus/client_golang/prometheus"
2825
)
2926

@@ -131,17 +128,6 @@ func (s *Server) Scrape(ch chan<- prometheus.Metric, disableSettingsMetrics bool
131128
err = fmt.Errorf("queryNamespaceMappings returned %d errors", len(errMap))
132129
}
133130

134-
{
135-
pgdb := collector.NewPGDatabaseCollector()
136-
metrics, err := pgdb.Update(context.TODO(), s.db, s.String())
137-
if err != nil {
138-
log.Printf("Failed to scrape pg_database metrics: %s", err)
139-
}
140-
for _, m := range metrics {
141-
ch <- m
142-
}
143-
}
144-
145131
return err
146132
}
147133

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

+24-14
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
// Copyright 2021 The Prometheus Authors
1+
// Copyright 2022 The Prometheus Authors
22
// Licensed under the Apache License, Version 2.0 (the "License");
33
// you may not use this file except in compliance with the License.
44
// You may obtain a copy of the License at
@@ -15,15 +15,21 @@ package collector
1515

1616
import (
1717
"context"
18-
"database/sql"
1918

19+
"github.com/go-kit/log"
2020
"github.com/prometheus/client_golang/prometheus"
2121
)
2222

23-
type PGDatabaseCollector struct{}
23+
func init() {
24+
registerCollector("database", defaultEnabled, NewPGDatabaseCollector)
25+
}
26+
27+
type PGDatabaseCollector struct {
28+
log log.Logger
29+
}
2430

25-
func NewPGDatabaseCollector() *PGDatabaseCollector {
26-
return &PGDatabaseCollector{}
31+
func NewPGDatabaseCollector(logger log.Logger) (Collector, error) {
32+
return &PGDatabaseCollector{log: logger}, nil
2733
}
2834

2935
var pgDatabase = map[string]*prometheus.Desc{
@@ -34,30 +40,34 @@ var pgDatabase = map[string]*prometheus.Desc{
3440
),
3541
}
3642

37-
func (PGDatabaseCollector) Update(ctx context.Context, db *sql.DB, server string) ([]prometheus.Metric, error) {
38-
metrics := []prometheus.Metric{}
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+
}
3948
rows, err := db.QueryContext(ctx,
4049
`SELECT pg_database.datname
4150
,pg_database_size(pg_database.datname)
4251
FROM pg_database;`)
4352
if err != nil {
44-
return metrics, err
53+
return err
4554
}
4655
defer rows.Close()
4756

4857
for rows.Next() {
4958
var datname string
5059
var size int64
5160
if err := rows.Scan(&datname, &size); err != nil {
52-
return metrics, err
61+
return err
5362
}
54-
metrics = append(metrics, prometheus.MustNewConstMetric(
63+
64+
ch <- prometheus.MustNewConstMetric(
5565
pgDatabase["size_bytes"],
56-
prometheus.GaugeValue, float64(size), datname, server,
57-
))
66+
prometheus.GaugeValue, float64(size), datname, server.GetName(),
67+
)
5868
}
5969
if err := rows.Err(); err != nil {
60-
return metrics, err
70+
return err
6171
}
62-
return metrics, nil
72+
return nil
6373
}

0 commit comments

Comments
 (0)