dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove userinfo from url tag in prometheus input (#3743)

This commit is contained in:
Philipp Weber 2018-02-05 20:16:00 +01:00 committed by Daniel Nelson
parent 248ac6d92b
commit 0cb1b147e6
4 changed files with 42 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/inputs/prometheus/README.md
View File

@ -67,7 +67,7 @@ Measurement names are based on the Metric Family and tags are created for each
label. The value is added to a field named based on the metric type. label. The value is added to a field named based on the metric type.
All metrics receive the `url` tag indicating the related URL specified in the All metrics receive the `url` tag indicating the related URL specified in the
Telegraf configuration. If using Kubernetes service discovery the `address` Telegraf configuration. If using Kubernetes service discovery the `address`
tag is also added indicating the discovered ip address. tag is also added indicating the discovered ip address.
### Example Output: ### Example Output:

68
plugins/inputs/prometheus/prometheus.go
View File

@ -20,7 +20,7 @@ const acceptHeader = `application/vnd.google.protobuf;proto=io.prometheus.client
type Prometheus struct { type Prometheus struct {
// An array of urls to scrape metrics from. // An array of urls to scrape metrics from.
Urls []string URLs []string `toml:"urls"`
// An array of Kubernetes services to scrape metrics from. // An array of Kubernetes services to scrape metrics from.
KubernetesServices []string KubernetesServices []string
@ -73,12 +73,12 @@ func (p *Prometheus) Description() string {
var ErrProtocolError = errors.New("prometheus protocol error") var ErrProtocolError = errors.New("prometheus protocol error")
func (p *Prometheus) AddressToURL(u *url.URL, address string) string { func (p *Prometheus) AddressToURL(u *url.URL, address string) *url.URL {
host := address host := address
if u.Port() != "" { if u.Port() != "" {
host = address + ":" + u.Port() host = address + ":" + u.Port()
} }
reconstructedUrl := url.URL{ reconstructedURL := &url.URL{
Scheme: u.Scheme, Scheme: u.Scheme,
Opaque: u.Opaque, Opaque: u.Opaque,
User: u.User, User: u.User,
@ -89,36 +89,42 @@ func (p *Prometheus) AddressToURL(u *url.URL, address string) string {
Fragment: u.Fragment, Fragment: u.Fragment,
Host: host, Host: host,
} }
return reconstructedUrl.String() return reconstructedURL
} }
type UrlAndAddress struct { type URLAndAddress struct {
OriginalUrl string OriginalURL *url.URL
Url string URL *url.URL
Address string Address string
} }
func (p *Prometheus) GetAllURLs() ([]UrlAndAddress, error) { func (p *Prometheus) GetAllURLs() ([]URLAndAddress, error) {
allUrls := make([]UrlAndAddress, 0) allURLs := make([]URLAndAddress, 0)
for _, url := range p.Urls { for _, u := range p.URLs {
allUrls = append(allUrls, UrlAndAddress{Url: url, OriginalUrl: url}) URL, err := url.Parse(u)
if err != nil {
log.Printf("prometheus: Could not parse %s, skipping it. Error: %s", u, err)
continue
}
allURLs = append(allURLs, URLAndAddress{URL: URL, OriginalURL: URL})
} }
for _, service := range p.KubernetesServices { for _, service := range p.KubernetesServices {
u, err := url.Parse(service) URL, err := url.Parse(service)
if err != nil { if err != nil {
return nil, err return nil, err
} }
resolvedAddresses, err := net.LookupHost(u.Hostname()) resolvedAddresses, err := net.LookupHost(URL.Hostname())
if err != nil { if err != nil {
log.Printf("prometheus: Could not resolve %s, skipping it. Error: %s", u.Host, err) log.Printf("prometheus: Could not resolve %s, skipping it. Error: %s", URL.Host, err)
continue continue
} }
for _, resolved := range resolvedAddresses { for _, resolved := range resolvedAddresses {
serviceUrl := p.AddressToURL(u, resolved) serviceURL := p.AddressToURL(URL, resolved)
allUrls = append(allUrls, UrlAndAddress{Url: serviceUrl, Address: resolved, OriginalUrl: service}) allURLs = append(allURLs, URLAndAddress{URL: serviceURL, Address: resolved, OriginalURL: URL})
} }
} }
return allUrls, nil return allURLs, nil
} }
// Reads stats from all configured servers accumulates stats. // Reads stats from all configured servers accumulates stats.
@ -134,16 +140,16 @@ func (p *Prometheus) Gather(acc telegraf.Accumulator) error {
var wg sync.WaitGroup var wg sync.WaitGroup
allUrls, err := p.GetAllURLs() allURLs, err := p.GetAllURLs()
if err != nil { if err != nil {
return err return err
} }
for _, url := range allUrls { for _, URL := range allURLs {
wg.Add(1) wg.Add(1)
go func(serviceUrl UrlAndAddress) { go func(serviceURL URLAndAddress) {
defer wg.Done() defer wg.Done()
acc.AddError(p.gatherURL(serviceUrl, acc)) acc.AddError(p.gatherURL(serviceURL, acc))
}(url) }(URL)
} }
wg.Wait() wg.Wait()
@ -178,8 +184,8 @@ func (p *Prometheus) createHttpClient() (*http.Client, error) {
return client, nil return client, nil
} }
func (p *Prometheus) gatherURL(url UrlAndAddress, acc telegraf.Accumulator) error { func (p *Prometheus) gatherURL(u URLAndAddress, acc telegraf.Accumulator) error {
var req, err = http.NewRequest("GET", url.Url, nil) var req, err = http.NewRequest("GET", u.URL.String(), nil)
req.Header.Add("Accept", acceptHeader) req.Header.Add("Accept", acceptHeader)
var token []byte var token []byte
var resp *http.Response var resp *http.Response
@ -194,11 +200,11 @@ func (p *Prometheus) gatherURL(url UrlAndAddress, acc telegraf.Accumulator) erro
resp, err = p.client.Do(req) resp, err = p.client.Do(req)
if err != nil { if err != nil {
return fmt.Errorf("error making HTTP request to %s: %s", url.Url, err) return fmt.Errorf("error making HTTP request to %s: %s", u.URL, err)
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return fmt.Errorf("%s returned HTTP status %s", url.Url, resp.Status) return fmt.Errorf("%s returned HTTP status %s", u.URL, resp.Status)
} }
body, err := ioutil.ReadAll(resp.Body) body, err := ioutil.ReadAll(resp.Body)
@ -209,14 +215,16 @@ func (p *Prometheus) gatherURL(url UrlAndAddress, acc telegraf.Accumulator) erro
metrics, err := Parse(body, resp.Header) metrics, err := Parse(body, resp.Header)
if err != nil { if err != nil {
return fmt.Errorf("error reading metrics for %s: %s", return fmt.Errorf("error reading metrics for %s: %s",
url.Url, err) u.URL, err)
} }
// Add (or not) collected metrics // Add (or not) collected metrics
for _, metric := range metrics { for _, metric := range metrics {
tags := metric.Tags() tags := metric.Tags()
tags["url"] = url.OriginalUrl // strip user and password from URL
if url.Address != "" { u.OriginalURL.User = nil
tags["address"] = url.Address tags["url"] = u.OriginalURL.String()
if u.Address != "" {
tags["address"] = u.Address
} }
switch metric.Type() { switch metric.Type() {

4
plugins/inputs/prometheus/prometheus_test.go
View File

@ -37,7 +37,7 @@ func TestPrometheusGeneratesMetrics(t *testing.T) {
defer ts.Close() defer ts.Close()
p := &Prometheus{ p := &Prometheus{
Urls: []string{ts.URL}, URLs: []string{ts.URL},
} }
var acc testutil.Accumulator var acc testutil.Accumulator
@ -89,7 +89,7 @@ func TestPrometheusGeneratesMetricsAlthoughFirstDNSFails(t *testing.T) {
defer ts.Close() defer ts.Close()
p := &Prometheus{ p := &Prometheus{
Urls: []string{ts.URL}, URLs: []string{ts.URL},
KubernetesServices: []string{"http://random.telegraf.local:88/metrics"}, KubernetesServices: []string{"http://random.telegraf.local:88/metrics"},
} }

2
plugins/outputs/prometheus_client/prometheus_client_test.go
View File

@ -631,7 +631,7 @@ func setupPrometheus() (*PrometheusClient, *prometheus_input.Prometheus, error)
time.Sleep(time.Millisecond * 200) time.Sleep(time.Millisecond * 200)
p := &prometheus_input.Prometheus{ p := &prometheus_input.Prometheus{
Urls: []string{"http://localhost:9127/metrics"}, URLs: []string{"http://localhost:9127/metrics"},
} }
return pTesting, p, nil return pTesting, p, nil