package cassandra import ( "encoding/json" "errors" "fmt" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/plugins/inputs" "io/ioutil" "net/http" "net/url" "strings" ) 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 Cassandra struct { jClient JolokiaClient Context string Servers []string Metrics []string } type javaMetric struct { host string metric string acc telegraf.Accumulator } type cassandraMetric struct { host string metric string acc telegraf.Accumulator } type jmxMetric interface { addTagsFields(out map[string]interface{}) } func newJavaMetric(host string, metric string, acc telegraf.Accumulator) *javaMetric { return &javaMetric{host: host, metric: metric, acc: acc} } func newCassandraMetric(host string, metric string, acc telegraf.Accumulator) *cassandraMetric { return &cassandraMetric{host: host, metric: metric, acc: acc} } func addValuesAsFields(values map[string]interface{}, fields map[string]interface{}, mname string) { for k, v := range values { if v != nil { fields[mname+"_"+k] = v } } } func parseJmxMetricRequest(mbean string) map[string]string { tokens := make(map[string]string) classAndPairs := strings.Split(mbean, ":") if classAndPairs[0] == "org.apache.cassandra.metrics" { tokens["class"] = "cassandra" } else if classAndPairs[0] == "java.lang" { tokens["class"] = "java" } else { return tokens } pairs := strings.Split(classAndPairs[1], ",") for _, pair := range pairs { p := strings.Split(pair, "=") tokens[p[0]] = p[1] } return tokens } func addTokensToTags(tokens map[string]string, tags map[string]string) { for k, v := range tokens { if k == "name" { tags["mname"] = v // name seems to a reserved word in influxdb } else if k == "class" || k == "type" { continue // class and type are used in the metric name } else { tags[k] = v } } } func (j javaMetric) addTagsFields(out map[string]interface{}) { tags := make(map[string]string) fields := make(map[string]interface{}) a := out["request"].(map[string]interface{}) attribute := a["attribute"].(string) mbean := a["mbean"].(string) tokens := parseJmxMetricRequest(mbean) addTokensToTags(tokens, tags) tags["cassandra_host"] = j.host if _, ok := tags["mname"]; !ok { //Queries for a single value will not return a "name" tag in the response. tags["mname"] = attribute } if values, ok := out["value"]; ok { switch t := values.(type) { case map[string]interface{}: addValuesAsFields(values.(map[string]interface{}), fields, attribute) case interface{}: fields[attribute] = t } j.acc.AddFields(tokens["class"]+tokens["type"], fields, tags) } else { j.acc.AddError(fmt.Errorf("Missing key 'value' in '%s' output response\n%v\n", j.metric, out)) } } func addCassandraMetric(mbean string, c cassandraMetric, values map[string]interface{}) { tags := make(map[string]string) fields := make(map[string]interface{}) tokens := parseJmxMetricRequest(mbean) addTokensToTags(tokens, tags) tags["cassandra_host"] = c.host addValuesAsFields(values, fields, tags["mname"]) c.acc.AddFields(tokens["class"]+tokens["type"], fields, tags) } func (c cassandraMetric) addTagsFields(out map[string]interface{}) { r := out["request"] tokens := parseJmxMetricRequest(r.(map[string]interface{})["mbean"].(string)) // Requests with wildcards for keyspace or table names will return nested // maps in the json response if (tokens["type"] == "Table" || tokens["type"] == "ColumnFamily") && (tokens["keyspace"] == "*" || tokens["scope"] == "*") { if valuesMap, ok := out["value"]; ok { for k, v := range valuesMap.(map[string]interface{}) { addCassandraMetric(k, c, v.(map[string]interface{})) } } else { c.acc.AddError(fmt.Errorf("Missing key 'value' in '%s' output response\n%v\n", c.metric, out)) return } } else { if values, ok := out["value"]; ok { addCassandraMetric(r.(map[string]interface{})["mbean"].(string), c, values.(map[string]interface{})) } else { c.acc.AddError(fmt.Errorf("Missing key 'value' in '%s' output response\n%v\n", c.metric, out)) return } } } func (j *Cassandra) SampleConfig() string { return ` # This is the context root used to compose the jolokia url context = "/jolokia/read" ## List of cassandra servers exposing jolokia read service servers = ["myuser:mypassword@10.10.10.1:8778","10.10.10.2:8778",":8778"] ## List of metrics collected on above servers ## Each metric consists of a jmx path. ## This will collect all heap memory usage metrics from the jvm and ## ReadLatency metrics for all keyspaces and tables. ## "type=Table" in the query works with Cassandra3.0. Older versions might ## need to use "type=ColumnFamily" metrics = [ "/java.lang:type=Memory/HeapMemoryUsage", "/org.apache.cassandra.metrics:type=Table,keyspace=*,scope=*,name=ReadLatency" ] ` } func (j *Cassandra) Description() string { return "Read Cassandra metrics through Jolokia" } func (j *Cassandra) 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 } 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 parseServerTokens(server string) map[string]string { serverTokens := make(map[string]string) hostAndUser := strings.Split(server, "@") hostPort := "" userPasswd := "" if len(hostAndUser) == 2 { hostPort = hostAndUser[1] userPasswd = hostAndUser[0] } else { hostPort = hostAndUser[0] } hostTokens := strings.Split(hostPort, ":") serverTokens["host"] = hostTokens[0] serverTokens["port"] = hostTokens[1] if userPasswd != "" { userTokens := strings.Split(userPasswd, ":") serverTokens["user"] = userTokens[0] serverTokens["passwd"] = userTokens[1] } return serverTokens } func (c *Cassandra) Gather(acc telegraf.Accumulator) error { context := c.Context servers := c.Servers metrics := c.Metrics for _, server := range servers { for _, metric := range metrics { serverTokens := parseServerTokens(server) var m jmxMetric if strings.HasPrefix(metric, "/java.lang:") { m = newJavaMetric(serverTokens["host"], metric, acc) } else if strings.HasPrefix(metric, "/org.apache.cassandra.metrics:") { m = newCassandraMetric(serverTokens["host"], metric, acc) } else { // unsupported metric type acc.AddError(fmt.Errorf("E! Unsupported Cassandra metric [%s], skipping", metric)) continue } // Prepare URL requestUrl, err := url.Parse("http://" + serverTokens["host"] + ":" + serverTokens["port"] + context + metric) if err != nil { acc.AddError(err) continue } if serverTokens["user"] != "" && serverTokens["passwd"] != "" { requestUrl.User = url.UserPassword(serverTokens["user"], serverTokens["passwd"]) } out, err := c.getAttr(requestUrl) if err != nil { acc.AddError(err) continue } if out["status"] != 200.0 { acc.AddError(fmt.Errorf("URL returned with status %v - %s\n", out["status"], requestUrl)) continue } m.addTagsFields(out) } } return nil } func init() { inputs.Add("cassandra", func() telegraf.Input { return &Cassandra{jClient: &JolokiaClientImpl{client: &http.Client{}}} }) }