Feature/jolokia plugin

README.md

@@ -173,6 +173,7 @@ Telegraf currently has support for collecting metrics from:
 * exec (generic JSON-emitting executable plugin)
 * haproxy
 * httpjson (generic JSON-emitting http service plugin)
+* jolokia (remote JMX with JSON over HTTP)
 * kafka_consumer
 * leofs
 * lustre2

plugins/all/all.go

@@ -9,6 +9,7 @@ import (
  _ "github.com/influxdb/telegraf/plugins/exec"
  _ "github.com/influxdb/telegraf/plugins/haproxy"
  _ "github.com/influxdb/telegraf/plugins/httpjson"
+ _ "github.com/influxdb/telegraf/plugins/jolokia"
  _ "github.com/influxdb/telegraf/plugins/kafka_consumer"
  _ "github.com/influxdb/telegraf/plugins/leofs"
  _ "github.com/influxdb/telegraf/plugins/lustre2"

plugins/jolokia/README.md

@@ -0,0 +1,51 @@
+# Telegraf plugin: Jolokia
+
+#### Plugin arguments:
+- **context** string: Context root used of jolokia url
+- **servers** []Server: List of servers
+ + **name** string: Server's logical name
+ + **host** string: Server's ip address or hostname
+ + **port** string: Server's listening port
+- **metrics** []Metric
+ + **name** string: Name of the measure
+ + **jmx** string: Jmx path that identifies mbeans attributes
+ + **pass** []string: Attributes to retain when collecting values
+ + **drop** []string: Attributes to drop when collecting values
+
+#### Description
+
+The Jolokia plugin collects JVM metrics exposed as MBean's attributes through jolokia REST endpoint. All metrics
+are collected for each server configured.
+
+See: https://jolokia.org/
+
+# Measurements:
+Jolokia plugin produces one measure for each metric configured, adding Server's `name`, `host` and `port` as tags.
+
+Given a configuration like:
+
+```ini
+[jolokia]
+
+[[jolokia.servers]]
+ name = "as-service-1"
+ host = "127.0.0.1"
+ port = "8080"
+
+[[jolokia.servers]]
+ name = "as-service-2"
+ host = "127.0.0.1"
+ port = "8180"
+
+[[jolokia.metrics]]
+ name = "heap_memory_usage"
+ jmx = "/java.lang:type=Memory/HeapMemoryUsage"
+ pass = ["used", "max"]
+```
+
+The collected metrics will be:
+
+```
+jolokia_heap_memory_usage name=as-service-1,host=127.0.0.1,port=8080 used=xxx,max=yyy
+jolokia_heap_memory_usage name=as-service-2,host=127.0.0.1,port=8180 used=vvv,max=zzz
+```

plugins/jolokia/jolokia.go

@@ -0,0 +1,223 @@
+package jolokia
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io/ioutil"
+ "net/http"
+ "net/url"
+ "strings"
+
+ "github.com/influxdb/telegraf/plugins"
+)
+
+type Server struct {
+ Name string
+ Host string
+ Port string
+}
+
+type Metric struct {
+ Name string
+ Jmx string
+ Pass []string
+ Drop []string
+}
+
+type JolokiaClient interface {
+ MakeRequest(req *http.Request) (*http.Response, error)
+}
+
+type JolokiaClientImpl struct {
+ client *http.Client
+}
+
+func (c JolokiaClientImpl) MakeRequest(req *http.Request) (*http.Response, error) {
+ return c.client.Do(req)
+}
+
+type Jolokia struct {
+ jClient JolokiaClient
+ Context string
+ Servers []Server
+ Metrics []Metric
+ Tags map[string]string
+}
+
+func (j *Jolokia) SampleConfig() string {
+ return `
+ # This is the context root used to compose the jolokia url
+ context = "/jolokia/read"
+
+ # Tags added to each measurements
+ [jolokia.tags]
+ group = "as"
+
+ # List of servers exposing jolokia read service
+ [[jolokia.servers]]
+ name = "stable"
+ host = "192.168.103.2"
+ port = "8180"
+
+ # List of metrics collected on above servers
+ # Each metric consists in a name, a jmx path and either a pass or drop slice attributes
+ # This collect all heap memory usage metrics
+ [[jolokia.metrics]]
+ name = "heap_memory_usage"
+ jmx = "/java.lang:type=Memory/HeapMemoryUsage"
+
+
+ # This drops the 'committed' value from Eden space measurement
+ [[jolokia.metrics]]
+ name = "memory_eden"
+ jmx = "/java.lang:type=MemoryPool,name=PS Eden Space/Usage"
+ drop = [ "committed" ]
+
+
+ # This passes only DaemonThreadCount and ThreadCount
+ [[jolokia.metrics]]
+ name = "heap_threads"
+ jmx = "/java.lang:type=Threading"
+ pass = [
+ "DaemonThreadCount",
+ "ThreadCount"
+ ]
+`
+}
+
+func (j *Jolokia) Description() string {
+ return "Read JMX metrics through Jolokia"
+}
+
+func (j *Jolokia) getAttr(requestUrl *url.URL) (map[string]interface{}, error) {
+ // Create + send request
+ req, err := http.NewRequest("GET", requestUrl.String(), nil)
+ if err != nil {
+ return nil, err
+ }
+
+ resp, err := j.jClient.MakeRequest(req)
+ if err != nil {
+ return nil, err
+ }
+
+ if err != nil {
+ return nil, err
+ }
+ defer resp.Body.Close()
+
+ // Process response
+ if resp.StatusCode != http.StatusOK {
+ err = fmt.Errorf("Response from url \"%s\" has status code %d (%s), expected %d (%s)",
+ requestUrl,
+ resp.StatusCode,
+ http.StatusText(resp.StatusCode),
+ http.StatusOK,
+ http.StatusText(http.StatusOK))
+ return nil, err
+ }
+
+ // read body
+ body, err := ioutil.ReadAll(resp.Body)
+ if err != nil {
+ return nil, err
+ }
+
+ // Unmarshal json
+ var jsonOut map[string]interface{}
+ if err = json.Unmarshal([]byte(body), &jsonOut); err != nil {
+ return nil, errors.New("Error decoding JSON response")
+ }
+
+ return jsonOut, nil
+}
+
+func (m *Metric) shouldPass(field string) bool {
+
+ if m.Pass != nil {
+
+ for _, pass := range m.Pass {
+ if strings.HasPrefix(field, pass) {
+ return true
+ }
+ }
+
+ return false
+ }
+
+ if m.Drop != nil {
+
+ for _, drop := range m.Drop {
+ if strings.HasPrefix(field, drop) {
+ return false
+ }
+ }
+
+ return true
+ }
+
+ return true
+}
+
+func (m *Metric) filterFields(fields map[string]interface{}) map[string]interface{} {
+
+ for field, _ := range fields {
+ if !m.shouldPass(field) {
+ delete(fields, field)
+ }
+ }
+
+ return fields
+}
+
+func (j *Jolokia) Gather(acc plugins.Accumulator) error {
+
+ context := j.Context //"/jolokia/read"
+ servers := j.Servers
+ metrics := j.Metrics
+ tags := j.Tags
+
+ if tags == nil {
+ tags = map[string]string{}
+ }
+
+ for _, server := range servers {
+ for _, metric := range metrics {
+
+ measurement := metric.Name
+ jmxPath := metric.Jmx
+
+ tags["server"] = server.Name
+ tags["port"] = server.Port
+ tags["host"] = server.Host
+
+ // Prepare URL
+ requestUrl, err := url.Parse("http://" + server.Host + ":" + server.Port + context + jmxPath)
+ if err != nil {
+ return err
+ }
+
+ out, _ := j.getAttr(requestUrl)
+
+ if values, ok := out["value"]; ok {
+ switch values.(type) {
+ case map[string]interface{}:
+ acc.AddFields(measurement, metric.filterFields(values.(map[string]interface{})), tags)
+ case interface{}:
+ acc.Add(measurement, values.(interface{}), tags)
+ }
+ } else {
+ fmt.Printf("Missing key 'value' in '%s' output response\n", requestUrl.String())
+ }
+ }
+ }
+
+ return nil
+}
+
+func init() {
+ plugins.Add("jolokia", func() plugins.Plugin {
+ return &Jolokia{jClient: &JolokiaClientImpl{client: &http.Client{}}}
+ })
+}