forked from kloudsense/cloudera_exporter
-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathcloudera_exporter.go
237 lines (191 loc) · 7.62 KB
/
cloudera_exporter.go
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
/*
*
* title :cloudera_exporter.go
* description :Main code file of the Cloudera Exporter for Prometheus DB
* author :Raul Barroso and Alejandro Villegas
* date :2018/10/05
*
*/
package main
/* ======================================================================
* Dependencies and libraries
* ====================================================================== */
import (
"context"
"fmt"
// Go Default libraries
"net/http"
"os"
"path"
"runtime"
"strconv"
"strings"
"time"
// Own libraries
cl "keedio/cloudera_exporter/collector"
cp "keedio/cloudera_exporter/config_parser"
log "keedio/cloudera_exporter/logger"
// Go external libraries
"gopkg.in/alecthomas/kingpin.v2"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
// Go Prometheus libraries
"github.com/prometheus/common/version"
)
/* ======================================================================
* Global variables
* ====================================================================== */
// Exporter Configuration Struct
var config *cp.CE_config
// Timeout Offset for Prometheus TimeStamping
var timeoutOffset = 0.0
// HTML Code por Landing Page
var metrics_path="/metrics"
var landingPage = []byte(`<html>
<head><title>Cloudera Manager exporter</title></head>
<body>
<h1>Cloudera Manager exporter</h1>
<h3> by KEEDIO - Big Data Facilitators</h3>
<p><a href='` + metrics_path + `'>Metrics</a></p>
</body>
</html>
`)
/* ======================================================================
* Functions
* ====================================================================== */
// Creates and initialize a Prometheus Collector
func init() {
set_version_properties()
prometheus.MustRegister(version.NewCollector("kbdi"))
}
// Create and returns a Handler for the Collector
func newHandler(metrics cl.Metrics, scrapers []cl.Scraper) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
// Use request context for cancellation when connection gets closed.
ctx := r.Context()
// If a timeout is configured via the Prometheus header, add it to the context.
if v := r.Header.Get("X-Prometheus-Scrape-Timeout-Seconds"); v != "" {
timeoutSeconds, err := strconv.ParseFloat(v, 64)
if err != nil {
log.Err_msg("Failed to parse timeout from Prometheus header: %s", err.Error())
} else {
if timeoutOffset >= timeoutSeconds {
// Ignore timeout offset if it doesn't leave time to scrape.
log.Err_msg("Timeout offset (--timeout-offset=%.2f) should be lower than prometheus scrape time (X-Prometheus-Scrape-Timeout-Seconds=%.2f).", timeoutOffset, timeoutSeconds)
} else {
// Subtract timeout offset from timeout.
timeoutSeconds -= timeoutOffset
}
// Create new timeout context with request context as parent.
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, time.Duration(timeoutSeconds * float64(time.Second)))
defer cancel()
// Overwrite request with timeout context.
r = r.WithContext(ctx)
}
}
// Create Prometheus registry with filtererd scrapers
registry := prometheus.NewRegistry()
// Register the collector with the data connection struct in the registry
registry.MustRegister(cl.New(ctx, config.Connection, metrics, scrapers))
gatherers := prometheus.Gatherers { prometheus.DefaultGatherer, registry }
// Delegate http serving to Prometheus client library, which will call collector.Collect.
h := promhttp.HandlerFor(gatherers, promhttp.HandlerOpts{})
h.ServeHTTP(w, r)
}
}
// Set the version properties of the Cloudera Exporter
func set_version_properties() {
version.Version="1.3"
version.Revision="PRO"
version.Branch="Master"
version.BuildUser="Keedio"
currentTime := time.Now()
version.BuildDate=currentTime.String()
}
// Prepare and parse the execution flags
func parse_exec_flags () {
kingpin.Version(version.Print("cloudera_exporter"))
kingpin.HelpFlag.Short('h')
kingpin.Parse()
}
// Register scrapers enabled.
func register_scrapers (config *cp.CE_config) []cl.Scraper{
enabledScrapers := []cl.Scraper{}
log.Info_msg("Enabled scrapers:")
for scraper, enabled := range config.Scrapers.Scrapers {
if enabled {
log.Info_msg(" -> %s", strings.Title(strings.Replace(scraper.Name(), "_", " ", -1)))
enabledScrapers = append(enabledScrapers, scraper)
}
}
return enabledScrapers
}
// Read the flags and the config file and set all the values of the
// Configuration Structure
func parse_flags_and_config_file() error {
var err error
// Parse flags and config file
configFile := kingpin.Flag("config-file", "Path to ini file.", ).Default(path.Join(os.Getenv("HOME"), "config.ini")).String()
arg_host := *(kingpin.Flag("web.listen-address", "Listent Address.",).Default("").String())
arg_num_procs := *(kingpin.Flag("num-procs", "Number Processes for parallel execution",).Default("0").Int())
arg_log_level := *(kingpin.Flag("log-level", "Debug Log Mode",).Default("0").Int())
timeoutOffset = *(kingpin.Flag("timeout-offset", "Time to subtract from timeout in seconds.", ).Default("0.25").Float64())
parse_exec_flags()
if config, err = cp.Parse_config(*configFile); err != nil {
return err
}
// If host, num_procs or log_level are defined in the execution flags, they
// have priority over the configuration file
if arg_host != "" {
config.Connection.Host = arg_host
}
if arg_num_procs != 0 {
config.Num_procs = arg_num_procs
}
if arg_log_level != 0 {
config.Log_level = arg_log_level
}
// Check if Api_version is defined on the config file, else, the version is
// obtained by Cloudera Manager API
if config.Connection.Api_version == "" {
if config.Connection.Api_version, err = cl.Get_api_cloudera_version(nil, config.Connection); err != nil {
return err
}
}
cl.SendConf(config)
return nil
}
// Main function
func main(){
// Starting Logging
log.Init(os.Stdout, os.Stdout, os.Stdout, os.Stderr, os.Stdout, 0)
log.Info_msg("================================================================================")
log.Info_msg("Starting Keedio Cloudera's Metrics Exporter")
// Setting code version properties
log.Info_msg("Exporter Version: %s", version.Version)
// Parse Flags and config file
if err := parse_flags_and_config_file(); err != nil {
log.Err_msg(err.Error())
return
}
log.Init(os.Stdout, os.Stdout, os.Stdout, os.Stderr, os.Stdout, config.Log_level)
//Parallel Execution
runtime.GOMAXPROCS(config.Num_procs)
log.Info_msg("Cores allocated: %s", strconv.Itoa(config.Num_procs))
// Run info
log.Info_msg("Build context %s", version.BuildContext())
// Exporter creation
log.Info_msg("Registering Handlers")
handlerFunc := newHandler(cl.NewMetrics(), register_scrapers(config))
http.Handle(metrics_path, promhttp.InstrumentMetricHandler(prometheus.DefaultRegisterer, handlerFunc))
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.Write(landingPage) })
log.Ok_msg("Landing Page and Handlers are running")
// Exporter HTTP connection
log.Info_msg("Target to scraping metrics from: %s:%s", config.Connection.Host, config.Connection.Port)
ip := func () string {if config.Deploy_ip == "" { return "0.0.0.0" } else { return config.Deploy_ip }}
log.Info_msg("Metrics published on: %s:%d", ip(), config.Deploy_port)
log.Ok_msg("Keedio's Cloudera Exporter running")
log.Err_msg(http.ListenAndServe(fmt.Sprintf("%s:%d", config.Deploy_ip, config.Deploy_port), nil).Error())
return
}