Linter fixes for plugins/inputs/[c]* (influxdata#9194)

* Linter fixes for plugins/inputs/[c]*

* Linter fixes for plugins/inputs/[c]*

Co-authored-by: Pawel Zak <Pawel Zak>

Author: zak-pawel

plugins/inputs/cassandra/README.md

@@ -19,10 +19,26 @@ Cassandra plugin produces one or more measurements for each metric configured, a
 Given a configuration like:
 
 ```toml
+# Read Cassandra metrics through Jolokia
 [[inputs.cassandra]]
+  ## DEPRECATED: The cassandra plugin has been deprecated.  Please use the
+  ## jolokia2 plugin instead.
+  ##
+  ## see https://github.com/influxdata/telegraf/tree/master/plugins/inputs/jolokia2
+
   context = "/jolokia/read"
-  servers = [":8778"]
-  metrics = ["/java.lang:type=Memory/HeapMemoryUsage"]
+  ## List of cassandra servers exposing jolokia read service
+  servers = ["myuser:[email protected]: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"
+  ]
 ```
 
 The collected metrics will be:

plugins/inputs/cassandra/cassandra.go

@@ -49,13 +49,11 @@ type jmxMetric interface {
 	addTagsFields(out map[string]interface{})
 }
 
-func newJavaMetric(host string, metric string,
-	acc telegraf.Accumulator) *javaMetric {
+func newJavaMetric(acc telegraf.Accumulator, host string, metric string) *javaMetric {
 	return &javaMetric{host: host, metric: metric, acc: acc}
 }
 
-func newCassandraMetric(host string, metric string,
-	acc telegraf.Accumulator) *cassandraMetric {
+func newCassandraMetric(acc telegraf.Accumulator, host string, metric string) *cassandraMetric {
 	return &cassandraMetric{host: host, metric: metric, acc: acc}
 }
 
@@ -72,13 +70,15 @@ func addValuesAsFields(values map[string]interface{}, fields map[string]interfac
 func parseJmxMetricRequest(mbean string) map[string]string {
 	tokens := make(map[string]string)
 	classAndPairs := strings.Split(mbean, ":")
-	if classAndPairs[0] == "org.apache.cassandra.metrics" {
+	switch classAndPairs[0] {
+	case "org.apache.cassandra.metrics":
 		tokens["class"] = "cassandra"
-	} else if classAndPairs[0] == "java.lang" {
+	case "java.lang":
 		tokens["class"] = "java"
-	} else {
+	default:
 		return tokens
 	}
+
 	pairs := strings.Split(classAndPairs[1], ",")
 	for _, pair := range pairs {
 		p := strings.Split(pair, "=")
@@ -147,22 +147,21 @@ func (c cassandraMetric) addTagsFields(out map[string]interface{}) {
 	// 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 {
+		valuesMap, ok := out["value"]
+		if !ok {
 			c.acc.AddError(fmt.Errorf("missing key 'value' in '%s' output response: %v", c.metric, out))
 			return
 		}
+		for k, v := range valuesMap.(map[string]interface{}) {
+			addCassandraMetric(k, c, v.(map[string]interface{}))
+		}
 	} else {
-		if values, ok := out["value"]; ok {
-			addCassandraMetric(r.(map[string]interface{})["mbean"].(string),
-				c, values.(map[string]interface{}))
-		} else {
+		values, ok := out["value"]
+		if !ok {
 			c.acc.AddError(fmt.Errorf("missing key 'value' in '%s' output response: %v", c.metric, out))
 			return
 		}
+		addCassandraMetric(r.(map[string]interface{})["mbean"].(string), c, values.(map[string]interface{}))
 	}
 }
 
@@ -277,10 +276,10 @@ func (c *Cassandra) Gather(acc telegraf.Accumulator) error {
 
 			var m jmxMetric
 			if strings.HasPrefix(metric, "/java.lang:") {
-				m = newJavaMetric(serverTokens["host"], metric, acc)
+				m = newJavaMetric(acc, serverTokens["host"], metric)
 			} else if strings.HasPrefix(metric,
 				"/org.apache.cassandra.metrics:") {
-				m = newCassandraMetric(serverTokens["host"], metric, acc)
+				m = newCassandraMetric(acc, serverTokens["host"], metric)
 			} else {
 				// unsupported metric type
 				acc.AddError(fmt.Errorf("unsupported Cassandra metric [%s], skipping", metric))

plugins/inputs/ceph/README.md

@@ -45,7 +45,7 @@ the cluster.  The currently supported commands are:
 ### Configuration:
 
 ```toml
-# Collects performance metrics from the MON and OSD nodes in a Ceph storage cluster.
+# Collects performance metrics from the MON, OSD, MDS and RGW nodes in a Ceph storage cluster.
 [[inputs.ceph]]
   ## This is the recommended interval to poll.  Too frequent and you will lose
   ## data points due to timeouts during rebalancing and recovery

plugins/inputs/ceph/ceph.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
-	"log"
 	"os/exec"
 	"path/filepath"
 	"strings"
@@ -28,17 +27,19 @@ const (
 )
 
 type Ceph struct {
-	CephBinary             string
-	OsdPrefix              string
-	MonPrefix              string
-	MdsPrefix              string
-	RgwPrefix              string
-	SocketDir              string
-	SocketSuffix           string
-	CephUser               string
-	CephConfig             string
-	GatherAdminSocketStats bool
-	GatherClusterStats     bool
+	CephBinary             string `toml:"ceph_binary"`
+	OsdPrefix              string `toml:"osd_prefix"`
+	MonPrefix              string `toml:"mon_prefix"`
+	MdsPrefix              string `toml:"mds_prefix"`
+	RgwPrefix              string `toml:"rgw_prefix"`
+	SocketDir              string `toml:"socket_dir"`
+	SocketSuffix           string `toml:"socket_suffix"`
+	CephUser               string `toml:"ceph_user"`
+	CephConfig             string `toml:"ceph_config"`
+	GatherAdminSocketStats bool   `toml:"gather_admin_socket_stats"`
+	GatherClusterStats     bool   `toml:"gather_cluster_stats"`
+
+	Log telegraf.Logger `toml:"-"`
 }
 
 func (c *Ceph) Description() string {
@@ -67,7 +68,14 @@ var sampleConfig = `
   ## suffix used to identify socket files
   socket_suffix = "asok"
 
-  ## Ceph user to authenticate as
+  ## Ceph user to authenticate as, ceph will search for the corresponding keyring
+  ## e.g. client.admin.keyring in /etc/ceph, or the explicit path defined in the
+  ## client section of ceph.conf for example:
+  ##
+  ##     [client.telegraf]
+  ##         keyring = /etc/ceph/client.telegraf.keyring
+  ##
+  ## Consult the ceph documentation for more detail on keyring generation.
   ceph_user = "client.admin"
 
   ## Ceph configuration to use to locate the cluster
@@ -76,7 +84,8 @@ var sampleConfig = `
   ## Whether to gather statistics via the admin socket
   gather_admin_socket_stats = true
 
-  ## Whether to gather statistics via ceph commands
+  ## Whether to gather statistics via ceph commands, requires ceph_user and ceph_config
+  ## to be specified
   gather_cluster_stats = false
 `
 
@@ -112,14 +121,14 @@ func (c *Ceph) gatherAdminSocketStats(acc telegraf.Accumulator) error {
 			acc.AddError(fmt.Errorf("error reading from socket '%s': %v", s.socket, err))
 			continue
 		}
-		data, err := parseDump(dump)
+		data, err := c.parseDump(dump)
 		if err != nil {
 			acc.AddError(fmt.Errorf("error parsing dump from socket '%s': %v", s.socket, err))
 			continue
 		}
 		for tag, metrics := range data {
 			acc.AddFields(measurement,
-				map[string]interface{}(metrics),
+				metrics,
 				map[string]string{"type": s.sockType, "id": s.sockID, "collection": tag})
 		}
 	}
@@ -138,7 +147,7 @@ func (c *Ceph) gatherClusterStats(acc telegraf.Accumulator) error {
 
 	// For each job, execute against the cluster, parse and accumulate the data points
 	for _, job := range jobs {
-		output, err := c.exec(job.command)
+		output, err := c.execute(job.command)
 		if err != nil {
 			return fmt.Errorf("error executing command: %v", err)
 		}
@@ -171,15 +180,17 @@ func init() {
 
 var perfDump = func(binary string, socket *socket) (string, error) {
 	cmdArgs := []string{"--admin-daemon", socket.socket}
-	if socket.sockType == typeOsd {
+
+	switch socket.sockType {
+	case typeOsd:
 		cmdArgs = append(cmdArgs, "perf", "dump")
-	} else if socket.sockType == typeMon {
+	case typeMon:
 		cmdArgs = append(cmdArgs, "perfcounters_dump")
-	} else if socket.sockType == typeMds {
+	case typeMds:
 		cmdArgs = append(cmdArgs, "perf", "dump")
-	} else if socket.sockType == typeRgw {
+	case typeRgw:
 		cmdArgs = append(cmdArgs, "perf", "dump")
-	} else {
+	default:
 		return "", fmt.Errorf("ignoring unknown socket type: %s", socket.sockType)
 	}
 
@@ -268,23 +279,23 @@ type taggedMetricMap map[string]metricMap
 
 // Parses a raw JSON string into a taggedMetricMap
 // Delegates the actual parsing to newTaggedMetricMap(..)
-func parseDump(dump string) (taggedMetricMap, error) {
+func (c *Ceph) parseDump(dump string) (taggedMetricMap, error) {
 	data := make(map[string]interface{})
 	err := json.Unmarshal([]byte(dump), &data)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse json: '%s': %v", dump, err)
 	}
 
-	return newTaggedMetricMap(data), nil
+	return c.newTaggedMetricMap(data), nil
 }
 
 // Builds a TaggedMetricMap out of a generic string map.
 // The top-level key is used as a tag and all sub-keys are flattened into metrics
-func newTaggedMetricMap(data map[string]interface{}) taggedMetricMap {
+func (c *Ceph) newTaggedMetricMap(data map[string]interface{}) taggedMetricMap {
 	tmm := make(taggedMetricMap)
 	for tag, datapoints := range data {
 		mm := make(metricMap)
-		for _, m := range flatten(datapoints) {
+		for _, m := range c.flatten(datapoints) {
 			mm[m.name()] = m.value
 		}
 		tmm[tag] = mm
@@ -296,7 +307,7 @@ func newTaggedMetricMap(data map[string]interface{}) taggedMetricMap {
 // Nested keys are flattened into ordered slices associated with a metric value.
 // The key slices are treated as stacks, and are expected to be reversed and concatenated
 // when passed as metrics to the accumulator. (see (*metric).name())
-func flatten(data interface{}) []*metric {
+func (c *Ceph) flatten(data interface{}) []*metric {
 	var metrics []*metric
 
 	switch val := data.(type) {
@@ -305,20 +316,20 @@ func flatten(data interface{}) []*metric {
 	case map[string]interface{}:
 		metrics = make([]*metric, 0, len(val))
 		for k, v := range val {
-			for _, m := range flatten(v) {
+			for _, m := range c.flatten(v) {
 				m.pathStack = append(m.pathStack, k)
 				metrics = append(metrics, m)
 			}
 		}
 	default:
-		log.Printf("I! [inputs.ceph] ignoring unexpected type '%T' for value %v", val, val)
+		c.Log.Infof("ignoring unexpected type '%T' for value %v", val, val)
 	}
 
 	return metrics
 }
 
-// exec executes the 'ceph' command with the supplied arguments, returning JSON formatted output
-func (c *Ceph) exec(command string) (string, error) {
+// execute executes the 'ceph' command with the supplied arguments, returning JSON formatted output
+func (c *Ceph) execute(command string) (string, error) {
 	cmdArgs := []string{"--conf", c.CephConfig, "--name", c.CephUser, "--format", "json"}
 	cmdArgs = append(cmdArgs, strings.Split(command, " ")...)
 

plugins/inputs/ceph/ceph_test.go

@@ -9,8 +9,9 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf/testutil"
 )
 
 const (
@@ -29,28 +30,32 @@ func TestParseSockId(t *testing.T) {
 }
 
 func TestParseMonDump(t *testing.T) {
-	dump, err := parseDump(monPerfDump)
+	c := &Ceph{Log: testutil.Logger{}}
+	dump, err := c.parseDump(monPerfDump)
 	require.NoError(t, err)
 	require.InEpsilon(t, int64(5678670180), dump["cluster"]["osd_kb_used"], epsilon)
 	require.InEpsilon(t, 6866.540527000, dump["paxos"]["store_state_latency.sum"], epsilon)
 }
 
 func TestParseOsdDump(t *testing.T) {
-	dump, err := parseDump(osdPerfDump)
+	c := &Ceph{Log: testutil.Logger{}}
+	dump, err := c.parseDump(osdPerfDump)
 	require.NoError(t, err)
 	require.InEpsilon(t, 552132.109360000, dump["filestore"]["commitcycle_interval.sum"], epsilon)
 	require.Equal(t, float64(0), dump["mutex-FileJournal::finisher_lock"]["wait.avgcount"])
 }
 
 func TestParseMdsDump(t *testing.T) {
-	dump, err := parseDump(mdsPerfDump)
+	c := &Ceph{Log: testutil.Logger{}}
+	dump, err := c.parseDump(mdsPerfDump)
 	require.NoError(t, err)
 	require.InEpsilon(t, 2408386.600934982, dump["mds"]["reply_latency.sum"], epsilon)
 	require.Equal(t, float64(0), dump["throttle-write_buf_throttle"]["wait.avgcount"])
 }
 
 func TestParseRgwDump(t *testing.T) {
-	dump, err := parseDump(rgwPerfDump)
+	c := &Ceph{Log: testutil.Logger{}}
+	dump, err := c.parseDump(rgwPerfDump)
 	require.NoError(t, err)
 	require.InEpsilon(t, 0.002219876, dump["rgw"]["get_initial_lat.sum"], epsilon)
 	require.Equal(t, float64(0), dump["rgw"]["put_initial_lat.avgcount"])

plugins/inputs/cgroup/README.md

@@ -44,12 +44,19 @@ All measurements have the following tags:
 ### Configuration:
 
 ```toml
+# Read specific statistics per cgroup
 # [[inputs.cgroup]]
+  ## Directories in which to look for files, globs are supported.
+  ## Consider restricting paths to the set of cgroups you really
+  ## want to monitor if you have a large number of cgroups, to avoid
+  ## any cardinality issues.
   # paths = [
-  #   "/sys/fs/cgroup/memory",           # root cgroup
-  #   "/sys/fs/cgroup/memory/child1",    # container cgroup
-  #   "/sys/fs/cgroup/memory/child2/*",  # all children cgroups under child2, but not child2 itself
+  #   "/sys/fs/cgroup/memory",
+  #   "/sys/fs/cgroup/memory/child1",
+  #   "/sys/fs/cgroup/memory/child2/*",
   # ]
+  ## cgroup stat fields, as file names, globs are supported.
+  ## these file names are appended to each path from above.
   # files = ["memory.*usage*", "memory.limit_in_bytes"]
 ```
 

plugins/inputs/cgroup/cgroup_linux.go

@@ -25,15 +25,15 @@ func (g *CGroup) Gather(acc telegraf.Accumulator) error {
 			acc.AddError(dir.err)
 			continue
 		}
-		if err := g.gatherDir(dir.path, acc); err != nil {
+		if err := g.gatherDir(acc, dir.path); err != nil {
 			acc.AddError(err)
 		}
 	}
 
 	return nil
 }
 
-func (g *CGroup) gatherDir(dir string, acc telegraf.Accumulator) error {
+func (g *CGroup) gatherDir(acc telegraf.Accumulator, dir string) error {
 	fields := make(map[string]interface{})
 
 	list := make(chan pathInfo)
@@ -72,8 +72,8 @@ type pathInfo struct {
 	err  error
 }
 
-func isDir(path string) (bool, error) {
-	result, err := os.Stat(path)
+func isDir(pathToCheck string) (bool, error) {
+	result, err := os.Stat(pathToCheck)
 	if err != nil {
 		return false, err
 	}