Add new iptables plugin

The iptables plugin aims at monitoring bytes and packet counters
matching a given set of iptables rules.

Typically the user would set a dedicated monitoring chain into a given
iptables table, and add the rules to monitor to this chain. The plugin
will allow to focus on the counters for this particular table/chain.

CHANGELOG.md

@@ -9,6 +9,7 @@
 - [#1572](https://github.com/influxdata/telegraf/pull/1572): mesos improvements.
 - [#1513](https://github.com/influxdata/telegraf/issues/1513): Add Ceph Cluster Performance Statistics
 - [#1650](https://github.com/influxdata/telegraf/issues/1650): Ability to configure response_timeout in httpjson input.
+- [#1471](https://github.com/influxdata/telegraf/pull/1471): iptables input plugin.
 
 ### Bugfixes
 

README.md

@@ -161,6 +161,7 @@ Currently implemented sources:
 * [httpjson](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/httpjson) (generic JSON-emitting http service plugin)
 * [influxdb](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/influxdb)
 * [ipmi_sensor](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/ipmi_sensor)
+* [iptables](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/iptables)
 * [jolokia](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/jolokia)
 * [leofs](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/leofs)
 * [lustre2](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/lustre2)

etc/telegraf.conf

@@ -910,6 +910,18 @@
 #   ##
 #   servers = ["USERID:PASSW0RD@lan(192.168.1.1)"]
 
+# # Gather packets and bytes throughput from iptables
+# [[inputs.iptables]]
+#   ## iptables require root access on most systems.
+#   ## Setting 'use_sudo' to true will make use of sudo to run iptables.
+#   ## Users must configure sudo to allow telegraf user to run iptables.
+#   ## iptables can be restricted to only use list command  "iptables -nvL"
+#   use_sudo = false
+#   ## define the table to monitor:
+#   table = "filter"
+#   ## Defines the chains to monitor:
+#   chains = [ "INPUT" ]
+
 
 # # Read JMX metrics through Jolokia
 # [[inputs.jolokia]]

plugins/inputs/all/all.go

@@ -27,6 +27,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/httpjson"
 	_ "github.com/influxdata/telegraf/plugins/inputs/influxdb"
 	_ "github.com/influxdata/telegraf/plugins/inputs/ipmi_sensor"
+	_ "github.com/influxdata/telegraf/plugins/inputs/iptables"
 	_ "github.com/influxdata/telegraf/plugins/inputs/jolokia"
 	_ "github.com/influxdata/telegraf/plugins/inputs/kafka_consumer"
 	_ "github.com/influxdata/telegraf/plugins/inputs/leofs"

plugins/inputs/iptables/README.md

@@ -0,0 +1,74 @@
+# Iptables Plugin
+
+The iptables plugin gathers packets and bytes counters for rules within a set of table and chain from the Linux's iptables firewall.
+
+Rules are identified through associated comment. Rules without comment are ignored.
+
+The iptables command requires CAP_NET_ADMIN and CAP_NET_RAW capabilities. You have several options to grant telegraf to run iptables:
+
+* Run telegraf as root. This is strongly discouraged.
+* Configure systemd to run telegraf with CAP_NET_ADMIN and CAP_NET_RAW. This is the simplest and recommended option.
+* Configure sudo to grant telegraf to run iptables. This is the most restrictive option, but require sudo setup.
+
+### Using systemd capabilities
+
+You may run `systemctl edit telegraf.service` and add the following:
+
+```
+[Service]
+CapabilityBoundingSet=CAP_NET_RAW CAP_NET_ADMIN
+AmbientCapabilities=CAP_NET_RAW CAP_NET_ADMIN
+```
+
+Since telegraf will fork a process to run iptables, `AmbientCapabilities` is required to transmit the capabilities bounding set to the forked process.
+
+### Using sudo
+
+You may edit your sudo configuration with the following:
+
+```sudo
+telegraf ALL=(root) NOPASSWD: /usr/bin/iptables -nvL *
+```
+
+### Configuration:
+
+```toml
+  # use sudo to run iptables
+  use_sudo = false
+  # defines the table to monitor:
+  table = "filter"
+  # defines the chains to monitor:
+  chains = [ "INPUT" ]
+```
+
+### Measurements & Fields:
+
+
+- iptables
+    - pkts (integer, count)
+    - bytes (integer, bytes)
+
+### Tags:
+
+- All measurements have the following tags:
+    - table
+    - chain
+    - ruleid
+
+The `ruleid` is the comment associated to the rule.
+
+### Example Output:
+
+```
+$ iptables -nvL INPUT
+Chain INPUT (policy DROP 0 packets, 0 bytes)
+pkts bytes target     prot opt in     out     source               destination
+100   1024   ACCEPT     tcp  --  *      *       192.168.0.0/24       0.0.0.0/0            tcp dpt:22 /* ssh */
+ 42   2048   ACCEPT     tcp  --  *      *       192.168.0.0/24       0.0.0.0/0            tcp dpt:80 /* httpd */
+```
+
+```
+$ ./telegraf -config telegraf.conf -input-filter iptables -test
+iptables,table=filter,chain=INPUT,ruleid=ssh pkts=100i,bytes=1024i 1453831884664956455
+iptables,table=filter,chain=INPUT,ruleid=httpd pkts=42i,bytes=2048i 1453831884664956455
+```

plugins/inputs/iptables/iptables.go

@@ -0,0 +1,128 @@
+// +build linux
+
+package iptables
+
+import (
+	"errors"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+// Iptables is a telegraf plugin to gather packets and bytes throughput from Linux's iptables packet filter.
+type Iptables struct {
+	UseSudo bool
+	Table   string
+	Chains  []string
+	lister  chainLister
+}
+
+// Description returns a short description of the plugin.
+func (ipt *Iptables) Description() string {
+	return "Gather packets and bytes throughput from iptables"
+}
+
+// SampleConfig returns sample configuration options.
+func (ipt *Iptables) SampleConfig() string {
+	return `
+  ## iptables require root access on most systems.
+  ## Setting 'use_sudo' to true will make use of sudo to run iptables.
+  ## Users must configure sudo to allow telegraf user to run iptables with no password.
+  ## iptables can be restricted to only list command  "iptables -nvL"
+  use_sudo = false
+  ## defines the table to monitor:
+  table = "filter"
+  ## defines the chains to monitor:
+  chains = [ "INPUT" ]
+`
+}
+
+// Gather gathers iptables packets and bytes throughput from the configured tables and chains.
+func (ipt *Iptables) Gather(acc telegraf.Accumulator) error {
+	if ipt.Table == "" || len(ipt.Chains) == 0 {
+		return nil
+	}
+	// best effort : we continue through the chains even if an error is encountered,
+	// but we keep track of the last error.
+	var err error
+	for _, chain := range ipt.Chains {
+		data, e := ipt.lister(ipt.Table, chain)
+		if e != nil {
+			err = e
+			continue
+		}
+		e = ipt.parseAndGather(data, acc)
+		if e != nil {
+			err = e
+			continue
+		}
+	}
+	return err
+}
+
+func (ipt *Iptables) chainList(table, chain string) (string, error) {
+	iptablePath, err := exec.LookPath("iptables")
+	if err != nil {
+		return "", err
+	}
+	var args []string
+	name := iptablePath
+	if ipt.UseSudo {
+		name = "sudo"
+		args = append(args, iptablePath)
+	}
+	args = append(args, "-nvL", chain, "-t", table, "-x")
+	c := exec.Command(name, args...)
+	out, err := c.Output()
+	return string(out), err
+}
+
+const measurement = "iptables"
+
+var errParse = errors.New("Cannot parse iptables list information")
+var chainNameRe = regexp.MustCompile(`^Chain\s+(\S+)`)
+var fieldsHeaderRe = regexp.MustCompile(`^\s*pkts\s+bytes\s+`)
+var valuesRe = regexp.MustCompile(`^\s*([0-9]+)\s+([0-9]+)\s+.*?(/\*\s(.*)\s\*/)?$`)
+
+func (ipt *Iptables) parseAndGather(data string, acc telegraf.Accumulator) error {
+	lines := strings.Split(data, "\n")
+	if len(lines) < 3 {
+		return nil
+	}
+	mchain := chainNameRe.FindStringSubmatch(lines[0])
+	if mchain == nil {
+		return errParse
+	}
+	if !fieldsHeaderRe.MatchString(lines[1]) {
+		return errParse
+	}
+	for _, line := range lines[2:] {
+		mv := valuesRe.FindAllStringSubmatch(line, -1)
+		// best effort : if line does not match or rule is not commented forget about it
+		if len(mv) == 0 || len(mv[0]) != 5 || mv[0][4] == "" {
+			continue
+		}
+		tags := map[string]string{"table": ipt.Table, "chain": mchain[1], "ruleid": mv[0][4]}
+		fields := make(map[string]interface{})
+		// since parse error is already catched by the regexp,
+		// we never enter ther error case here => no error check (but still need a test to cover the case)
+		fields["pkts"], _ = strconv.ParseUint(mv[0][1], 10, 64)
+		fields["bytes"], _ = strconv.ParseUint(mv[0][2], 10, 64)
+		acc.AddFields(measurement, fields, tags)
+	}
+	return nil
+}
+
+type chainLister func(table, chain string) (string, error)
+
+func init() {
+	inputs.Add("iptables", func() telegraf.Input {
+		ipt := new(Iptables)
+		ipt.lister = ipt.chainList
+		return ipt
+	})
+}

plugins/inputs/iptables/iptables_nocompile.go

@@ -0,0 +1,3 @@
+// +build !linux
+
+package iptables

plugins/inputs/iptables/iptables_test.go

@@ -0,0 +1,206 @@
+// +build linux
+
+package iptables
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+
+	"github.com/influxdata/telegraf/testutil"
+)
+
+func TestIptables_Gather(t *testing.T) {
+	tests := []struct {
+		table  string
+		chains []string
+		values []string
+		tags   []map[string]string
+		fields [][]map[string]interface{}
+		err    error
+	}{
+		{ // 1 - no configured table => no results
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+		                `},
+		},
+		{ // 2 - no configured chains => no results
+			table: "filter",
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+		                `},
+		},
+		{ // 3 - pkts and bytes are gathered as integers
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0   /* foobar */
+		                `},
+			tags: []map[string]string{map[string]string{"table": "filter", "chain": "INPUT", "ruleid": "foobar"}},
+			fields: [][]map[string]interface{}{
+				{map[string]interface{}{"pkts": uint64(57), "bytes": uint64(4520)}},
+			},
+		},
+		{ // 4 - missing fields header => no results
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)`},
+		},
+		{ // 5 - invalid chain header => error
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{
+				`INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+		                `},
+			err: errParse,
+		},
+		{ // 6 - invalid fields header => error
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+
+		                57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+		                `},
+			err: errParse,
+		},
+		{ // 7 - invalid integer value => best effort, no error
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                K     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+            `},
+		},
+		{ // 8 - Multiple rows, multipe chains => no error
+			table:  "filter",
+			chains: []string{"INPUT", "FORWARD"},
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                100     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0
+		                200     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0  /* foo */
+		                `,
+				`Chain FORWARD (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+		                300     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0  /* bar */
+		                400     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0 
+		                500     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0 /* foobar */
+		                `,
+			},
+			tags: []map[string]string{
+				map[string]string{"table": "filter", "chain": "INPUT", "ruleid": "foo"},
+				map[string]string{"table": "filter", "chain": "FORWARD", "ruleid": "bar"},
+				map[string]string{"table": "filter", "chain": "FORWARD", "ruleid": "foobar"},
+			},
+			fields: [][]map[string]interface{}{
+				{map[string]interface{}{"pkts": uint64(200), "bytes": uint64(4520)}},
+				{map[string]interface{}{"pkts": uint64(300), "bytes": uint64(4520)}},
+				{map[string]interface{}{"pkts": uint64(500), "bytes": uint64(4520)}},
+			},
+		},
+		{ // 9 - comments are used as ruleid if any
+			table:  "filter",
+			chains: []string{"INPUT"},
+			values: []string{
+				`Chain INPUT (policy ACCEPT 58 packets, 5096 bytes)
+		                pkts      bytes target     prot opt in     out     source               destination
+                        57     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0             tcp dpt:22 /* foobar */
+                        100     4520 RETURN     tcp  --  *      *       0.0.0.0/0            0.0.0.0/0    tcp dpt:80
+		                `},
+			tags: []map[string]string{
+				map[string]string{"table": "filter", "chain": "INPUT", "ruleid": "foobar"},
+			},
+			fields: [][]map[string]interface{}{
+				{map[string]interface{}{"pkts": uint64(57), "bytes": uint64(4520)}},
+			},
+		},
+	}
+
+	for i, tt := range tests {
+		i++
+		ipt := &Iptables{
+			Table:  tt.table,
+			Chains: tt.chains,
+			lister: func(table, chain string) (string, error) {
+				if len(tt.values) > 0 {
+					v := tt.values[0]
+					tt.values = tt.values[1:]
+					return v, nil
+				}
+				return "", nil
+			},
+		}
+		acc := new(testutil.Accumulator)
+		err := ipt.Gather(acc)
+		if !reflect.DeepEqual(tt.err, err) {
+			t.Errorf("%d: expected error '%#v' got '%#v'", i, tt.err, err)
+		}
+		if tt.table == "" {
+			n := acc.NFields()
+			if n != 0 {
+				t.Errorf("%d: expected 0 fields if empty table got %d", i, n)
+			}
+			continue
+		}
+		if len(tt.chains) == 0 {
+			n := acc.NFields()
+			if n != 0 {
+				t.Errorf("%d: expected 0 fields if empty chains got %d", i, n)
+			}
+			continue
+		}
+		if len(tt.tags) == 0 {
+			n := acc.NFields()
+			if n != 0 {
+				t.Errorf("%d: expected 0 values got %d", i, n)
+			}
+			continue
+		}
+		n := 0
+		for j, tags := range tt.tags {
+			for k, fields := range tt.fields[j] {
+				if len(acc.Metrics) < n+1 {
+					t.Errorf("%d: expected at least %d values got %d", i, n+1, len(acc.Metrics))
+					break
+				}
+				m := acc.Metrics[n]
+				if !reflect.DeepEqual(m.Measurement, measurement) {
+					t.Errorf("%d %d %d: expected measurement '%#v' got '%#v'\n", i, j, k, measurement, m.Measurement)
+				}
+				if !reflect.DeepEqual(m.Tags, tags) {
+					t.Errorf("%d %d %d: expected tags\n%#v got\n%#v\n", i, j, k, tags, m.Tags)
+				}
+				if !reflect.DeepEqual(m.Fields, fields) {
+					t.Errorf("%d %d %d: expected fields\n%#v got\n%#v\n", i, j, k, fields, m.Fields)
+				}
+				n++
+			}
+		}
+	}
+}
+
+func TestIptables_Gather_listerError(t *testing.T) {
+	errFoo := errors.New("error foobar")
+	ipt := &Iptables{
+		Table:  "nat",
+		Chains: []string{"foo", "bar"},
+		lister: func(table, chain string) (string, error) {
+			return "", errFoo
+		},
+	}
+	acc := new(testutil.Accumulator)
+	err := ipt.Gather(acc)
+	if !reflect.DeepEqual(err, errFoo) {
+		t.Errorf("Expected error %#v got\n%#v\n", errFoo, err)
+	}
+}