1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
package main
import (
"fmt"
"strconv"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/client_golang/prometheus"
)
// Run executes a single Query on a single connection
func (q *Query) Run(conn *connection) error {
if q.log == nil {
q.log = log.NewNopLogger()
}
if q.desc == nil {
return fmt.Errorf("metrics descriptor is nil")
}
if q.Query == "" {
return fmt.Errorf("query is empty")
}
if conn == nil || conn.conn == nil {
return fmt.Errorf("db connection not initialized (should not happen)")
}
// execute query
rows, err := conn.conn.Queryx(q.Query)
if err != nil {
failedScrapes.WithLabelValues(conn.driver, conn.host, conn.database, conn.user, q.jobName, q.Name).Set(1.0)
return err
}
defer rows.Close()
updated := 0
metrics := make([]prometheus.Metric, 0, len(q.metrics))
for rows.Next() {
res := make(map[string]interface{})
err := rows.MapScan(res)
if err != nil {
level.Error(q.log).Log("msg", "Failed to scan", "err", err, "host", conn.host, "db", conn.database)
failedScrapes.WithLabelValues(conn.driver, conn.host, conn.database, conn.user, q.jobName, q.Name).Set(1.0)
continue
}
m, err := q.updateMetrics(conn, res)
if err != nil {
level.Error(q.log).Log("msg", "Failed to update metrics", "err", err, "host", conn.host, "db", conn.database)
failedScrapes.WithLabelValues(conn.driver, conn.host, conn.database, conn.user, q.jobName, q.Name).Set(1.0)
continue
}
metrics = append(metrics, m...)
updated++
failedScrapes.WithLabelValues(conn.driver, conn.host, conn.database, conn.user, q.jobName, q.Name).Set(0.0)
}
if updated < 1 {
if q.AllowZeroRows {
failedScrapes.WithLabelValues(conn.driver, conn.host, conn.database, conn.user, q.jobName, q.Name).Set(0.0)
} else {
return fmt.Errorf("zero rows returned")
}
}
// update the metrics cache
q.Lock()
q.metrics[conn] = metrics
q.Unlock()
return nil
}
// updateMetrics parses the result set and returns a slice of const metrics
func (q *Query) updateMetrics(conn *connection, res map[string]interface{}) ([]prometheus.Metric, error) {
updated := 0
metrics := make([]prometheus.Metric, 0, len(q.Values))
for _, valueName := range q.Values {
m, err := q.updateMetric(conn, res, valueName)
if err != nil {
level.Error(q.log).Log(
"msg", "Failed to update metric",
"value", valueName,
"err", err,
"host", conn.host,
"db", conn.database,
)
continue
}
metrics = append(metrics, m)
updated++
}
if updated < 1 {
return nil, fmt.Errorf("zero values found")
}
return metrics, nil
}
// updateMetrics parses a single row and returns a const metric
func (q *Query) updateMetric(conn *connection, res map[string]interface{}, valueName string) (prometheus.Metric, error) {
var value float64
if i, ok := res[valueName]; ok {
switch f := i.(type) {
case int:
value = float64(f)
case int32:
value = float64(f)
case int64:
value = float64(f)
case uint:
value = float64(f)
case uint32:
value = float64(f)
case uint64:
value = float64(f)
case float32:
value = float64(f)
case float64:
value = float64(f)
case []uint8:
val, err := strconv.ParseFloat(string(f), 64)
if err != nil {
return nil, fmt.Errorf("column '%s' must be type float, is '%T' (val: %s)", valueName, i, f)
}
value = val
case string:
val, err := strconv.ParseFloat(f, 64)
if err != nil {
return nil, fmt.Errorf("column '%s' must be type float, is '%T' (val: %s)", valueName, i, f)
}
value = val
default:
return nil, fmt.Errorf("column '%s' must be type float, is '%T' (val: %s)", valueName, i, f)
}
} else {
level.Warn(q.log).Log(
"msg", "Column not found in query result",
"column", valueName,
"resultColumns", res,
)
}
// make space for all defined variable label columns and the "static" labels
// added below
labels := make([]string, 0, len(q.Labels)+5)
for _, label := range q.Labels {
// we need to fill every spot in the slice or the key->value mapping
// won't match up in the end.
//
// ORDER MATTERS!
lv := ""
if i, ok := res[label]; ok {
switch str := i.(type) {
case string:
lv = str
case []uint8:
lv = string(str)
default:
return nil, fmt.Errorf("column '%s' must be type text (string)", label)
}
}
labels = append(labels, lv)
}
labels = append(labels, conn.driver)
labels = append(labels, conn.host)
labels = append(labels, conn.database)
labels = append(labels, conn.user)
labels = append(labels, valueName)
// create a new immutable const metric that can be cached and returned on
// every scrape. Remember that the order of the lable values in the labels
// slice must match the order of the label names in the descriptor!
return prometheus.NewConstMetric(q.desc, prometheus.GaugeValue, value, labels...)
}
|
