Add dmcache input plugin

plugins/inputs/all/all.go

@@ -14,6 +14,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,11 @@
+# DMCache Input Plugin
+
+This plugin provide a native collection for dmsetup based statistics for dm-cache
+
+## Configuration
+
+```
+[[inputs.dmcache]]
+## Whether to report per-device stats or not
+	per_device = true
+```

plugins/inputs/dmcache/dmcache.go

@@ -0,0 +1,49 @@
+package dmcache
+
+import (
+	"os/exec"
+	"strings"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+type DMCache struct {
+	PerDevice bool `toml:"per_device"`
+	rawStatus 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,
+			rawStatus: func() ([]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 nil, nil
+				}
+
+				status := strings.Split(string(out), "\n")
+				status = status[:len(status)-1] // removing last empty line
+
+				return status, nil
+			},
+		}
+	})
+}

plugins/inputs/dmcache/dmcache_linux.go

@@ -0,0 +1,131 @@
+// +build linux
+
+package dmcache
+
+import (
+	"strconv"
+	"strings"
+
+	"github.com/influxdata/telegraf"
+)
+
+const metricName = "dmcache"
+
+var fieldNames = [...]string{
+	"metadata_used",
+	"metadata_free",
+	"cache_used",
+	"cache_free",
+	"read_hits",
+	"read_misses",
+	"write_hits",
+	"write_misses",
+	"demotions",
+	"promotions",
+	"dirty",
+}
+
+func (c *DMCache) Gather(acc telegraf.Accumulator) error {
+	outputLines, err := c.rawStatus()
+	if err != nil {
+		return err
+	}
+
+	var total map[string]interface{}
+	if !c.PerDevice {
+		total = make(map[string]interface{})
+	}
+
+	for _, s := range outputLines {
+		fields := make(map[string]interface{})
+		data, err := parseDMSetupStatus(s)
+		if err != nil {
+			return err
+		}
+
+		for _, f := range fieldNames {
+			fields[f] = calculateSize(data, f)
+		}
+
+		if c.PerDevice {
+			tags := map[string]string{"device": data["device"].(string)}
+			acc.AddFields(metricName, fields, tags)
+		} else {
+			aggregateStats(total, fields)
+		}
+	}
+
+	if !c.PerDevice {
+		acc.AddFields(metricName, total, nil)
+	}
+
+	return nil
+}
+
+func parseDMSetupStatus(line string) (status map[string]interface{}, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			status = nil
+			err = r.(error)
+		}
+	}()
+
+	values := strings.Split(line, " ")
+	status = make(map[string]interface{})
+
+	status["device"] = values[0][:len(values[0])-1]
+	status["length"] = toInt(values[2])
+	status["target"] = values[3]
+	status["metadata_blocksize"] = toInt(values[4])
+	status["metadata_used"] = toInt(strings.Split(values[5], "/")[0])
+	status["metadata_total"] = toInt(strings.Split(values[5], "/")[1])
+	status["cache_blocksize"] = toInt(values[6])
+	status["cache_used"] = toInt(strings.Split(values[7], "/")[0])
+	status["cache_total"] = toInt(strings.Split(values[7], "/")[1])
+	status["read_hits"] = toInt(values[8])
+	status["read_misses"] = toInt(values[9])
+	status["write_hits"] = toInt(values[10])
+	status["write_misses"] = toInt(values[11])
+	status["demotions"] = toInt(values[12])
+	status["promotions"] = toInt(values[13])
+	status["dirty"] = toInt(values[14])
+	status["blocksize"] = 512
+
+	return status, nil
+}
+
+func toInt(s string) int {
+	i, err := strconv.Atoi(s)
+	if err != nil {
+		panic(err)
+	}
+	return i
+}
+
+func calculateSize(data map[string]interface{}, key string) (value int) {
+	if key == "metadata_free" {
+		value = data["metadata_total"].(int) - data["metadata_used"].(int)
+	} else if key == "cache_free" {
+		value = data["cache_total"].(int) - data["cache_used"].(int) - data["dirty"].(int)
+	} else {
+		value = data[key].(int)
+	}
+
+	if key == "metadata_free" || key == "metadata_used" {
+		value = value * data["blocksize"].(int) * data["metadata_blocksize"].(int)
+	} else {
+		value = value * data["blocksize"].(int) * data["cache_blocksize"].(int)
+	}
+
+	return
+}
+
+func aggregateStats(total, fields map[string]interface{}) {
+	for key, value := range fields {
+		if _, ok := total[key]; ok {
+			total[key] = total[key].(int) + value.(int)
+		} else {
+			total[key] = value.(int)
+		}
+	}
+}

plugins/inputs/dmcache/dmcache_notlinux.go

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

plugins/inputs/dmcache/dmcache_test.go

@@ -0,0 +1,94 @@
+// +build linux
+
+package dmcache
+
+import (
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func output2Devices() ([]string, error) {
+	return []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",
+	}, nil
+}
+
+var dmc1 = &DMCache{
+	PerDevice: true,
+	rawStatus: output2Devices,
+}
+
+func TestDMCacheStats_1(t *testing.T) {
+	var acc testutil.Accumulator
+
+	err := dmc1.Gather(&acc)
+	require.NoError(t, err)
+
+	tags1 := map[string]string{
+		"device": "cs-1",
+	}
+	fields1 := map[string]interface{}{
+		"metadata_used": 4169728,
+		"metadata_free": 6144425984,
+		"cache_used":    1835008,
+		"cache_free":    121885163520,
+		"read_hits":     36438016,
+		"read_misses":   92443246592,
+		"write_hits":    3932160,
+		"write_misses":  12058624,
+		"demotions":     0,
+		"promotions":    1835008,
+		"dirty":         0,
+	}
+	acc.AssertContainsTaggedFields(t, "dmcache", fields1, tags1)
+
+	tags2 := map[string]string{
+		"device": "cs-2",
+	}
+	fields2 := map[string]interface{}{
+		"metadata_used": 296353792,
+		"metadata_free": 5072355328,
+		"cache_used":    1703936,
+		"cache_free":    1594303578112,
+		"read_hits":     157876224,
+		"read_misses":   18743296,
+		"write_hits":    17367040,
+		"write_misses":  34385559552,
+		"demotions":     0,
+		"promotions":    0,
+		"dirty":         0,
+	}
+	acc.AssertContainsTaggedFields(t, "dmcache", fields2, tags2)
+}
+
+var dmc2 = &DMCache{
+	PerDevice: false,
+	rawStatus: output2Devices,
+}
+
+func TestDMCacheStats_2(t *testing.T) {
+	var acc testutil.Accumulator
+
+	err := dmc2.Gather(&acc)
+	require.NoError(t, err)
+
+	tags := map[string]string{}
+
+	fields := map[string]interface{}{
+		"metadata_used": 300523520,
+		"metadata_free": 11216781312,
+		"cache_used":    3538944,
+		"cache_free":    1716188741632,
+		"read_hits":     194314240,
+		"read_misses":   92461989888,
+		"write_hits":    21299200,
+		"write_misses":  34397618176,
+		"demotions":     0,
+		"promotions":    1835008,
+		"dirty":         0,
+	}
+	acc.AssertContainsTaggedFields(t, "dmcache", fields, tags)
+}