Add dmcache input plugin (#1667)

CHANGELOG.md

@@ -64,6 +64,7 @@ be deprecated eventually.
 - [#2587](https://github.com/influxdata/telegraf/pull/2587): Add json timestamp units configurability
 - [#2597](https://github.com/influxdata/telegraf/issues/2597): Add support for Linux sysctl-fs metrics.
 - [#2425](https://github.com/influxdata/telegraf/pull/2425): Support to include/exclude docker container labels as tags
+- [#1667](https://github.com/influxdata/telegraf/pull/1667): dmcache input plugin
 
 ### Bugfixes
 

plugins/inputs/all/all.go

@@ -15,6 +15,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/couchbase"
 	_ "github.com/influxdata/telegraf/plugins/inputs/couchdb"
 	_ "github.com/influxdata/telegraf/plugins/inputs/disque"
+	_ "github.com/influxdata/telegraf/plugins/inputs/dmcache"
 	_ "github.com/influxdata/telegraf/plugins/inputs/dns_query"
 	_ "github.com/influxdata/telegraf/plugins/inputs/docker"
 	_ "github.com/influxdata/telegraf/plugins/inputs/dovecot"

plugins/inputs/dmcache/README.md

@@ -0,0 +1,47 @@
+# DMCache Input Plugin
+
+This plugin provide a native collection for dmsetup based statistics for dm-cache.
+
+This plugin requires sudo, that is why you should setup and be sure that the telegraf is able to execute sudo without a password.
+
+`sudo /sbin/dmsetup status --target cache` is the full command that telegraf will run for debugging purposes.
+
+### Configuration
+
+```toml
+[[inputs.dmcache]]
+  ## Whether to report per-device stats or not
+  per_device = true
+```
+
+### Measurements & Fields:
+
+- dmcache
+    - length
+    - target
+    - metadata_blocksize
+    - metadata_used
+    - metadata_total
+    - cache_blocksize
+    - cache_used
+    - cache_total
+    - read_hits
+    - read_misses
+    - write_hits
+    - write_misses
+    - demotions
+    - promotions
+    - dirty
+
+### Tags:
+
+- All measurements have the following tags:
+    - device
+
+### Example Output:
+
+```
+$ ./telegraf --test --config /etc/telegraf/telegraf.conf --input-filter dmcache
+* Plugin: inputs.dmcache, Collection 1
+> dmcache,device=example cache_blocksize=0i,read_hits=995134034411520i,read_misses=916807089127424i,write_hits=195107267543040i,metadata_used=12861440i,write_misses=563725346013184i,promotions=3265223720960i,dirty=0i,metadata_blocksize=0i,cache_used=1099511627776ii,cache_total=0i,length=0i,metadata_total=1073741824i,demotions=3265223720960i 1491482035000000000
+```

plugins/inputs/dmcache/dmcache.go

@@ -0,0 +1,33 @@
+package dmcache
+
+import (
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+type DMCache struct {
+	PerDevice        bool `toml:"per_device"`
+	getCurrentStatus func() ([]string, error)
+}
+
+var sampleConfig = `
+  ## Whether to report per-device stats or not
+  per_device = true
+`
+
+func (c *DMCache) SampleConfig() string {
+	return sampleConfig
+}
+
+func (c *DMCache) Description() string {
+	return "Provide a native collection for dmsetup based statistics for dm-cache"
+}
+
+func init() {
+	inputs.Add("dmcache", func() telegraf.Input {
+		return &DMCache{
+			PerDevice:        true,
+			getCurrentStatus: dmSetupStatus,
+		}
+	})
+}

plugins/inputs/dmcache/dmcache_linux.go

@@ -0,0 +1,190 @@
+// +build linux
+
+package dmcache
+
+import (
+	"os/exec"
+	"strconv"
+	"strings"
+
+	"errors"
+
+	"github.com/influxdata/telegraf"
+)
+
+const metricName = "dmcache"
+
+type cacheStatus struct {
+	device            string
+	length            int
+	target            string
+	metadataBlocksize int
+	metadataUsed      int
+	metadataTotal     int
+	cacheBlocksize    int
+	cacheUsed         int
+	cacheTotal        int
+	readHits          int
+	readMisses        int
+	writeHits         int
+	writeMisses       int
+	demotions         int
+	promotions        int
+	dirty             int
+}
+
+func (c *DMCache) Gather(acc telegraf.Accumulator) error {
+	outputLines, err := c.getCurrentStatus()
+	if err != nil {
+		return err
+	}
+
+	totalStatus := cacheStatus{}
+
+	for _, s := range outputLines {
+		status, err := parseDMSetupStatus(s)
+		if err != nil {
+			return err
+		}
+
+		if c.PerDevice {
+			tags := map[string]string{"device": status.device}
+			acc.AddFields(metricName, toFields(status), tags)
+		}
+		aggregateStats(&totalStatus, status)
+	}
+
+	acc.AddFields(metricName, toFields(totalStatus), map[string]string{"device": "all"})
+
+	return nil
+}
+
+func parseDMSetupStatus(line string) (cacheStatus, error) {
+	var err error
+	parseError := errors.New("Output from dmsetup could not be parsed")
+	status := cacheStatus{}
+	values := strings.Fields(line)
+	if len(values) < 15 {
+		return cacheStatus{}, parseError
+	}
+
+	status.device = strings.TrimRight(values[0], ":")
+	status.length, err = strconv.Atoi(values[2])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.target = values[3]
+	status.metadataBlocksize, err = strconv.Atoi(values[4])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	metadata := strings.Split(values[5], "/")
+	if len(metadata) != 2 {
+		return cacheStatus{}, parseError
+	}
+	status.metadataUsed, err = strconv.Atoi(metadata[0])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.metadataTotal, err = strconv.Atoi(metadata[1])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.cacheBlocksize, err = strconv.Atoi(values[6])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	cache := strings.Split(values[7], "/")
+	if len(cache) != 2 {
+		return cacheStatus{}, parseError
+	}
+	status.cacheUsed, err = strconv.Atoi(cache[0])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.cacheTotal, err = strconv.Atoi(cache[1])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.readHits, err = strconv.Atoi(values[8])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.readMisses, err = strconv.Atoi(values[9])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.writeHits, err = strconv.Atoi(values[10])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.writeMisses, err = strconv.Atoi(values[11])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.demotions, err = strconv.Atoi(values[12])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.promotions, err = strconv.Atoi(values[13])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+	status.dirty, err = strconv.Atoi(values[14])
+	if err != nil {
+		return cacheStatus{}, err
+	}
+
+	return status, nil
+}
+
+func aggregateStats(totalStatus *cacheStatus, status cacheStatus) {
+	totalStatus.length += status.length
+	totalStatus.metadataBlocksize += status.metadataBlocksize
+	totalStatus.metadataUsed += status.metadataUsed
+	totalStatus.metadataTotal += status.metadataTotal
+	totalStatus.cacheBlocksize += status.cacheBlocksize
+	totalStatus.cacheUsed += status.cacheUsed
+	totalStatus.cacheTotal += status.cacheTotal
+	totalStatus.readHits += status.readHits
+	totalStatus.readMisses += status.readMisses
+	totalStatus.writeHits += status.writeHits
+	totalStatus.writeMisses += status.writeMisses
+	totalStatus.demotions += status.demotions
+	totalStatus.promotions += status.promotions
+	totalStatus.dirty += status.dirty
+}
+
+func toFields(status cacheStatus) map[string]interface{} {
+	fields := make(map[string]interface{})
+	fields["length"] = status.length
+	fields["metadata_blocksize"] = status.metadataBlocksize
+	fields["metadata_used"] = status.metadataUsed
+	fields["metadata_total"] = status.metadataTotal
+	fields["cache_blocksize"] = status.cacheBlocksize
+	fields["cache_used"] = status.cacheUsed
+	fields["cache_total"] = status.cacheTotal
+	fields["read_hits"] = status.readHits
+	fields["read_misses"] = status.readMisses
+	fields["write_hits"] = status.writeHits
+	fields["write_misses"] = status.writeMisses
+	fields["demotions"] = status.demotions
+	fields["promotions"] = status.promotions
+	fields["dirty"] = status.dirty
+	return fields
+}
+
+func dmSetupStatus() ([]string, error) {
+	out, err := exec.Command("/bin/sh", "-c", "sudo /sbin/dmsetup status --target cache").Output()
+	if err != nil {
+		return nil, err
+	}
+	if string(out) == "No devices found\n" {
+		return []string{}, nil
+	}
+
+	outString := strings.TrimRight(string(out), "\n")
+	status := strings.Split(outString, "\n")
+
+	return status, nil
+}

plugins/inputs/dmcache/dmcache_notlinux.go

@@ -0,0 +1,15 @@
+// +build !linux
+
+package dmcache
+
+import (
+	"github.com/influxdata/telegraf"
+)
+
+func (c *DMCache) Gather(acc telegraf.Accumulator) error {
+	return nil
+}
+
+func dmSetupStatus() ([]string, error) {
+	return []string{}, nil
+}

plugins/inputs/dmcache/dmcache_test.go

@@ -0,0 +1,169 @@
+package dmcache
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	measurement              = "dmcache"
+	badFormatOutput          = []string{"cs-1: 0 4883791872 cache 8 1018/1501122 512 7/464962 139 352643 "}
+	good2DevicesFormatOutput = []string{
+		"cs-1: 0 4883791872 cache 8 1018/1501122 512 7/464962 139 352643 15 46 0 7 0 1 writeback 2 migration_threshold 2048 mq 10 random_threshold 4 sequential_threshold 512 discard_promote_adjustment 1 read_promote_adjustment 4 write_promote_adjustment 8",
+		"cs-2: 0 4294967296 cache 8 72352/1310720 128 26/24327168 2409 286 265 524682 0 0 0 1 writethrough 2 migration_threshold 2048 mq 10 random_threshold 4 sequential_threshold 512 discard_promote_adjustment 1 read_promote_adjustment 4 write_promote_adjustment 8",
+	}
+)
+
+func TestPerDeviceGoodOutput(t *testing.T) {
+	var acc testutil.Accumulator
+	var plugin = &DMCache{
+		PerDevice: true,
+		getCurrentStatus: func() ([]string, error) {
+			return good2DevicesFormatOutput, nil
+		},
+	}
+
+	err := plugin.Gather(&acc)
+	require.NoError(t, err)
+
+	tags1 := map[string]string{
+		"device": "cs-1",
+	}
+	fields1 := map[string]interface{}{
+		"length":             4883791872,
+		"metadata_blocksize": 8,
+		"metadata_used":      1018,
+		"metadata_total":     1501122,
+		"cache_blocksize":    512,
+		"cache_used":         7,
+		"cache_total":        464962,
+		"read_hits":          139,
+		"read_misses":        352643,
+		"write_hits":         15,
+		"write_misses":       46,
+		"demotions":          0,
+		"promotions":         7,
+		"dirty":              0,
+	}
+	acc.AssertContainsTaggedFields(t, measurement, fields1, tags1)
+
+	tags2 := map[string]string{
+		"device": "cs-2",
+	}
+	fields2 := map[string]interface{}{
+		"length":             4294967296,
+		"metadata_blocksize": 8,
+		"metadata_used":      72352,
+		"metadata_total":     1310720,
+		"cache_blocksize":    128,
+		"cache_used":         26,
+		"cache_total":        24327168,
+		"read_hits":          2409,
+		"read_misses":        286,
+		"write_hits":         265,
+		"write_misses":       524682,
+		"demotions":          0,
+		"promotions":         0,
+		"dirty":              0,
+	}
+	acc.AssertContainsTaggedFields(t, measurement, fields2, tags2)
+
+	tags3 := map[string]string{
+		"device": "all",
+	}
+
+	fields3 := map[string]interface{}{
+		"length":             9178759168,
+		"metadata_blocksize": 16,
+		"metadata_used":      73370,
+		"metadata_total":     2811842,
+		"cache_blocksize":    640,
+		"cache_used":         33,
+		"cache_total":        24792130,
+		"read_hits":          2548,
+		"read_misses":        352929,
+		"write_hits":         280,
+		"write_misses":       524728,
+		"demotions":          0,
+		"promotions":         7,
+		"dirty":              0,
+	}
+	acc.AssertContainsTaggedFields(t, measurement, fields3, tags3)
+}
+
+func TestNotPerDeviceGoodOutput(t *testing.T) {
+	var acc testutil.Accumulator
+	var plugin = &DMCache{
+		PerDevice: false,
+		getCurrentStatus: func() ([]string, error) {
+			return good2DevicesFormatOutput, nil
+		},
+	}
+
+	err := plugin.Gather(&acc)
+	require.NoError(t, err)
+
+	tags := map[string]string{
+		"device": "all",
+	}
+
+	fields := map[string]interface{}{
+		"length":             9178759168,
+		"metadata_blocksize": 16,
+		"metadata_used":      73370,
+		"metadata_total":     2811842,
+		"cache_blocksize":    640,
+		"cache_used":         33,
+		"cache_total":        24792130,
+		"read_hits":          2548,
+		"read_misses":        352929,
+		"write_hits":         280,
+		"write_misses":       524728,
+		"demotions":          0,
+		"promotions":         7,
+		"dirty":              0,
+	}
+	acc.AssertContainsTaggedFields(t, measurement, fields, tags)
+}
+
+func TestNoDevicesOutput(t *testing.T) {
+	var acc testutil.Accumulator
+	var plugin = &DMCache{
+		PerDevice: true,
+		getCurrentStatus: func() ([]string, error) {
+			return []string{}, nil
+		},
+	}
+
+	err := plugin.Gather(&acc)
+	require.NoError(t, err)
+}
+
+func TestErrorDuringGettingStatus(t *testing.T) {
+	var acc testutil.Accumulator
+	var plugin = &DMCache{
+		PerDevice: true,
+		getCurrentStatus: func() ([]string, error) {
+			return nil, errors.New("dmsetup doesn't exist")
+		},
+	}
+
+	err := plugin.Gather(&acc)
+	require.Error(t, err)
+}
+
+func TestBadFormatOfStatus(t *testing.T) {
+	var acc testutil.Accumulator
+	var plugin = &DMCache{
+		PerDevice: true,
+		getCurrentStatus: func() ([]string, error) {
+			return badFormatOutput, nil
+		},
+	}
+
+	err := plugin.Gather(&acc)
+	require.Error(t, err)
+}