jolokia: use always POST

code refactor to use same prepareRequest method
for both 'agent' and 'proxy' mode

closes #1031
closes #1050
closes #473
This commit is contained in:
Simone Aiello 2016-04-15 00:38:46 +02:00 committed by Cameron Sparr
parent cf5980ace2
commit 18636ea628
4 changed files with 116 additions and 136 deletions

View File

@ -82,8 +82,10 @@ https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#g
- Possible breaking change for the librato and graphite outputs. Telegraf will - Possible breaking change for the librato and graphite outputs. Telegraf will
no longer insert field names when the field is simply named `value`. This is no longer insert field names when the field is simply named `value`. This is
because the `value` field is redundant in the graphite/librato context. because the `value` field is redundant in the graphite/librato context.
- Breaking change in jolokia plugin. See https://github.com/influxdata/telegraf/blob/master/plugins/inputs/jolokia/README.md
### Features ### Features
- [#1031](https://github.com/influxdata/telegraf/pull/1031): Jolokia plugin proxy mode. Thanks @saiello!
- [#1009](https://github.com/influxdata/telegraf/pull/1009): Cassandra input plugin. Thanks @subhachandrachandra! - [#1009](https://github.com/influxdata/telegraf/pull/1009): Cassandra input plugin. Thanks @subhachandrachandra!
- [#976](https://github.com/influxdata/telegraf/pull/976): Reduce allocations in the UDP and statsd inputs. - [#976](https://github.com/influxdata/telegraf/pull/976): Reduce allocations in the UDP and statsd inputs.
- [#979](https://github.com/influxdata/telegraf/pull/979): Reduce allocations in the TCP listener. - [#979](https://github.com/influxdata/telegraf/pull/979): Reduce allocations in the TCP listener.
@ -96,6 +98,7 @@ because the `value` field is redundant in the graphite/librato context.
- [#1008](https://github.com/influxdata/telegraf/pull/1008): Adding memstats metrics to the influxdb plugin. - [#1008](https://github.com/influxdata/telegraf/pull/1008): Adding memstats metrics to the influxdb plugin.
### Bugfixes ### Bugfixes
- [#1050](https://github.com/influxdata/telegraf/issues/1050): jolokia plugin - do not overwrite host tag. Thanks @saiello!
- [#968](https://github.com/influxdata/telegraf/issues/968): Processes plugin gets unknown state when spaces are in (command name) - [#968](https://github.com/influxdata/telegraf/issues/968): Processes plugin gets unknown state when spaces are in (command name)
- [#969](https://github.com/influxdata/telegraf/pull/969): ipmi_sensors: allow : in password. Thanks @awaw! - [#969](https://github.com/influxdata/telegraf/pull/969): ipmi_sensors: allow : in password. Thanks @awaw!
- [#972](https://github.com/influxdata/telegraf/pull/972): dovecot: remove extra newline in dovecot command. Thanks @mrannanj! - [#972](https://github.com/influxdata/telegraf/pull/972): dovecot: remove extra newline in dovecot command. Thanks @mrannanj!

View File

@ -54,4 +54,4 @@ are collected for each server configured.
See: https://jolokia.org/ See: https://jolokia.org/
# Measurements: # Measurements:
Jolokia plugin produces one measure for each metric configured, adding Server's `name`, `host` and `port` as tags. Jolokia plugin produces one measure for each metric configured, adding Server's `server_name`, `server_host` and `server_port` as tags.

View File

@ -1,6 +1,7 @@
package jolokia package jolokia
import ( import (
"bytes"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
@ -8,7 +9,6 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"time" "time"
"bytes"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/inputs"
@ -23,10 +23,10 @@ type Server struct {
} }
type Metric struct { type Metric struct {
Name string Name string
Mbean string Mbean string
Attribute string Attribute string
Path string Path string
} }
type JolokiaClient interface { type JolokiaClient interface {
@ -44,28 +44,28 @@ func (c JolokiaClientImpl) MakeRequest(req *http.Request) (*http.Response, error
type Jolokia struct { type Jolokia struct {
jClient JolokiaClient jClient JolokiaClient
Context string Context string
Mode string Mode string
Servers []Server Servers []Server
Metrics []Metric Metrics []Metric
Proxy Server Proxy Server
} }
func (j *Jolokia) SampleConfig() string { const sampleConfig = `
return ` ## This is the context root used to compose the jolokia url
# This is the context root used to compose the jolokia url
context = "/jolokia" context = "/jolokia"
# This specifies the mode used ## This specifies the mode used
# mode = "proxy" # mode = "proxy"
# #
# When in proxy mode this section is used to specify further proxy address configurations. ## When in proxy mode this section is used to specify further
# Remember to change servers addresses ## proxy address configurations.
## Remember to change host address to fit your environment.
# [inputs.jolokia.proxy] # [inputs.jolokia.proxy]
# host = "127.0.0.1" # host = "127.0.0.1"
# port = "8080" # port = "8080"
# List of servers exposing jolokia read service ## List of servers exposing jolokia read service
[[inputs.jolokia.servers]] [[inputs.jolokia.servers]]
name = "as-server-01" name = "as-server-01"
host = "127.0.0.1" host = "127.0.0.1"
@ -86,14 +86,17 @@ func (j *Jolokia) SampleConfig() string {
[[inputs.jolokia.metrics]] [[inputs.jolokia.metrics]]
name = "thread_count" name = "thread_count"
mbean = "java.lang:type=Threading" mbean = "java.lang:type=Threading"
attribute = "TotalStartedThreadCount,ThreadCount,DaemonThreadCount,PeakThreadCount" attribute = "TotalStartedThreadCount,ThreadCount,DaemonThreadCount,PeakThreadCount"
## This collect number of class loaded/unloaded counts metrics. ## This collect number of class loaded/unloaded counts metrics.
[[inputs.jolokia.metrics]] [[inputs.jolokia.metrics]]
name = "class_count" name = "class_count"
mbean = "java.lang:type=ClassLoading" mbean = "java.lang:type=ClassLoading"
attribute = "LoadedClassCount,UnloadedClassCount,TotalLoadedClassCount" attribute = "LoadedClassCount,UnloadedClassCount,TotalLoadedClassCount"
` `
func (j *Jolokia) SampleConfig() string {
return sampleConfig
} }
func (j *Jolokia) Description() string { func (j *Jolokia) Description() string {
@ -133,7 +136,8 @@ func (j *Jolokia) doRequest(req *http.Request) (map[string]interface{}, error) {
if status, ok := jsonOut["status"]; ok { if status, ok := jsonOut["status"]; ok {
if status != float64(200) { if status != float64(200) {
return nil, fmt.Errorf("Not expected status value in response body: %3.f", status) return nil, fmt.Errorf("Not expected status value in response body: %3.f",
status)
} }
} else { } else {
return nil, fmt.Errorf("Missing status in response body") return nil, fmt.Errorf("Missing status in response body")
@ -142,148 +146,122 @@ func (j *Jolokia) doRequest(req *http.Request) (map[string]interface{}, error) {
return jsonOut, nil return jsonOut, nil
} }
func (j *Jolokia) getAttr(requestUrl *url.URL) (map[string]interface{}, error) { func (j *Jolokia) prepareRequest(server Server, metric Metric) (*http.Request, error) {
// Create + send request var jolokiaUrl *url.URL
req, err := http.NewRequest("GET", requestUrl.String(), nil)
if err != nil {
return nil, err
}
return j.doRequest(req)
}
func (j *Jolokia) collectMeasurement(measurement string, out map[string]interface{}, fields map[string]interface{}) {
if values, ok := out["value"]; ok {
switch t := values.(type) {
case map[string]interface{}:
for k, v := range t {
fields[measurement+"_"+k] = v
}
case interface{}:
fields[measurement] = t
}
} else {
fmt.Printf("Missing key 'value' in output response\n")
}
}
func (j *Jolokia) Gather(acc telegraf.Accumulator) error {
context := j.Context // Usually "/jolokia" context := j.Context // Usually "/jolokia"
servers := j.Servers
metrics := j.Metrics
tags := make(map[string]string)
mode := j.Mode
if( mode == "agent" || mode == ""){ // Create bodyContent
bodyContent := map[string]interface{}{
"type": "read",
"mbean": metric.Mbean,
}
for _, server := range servers { if metric.Attribute != "" {
tags["server"] = server.Name bodyContent["attribute"] = metric.Attribute
tags["port"] = server.Port if metric.Path != "" {
tags["host"] = server.Host bodyContent["path"] = metric.Path
fields := make(map[string]interface{}) }
for _, metric := range metrics { }
measurement := metric.Name // Add target, only in proxy mode
jmxPath := "/" + metric.Mbean if j.Mode == "proxy" {
if metric.Attribute != "" { serviceUrl := fmt.Sprintf("service:jmx:rmi:///jndi/rmi://%s:%s/jmxrmi",
jmxPath = jmxPath + "/" + metric.Attribute server.Host, server.Port)
if metric.Path != "" { target := map[string]string{
jmxPath = jmxPath + "/" + metric.Path "url": serviceUrl,
}
}
// Prepare URL
requestUrl, err := url.Parse("http://" + server.Host + ":" +
server.Port + context + "/read" + jmxPath)
if err != nil {
return err
}
if server.Username != "" || server.Password != "" {
requestUrl.User = url.UserPassword(server.Username, server.Password)
}
out, _ := j.getAttr(requestUrl)
j.collectMeasurement(measurement, out, fields)
}
acc.AddFields("jolokia", fields, tags)
} }
} else if ( mode == "proxy") { if server.Username != "" {
target["user"] = server.Username
}
if server.Password != "" {
target["password"] = server.Password
}
bodyContent["target"] = target
proxy := j.Proxy proxy := j.Proxy
// Prepare ProxyURL // Prepare ProxyURL
proxyURL, err := url.Parse("http://" + proxy.Host + ":" + proxyUrl, err := url.Parse("http://" + proxy.Host + ":" + proxy.Port + context)
proxy.Port + context)
if err != nil { if err != nil {
return err return nil, err
} }
if proxy.Username != "" || proxy.Password != "" { if proxy.Username != "" || proxy.Password != "" {
proxyURL.User = url.UserPassword(proxy.Username, proxy.Password) proxyUrl.User = url.UserPassword(proxy.Username, proxy.Password)
} }
for _, server := range servers { jolokiaUrl = proxyUrl
tags["server"] = server.Name
tags["port"] = server.Port
tags["host"] = server.Host
fields := make(map[string]interface{})
for _, metric := range metrics {
measurement := metric.Name } else {
// Prepare URL serverUrl, err := url.Parse("http://" + server.Host + ":" + server.Port + context)
serviceUrl := fmt.Sprintf("service:jmx:rmi:///jndi/rmi://%s:%s/jmxrmi", server.Host, server.Port) if err != nil {
return nil, err
}
if server.Username != "" || server.Password != "" {
serverUrl.User = url.UserPassword(server.Username, server.Password)
}
target := map[string]string{ jolokiaUrl = serverUrl
"url": serviceUrl, }
}
if server.Username != "" { requestBody, err := json.Marshal(bodyContent)
target["user"] = server.Username
}
if server.Password != "" { req, err := http.NewRequest("POST", jolokiaUrl.String(), bytes.NewBuffer(requestBody))
target["password"] = server.Password
}
// Create + send request if err != nil {
bodyContent := map[string]interface{}{ return nil, err
"type": "read", }
"mbean": metric.Mbean,
"target": target,
}
if metric.Attribute != "" { req.Header.Add("Content-type", "application/json")
bodyContent["attribute"] = metric.Attribute
if metric.Path != "" {
bodyContent["path"] = metric.Path
}
}
requestBody, err := json.Marshal(bodyContent) return req, nil
}
req, err := http.NewRequest("POST", proxyURL.String(), bytes.NewBuffer(requestBody)) func (j *Jolokia) Gather(acc telegraf.Accumulator) error {
servers := j.Servers
metrics := j.Metrics
tags := make(map[string]string)
if err != nil { for _, server := range servers {
return err tags["server_name"] = server.Name
} tags["server_port"] = server.Port
tags["server_host"] = server.Host
fields := make(map[string]interface{})
req.Header.Add("Content-type", "application/json") for _, metric := range metrics {
measurement := metric.Name
req, err := j.prepareRequest(server, metric)
if err != nil {
return err
}
out, err := j.doRequest(req)
if err != nil {
fmt.Printf("Error handling response: %s\n", err)
} else {
if values, ok := out["value"]; ok {
switch t := values.(type) {
case map[string]interface{}:
for k, v := range t {
fields[measurement+"_"+k] = v
}
case interface{}:
fields[measurement] = t
}
} else {
fmt.Printf("Missing key 'value' in output response\n")
}
out, err := j.doRequest(req)
if err != nil {
fmt.Printf("Error handling response: %s\n", err)
}else {
j.collectMeasurement(measurement, out, fields)
}
} }
acc.AddFields("jolokia", fields, tags)
} }
acc.AddFields("jolokia", fields, tags)
} }
return nil return nil

View File

@ -48,7 +48,7 @@ const empty = ""
var Servers = []Server{Server{Name: "as1", Host: "127.0.0.1", Port: "8080"}} var Servers = []Server{Server{Name: "as1", Host: "127.0.0.1", Port: "8080"}}
var HeapMetric = Metric{Name: "heap_memory_usage", var HeapMetric = Metric{Name: "heap_memory_usage",
Mbean: "java.lang:type=Memory", Attribute: "HeapMemoryUsage" } Mbean: "java.lang:type=Memory", Attribute: "HeapMemoryUsage"}
var UsedHeapMetric = Metric{Name: "heap_memory_usage", var UsedHeapMetric = Metric{Name: "heap_memory_usage",
Mbean: "java.lang:type=Memory", Attribute: "HeapMemoryUsage"} Mbean: "java.lang:type=Memory", Attribute: "HeapMemoryUsage"}
@ -96,9 +96,9 @@ func TestHttpJsonMultiValue(t *testing.T) {
"heap_memory_usage_used": 203288528.0, "heap_memory_usage_used": 203288528.0,
} }
tags := map[string]string{ tags := map[string]string{
"host": "127.0.0.1", "server_host": "127.0.0.1",
"port": "8080", "server_port": "8080",
"server": "as1", "server_name": "as1",
} }
acc.AssertContainsTaggedFields(t, "jolokia", fields, tags) acc.AssertContainsTaggedFields(t, "jolokia", fields, tags)
} }
@ -117,7 +117,6 @@ func TestHttpJsonOn404(t *testing.T) {
assert.Equal(t, 0, len(acc.Metrics)) assert.Equal(t, 0, len(acc.Metrics))
} }
// Test that the proper values are ignored or collected // Test that the proper values are ignored or collected
func TestHttpInvalidJson(t *testing.T) { func TestHttpInvalidJson(t *testing.T) {