forked from msiedlarek/nifi_exporter
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.go
144 lines (126 loc) · 4.5 KB
/
main.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
package main
import (
"fmt"
"io/ioutil"
"net/http"
"os"
"reflect"
"strings"
"github.com/go-playground/locales/en"
ut "github.com/go-playground/universal-translator"
"github.com/juju/errors"
"github.com/msiedlarek/nifi_exporter/nifi/client"
"github.com/msiedlarek/nifi_exporter/nifi/collectors"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
log "github.com/sirupsen/logrus"
"gopkg.in/go-playground/validator.v9"
validator_en "gopkg.in/go-playground/validator.v9/translations/en"
"gopkg.in/yaml.v2"
)
type Configuration struct {
Exporter struct {
ListenAddress string `yaml:"listenAddress" validate:"required"`
} `yaml:"exporter" validate:"required"`
Nodes []struct {
URL string `yaml:"url" validate:"required,url"`
CACertificates string `yaml:"caCertificates"`
Username string `yaml:"username"`
Password string `yaml:"password"`
Labels map[string]string `yaml:"labels"`
} `yaml:"nodes" validate:"required,dive"`
}
func main() {
if len(os.Args) != 2 {
fmt.Fprintf(os.Stderr, "Usage: %s CONFIG_PATH", os.Args[0])
os.Exit(2)
}
configPath := os.Args[1]
log.Info("Starting nifi_exporter...")
config, err := loadConfig(configPath)
if err != nil {
log.Fatal(err)
}
if err := start(config); err != nil {
log.Fatal(err)
}
}
func loadConfig(configPath string) (*Configuration, error) {
log.WithField("path", configPath).Info("Loading configuration file...")
configYaml, err := ioutil.ReadFile(configPath)
if err != nil {
return nil, errors.Annotate(err, "Couldn't read config file")
}
var config Configuration
if err := yaml.Unmarshal(configYaml, &config); err != nil {
return nil, errors.Annotate(err, "Couldn't parse config file")
}
locale := en.New()
universalTranslator := ut.New(locale, locale)
translator, found := universalTranslator.GetTranslator(locale.Locale())
if !found {
return nil, errors.New("Couldn't initialize validation error translator")
}
validate := validator.New()
validate.RegisterTagNameFunc(func(field reflect.StructField) string {
return field.Tag.Get("yaml")
})
validator_en.RegisterDefaultTranslations(validate, translator)
if err := validate.Struct(&config); err != nil {
validationErrors := err.(validator.ValidationErrors)
for i := range validationErrors {
fieldError := validationErrors[i]
log.WithFields(log.Fields{
"field": strings.SplitN(fieldError.Namespace(), ".", 2)[1],
"error": fieldError.Translate(translator),
}).Error("Invalid configuration.")
}
return nil, errors.New("Configuration file validation failed.")
}
log.WithField("path", configPath).Info("Configuration file successfully loaded.")
return &config, nil
}
func start(config *Configuration) error {
for i := range config.Nodes {
node := &config.Nodes[i]
api, err := client.NewClient(node.URL, node.Username, node.Password, node.CACertificates)
if err != nil {
return errors.Annotate(err, "Couldn't create Prometheus API client")
}
log.WithFields(log.Fields{
"labels": node.Labels,
"url": node.URL,
"username": node.Username,
}).Info("Registering NiFi node...")
if err := prometheus.DefaultRegisterer.Register(collectors.NewDiagnosticsCollector(api, node.Labels)); err != nil {
return errors.Annotate(err, "Couldn't register system diagnostics collector.")
}
if err := prometheus.DefaultRegisterer.Register(collectors.NewCountersCollector(api, node.Labels)); err != nil {
return errors.Annotate(err, "Couldn't register counters collector.")
}
if err := prometheus.DefaultRegisterer.Register(collectors.NewProcessGroupsCollector(api, node.Labels)); err != nil {
return errors.Annotate(err, "Couldn't register process groups collector.")
}
if err := prometheus.DefaultRegisterer.Register(collectors.NewConnectionsCollector(api, node.Labels)); err != nil {
return errors.Annotate(err, "Couldn't register connections collector.")
}
}
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(`<html>
<head><title>NiFi Exporter</title></head>
<body>
<h1>NiFi Exporter</h1>
<p><a href="/metrics">Metrics</a></p>
</body>
</html>`))
})
http.Handle("/metrics", promhttp.Handler())
log.WithField("address", config.Exporter.ListenAddress).Infof(
"Listening on: http://%s/metrics",
config.Exporter.ListenAddress,
)
if err := http.ListenAndServe(config.Exporter.ListenAddress, nil); err != nil {
return errors.Annotate(err, "Couldn't start HTTP server.")
}
return nil
}