Add bond input plugin (#3424)

plugins/inputs/all/all.go

@@ -5,6 +5,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/amqp_consumer"
 	_ "github.com/influxdata/telegraf/plugins/inputs/apache"
 	_ "github.com/influxdata/telegraf/plugins/inputs/bcache"
+	_ "github.com/influxdata/telegraf/plugins/inputs/bond"
 	_ "github.com/influxdata/telegraf/plugins/inputs/cassandra"
 	_ "github.com/influxdata/telegraf/plugins/inputs/ceph"
 	_ "github.com/influxdata/telegraf/plugins/inputs/cgroup"

plugins/inputs/bond/README.md

+# Bond Input Plugin
+
+The Bond Input plugin collects bond interface status, bond's slaves interfaces
+status and failures count of bond's slaves interfaces.
+The plugin collects these metrics from `/proc/net/bonding/*` files.
+
+### Configuration:
+
+```toml
+[[inputs.bond]]
+  ## Sets 'proc' directory path
+  ## If not specified, then default is /proc
+  # host_proc = "/proc"
+
+  ## By default, telegraf gather stats for all bond interfaces
+  ## Setting interfaces will restrict the stats to the specified
+  ## bond interfaces.
+  # bond_interfaces = ["bond0"]
+```
+
+### Measurements & Fields:
+
+- bond
+  - active_slave (for active-backup mode)
+  - status
+
+- bond_slave
+  - failures
+  - status
+
+### Description:
+
+```
+active_slave
+  Currently active slave interface for active-backup mode.
+
+status
+  Status of bond interface or bonds's slave interface (down = 0, up = 1).
+
+failures
+  Amount of failures for bond's slave interface.
+```
+
+### Tags:
+
+- bond
+  - bond
+
+- bond_slave
+  - bond
+  - interface
+
+### Example output:
+
+Configuration:
+
+```
+[[inputs.bond]]
+  ## Sets 'proc' directory path
+  ## If not specified, then default is /proc
+  host_proc = "/proc"
+
+  ## By default, telegraf gather stats for all bond interfaces
+  ## Setting interfaces will restrict the stats to the specified
+  ## bond interfaces.
+  bond_interfaces = ["bond0", "bond1"]
+```
+
+Run:
+
+```
+telegraf --config telegraf.conf --input-filter bond --test
+```
+
+Output:
+
+```
+* Plugin: inputs.bond, Collection 1
+> bond,bond=bond1,host=local active_slave="eth0",status=1i 1509704525000000000
+> bond_slave,bond=bond1,interface=eth0,host=local status=1i,failures=0i 1509704525000000000
+> bond_slave,host=local,bond=bond1,interface=eth1 status=1i,failures=0i 1509704525000000000
+> bond,bond=bond0,host=isvetlov-mac.local status=1i 1509704525000000000
+> bond_slave,bond=bond0,interface=eth1,host=local status=1i,failures=0i 1509704525000000000
+> bond_slave,bond=bond0,interface=eth2,host=local status=1i,failures=0i 1509704525000000000
+```

plugins/inputs/bond/bond.go

+package bond
+
+import (
+	"bufio"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+// default host proc path
+const defaultHostProc = "/proc"
+
+// env host proc variable name
+const envProc = "HOST_PROC"
+
+type Bond struct {
+	HostProc       string   `toml:"host_proc"`
+	BondInterfaces []string `toml:"bond_interfaces"`
+}
+
+var sampleConfig = `
+  ## Sets 'proc' directory path
+  ## If not specified, then default is /proc
+  # host_proc = "/proc"
+
+  ## By default, telegraf gather stats for all bond interfaces
+  ## Setting interfaces will restrict the stats to the specified
+  ## bond interfaces.
+  # bond_interfaces = ["bond0"]
+`
+
+func (bond *Bond) Description() string {
+	return "Collect bond interface status, slaves statuses and failures count"
+}
+
+func (bond *Bond) SampleConfig() string {
+	return sampleConfig
+}
+
+func (bond *Bond) Gather(acc telegraf.Accumulator) error {
+	// load proc path, get default value if config value and env variable are empty
+	bond.loadPath()
+	// list bond interfaces from bonding directory or gather all interfaces.
+	bondNames, err := bond.listInterfaces()
+	if err != nil {
+		return err
+	}
+	for _, bondName := range bondNames {
+		bondAbsPath := bond.HostProc + "/net/bonding/" + bondName
+		file, err := ioutil.ReadFile(bondAbsPath)
+		if err != nil {
+			acc.AddError(fmt.Errorf("error inspecting '%s' interface: %v", bondAbsPath, err))
+			continue
+		}
+		rawFile := strings.TrimSpace(string(file))
+		err = bond.gatherBondInterface(bondName, rawFile, acc)
+		if err != nil {
+			acc.AddError(fmt.Errorf("error inspecting '%s' interface: %v", bondName, err))
+		}
+	}
+	return nil
+}
+
+func (bond *Bond) gatherBondInterface(bondName string, rawFile string, acc telegraf.Accumulator) error {
+	splitIndex := strings.Index(rawFile, "Slave Interface:")
+	if splitIndex == -1 {
+		splitIndex = len(rawFile)
+	}
+	bondPart := rawFile[:splitIndex]
+	slavePart := rawFile[splitIndex:]
+
+	err := bond.gatherBondPart(bondName, bondPart, acc)
+	if err != nil {
+		return err
+	}
+	err = bond.gatherSlavePart(bondName, slavePart, acc)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (bond *Bond) gatherBondPart(bondName string, rawFile string, acc telegraf.Accumulator) error {
+	fields := make(map[string]interface{})
+	tags := map[string]string{
+		"bond": bondName,
+	}
+
+	scanner := bufio.NewScanner(strings.NewReader(rawFile))
+	for scanner.Scan() {
+		line := scanner.Text()
+		stats := strings.Split(line, ":")
+		if len(stats) < 2 {
+			continue
+		}
+		name := strings.TrimSpace(stats[0])
+		value := strings.TrimSpace(stats[1])
+		if strings.Contains(name, "Currently Active Slave") {
+			fields["active_slave"] = value
+		}
+		if strings.Contains(name, "MII Status") {
+			fields["status"] = 0
+			if value == "up" {
+				fields["status"] = 1
+			}
+			acc.AddFields("bond", fields, tags)
+			return nil
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return err
+	}
+	return fmt.Errorf("Couldn't find status info for '%s' ", bondName)
+}
+
+func (bond *Bond) gatherSlavePart(bondName string, rawFile string, acc telegraf.Accumulator) error {
+	var slave string
+	var status int
+
+	scanner := bufio.NewScanner(strings.NewReader(rawFile))
+	for scanner.Scan() {
+		line := scanner.Text()
+		stats := strings.Split(line, ":")
+		if len(stats) < 2 {
+			continue
+		}
+		name := strings.TrimSpace(stats[0])
+		value := strings.TrimSpace(stats[1])
+		if strings.Contains(name, "Slave Interface") {
+			slave = value
+		}
+		if strings.Contains(name, "MII Status") {
+			status = 0
+			if value == "up" {
+				status = 1
+			}
+		}
+		if strings.Contains(name, "Link Failure Count") {
+			count, err := strconv.Atoi(value)
+			if err != nil {
+				return err
+			}
+			fields := map[string]interface{}{
+				"status":   status,
+				"failures": count,
+			}
+			tags := map[string]string{
+				"bond":      bondName,
+				"interface": slave,
+			}
+			acc.AddFields("bond_slave", fields, tags)
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// loadPath can be used to read path firstly from config
+// if it is empty then try read from env variable
+func (bond *Bond) loadPath() {
+	if bond.HostProc == "" {
+		bond.HostProc = proc(envProc, defaultHostProc)
+	}
+}
+
+// proc can be used to read file paths from env
+func proc(env, path string) string {
+	// try to read full file path
+	if p := os.Getenv(env); p != "" {
+		return p
+	}
+	// return default path
+	return path
+}
+
+func (bond *Bond) listInterfaces() ([]string, error) {
+	var interfaces []string
+	if len(bond.BondInterfaces) > 0 {
+		interfaces = bond.BondInterfaces
+	} else {
+		paths, err := filepath.Glob(bond.HostProc + "/net/bonding/*")
+		if err != nil {
+			return nil, err
+		}
+		for _, p := range paths {
+			interfaces = append(interfaces, filepath.Base(p))
+		}
+	}
+	return interfaces, nil
+}
+
+func init() {
+	inputs.Add("bond", func() telegraf.Input {
+		return &Bond{}
+	})
+}

plugins/inputs/bond/bond_test.go

+package bond
+
+import (
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+)
+
+var sampleTest802 = `
+Ethernet Channel Bonding Driver: v3.5.0 (November 4, 2008)
+
+Bonding Mode: IEEE 802.3ad Dynamic link aggregation
+Transmit Hash Policy: layer2 (0)
+MII Status: up
+MII Polling Interval (ms): 100
+Up Delay (ms): 0
+Down Delay (ms): 0
+
+802.3ad info
+LACP rate: fast
+Aggregator selection policy (ad_select): stable
+bond bond0 has no active aggregator
+
+Slave Interface: eth1
+MII Status: up
+Link Failure Count: 0
+Permanent HW addr: 00:0c:29:f5:b7:11
+Aggregator ID: N/A
+
+Slave Interface: eth2
+MII Status: up
+Link Failure Count: 3
+Permanent HW addr: 00:0c:29:f5:b7:1b
+Aggregator ID: N/A
+`
+
+var sampleTestAB = `
+Ethernet Channel Bonding Driver: v3.6.0 (September 26, 2009)
+
+Bonding Mode: fault-tolerance (active-backup)
+Primary Slave: eth2 (primary_reselect always)
+Currently Active Slave: eth2
+MII Status: up
+MII Polling Interval (ms): 100
+Up Delay (ms): 0
+Down Delay (ms): 0
+
+Slave Interface: eth3
+MII Status: down
+Speed: 1000 Mbps
+Duplex: full
+Link Failure Count: 2
+Permanent HW addr:
+Slave queue ID: 0
+
+Slave Interface: eth2
+MII Status: up
+Speed: 100 Mbps
+Duplex: full
+Link Failure Count: 0
+Permanent HW addr:
+`
+
+func TestGatherBondInterface(t *testing.T) {
+	var acc testutil.Accumulator
+	bond := &Bond{}
+
+	bond.gatherBondInterface("bond802", sampleTest802, &acc)
+	acc.AssertContainsTaggedFields(t, "bond", map[string]interface{}{"status": 1}, map[string]string{"bond": "bond802"})
+	acc.AssertContainsTaggedFields(t, "bond_slave", map[string]interface{}{"failures": 0, "status": 1}, map[string]string{"bond": "bond802", "interface": "eth1"})
+	acc.AssertContainsTaggedFields(t, "bond_slave", map[string]interface{}{"failures": 3, "status": 1}, map[string]string{"bond": "bond802", "interface": "eth2"})
+
+	bond.gatherBondInterface("bondAB", sampleTestAB, &acc)
+	acc.AssertContainsTaggedFields(t, "bond", map[string]interface{}{"active_slave": "eth2", "status": 1}, map[string]string{"bond": "bondAB"})
+	acc.AssertContainsTaggedFields(t, "bond_slave", map[string]interface{}{"failures": 2, "status": 0}, map[string]string{"bond": "bondAB", "interface": "eth3"})
+	acc.AssertContainsTaggedFields(t, "bond_slave", map[string]interface{}{"failures": 0, "status": 1}, map[string]string{"bond": "bondAB", "interface": "eth2"})
+}