beanstalkd (new plugin)

Makefile

@@ -61,6 +61,7 @@ docker-run:
 	docker run --name mqtt -p "1883:1883" -d ncarlier/mqtt
 	docker run --name riemann -p "5555:5555" -d stealthly/docker-riemann
 	docker run --name nats -p "4222:4222" -d nats
+	docker run --name beanstalkd -p "11300:11300" -d schickling/beanstalkd
 
 # Run docker containers necessary for CircleCI unit tests
 docker-run-circle:
@@ -75,11 +76,12 @@ docker-run-circle:
 	docker run --name mqtt -p "1883:1883" -d ncarlier/mqtt
 	docker run --name riemann -p "5555:5555" -d stealthly/docker-riemann
 	docker run --name nats -p "4222:4222" -d nats
+	docker run --name beanstalkd -p "11300:11300" -d schickling/beanstalkd
 
 # Kill all docker containers, ignore errors
 docker-kill:
-	-docker kill nsq aerospike redis rabbitmq postgres memcached mysql kafka mqtt riemann nats elasticsearch
-	-docker rm nsq aerospike redis rabbitmq postgres memcached mysql kafka mqtt riemann nats elasticsearch
+	docker kill nsq aerospike redis rabbitmq postgres memcached mysql kafka mqtt riemann nats beanstalkd elasticsearch
+	docker rm nsq aerospike redis rabbitmq postgres memcached mysql kafka mqtt riemann nats beanstalkd elasticsearch
 
 # Run full unit tests using docker containers (includes setup and teardown)
 test: vet docker-kill docker-run

README.md

@@ -95,6 +95,78 @@ telegraf --config telegraf.conf -input-filter cpu:mem -output-filter influxdb
 See the [configuration guide](docs/CONFIGURATION.md) for a rundown of the more advanced
 configuration options.
 
+## Supported Input Plugins
+
+Telegraf currently has support for collecting metrics from many sources. For
+more information on each, please look at the directory of the same name in
+`plugins/inputs`.
+
+Currently implemented sources:
+
+* [aws cloudwatch](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/cloudwatch)
+* [aerospike](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/aerospike)
+* [apache](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/apache)
+* [bcache](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/bcache)
+* [beanstalkd](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/beanstalkd)
+* [cassandra](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/cassandra)
+* [ceph](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/ceph)
+* [chrony](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/chrony)
+* [consul](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/consul)
+* [conntrack](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/conntrack)
+* [couchbase](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/couchbase)
+* [couchdb](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/couchdb)
+* [disque](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/disque)
+* [dns query time](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/dns_query)
+* [docker](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/docker)
+* [dovecot](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/dovecot)
+* [elasticsearch](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/elasticsearch)
+* [exec](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/exec) (generic executable plugin, support JSON, influx, graphite and nagios)
+* [filestat](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/filestat)
+* [haproxy](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/haproxy)
+* [hddtemp](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/hddtemp)
+* [http_response](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/http_response)
+* [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)
+* [mailchimp](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mailchimp)
+* [memcached](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/memcached)
+* [mesos](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mesos)
+* [mongodb](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mongodb)
+* [mysql](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mysql)
+* [net_response](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/net_response)
+* [nginx](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nginx)
+* [nsq](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nsq)
+* [nstat](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/nstat)
+* [ntpq](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/ntpq)
+* [phpfpm](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/phpfpm)
+* [phusion passenger](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/passenger)
+* [ping](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/ping)
+* [postgresql](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/postgresql)
+* [postgresql_extensible](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/postgresql_extensible)
+* [powerdns](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/powerdns)
+* [procstat](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/procstat)
+* [prometheus](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/prometheus)
+* [puppetagent](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/puppetagent)
+* [rabbitmq](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/rabbitmq)
+* [raindrops](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/raindrops)
+* [redis](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/redis)
+* [rethinkdb](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/rethinkdb)
+* [riak](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/riak)
+* [sensors](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/sensors)
+* [snmp](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/snmp)
+* [snmp_legacy](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/snmp_legacy)
+* [sql server](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/sqlserver) (microsoft)
+* [twemproxy](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/twemproxy)
+* [varnish](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/varnish)
+* [zfs](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/zfs)
+* [zookeeper](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/zookeeper)
+* [win_perf_counters ](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/win_perf_counters) (windows performance counters)
+* [sysstat](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/sysstat)
+* [system](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/system)
 ## Input Plugins
 
 * [aerospike](./plugins/inputs/aerospike)

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/beanstalkd"
 	_ "github.com/influxdata/telegraf/plugins/inputs/cassandra"
 	_ "github.com/influxdata/telegraf/plugins/inputs/ceph"
 	_ "github.com/influxdata/telegraf/plugins/inputs/cgroup"

plugins/inputs/beanstalkd/README.md

@@ -0,0 +1,78 @@
+# Telegraf Plugin: Beanstalkd
+
+### Configuration:
+
+```
+# # Read metrics from one or many beanstalkd servers
+[[inputs.beanstalkd]]
+#   ## An array of address to gather stats about. Specify an ip on hostname
+#   ## with optional port. ie localhost, 10.0.0.1:11300, etc.
+   servers = ["localhost:11300"]
+```
+
+### Measurements & Fields:
+
+- Measurement
+    - current-jobs-urgent
+    - current-jobs-ready
+    - current-jobs-reserved
+    - current-jobs-delayed
+    - current-jobs-buried
+    - cmd-put
+    - cmd-peek
+    - cmd-peek-ready
+    - cmd-peek-delayed
+    - cmd-peek-buried
+    - cmd-reserve
+    - cmd-reserve-with-timeout
+    - cmd-delete
+    - cmd-release
+    - cmd-use
+    - cmd-watch
+    - cmd-ignore
+    - cmd-bury
+    - cmd-kick
+    - cmd-touch
+    - cmd-stats
+    - cmd-stats-job
+    - cmd-stats-tube
+    - cmd-list-tubes
+    - cmd-list-tube-used
+    - cmd-list-tubes-watched
+    - cmd-pause-tube
+    - job-timeouts
+    - total-jobs
+    - current-tubes
+    - current-connections
+    - current-producers
+    - current-workers
+    - current-waiting
+    - total-connections
+    - uptime
+    - binlog-oldest-index
+    - binlog-current-index
+    - binlog-records-migrated
+    - binlog-records-written
+    - binlog-max-size"
+
+### Example Output:
+
+Using this configuration:
+```
+# # Read metrics from one or many beanstalkd servers
+[[inputs.beanstalkd]]
+#   ## An array of address to gather stats about. Specify an ip on hostname
+#   ## with optional port. ie localhost, 10.0.0.1:11300, etc.
+   servers = ["localhost:11300"]
+```
+
+When run with:
+```
+./telegraf -config telegraf.conf -input-filter beanstalkd -test
+```
+
+It produces:
+```
+* Plugin: inputs.beanstalkd, Collection 1
+> beanstalkd,host=irrlab,server=localhost:11300 binlog-current-index=10i,binlog-max-size=10485760i,binlog-oldest-index=10i,binlog-records-migrated=0i,binlog-records-written=1838i,cmd-bury=0i,cmd-delete=919i,cmd-ignore=1255i,cmd-kick=0i,cmd-list-tube-used=0i,cmd-list-tubes=0i,cmd-list-tubes-watched=0i,cmd-pause-tube=0i,cmd-peek=0i,cmd-peek-buried=0i,cmd-peek-delayed=0i,cmd-peek-ready=0i,cmd-put=919i,cmd-release=0i,cmd-reserve=1255i,cmd-reserve-with-timeout=0i,cmd-stats=341i,cmd-stats-job=0i,cmd-stats-tube=0i,cmd-touch=0i,cmd-use=919i,cmd-watch=1255i,current-connections=1i,current-jobs-buried=0i,current-jobs-delayed=0i,current-jobs-ready=0i,current-jobs-reserved=0i,current-jobs-urgent=0i,current-producers=0i,current-tubes=1i,current-waiting=0i,current-workers=0i,job-timeouts=0i,total-connections=1595i,total-jobs=919i,uptime=15736i 1476809911000000000
+```

plugins/inputs/beanstalkd/beanstalkd.go

@@ -0,0 +1,193 @@
+package beanstalkd
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"net"
+	"strconv"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal/errchan"
+	"github.com/influxdata/telegraf/plugins/inputs"
+)
+
+// Beanstalkd is a Beanstalkd plugin
+type Beanstalkd struct {
+	Servers []string
+}
+
+var sampleConfig = `
+  ## An array of address to gather stats about. Specify an ip on hostname
+  ## with optional port. ie localhost, 10.0.0.1:11300, etc.
+  servers = ["localhost:11300"]
+`
+
+var defaultTimeout = 5 * time.Second
+
+// The list of metrics that should be sent
+var sendMetrics = []string{
+	"current-jobs-urgent",
+	"current-jobs-ready",
+	"current-jobs-reserved",
+	"current-jobs-delayed",
+	"current-jobs-buried",
+	"cmd-put",
+	"cmd-peek",
+	"cmd-peek-ready",
+	"cmd-peek-delayed",
+	"cmd-peek-buried",
+	"cmd-reserve",
+	"cmd-reserve-with-timeout",
+	"cmd-delete",
+	"cmd-release",
+	"cmd-use",
+	"cmd-watch",
+	"cmd-ignore",
+	"cmd-bury",
+	"cmd-kick",
+	"cmd-touch",
+	"cmd-stats",
+	"cmd-stats-job",
+	"cmd-stats-tube",
+	"cmd-list-tubes",
+	"cmd-list-tube-used",
+	"cmd-list-tubes-watched",
+	"cmd-pause-tube",
+	"job-timeouts",
+	"total-jobs",
+	"current-tubes",
+	"current-connections",
+	"current-producers",
+	"current-workers",
+	"current-waiting",
+	"total-connections",
+	"uptime",
+	"binlog-oldest-index",
+	"binlog-current-index",
+	"binlog-records-migrated",
+	"binlog-records-written",
+	"binlog-max-size",
+}
+
+// SampleConfig returns sample configuration message
+func (m *Beanstalkd) SampleConfig() string {
+	return sampleConfig
+}
+
+// Description returns description of Beanstalkd plugin
+func (m *Beanstalkd) Description() string {
+	return "Read metrics from one or many Beanstalkd servers"
+}
+
+// Gather reads stats from all configured servers accumulates stats
+func (m *Beanstalkd) Gather(acc telegraf.Accumulator) error {
+	if len(m.Servers) == 0 {
+		return m.gatherServer(":11300", acc)
+	}
+
+	errChan := errchan.New(len(m.Servers))
+	for _, serverAddress := range m.Servers {
+		errChan.C <- m.gatherServer(serverAddress, acc)
+	}
+
+	return errChan.Error()
+}
+
+func (m *Beanstalkd) gatherServer(
+	address string,
+	acc telegraf.Accumulator,
+) error {
+	var conn net.Conn
+	var err error
+	_, _, err = net.SplitHostPort(address)
+	if err != nil {
+		address = address + ":11300"
+	}
+
+	conn, err = net.DialTimeout("tcp", address, defaultTimeout)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	if conn == nil {
+		return fmt.Errorf("Failed to create net connection")
+	}
+
+	// Extend connection
+	conn.SetDeadline(time.Now().Add(defaultTimeout))
+
+	// Read and write buffer
+	rw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn))
+
+	// Send command
+	if _, err := fmt.Fprint(rw, "stats\r\n"); err != nil {
+		return err
+	}
+	if err := rw.Flush(); err != nil {
+		return err
+	}
+
+	values, err := parseResponse(rw.Reader)
+	if err != nil {
+		return err
+	}
+
+	// Add server address as a tag
+	tags := map[string]string{"server": address}
+
+	// Process values
+	fields := make(map[string]interface{})
+	for _, key := range sendMetrics {
+		if value, ok := values[key]; ok {
+			// Mostly it is the number
+			if iValue, errParse := strconv.ParseInt(value, 10, 64); errParse == nil {
+				fields[key] = iValue
+			} else {
+				fields[key] = value
+			}
+		}
+	}
+	acc.AddFields("beanstalkd", fields, tags)
+	return nil
+}
+
+func parseResponse(r *bufio.Reader) (map[string]string, error) {
+	values := make(map[string]string)
+
+	for {
+		// Read line
+		line, _, errRead := r.ReadLine()
+		if errRead != nil {
+			return values, errRead
+		}
+		// Done
+		if bytes.Equal(line, []byte("")) {
+			break
+		}
+
+		// Read values
+		s := bytes.SplitN(line, []byte(": "), 2)
+
+		if len(s) != 2 {
+			s := bytes.SplitN(line, []byte(" "), 2)
+			if bytes.Equal(s[0], []byte("---")) || bytes.Equal(s[0], []byte("OK")) {
+				continue
+			} else {
+				return values, fmt.Errorf("unexpected line in stats response: %q", line)
+			}
+		}
+
+		// Save values
+		values[string(s[0])] = string(s[1])
+	}
+	return values, nil
+}
+
+func init() {
+	inputs.Add("beanstalkd", func() telegraf.Input {
+		return &Beanstalkd{}
+	})
+}