Load external .so plugins

support for the Go 1.8 shared object feature of loading external plugins. this support relies on the developer defining a `Plugin` symbol in their .so file that is a telegraf plugin interface. So instead of the plugin developer "Adding" their own plugin to the telegraf registry, telegraf loads the .so, looks up the Plugin symbol, and then adds it if it finds it. The name of the plugin is determined by telegraf, and is namespaced based on the filename and path. see #1717
create telegraf.d directory in tarball
2017-03-09 11:49:16 +00:00 · 2017-03-09 11:41:08 +00:00 · 2017-03-09 11:21:03 +00:00 · 2017-03-09 11:19:03 +00:00 · 2017-03-08 14:08:55 -08:00 · 2017-03-08 16:38:36 +00:00
56 changed files with 1224 additions and 525 deletions
--- a/.github/ISSUE_TEMPLATE.md
+++ b/.github/ISSUE_TEMPLATE.md
@@ -1,7 +1,7 @@
 ## Directions

 GitHub Issues are reserved for actionable bug reports and feature requests.
-General questions should be sent to the [InfluxDB mailing list](https://groups.google.com/forum/#!forum/influxdb).
+General questions should be asked at the [InfluxData Community](https://community.influxdata.com) site.

 Before opening an issue, search for similar bug reports or feature requests on GitHub Issues.
 If no similar issue can be found, fill out either the "Bug Report" or the "Feature Request" section below.
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@

 ### Release Notes

+- Users of the windows `ping` plugin will need to drop or migrate their
+measurements in order to continue using the plugin. The reason for this is that
+the windows plugin was outputting a different type than the linux plugin. This
+made it impossible to use the `ping` plugin for both windows and linux
+machines.
+
 - Ceph: the `ceph_pgmap_state` metric content has been modified to use a unique field `count`, with each state expressed as a `state` tag.

 Telegraf < 1.3:
@@ -27,8 +33,15 @@ The previous riemann output will still be available using
 `outputs.riemann_legacy` if needed, but that will eventually be deprecated.
 It is highly recommended that all users migrate to the new riemann output plugin.

+- Generic [socket_listener](./plugins/inputs/socket_listener) and
+[socket_writer](./plugins/outputs/socket_writer) plugins have been implemented
+for receiving and sending UDP, TCP, unix, & unix-datagram data. These plugins
+will replace udp_listener and tcp_listener, which are still available but will
+be deprecated eventually.
+
 ### Features

+- [#2094](https://github.com/influxdata/telegraf/pull/2094): Add generic socket listener & writer.
 - [#2204](https://github.com/influxdata/telegraf/pull/2204): Extend http_response to support searching for a substring in response. Return 1 if found, else 0.
 - [#2137](https://github.com/influxdata/telegraf/pull/2137): Added userstats to mysql input plugin.
 - [#2179](https://github.com/influxdata/telegraf/pull/2179): Added more InnoDB metric to MySQL plugin.
@@ -41,18 +54,30 @@ It is highly recommended that all users migrate to the new riemann output plugin
 - [#2201](https://github.com/influxdata/telegraf/pull/2201): Add lock option to the IPtables input plugin.
 - [#2244](https://github.com/influxdata/telegraf/pull/2244): Support ipmi_sensor plugin querying local ipmi sensors.
 - [#2339](https://github.com/influxdata/telegraf/pull/2339): Increment gather_errors for all errors emitted by inputs.
+- [#2071](https://github.com/influxdata/telegraf/issues/2071): Use official docker SDK.
+- [#1678](https://github.com/influxdata/telegraf/pull/1678): Add AMQP consumer input plugin

 ### Bugfixes

 - [#2077](https://github.com/influxdata/telegraf/issues/2077): SQL Server Input - Arithmetic overflow error converting numeric to data type int.
 - [#2262](https://github.com/influxdata/telegraf/issues/2262): Flush jitter can inhibit metric collection.
- [#2287](https://github.com/influxdata/telegraf/issues/2287): Kubernetes input: Handle null startTime for stopped pods
- [#1636](https://github.com/influxdata/telegraf/issues/1636): procstat - stop caching PIDs.
 - [#2318](https://github.com/influxdata/telegraf/issues/2318): haproxy input - Add missing fields.
 - [#2287](https://github.com/influxdata/telegraf/issues/2287): Kubernetes input: Handle null startTime for stopped pods.
 - [#2356](https://github.com/influxdata/telegraf/issues/2356): cpu input panic when /proc/stat is empty.
 - [#2341](https://github.com/influxdata/telegraf/issues/2341): telegraf swallowing panics in --test mode.
 - [#2358](https://github.com/influxdata/telegraf/pull/2358): Create pidfile with 644 permissions & defer file deletion.
+- [#2282](https://github.com/influxdata/telegraf/issues/2282): Reloading telegraf freezes prometheus output.
+- [#2390](https://github.com/influxdata/telegraf/issues/2390): Empty tag value causes error on InfluxDB output.
+- [#2380](https://github.com/influxdata/telegraf/issues/2380): buffer_size field value is negative number from "internal" plugin.
+- [#2414](https://github.com/influxdata/telegraf/issues/2414): Missing error handling in the MySQL plugin leads to segmentation violation.
+- [#2462](https://github.com/influxdata/telegraf/pull/2462): Fix type conflict in windows ping plugin.
+- [#2178](https://github.com/influxdata/telegraf/issues/2178): logparser: regexp with lookahead.
+- [#2466](https://github.com/influxdata/telegraf/issues/2466): Telegraf can crash in LoadDirectory on 0600 files.
+- [#2215](https://github.com/influxdata/telegraf/issues/2215): Iptables input: document better that rules without a comment are ignored.
+- [#2483](https://github.com/influxdata/telegraf/pull/2483): Fix win_perf_counters capping values at 100.
+- [#2498](https://github.com/influxdata/telegraf/pull/2498): Exporting Ipmi.Path to be set by config.
+- [#2500](https://github.com/influxdata/telegraf/pull/2500): Remove warning if parse empty content
+- [#2513](https://github.com/influxdata/telegraf/issues/2513): create /etc/telegraf/telegraf.d directory in tarball.

 ## v1.2.1 [2017-02-01]

@@ -112,7 +137,6 @@ plugins, not just statsd.
 - [#1980](https://github.com/influxdata/telegraf/issues/1980): Hide username/password from elasticsearch error log messages.
 - [#2097](https://github.com/influxdata/telegraf/issues/2097): Configurable HTTP timeouts in Jolokia plugin
 - [#2255](https://github.com/influxdata/telegraf/pull/2255): Allow changing jolokia attribute delimiter
- [#2094](https://github.com/influxdata/telegraf/pull/2094): Add generic socket listener & writer.

 ### Bugfixes

--- a/10
+++ b/10
@@ -9,10 +9,7 @@ github.com/couchbase/go-couchbase bfe555a140d53dc1adf390f1a1d4b0fd4ceadb28
 github.com/couchbase/gomemcached 4a25d2f4e1dea9ea7dd76dfd943407abf9b07d29
 github.com/couchbase/goutils 5823a0cbaaa9008406021dc5daf80125ea30bba6
 github.com/davecgh/go-spew 346938d642f2ec3594ed81d874461961cd0faa76
-github.com/docker/distribution fb0bebc4b64e3881cc52a2478d749845ed76d2a8
-github.com/docker/engine-api 4290f40c056686fcaa5c9caf02eac1dde9315adf
-github.com/docker/go-connections 9670439d95da2651d9dfc7acc5d2ed92d3f25ee6
-github.com/docker/go-units 0dadbb0345b35ec7ef35e228dabb8de89a65bf52
+github.com/docker/docker b89aff1afa1f61993ab2ba18fd62d9375a195f5d
 github.com/eapache/go-resiliency b86b1ec0dd4209a588dc1285cdd471e73525c0b3
 github.com/eapache/go-xerial-snappy bb955e01b9346ac19dc29eb16586c90ded99a98c
 github.com/eapache/queue 44cc805cf13205b55f69e14bcb69867d1ae92f98
@@ -25,8 +22,7 @@ github.com/gorilla/mux 392c28fe23e1c45ddba891b0320b3b5df220beea
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/hashicorp/consul 63d2fc68239b996096a1c55a0d4b400ea4c2583f
 github.com/hpcloud/tail 915e5feba042395f5fda4dbe9c0e99aeab3088b3
-github.com/influxdata/config 8ec4638a81500c20be24855812bc8498ebe2dc92
-github.com/influxdata/toml ad49a5c2936f96b8f5943c3fdba47630ccf45a0d
+github.com/influxdata/toml 5d1d907f22ead1cd47adde17ceec5bda9cacaf8f
 github.com/influxdata/wlog 7c63b0a71ef8300adc255344d275e10e5c3a71ec
 github.com/jackc/pgx c8080fc4a1bfa44bf90383ad0fdce2f68b7d313c
 github.com/kardianos/osext c2c54e542fb797ad986b31721e1baedf214ca413
@@ -48,7 +44,7 @@ github.com/prometheus/common dd2f054febf4a6c00f2343686efb775948a8bff4
 github.com/prometheus/procfs 1878d9fbb537119d24b21ca07effd591627cd160
 github.com/rcrowley/go-metrics 1f30fe9094a513ce4c700b9a54458bbb0c96996c
 github.com/samuel/go-zookeeper 1d7be4effb13d2d908342d349d71a284a7542693
-github.com/shirou/gopsutil 77b5d0080adb6f028e457906f1944d9fcca34442
+github.com/shirou/gopsutil d371ba1293cb48fedc6850526ea48b3846c54f2c
 github.com/soniah/gosnmp 5ad50dc75ab389f8a1c9f8a67d3a1cd85f67ed15
 github.com/streadway/amqp 63795daa9a446c920826655f26ba31c81c860fd6
 github.com/stretchr/testify 4d4bfba8f1d1027c4fdbe371823030df51419987
--- a/3
+++ b/3
@@ -15,8 +15,7 @@ windows: prepare-windows build-windows

 # Only run the build (no dependency grabbing)
 build:
-	go install -ldflags \
-		"-X main.version=$(VERSION) -X main.commit=$(COMMIT) -X main.branch=$(BRANCH)" ./...
+	go install -ldflags "-X main.version=$(VERSION) -X main.commit=$(COMMIT) -X main.branch=$(BRANCH)" ./...

 build-windows:
 	GOOS=windows GOARCH=amd64 go build -o telegraf.exe -ldflags \
--- a/README.md
+++ b/README.md
@@ -43,7 +43,7 @@ Ansible role: https://github.com/rossmcdonald/telegraf

 Telegraf manages dependencies via [gdm](https://github.com/sparrc/gdm),
 which gets installed via the Makefile
-if you don't have it already. You also must build with golang version 1.5+.
+if you don't have it already. You also must build with golang version 1.8+.

 1. [Install Go](https://golang.org/doc/install)
 2. [Setup your GOPATH](https://golang.org/doc/code.html#GOPATH)
@@ -97,12 +97,14 @@ configuration options.

 ## Input Plugins

-* [aws cloudwatch](./plugins/inputs/cloudwatch)
 * [aerospike](./plugins/inputs/aerospike)
+* [amqp_consumer](./plugins/inputs/amqp_consumer) (rabbitmq)
 * [apache](./plugins/inputs/apache)
+* [aws cloudwatch](./plugins/inputs/cloudwatch)
 * [bcache](./plugins/inputs/bcache)
 * [cassandra](./plugins/inputs/cassandra)
 * [ceph](./plugins/inputs/ceph)
+* [cgroup](./plugins/inputs/cgroup)
 * [chrony](./plugins/inputs/chrony)
 * [consul](./plugins/inputs/consul)
 * [conntrack](./plugins/inputs/conntrack)
@@ -184,8 +186,8 @@ Telegraf can also collect metrics via the following service plugins:
 * [statsd](./plugins/inputs/statsd)
 * [socket_listener](./plugins/inputs/socket_listener)
 * [tail](./plugins/inputs/tail)
-* [tcp_listener](./plugins/inputs/tcp_listener)
-* [udp_listener](./plugins/inputs/udp_listener)
+* [tcp_listener](./plugins/inputs/socket_listener)
+* [udp_listener](./plugins/inputs/socket_listener)
 * [webhooks](./plugins/inputs/webhooks)
  * [filestack](./plugins/inputs/webhooks/filestack)
  * [github](./plugins/inputs/webhooks/github)
@@ -220,9 +222,11 @@ Telegraf can also collect metrics via the following service plugins:
 * [nsq](./plugins/outputs/nsq)
 * [opentsdb](./plugins/outputs/opentsdb)
 * [prometheus](./plugins/outputs/prometheus_client)
-* [socket_writer](./plugins/outputs/socket_writer)
 * [riemann](./plugins/outputs/riemann)
 * [riemann_legacy](./plugins/outputs/riemann_legacy)
+* [socket_writer](./plugins/outputs/socket_writer)
+* [tcp](./plugins/outputs/socket_writer)
+* [udp](./plugins/outputs/socket_writer)

 ## Contributing

--- a/agent/agent.go
+++ b/agent/agent.go
@@ -191,6 +191,12 @@ func (a *Agent) Test() error {
 	}()

 	for _, input := range a.Config.Inputs {
+		if _, ok := input.Input.(telegraf.ServiceInput); ok {
+			fmt.Printf("\nWARNING: skipping plugin [[%s]]: service inputs not supported in --test mode\n",
+				input.Name())
+			continue
+		}
+
 		acc := NewAccumulator(input, metricC)
 		acc.SetPrecision(a.Config.Agent.Precision.Duration,
 			a.Config.Agent.Interval.Duration)
@@ -209,7 +215,7 @@ func (a *Agent) Test() error {
 		// Special instructions for some inputs. cpu, for example, needs to be
 		// run twice in order to return cpu usage percentages.
 		switch input.Name() {
-		case "cpu", "mongodb", "procstat":
+		case "inputs.cpu", "inputs.mongodb", "inputs.procstat":
 			time.Sleep(500 * time.Millisecond)
 			fmt.Printf("* Plugin: %s, Collection 2\n", input.Name())
 			if err := input.Input.Gather(acc); err != nil {
@@ -392,5 +398,6 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 	}

 	wg.Wait()
+	a.Close()
 	return nil
 }
--- a/circle.yml
+++ b/circle.yml
@@ -5,8 +5,8 @@ machine:
    - sudo service zookeeper stop
    - go version
    - sudo rm -rf /usr/local/go
-    - wget https://storage.googleapis.com/golang/go1.8rc3.linux-amd64.tar.gz
-    - sudo tar -C /usr/local -xzf go1.8rc3.linux-amd64.tar.gz
+    - wget https://storage.googleapis.com/golang/go1.8.linux-amd64.tar.gz
+    - sudo tar -C /usr/local -xzf go1.8.linux-amd64.tar.gz
    - go version

 dependencies:
--- a/cmd/telegraf/telegraf.go
+++ b/cmd/telegraf/telegraf.go
@@ -13,15 +13,20 @@ import (
 	"strings"
 	"syscall"

+	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/agent"
 	"github.com/influxdata/telegraf/internal/config"
 	"github.com/influxdata/telegraf/logger"
-	_ "github.com/influxdata/telegraf/plugins/aggregators/all"
+	"github.com/influxdata/telegraf/plugins/aggregators"
 	"github.com/influxdata/telegraf/plugins/inputs"
-	_ "github.com/influxdata/telegraf/plugins/inputs/all"
 	"github.com/influxdata/telegraf/plugins/outputs"
+	"github.com/influxdata/telegraf/plugins/processors"
+
+	_ "github.com/influxdata/telegraf/plugins/aggregators/all"
+	_ "github.com/influxdata/telegraf/plugins/inputs/all"
 	_ "github.com/influxdata/telegraf/plugins/outputs/all"
 	_ "github.com/influxdata/telegraf/plugins/processors/all"
+
 	"github.com/kardianos/service"
 )

@@ -53,25 +58,29 @@ var fUsage = flag.String("usage", "",
 	"print usage for a plugin, ie, 'telegraf -usage mysql'")
 var fService = flag.String("service", "",
 	"operate on the service")
-var fPlugins = flag.String("plugins", "",
+var fPlugins = flag.String("external-plugins", "",
 	"path to directory containing external plugins")

 // Telegraf version, populated linker.
 //   ie, -ldflags "-X main.version=`git describe --always --tags`"
 var (
-	version string
-	commit  string
-	branch  string
+	version   string
+	commit    string
+	branch    string
+	goversion string
 )

 func init() {
-	// If commit or branch are not set, make that clear.
+	if version == "" {
+		version = "unknown"
+	}
 	if commit == "" {
 		commit = "unknown"
 	}
 	if branch == "" {
 		branch = "unknown"
 	}
+	goversion = runtime.Version() + " " + runtime.GOOS + "/" + runtime.GOARCH
 }

 const usage = `Telegraf, The plugin-driven server agent for collecting and reporting metrics.
@@ -88,6 +97,9 @@ The commands & flags are:
  --config <file>     configuration file to load
  --test              gather metrics once, print them to stdout, and exit
  --config-directory  directory containing additional *.conf files
+  --external-plugins  directory containing *.so files, this directory will be
+                        searched recursively. Any Plugin found will be loaded
+                        and namespaced.
  --input-filter      filter the input plugins to enable, separator is :
  --output-filter     filter the output plugins to enable, separator is :
  --usage             print usage for a plugin, ie, 'telegraf --usage mysql'
@@ -193,7 +205,8 @@ func reloadLoop(
 			}
 		}()

-		log.Printf("I! Starting Telegraf (version %s)\n", version)
+		log.Printf("I! Starting Telegraf (version %s), Go version: %s\n",
+			version, goversion)
 		log.Printf("I! Loaded outputs: %s", strings.Join(c.OutputNames(), " "))
 		log.Printf("I! Loaded inputs: %s", strings.Join(c.InputNames(), " "))
 		log.Printf("I! Tags enabled: %s", c.ListTags())
@@ -253,8 +266,8 @@ func (p *program) Stop(s service.Service) error {

 // loadExternalPlugins loads external plugins from shared libraries (.so, .dll, etc.)
 // in the specified directory.
-func loadExternalPlugins(dir string) error {
-	return filepath.Walk(dir, func(pth string, info os.FileInfo, err error) error {
+func loadExternalPlugins(rootDir string) error {
+	return filepath.Walk(rootDir, func(pth string, info os.FileInfo, err error) error {
 		// Stop if there was an error.
 		if err != nil {
 			return err
@@ -271,30 +284,68 @@ func loadExternalPlugins(dir string) error {
 			return nil
 		}

+		// name will be the path to the plugin file beginning at the root
+		// directory, minus the extension.
+		// ie, if the plugin file is /opt/telegraf-plugins/group1/foo.so, name
+		//     will be "group1/foo"
+		name := strings.TrimPrefix(strings.TrimPrefix(pth, rootDir), string(os.PathSeparator))
+		name = strings.TrimSuffix(name, filepath.Ext(pth))
+		name = "external" + string(os.PathSeparator) + name
+
 		// Load plugin.
-		_, err = plugin.Open(pth)
+		p, err := plugin.Open(pth)
 		if err != nil {
-			return fmt.Errorf("error opening [%s]: %s", pth, err)
+			return fmt.Errorf("error loading [%s]: %s", pth, err)
+		}
+
+		s, err := p.Lookup("Plugin")
+		if err != nil {
+			fmt.Printf("ERROR Could not find 'Plugin' symbol in [%s]\n", pth)
+			return nil
+		}
+
+		switch tplugin := s.(type) {
+		case *telegraf.Input:
+			fmt.Printf("Adding external input plugin: %s\n", name)
+			inputs.Add(name, func() telegraf.Input { return *tplugin })
+		case *telegraf.Output:
+			fmt.Printf("Adding external output plugin: %s\n", name)
+			outputs.Add(name, func() telegraf.Output { return *tplugin })
+		case *telegraf.Processor:
+			fmt.Printf("Adding external processor plugin: %s\n", name)
+			processors.Add(name, func() telegraf.Processor { return *tplugin })
+		case *telegraf.Aggregator:
+			fmt.Printf("Adding external aggregator plugin: %s\n", name)
+			aggregators.Add(name, func() telegraf.Aggregator { return *tplugin })
+		default:
+			fmt.Printf("ERROR: 'Plugin' symbol from [%s] is not a telegraf interface, it has type: %T\n", pth, tplugin)
 		}

 		return nil
 	})
 }

+func printVersion() {
+	fmt.Printf(`Telegraf %s
+  branch:     %s
+  commit:     %s
+  go version: %s
+`, version, branch, commit, goversion)
+}
+
 func main() {
 	flag.Usage = func() { usageExit(0) }
 	flag.Parse()
 	args := flag.Args()
-
 	// Load external plugins, if requested.
 	if *fPlugins != "" {
 		pluginsDir, err := filepath.Abs(*fPlugins)
 		if err != nil {
-			log.Fatal("E! " + err.Error())
+			log.Fatal(err.Error())
 		}
-		log.Printf("I! Loading external plugins from: %s\n", pluginsDir)
+		fmt.Printf("Loading external plugins from: %s\n", pluginsDir)
 		if err := loadExternalPlugins(*fPlugins); err != nil {
-			log.Fatal("E! " + err.Error())
+			log.Fatal(err.Error())
 		}
 	}

@@ -317,7 +368,7 @@ func main() {
 	if len(args) > 0 {
 		switch args[0] {
 		case "version":
-			fmt.Printf("Telegraf v%s (git: %s %s)\n", version, branch, commit)
+			printVersion()
 			return
 		case "config":
 			config.PrintSampleConfig(
@@ -345,7 +396,7 @@ func main() {
 		}
 		return
 	case *fVersion:
-		fmt.Printf("Telegraf v%s (git: %s %s)\n", version, branch, commit)
+		printVersion()
 		return
 	case *fSampleConfig:
 		config.PrintSampleConfig(
--- a/docs/CONFIGURATION.md
+++ b/docs/CONFIGURATION.md
@@ -24,6 +24,16 @@ Environment variables can be used anywhere in the config file, simply prepend
 them with $. For strings the variable must be within quotes (ie, "$STR_VAR"),
 for numbers and booleans they should be plain (ie, $INT_VAR, $BOOL_VAR)

+## Configuration file locations
+
+The location of the configuration file can be set via the `--config` command
+line flag. Telegraf will also pick up all files matching the pattern `*.conf` if
+the `-config-directory` command line flag is used.
+
+On most systems, the default locations are `/etc/telegraf/telegraf.conf` for
+the main configuration file and `/etc/telegraf/telegraf.d` for the directory of
+configuration files.
+
 # Global Tags

 Global tags can be specified in the `[global_tags]` section of the config file
@@ -351,4 +361,4 @@ to the system load metrics due to the `namepass` parameter.

 [[outputs.file]]
  files = ["stdout"]
-```
+```
--- a/etc/telegraf_windows.conf
+++ b/etc/telegraf_windows.conf
@@ -117,7 +117,8 @@
    Instances = ["*"]
    Counters = [
      "% Idle Time",
-      "% Disk Time","% Disk Read Time",
+      "% Disk Time",
+      "% Disk Read Time",
      "% Disk Write Time",
      "Current Disk Queue Length",
      "% Free Space",
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -25,7 +25,6 @@ import (
 	"github.com/influxdata/telegraf/plugins/processors"
 	"github.com/influxdata/telegraf/plugins/serializers"

-	"github.com/influxdata/config"
 	"github.com/influxdata/toml"
 	"github.com/influxdata/toml/ast"
 )
@@ -40,6 +39,14 @@ var (

 	// envVarRe is a regex to find environment variables in the config file
 	envVarRe = regexp.MustCompile(`\$\w+`)
+
+	// addQuoteRe is a regex for finding and adding quotes around / characters
+	// when they are used for distinguishing external plugins.
+	// ie, a ReplaceAll() with this pattern will be used to turn this:
+	//     [[inputs.external/test/example]]
+	//  to
+	//     [[inputs."external/test/example"]]
+	addQuoteRe = regexp.MustCompile(`(\[?\[?inputs|outputs|processors|aggregators)\.(external\/[^.\]]+)`)
 )

 // Config specifies the URL/user/password for the database that telegraf
@@ -506,6 +513,10 @@ func PrintOutputConfig(name string) error {

 func (c *Config) LoadDirectory(path string) error {
 	walkfn := func(thispath string, info os.FileInfo, _ error) error {
+		if info == nil {
+			log.Printf("W! Telegraf is not permitted to read %s", thispath)
+			return nil
+		}
 		if info.IsDir() {
 			return nil
 		}
@@ -566,7 +577,7 @@ func (c *Config) LoadConfig(path string) error {
 			if !ok {
 				return fmt.Errorf("%s: invalid configuration", path)
 			}
-			if err = config.UnmarshalTable(subTable, c.Tags); err != nil {
+			if err = toml.UnmarshalTable(subTable, c.Tags); err != nil {
 				log.Printf("E! Could not parse [global_tags] config\n")
 				return fmt.Errorf("Error parsing %s, %s", path, err)
 			}
@@ -579,7 +590,7 @@ func (c *Config) LoadConfig(path string) error {
 		if !ok {
 			return fmt.Errorf("%s: invalid configuration", path)
 		}
-		if err = config.UnmarshalTable(subTable, c.Agent); err != nil {
+		if err = toml.UnmarshalTable(subTable, c.Agent); err != nil {
 			log.Printf("E! Could not parse [agent] config\n")
 			return fmt.Errorf("Error parsing %s, %s", path, err)
 		}
@@ -701,6 +712,9 @@ func parseFile(fpath string) (*ast.Table, error) {
 		}
 	}

+	// add quotes around external plugin paths.
+	contents = addQuoteRe.ReplaceAll(contents, []byte(`$1."$2"`))
+
 	return toml.Parse(contents)
 }

@@ -716,7 +730,7 @@ func (c *Config) addAggregator(name string, table *ast.Table) error {
 		return err
 	}

-	if err := config.UnmarshalTable(table, aggregator); err != nil {
+	if err := toml.UnmarshalTable(table, aggregator); err != nil {
 		return err
 	}

@@ -736,7 +750,7 @@ func (c *Config) addProcessor(name string, table *ast.Table) error {
 		return err
 	}

-	if err := config.UnmarshalTable(table, processor); err != nil {
+	if err := toml.UnmarshalTable(table, processor); err != nil {
 		return err
 	}

@@ -776,7 +790,7 @@ func (c *Config) addOutput(name string, table *ast.Table) error {
 		return err
 	}

-	if err := config.UnmarshalTable(table, output); err != nil {
+	if err := toml.UnmarshalTable(table, output); err != nil {
 		return err
 	}

@@ -817,7 +831,7 @@ func (c *Config) addInput(name string, table *ast.Table) error {
 		return err
 	}

-	if err := config.UnmarshalTable(table, input); err != nil {
+	if err := toml.UnmarshalTable(table, input); err != nil {
 		return err
 	}

@@ -909,7 +923,7 @@ func buildAggregator(name string, tbl *ast.Table) (*models.AggregatorConfig, err
 	conf.Tags = make(map[string]string)
 	if node, ok := tbl.Fields["tags"]; ok {
 		if subtbl, ok := node.(*ast.Table); ok {
-			if err := config.UnmarshalTable(subtbl, conf.Tags); err != nil {
+			if err := toml.UnmarshalTable(subtbl, conf.Tags); err != nil {
 				log.Printf("Could not parse tags for input %s\n", name)
 			}
 		}
@@ -1146,7 +1160,7 @@ func buildInput(name string, tbl *ast.Table) (*models.InputConfig, error) {
 	cp.Tags = make(map[string]string)
 	if node, ok := tbl.Fields["tags"]; ok {
 		if subtbl, ok := node.(*ast.Table); ok {
-			if err := config.UnmarshalTable(subtbl, cp.Tags); err != nil {
+			if err := toml.UnmarshalTable(subtbl, cp.Tags); err != nil {
 				log.Printf("E! Could not parse tags for input %s\n", name)
 			}
 		}
--- a/internal/models/running_output.go
+++ b/internal/models/running_output.go
@@ -122,9 +122,9 @@ func (ro *RunningOutput) AddMetric(m telegraf.Metric) {
 // Write writes all cached points to this output.
 func (ro *RunningOutput) Write() error {
 	nFails, nMetrics := ro.failMetrics.Len(), ro.metrics.Len()
+	ro.BufferSize.Set(int64(nFails + nMetrics))
 	log.Printf("D! Output [%s] buffer fullness: %d / %d metrics. ",
 		ro.Name, nFails+nMetrics, ro.MetricBufferLimit)
-	ro.BufferSize.Incr(int64(nFails + nMetrics))
 	var err error
 	if !ro.failMetrics.IsEmpty() {
 		// how many batches of failed writes we need to write.
@@ -176,7 +176,6 @@ func (ro *RunningOutput) write(metrics []telegraf.Metric) error {
 		log.Printf("D! Output [%s] wrote batch of %d metrics in %s\n",
 			ro.Name, nMetrics, elapsed)
 		ro.MetricsWritten.Incr(int64(nMetrics))
-		ro.BufferSize.Incr(-int64(nMetrics))
 		ro.WriteTime.Incr(elapsed.Nanoseconds())
 	}
 	return err
--- a/metric/metric.go
+++ b/metric/metric.go
@@ -44,13 +44,18 @@ func New(
 	// pre-allocate exact size of the tags slice
 	taglen := 0
 	for k, v := range tags {
-		// TODO check that length of tag key & value are > 0
+		if len(k) == 0 || len(v) == 0 {
+			continue
+		}
 		taglen += 2 + len(escape(k, "tagkey")) + len(escape(v, "tagval"))
 	}
 	m.tags = make([]byte, taglen)

 	i := 0
 	for k, v := range tags {
+		if len(k) == 0 || len(v) == 0 {
+			continue
+		}
 		m.tags[i] = ','
 		i++
 		i += copy(m.tags[i:], escape(k, "tagkey"))
--- a/metric/metric_test.go
+++ b/metric/metric_test.go
@@ -625,3 +625,26 @@ func TestNewMetricFailNaN(t *testing.T) {
 	_, err := New("cpu", tags, fields, now)
 	assert.NoError(t, err)
 }
+
+func TestEmptyTagValueOrKey(t *testing.T) {
+	now := time.Now()
+
+	tags := map[string]string{
+		"host":     "localhost",
+		"emptytag": "",
+		"":         "valuewithoutkey",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": float64(99),
+	}
+	m, err := New("cpu", tags, fields, now)
+
+	assert.True(t, m.HasTag("host"))
+	assert.False(t, m.HasTag("emptytag"))
+	assert.Equal(t,
+		fmt.Sprintf("cpu,host=localhost usage_idle=99 %d\n", now.UnixNano()),
+		m.String())
+
+	assert.NoError(t, err)
+
+}
--- a/metric/parse.go
+++ b/metric/parse.go
@@ -44,6 +44,9 @@ func Parse(buf []byte) ([]telegraf.Metric, error) {
 }

 func ParseWithDefaultTime(buf []byte, t time.Time) ([]telegraf.Metric, error) {
+	if len(buf) == 0 {
+		return []telegraf.Metric{}, nil
+	}
 	if len(buf) <= 6 {
 		return []telegraf.Metric{}, makeError("buffer too short", buf, 0)
 	}
--- a/plugins/inputs/all/all.go
+++ b/plugins/inputs/all/all.go
@@ -2,6 +2,7 @@ package all

 import (
 	_ "github.com/influxdata/telegraf/plugins/inputs/aerospike"
+	_ "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/cassandra"
--- a/plugins/inputs/amqp_consumer/README.md
+++ b/plugins/inputs/amqp_consumer/README.md
@@ -0,0 +1,47 @@
+# AMQP Consumer Input Plugin
+
+This plugin provides a consumer for use with AMQP 0-9-1, a promenent implementation of this protocol being [RabbitMQ](https://www.rabbitmq.com/).
+
+Metrics are read from a topic exchange using the configured queue and binding_key.
+
+Message payload should be formatted in one of the [Telegraf Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).
+
+For an introduction to AMQP see:
+- https://www.rabbitmq.com/tutorials/amqp-concepts.html
+- https://www.rabbitmq.com/getstarted.html
+
+The following defaults are known to work with RabbitMQ:
+
+```toml
+# AMQP consumer plugin
+[[inputs.amqp_consumer]]
+  ## AMQP url
+  url = "amqp://localhost:5672/influxdb"
+  ## AMQP exchange
+  exchange = "telegraf"
+  ## AMQP queue name
+  queue = "telegraf"
+  ## Binding Key
+  binding_key = "#"
+
+  ## Controls how many messages the server will try to keep on the network
+  ## for consumers before receiving delivery acks.
+  #prefetch_count = 50
+
+  ## Auth method. PLAIN and EXTERNAL are supported.
+  ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as
+  ## described here: https://www.rabbitmq.com/plugins.html
+  # auth_method = "PLAIN"
+  ## Optional SSL Config
+  # ssl_ca = "/etc/telegraf/ca.pem"
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Use SSL but skip chain & host verification
+  # insecure_skip_verify = false
+
+  ## Data format to output.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
+```
--- a/plugins/inputs/amqp_consumer/amqp_consumer.go
+++ b/plugins/inputs/amqp_consumer/amqp_consumer.go
@@ -0,0 +1,280 @@
+package amqp_consumer
+
+import (
+	"fmt"
+	"log"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/streadway/amqp"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	"github.com/influxdata/telegraf/plugins/parsers"
+)
+
+// AMQPConsumer is the top level struct for this plugin
+type AMQPConsumer struct {
+	URL string
+	// AMQP exchange
+	Exchange string
+	// Queue Name
+	Queue string
+	// Binding Key
+	BindingKey string `toml:"binding_key"`
+
+	// Controls how many messages the server will try to keep on the network
+	// for consumers before receiving delivery acks.
+	PrefetchCount int
+
+	// AMQP Auth method
+	AuthMethod string
+	// Path to CA file
+	SSLCA string `toml:"ssl_ca"`
+	// Path to host cert file
+	SSLCert string `toml:"ssl_cert"`
+	// Path to cert key file
+	SSLKey string `toml:"ssl_key"`
+	// Use SSL but skip chain & host verification
+	InsecureSkipVerify bool
+
+	parser parsers.Parser
+	conn   *amqp.Connection
+	wg     *sync.WaitGroup
+}
+
+type externalAuth struct{}
+
+func (a *externalAuth) Mechanism() string {
+	return "EXTERNAL"
+}
+func (a *externalAuth) Response() string {
+	return fmt.Sprintf("\000")
+}
+
+const (
+	DefaultAuthMethod    = "PLAIN"
+	DefaultPrefetchCount = 50
+)
+
+func (a *AMQPConsumer) SampleConfig() string {
+	return `
+  ## AMQP url
+  url = "amqp://localhost:5672/influxdb"
+  ## AMQP exchange
+  exchange = "telegraf"
+  ## AMQP queue name
+  queue = "telegraf"
+  ## Binding Key
+  binding_key = "#"
+
+  ## Maximum number of messages server should give to the worker.
+  prefetch_count = 50
+
+  ## Auth method. PLAIN and EXTERNAL are supported
+  ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as
+  ## described here: https://www.rabbitmq.com/plugins.html
+  # auth_method = "PLAIN"
+
+  ## Optional SSL Config
+  # ssl_ca = "/etc/telegraf/ca.pem"
+  # ssl_cert = "/etc/telegraf/cert.pem"
+  # ssl_key = "/etc/telegraf/key.pem"
+  ## Use SSL but skip chain & host verification
+  # insecure_skip_verify = false
+
+  ## Data format to output.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
+`
+}
+
+func (a *AMQPConsumer) Description() string {
+	return "AMQP consumer plugin"
+}
+
+func (a *AMQPConsumer) SetParser(parser parsers.Parser) {
+	a.parser = parser
+}
+
+// All gathering is done in the Start function
+func (a *AMQPConsumer) Gather(_ telegraf.Accumulator) error {
+	return nil
+}
+
+func (a *AMQPConsumer) createConfig() (*amqp.Config, error) {
+	// make new tls config
+	tls, err := internal.GetTLSConfig(
+		a.SSLCert, a.SSLKey, a.SSLCA, a.InsecureSkipVerify)
+	if err != nil {
+		return nil, err
+	}
+
+	// parse auth method
+	var sasl []amqp.Authentication // nil by default
+
+	if strings.ToUpper(a.AuthMethod) == "EXTERNAL" {
+		sasl = []amqp.Authentication{&externalAuth{}}
+	}
+
+	config := amqp.Config{
+		TLSClientConfig: tls,
+		SASL:            sasl, // if nil, it will be PLAIN
+	}
+	return &config, nil
+}
+
+// Start satisfies the telegraf.ServiceInput interface
+func (a *AMQPConsumer) Start(acc telegraf.Accumulator) error {
+	amqpConf, err := a.createConfig()
+	if err != nil {
+		return err
+	}
+
+	msgs, err := a.connect(amqpConf)
+	if err != nil {
+		return err
+	}
+
+	a.wg = &sync.WaitGroup{}
+	a.wg.Add(1)
+	go a.process(msgs, acc)
+
+	go func() {
+		err := <-a.conn.NotifyClose(make(chan *amqp.Error))
+		if err == nil {
+			return
+		}
+
+		log.Printf("I! AMQP consumer connection closed: %s; trying to reconnect", err)
+		for {
+			msgs, err := a.connect(amqpConf)
+			if err != nil {
+				log.Printf("E! AMQP connection failed: %s", err)
+				time.Sleep(10 * time.Second)
+				continue
+			}
+
+			a.wg.Add(1)
+			go a.process(msgs, acc)
+			break
+		}
+	}()
+
+	return nil
+}
+
+func (a *AMQPConsumer) connect(amqpConf *amqp.Config) (<-chan amqp.Delivery, error) {
+	conn, err := amqp.DialConfig(a.URL, *amqpConf)
+	if err != nil {
+		return nil, err
+	}
+	a.conn = conn
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, fmt.Errorf("Failed to open a channel: %s", err)
+	}
+
+	err = ch.ExchangeDeclare(
+		a.Exchange, // name
+		"topic",    // type
+		true,       // durable
+		false,      // auto-deleted
+		false,      // internal
+		false,      // no-wait
+		nil,        // arguments
+	)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to declare an exchange: %s", err)
+	}
+
+	q, err := ch.QueueDeclare(
+		a.Queue, // queue
+		true,    // durable
+		false,   // delete when unused
+		false,   // exclusive
+		false,   // no-wait
+		nil,     // arguments
+	)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to declare a queue: %s", err)
+	}
+
+	err = ch.QueueBind(
+		q.Name,       // queue
+		a.BindingKey, // binding-key
+		a.Exchange,   // exchange
+		false,
+		nil,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to bind a queue: %s", err)
+	}
+
+	err = ch.Qos(
+		a.PrefetchCount,
+		0,     // prefetch-size
+		false, // global
+	)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to set QoS: %s", err)
+	}
+
+	msgs, err := ch.Consume(
+		q.Name, // queue
+		"",     // consumer
+		false,  // auto-ack
+		false,  // exclusive
+		false,  // no-local
+		false,  // no-wait
+		nil,    // arguments
+	)
+	if err != nil {
+		return nil, fmt.Errorf("Failed establishing connection to queue: %s", err)
+	}
+
+	log.Println("I! Started AMQP consumer")
+	return msgs, err
+}
+
+// Read messages from queue and add them to the Accumulator
+func (a *AMQPConsumer) process(msgs <-chan amqp.Delivery, acc telegraf.Accumulator) {
+	defer a.wg.Done()
+	for d := range msgs {
+		metrics, err := a.parser.Parse(d.Body)
+		if err != nil {
+			log.Printf("E! %v: error parsing metric - %v", err, string(d.Body))
+		} else {
+			for _, m := range metrics {
+				acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time())
+			}
+		}
+
+		d.Ack(false)
+	}
+	log.Printf("I! AMQP consumer queue closed")
+}
+
+func (a *AMQPConsumer) Stop() {
+	err := a.conn.Close()
+	if err != nil && err != amqp.ErrClosed {
+		log.Printf("E! Error closing AMQP connection: %s", err)
+		return
+	}
+	a.wg.Wait()
+	log.Println("I! Stopped AMQP service")
+}
+
+func init() {
+	inputs.Add("amqp_consumer", func() telegraf.Input {
+		return &AMQPConsumer{
+			AuthMethod:    DefaultAuthMethod,
+			PrefetchCount: DefaultPrefetchCount,
+		}
+	})
+}
--- a/plugins/inputs/apache/README.md
+++ b/plugins/inputs/apache/README.md
@@ -4,7 +4,7 @@
 - **urls** []string: List of apache-status URLs to collect from. Default is "http://localhost/server-status?auto".
 - **username** string: Username for HTTP basic authentication
 - **password** string: Password for HTTP basic authentication
- **timeout** duration: time that the HTTP connection will remain waiting for response. Defalt 4 seconds ("4s")
+- **timeout** duration: time that the HTTP connection will remain waiting for response. Default 4 seconds ("4s")

 ##### Optional SSL Config

--- a/plugins/inputs/docker/README.md
+++ b/plugins/inputs/docker/README.md
@@ -16,12 +16,20 @@ for the stat structure can be found
 ```
 # Read metrics about docker containers
 [[inputs.docker]]
-  # Docker Endpoint
-  #   To use TCP, set endpoint = "tcp://[ip]:[port]"
-  #   To use environment variables (ie, docker-machine), set endpoint = "ENV"
+  ## Docker Endpoint
+  ##   To use TCP, set endpoint = "tcp://[ip]:[port]"
+  ##   To use environment variables (ie, docker-machine), set endpoint = "ENV"
  endpoint = "unix:///var/run/docker.sock"
-  # Only collect metrics for these containers, collect all if empty
+  ## Only collect metrics for these containers, collect all if empty
  container_names = []
+  ## Timeout for docker list, info, and stats commands
+  timeout = "5s"
+
+  ## Whether to report for each container per-device blkio (8:0, 8:1...) and
+  ## network (eth0, eth1, ...) stats or not
+  perdevice = true
+  ## Whether to report for each container total blkio and network stats or not
+  total = false
 ```

 ### Measurements & Fields:
--- a/plugins/inputs/docker/docker.go
+++ b/plugins/inputs/docker/docker.go
@@ -1,6 +1,7 @@
-package system
+package docker

 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -11,10 +12,9 @@ import (
 	"sync"
 	"time"

-	"golang.org/x/net/context"
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/client"

-	"github.com/docker/engine-api/client"
-	"github.com/docker/engine-api/types"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/inputs"
@@ -28,15 +28,46 @@ type Docker struct {
 	PerDevice      bool `toml:"perdevice"`
 	Total          bool `toml:"total"`

-	client      DockerClient
+	client      *client.Client
 	engine_host string
+
+	testing bool
 }

-// DockerClient interface, useful for testing
-type DockerClient interface {
-	Info(ctx context.Context) (types.Info, error)
-	ContainerList(ctx context.Context, options types.ContainerListOptions) ([]types.Container, error)
-	ContainerStats(ctx context.Context, containerID string, stream bool) (io.ReadCloser, error)
+// infoWrapper wraps client.Client.List for testing.
+func infoWrapper(c *client.Client, ctx context.Context) (types.Info, error) {
+	if c != nil {
+		return c.Info(ctx)
+	}
+	fc := FakeDockerClient{}
+	return fc.Info(ctx)
+}
+
+// listWrapper wraps client.Client.ContainerList for testing.
+func listWrapper(
+	c *client.Client,
+	ctx context.Context,
+	options types.ContainerListOptions,
+) ([]types.Container, error) {
+	if c != nil {
+		return c.ContainerList(ctx, options)
+	}
+	fc := FakeDockerClient{}
+	return fc.ContainerList(ctx, options)
+}
+
+// statsWrapper wraps client.Client.ContainerStats for testing.
+func statsWrapper(
+	c *client.Client,
+	ctx context.Context,
+	containerID string,
+	stream bool,
+) (types.ContainerStats, error) {
+	if c != nil {
+		return c.ContainerStats(ctx, containerID, stream)
+	}
+	fc := FakeDockerClient{}
+	return fc.ContainerStats(ctx, containerID, stream)
 }

 // KB, MB, GB, TB, PB...human friendly
@@ -80,7 +111,7 @@ func (d *Docker) SampleConfig() string { return sampleConfig }

 // Gather starts stats collection
 func (d *Docker) Gather(acc telegraf.Accumulator) error {
-	if d.client == nil {
+	if d.client == nil && !d.testing {
 		var c *client.Client
 		var err error
 		defaultHeaders := map[string]string{"User-Agent": "engine-api-cli-1.0"}
@@ -113,7 +144,7 @@ func (d *Docker) Gather(acc telegraf.Accumulator) error {
 	opts := types.ContainerListOptions{}
 	ctx, cancel := context.WithTimeout(context.Background(), d.Timeout.Duration)
 	defer cancel()
-	containers, err := d.client.ContainerList(ctx, opts)
+	containers, err := listWrapper(d.client, ctx, opts)
 	if err != nil {
 		return err
 	}
@@ -144,7 +175,7 @@ func (d *Docker) gatherInfo(acc telegraf.Accumulator) error {
 	// Get info from docker daemon
 	ctx, cancel := context.WithTimeout(context.Background(), d.Timeout.Duration)
 	defer cancel()
-	info, err := d.client.Info(ctx)
+	info, err := infoWrapper(d.client, ctx)
 	if err != nil {
 		return err
 	}
@@ -247,12 +278,12 @@ func (d *Docker) gatherContainer(

 	ctx, cancel := context.WithTimeout(context.Background(), d.Timeout.Duration)
 	defer cancel()
-	r, err := d.client.ContainerStats(ctx, container.ID, false)
+	r, err := statsWrapper(d.client, ctx, container.ID, false)
 	if err != nil {
 		return fmt.Errorf("Error getting docker stats: %s", err.Error())
 	}
-	defer r.Close()
-	dec := json.NewDecoder(r)
+	defer r.Body.Close()
+	dec := json.NewDecoder(r.Body)
 	if err = dec.Decode(&v); err != nil {
 		if err == io.EOF {
 			return nil
--- a/plugins/inputs/docker/docker_test.go
+++ b/plugins/inputs/docker/docker_test.go
@@ -1,18 +1,12 @@
-package system
+package docker

 import (
-	"io"
-	"io/ioutil"
-	"strings"
 	"testing"
 	"time"

-	"golang.org/x/net/context"
-
-	"github.com/docker/engine-api/types"
-	"github.com/docker/engine-api/types/registry"
 	"github.com/influxdata/telegraf/testutil"

+	"github.com/docker/docker/api/types"
 	"github.com/stretchr/testify/require"
 )

@@ -250,146 +244,14 @@ func testStats() *types.StatsJSON {
 	return stats
 }

-type FakeDockerClient struct {
-}
-
-func (d FakeDockerClient) Info(ctx context.Context) (types.Info, error) {
-	env := types.Info{
-		Containers:         108,
-		ContainersRunning:  98,
-		ContainersStopped:  6,
-		ContainersPaused:   3,
-		OomKillDisable:     false,
-		SystemTime:         "2016-02-24T00:55:09.15073105-05:00",
-		NEventsListener:    0,
-		ID:                 "5WQQ:TFWR:FDNG:OKQ3:37Y4:FJWG:QIKK:623T:R3ME:QTKB:A7F7:OLHD",
-		Debug:              false,
-		LoggingDriver:      "json-file",
-		KernelVersion:      "4.3.0-1-amd64",
-		IndexServerAddress: "https://index.docker.io/v1/",
-		MemTotal:           3840757760,
-		Images:             199,
-		CPUCfsQuota:        true,
-		Name:               "absol",
-		SwapLimit:          false,
-		IPv4Forwarding:     true,
-		ExperimentalBuild:  false,
-		CPUCfsPeriod:       true,
-		RegistryConfig: &registry.ServiceConfig{
-			IndexConfigs: map[string]*registry.IndexInfo{
-				"docker.io": {
-					Name:     "docker.io",
-					Mirrors:  []string{},
-					Official: true,
-					Secure:   true,
-				},
-			}, InsecureRegistryCIDRs: []*registry.NetIPNet{{IP: []byte{127, 0, 0, 0}, Mask: []byte{255, 0, 0, 0}}}, Mirrors: []string{}},
-		OperatingSystem:   "Linux Mint LMDE (containerized)",
-		BridgeNfIptables:  true,
-		HTTPSProxy:        "",
-		Labels:            []string{},
-		MemoryLimit:       false,
-		DriverStatus:      [][2]string{{"Pool Name", "docker-8:1-1182287-pool"}, {"Pool Blocksize", "65.54 kB"}, {"Backing Filesystem", "extfs"}, {"Data file", "/dev/loop0"}, {"Metadata file", "/dev/loop1"}, {"Data Space Used", "17.3 GB"}, {"Data Space Total", "107.4 GB"}, {"Data Space Available", "36.53 GB"}, {"Metadata Space Used", "20.97 MB"}, {"Metadata Space Total", "2.147 GB"}, {"Metadata Space Available", "2.127 GB"}, {"Udev Sync Supported", "true"}, {"Deferred Removal Enabled", "false"}, {"Data loop file", "/var/lib/docker/devicemapper/devicemapper/data"}, {"Metadata loop file", "/var/lib/docker/devicemapper/devicemapper/metadata"}, {"Library Version", "1.02.115 (2016-01-25)"}},
-		NFd:               19,
-		HTTPProxy:         "",
-		Driver:            "devicemapper",
-		NGoroutines:       39,
-		NCPU:              4,
-		DockerRootDir:     "/var/lib/docker",
-		NoProxy:           "",
-		BridgeNfIP6tables: true,
-	}
-	return env, nil
-}
-
-func (d FakeDockerClient) ContainerList(octx context.Context, options types.ContainerListOptions) ([]types.Container, error) {
-	container1 := types.Container{
-		ID:      "e2173b9478a6ae55e237d4d74f8bbb753f0817192b5081334dc78476296b7dfb",
-		Names:   []string{"/etcd"},
-		Image:   "quay.io/coreos/etcd:v2.2.2",
-		Command: "/etcd -name etcd0 -advertise-client-urls http://localhost:2379 -listen-client-urls http://0.0.0.0:2379",
-		Created: 1455941930,
-		Status:  "Up 4 hours",
-		Ports: []types.Port{
-			types.Port{
-				PrivatePort: 7001,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 4001,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 2380,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 2379,
-				PublicPort:  2379,
-				Type:        "tcp",
-				IP:          "0.0.0.0",
-			},
-		},
-		SizeRw:     0,
-		SizeRootFs: 0,
-	}
-	container2 := types.Container{
-		ID:      "b7dfbb9478a6ae55e237d4d74f8bbb753f0817192b5081334dc78476296e2173",
-		Names:   []string{"/etcd2"},
-		Image:   "quay.io:4443/coreos/etcd:v2.2.2",
-		Command: "/etcd -name etcd2 -advertise-client-urls http://localhost:2379 -listen-client-urls http://0.0.0.0:2379",
-		Created: 1455941933,
-		Status:  "Up 4 hours",
-		Ports: []types.Port{
-			types.Port{
-				PrivatePort: 7002,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 4002,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 2381,
-				PublicPort:  0,
-				Type:        "tcp",
-			},
-			types.Port{
-				PrivatePort: 2382,
-				PublicPort:  2382,
-				Type:        "tcp",
-				IP:          "0.0.0.0",
-			},
-		},
-		SizeRw:     0,
-		SizeRootFs: 0,
-	}
-
-	containers := []types.Container{container1, container2}
-	return containers, nil
-
-	//#{e6a96c84ca91a5258b7cb752579fb68826b68b49ff957487695cd4d13c343b44 titilambert/snmpsim /bin/sh -c 'snmpsimd --agent-udpv4-endpoint=0.0.0.0:31161 --process-user=root --process-group=user' 1455724831 Up 4 hours [{31161 31161 udp 0.0.0.0}] 0 0 [/snmp] map[]}]2016/02/24 01:05:01 Gathered metrics, (3s interval), from 1 inputs in 1.233836656s
-}
-
-func (d FakeDockerClient) ContainerStats(ctx context.Context, containerID string, stream bool) (io.ReadCloser, error) {
-	var stat io.ReadCloser
-	jsonStat := `{"read":"2016-02-24T11:42:27.472459608-05:00","memory_stats":{"stats":{},"limit":18935443456},"blkio_stats":{"io_service_bytes_recursive":[{"major":252,"minor":1,"op":"Read","value":753664},{"major":252,"minor":1,"op":"Write"},{"major":252,"minor":1,"op":"Sync"},{"major":252,"minor":1,"op":"Async","value":753664},{"major":252,"minor":1,"op":"Total","value":753664}],"io_serviced_recursive":[{"major":252,"minor":1,"op":"Read","value":26},{"major":252,"minor":1,"op":"Write"},{"major":252,"minor":1,"op":"Sync"},{"major":252,"minor":1,"op":"Async","value":26},{"major":252,"minor":1,"op":"Total","value":26}]},"cpu_stats":{"cpu_usage":{"percpu_usage":[17871,4959158,1646137,1231652,11829401,244656,369972,0],"usage_in_usermode":10000000,"total_usage":20298847},"system_cpu_usage":24052607520000000,"throttling_data":{}},"precpu_stats":{"cpu_usage":{"percpu_usage":[17871,4959158,1646137,1231652,11829401,244656,369972,0],"usage_in_usermode":10000000,"total_usage":20298847},"system_cpu_usage":24052599550000000,"throttling_data":{}}}`
-	stat = ioutil.NopCloser(strings.NewReader(jsonStat))
-	return stat, nil
-}
-
 func TestDockerGatherInfo(t *testing.T) {
 	var acc testutil.Accumulator
-	client := FakeDockerClient{}
-	d := Docker{client: client}
+	d := Docker{
+		client:  nil,
+		testing: true,
+	}

 	err := d.Gather(&acc)
-
 	require.NoError(t, err)

 	acc.AssertContainsTaggedFields(t,
--- a/plugins/inputs/docker/fake_client.go
+++ b/plugins/inputs/docker/fake_client.go
@@ -0,0 +1,143 @@
+package docker
+
+import (
+	"context"
+	"io/ioutil"
+	"strings"
+
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/registry"
+)
+
+type FakeDockerClient struct {
+}
+
+func (d FakeDockerClient) Info(ctx context.Context) (types.Info, error) {
+	env := types.Info{
+		Containers:         108,
+		ContainersRunning:  98,
+		ContainersStopped:  6,
+		ContainersPaused:   3,
+		OomKillDisable:     false,
+		SystemTime:         "2016-02-24T00:55:09.15073105-05:00",
+		NEventsListener:    0,
+		ID:                 "5WQQ:TFWR:FDNG:OKQ3:37Y4:FJWG:QIKK:623T:R3ME:QTKB:A7F7:OLHD",
+		Debug:              false,
+		LoggingDriver:      "json-file",
+		KernelVersion:      "4.3.0-1-amd64",
+		IndexServerAddress: "https://index.docker.io/v1/",
+		MemTotal:           3840757760,
+		Images:             199,
+		CPUCfsQuota:        true,
+		Name:               "absol",
+		SwapLimit:          false,
+		IPv4Forwarding:     true,
+		ExperimentalBuild:  false,
+		CPUCfsPeriod:       true,
+		RegistryConfig: &registry.ServiceConfig{
+			IndexConfigs: map[string]*registry.IndexInfo{
+				"docker.io": {
+					Name:     "docker.io",
+					Mirrors:  []string{},
+					Official: true,
+					Secure:   true,
+				},
+			}, InsecureRegistryCIDRs: []*registry.NetIPNet{{IP: []byte{127, 0, 0, 0}, Mask: []byte{255, 0, 0, 0}}}, Mirrors: []string{}},
+		OperatingSystem:   "Linux Mint LMDE (containerized)",
+		BridgeNfIptables:  true,
+		HTTPSProxy:        "",
+		Labels:            []string{},
+		MemoryLimit:       false,
+		DriverStatus:      [][2]string{{"Pool Name", "docker-8:1-1182287-pool"}, {"Pool Blocksize", "65.54 kB"}, {"Backing Filesystem", "extfs"}, {"Data file", "/dev/loop0"}, {"Metadata file", "/dev/loop1"}, {"Data Space Used", "17.3 GB"}, {"Data Space Total", "107.4 GB"}, {"Data Space Available", "36.53 GB"}, {"Metadata Space Used", "20.97 MB"}, {"Metadata Space Total", "2.147 GB"}, {"Metadata Space Available", "2.127 GB"}, {"Udev Sync Supported", "true"}, {"Deferred Removal Enabled", "false"}, {"Data loop file", "/var/lib/docker/devicemapper/devicemapper/data"}, {"Metadata loop file", "/var/lib/docker/devicemapper/devicemapper/metadata"}, {"Library Version", "1.02.115 (2016-01-25)"}},
+		NFd:               19,
+		HTTPProxy:         "",
+		Driver:            "devicemapper",
+		NGoroutines:       39,
+		NCPU:              4,
+		DockerRootDir:     "/var/lib/docker",
+		NoProxy:           "",
+		BridgeNfIP6tables: true,
+	}
+	return env, nil
+}
+
+func (d FakeDockerClient) ContainerList(octx context.Context, options types.ContainerListOptions) ([]types.Container, error) {
+	container1 := types.Container{
+		ID:      "e2173b9478a6ae55e237d4d74f8bbb753f0817192b5081334dc78476296b7dfb",
+		Names:   []string{"/etcd"},
+		Image:   "quay.io/coreos/etcd:v2.2.2",
+		Command: "/etcd -name etcd0 -advertise-client-urls http://localhost:2379 -listen-client-urls http://0.0.0.0:2379",
+		Created: 1455941930,
+		Status:  "Up 4 hours",
+		Ports: []types.Port{
+			types.Port{
+				PrivatePort: 7001,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 4001,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 2380,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 2379,
+				PublicPort:  2379,
+				Type:        "tcp",
+				IP:          "0.0.0.0",
+			},
+		},
+		SizeRw:     0,
+		SizeRootFs: 0,
+	}
+	container2 := types.Container{
+		ID:      "b7dfbb9478a6ae55e237d4d74f8bbb753f0817192b5081334dc78476296e2173",
+		Names:   []string{"/etcd2"},
+		Image:   "quay.io:4443/coreos/etcd:v2.2.2",
+		Command: "/etcd -name etcd2 -advertise-client-urls http://localhost:2379 -listen-client-urls http://0.0.0.0:2379",
+		Created: 1455941933,
+		Status:  "Up 4 hours",
+		Ports: []types.Port{
+			types.Port{
+				PrivatePort: 7002,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 4002,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 2381,
+				PublicPort:  0,
+				Type:        "tcp",
+			},
+			types.Port{
+				PrivatePort: 2382,
+				PublicPort:  2382,
+				Type:        "tcp",
+				IP:          "0.0.0.0",
+			},
+		},
+		SizeRw:     0,
+		SizeRootFs: 0,
+	}
+
+	containers := []types.Container{container1, container2}
+	return containers, nil
+
+	//#{e6a96c84ca91a5258b7cb752579fb68826b68b49ff957487695cd4d13c343b44 titilambert/snmpsim /bin/sh -c 'snmpsimd --agent-udpv4-endpoint=0.0.0.0:31161 --process-user=root --process-group=user' 1455724831 Up 4 hours [{31161 31161 udp 0.0.0.0}] 0 0 [/snmp] map[]}]2016/02/24 01:05:01 Gathered metrics, (3s interval), from 1 inputs in 1.233836656s
+}
+
+func (d FakeDockerClient) ContainerStats(ctx context.Context, containerID string, stream bool) (types.ContainerStats, error) {
+	var stat types.ContainerStats
+	jsonStat := `{"read":"2016-02-24T11:42:27.472459608-05:00","memory_stats":{"stats":{},"limit":18935443456},"blkio_stats":{"io_service_bytes_recursive":[{"major":252,"minor":1,"op":"Read","value":753664},{"major":252,"minor":1,"op":"Write"},{"major":252,"minor":1,"op":"Sync"},{"major":252,"minor":1,"op":"Async","value":753664},{"major":252,"minor":1,"op":"Total","value":753664}],"io_serviced_recursive":[{"major":252,"minor":1,"op":"Read","value":26},{"major":252,"minor":1,"op":"Write"},{"major":252,"minor":1,"op":"Sync"},{"major":252,"minor":1,"op":"Async","value":26},{"major":252,"minor":1,"op":"Total","value":26}]},"cpu_stats":{"cpu_usage":{"percpu_usage":[17871,4959158,1646137,1231652,11829401,244656,369972,0],"usage_in_usermode":10000000,"total_usage":20298847},"system_cpu_usage":24052607520000000,"throttling_data":{}},"precpu_stats":{"cpu_usage":{"percpu_usage":[17871,4959158,1646137,1231652,11829401,244656,369972,0],"usage_in_usermode":10000000,"total_usage":20298847},"system_cpu_usage":24052599550000000,"throttling_data":{}}}`
+	stat.Body = ioutil.NopCloser(strings.NewReader(jsonStat))
+	return stat, nil
+}
--- a/plugins/inputs/exec/exec_test.go
+++ b/plugins/inputs/exec/exec_test.go
@@ -37,6 +37,8 @@ const malformedJson = `
 `

 const lineProtocol = "cpu,host=foo,datacenter=us-east usage_idle=99,usage_busy=1\n"
+const lineProtocolEmpty = ""
+const lineProtocolShort = "ab"

 const lineProtocolMulti = `
 cpu,cpu=cpu0,host=foo,datacenter=us-east usage_idle=99,usage_busy=1
@@ -167,6 +169,33 @@ func TestLineProtocolParse(t *testing.T) {
 	acc.AssertContainsTaggedFields(t, "cpu", fields, tags)
 }

+func TestLineProtocolEmptyParse(t *testing.T) {
+	parser, _ := parsers.NewInfluxParser()
+	e := &Exec{
+		runner:   newRunnerMock([]byte(lineProtocolEmpty), nil),
+		Commands: []string{"line-protocol"},
+		parser:   parser,
+	}
+
+	var acc testutil.Accumulator
+	err := e.Gather(&acc)
+	require.NoError(t, err)
+}
+
+func TestLineProtocolShortParse(t *testing.T) {
+	parser, _ := parsers.NewInfluxParser()
+	e := &Exec{
+		runner:   newRunnerMock([]byte(lineProtocolShort), nil),
+		Commands: []string{"line-protocol"},
+		parser:   parser,
+	}
+
+	var acc testutil.Accumulator
+	err := e.Gather(&acc)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "buffer too short", "A buffer too short error was expected")
+}
+
 func TestLineProtocolParseMultiple(t *testing.T) {
 	parser, _ := parsers.NewInfluxParser()
 	e := &Exec{
--- a/plugins/inputs/ipmi_sensor/ipmi.go
+++ b/plugins/inputs/ipmi_sensor/ipmi.go
@@ -17,7 +17,7 @@ var (
 )

 type Ipmi struct {
-	path    string
+	Path    string
 	Servers []string
 }

@@ -44,7 +44,7 @@ func (m *Ipmi) Description() string {
 }

 func (m *Ipmi) Gather(acc telegraf.Accumulator) error {
-	if len(m.path) == 0 {
+	if len(m.Path) == 0 {
 		return fmt.Errorf("ipmitool not found: verify that ipmitool is installed and that ipmitool is in your PATH")
 	}

@@ -76,7 +76,7 @@ func (m *Ipmi) parse(acc telegraf.Accumulator, server string) error {
 	}

 	opts = append(opts, "sdr")
-	cmd := execCommand(m.path, opts...)
+	cmd := execCommand(m.Path, opts...)
 	out, err := internal.CombinedOutputTimeout(cmd, time.Second*5)
 	if err != nil {
 		return fmt.Errorf("failed to run command %s: %s - %s", strings.Join(cmd.Args, " "), err, string(out))
@@ -149,7 +149,7 @@ func init() {
 	m := Ipmi{}
 	path, _ := exec.LookPath("ipmitool")
 	if len(path) > 0 {
-		m.path = path
+		m.Path = path
 	}
 	inputs.Add("ipmi_sensor", func() telegraf.Input {
 		return &m
--- a/plugins/inputs/ipmi_sensor/ipmi_test.go
+++ b/plugins/inputs/ipmi_sensor/ipmi_test.go
@@ -14,7 +14,7 @@ import (
 func TestGather(t *testing.T) {
 	i := &Ipmi{
 		Servers: []string{"USERID:PASSW0RD@lan(192.168.1.1)"},
-		path:    "ipmitool",
+		Path:    "ipmitool",
 	}
 	// overwriting exec commands with mock commands
 	execCommand = fakeExecCommand
@@ -118,7 +118,7 @@ func TestGather(t *testing.T) {
 	}

 	i = &Ipmi{
-		path: "ipmitool",
+		Path: "ipmitool",
 	}

 	err = i.Gather(&acc)
--- a/plugins/inputs/iptables/README.md
+++ b/plugins/inputs/iptables/README.md
@@ -2,7 +2,11 @@

 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.
+Rules are identified through associated comment. **Rules without comment are ignored**.
+Indeed we need a unique ID for the rule and the rule number is not a constant: it may vary when rules are inserted/deleted at start-up or by automatic tools (interactive firewalls, fail2ban, ...).
+Also when the rule set is becoming big (hundreds of lines) most people are interested in monitoring only a small part of the rule set.
+
+Before using this plugin **you must ensure that the rules you want to monitor are named with a unique comment**. Comments are added using the `-m comment --comment "my comment"` iptables options.

 The iptables command requires CAP_NET_ADMIN and CAP_NET_RAW capabilities. You have several options to grant telegraf to run iptables:

--- a/plugins/inputs/iptables/iptables.go
+++ b/plugins/inputs/iptables/iptables.go
@@ -33,14 +33,16 @@ func (ipt *Iptables) SampleConfig() string {
  ## 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"
+  ## iptables can be restricted to only list command "iptables -nvL".
  use_sudo = false
  ## Setting 'use_lock' to true runs iptables with the "-w" option.
  ## Adjust your sudo settings appropriately if using this option ("iptables -wnvl")
  use_lock = false
  ## defines the table to monitor:
  table = "filter"
-  ## defines the chains to monitor:
+  ## defines the chains to monitor.
+  ## NOTE: iptables rules without a comment will not be monitored.
+  ## Read the plugin documentation for more information.
  chains = [ "INPUT" ]
 `
 }
--- a/plugins/inputs/logparser/grok/grok_test.go
+++ b/plugins/inputs/logparser/grok/grok_test.go
@@ -57,6 +57,43 @@ func Benchmark_ParseLine_CustomPattern(b *testing.B) {
 	benchM = m
 }

+// Test a very simple parse pattern.
+func TestSimpleParse(t *testing.T) {
+	p := &Parser{
+		Patterns: []string{"%{TESTLOG}"},
+		CustomPatterns: `
+			TESTLOG %{NUMBER:num:int} %{WORD:client}
+		`,
+	}
+	assert.NoError(t, p.Compile())
+
+	m, err := p.ParseLine(`142 bot`)
+	assert.NoError(t, err)
+	require.NotNil(t, m)
+
+	assert.Equal(t,
+		map[string]interface{}{
+			"num":    int64(142),
+			"client": "bot",
+		},
+		m.Fields())
+}
+
+// Verify that patterns with a regex lookahead fail at compile time.
+func TestParsePatternsWithLookahead(t *testing.T) {
+	p := &Parser{
+		Patterns: []string{"%{MYLOG}"},
+		CustomPatterns: `
+			NOBOT ((?!bot|crawl).)*
+			MYLOG %{NUMBER:num:int} %{NOBOT:client}
+		`,
+	}
+	assert.NoError(t, p.Compile())
+
+	_, err := p.ParseLine(`1466004605359052000 bot`)
+	assert.Error(t, err)
+}
+
 func TestMeasurementName(t *testing.T) {
 	p := &Parser{
 		Measurement: "my_web_log",
--- a/plugins/inputs/logparser/logparser.go
+++ b/plugins/inputs/logparser/logparser.go
@@ -226,6 +226,8 @@ func (l *LogParserPlugin) parser() {
 				if m != nil {
 					l.acc.AddFields(m.Name(), m.Fields(), m.Tags(), m.Time())
 				}
+			} else {
+				log.Println("E! Error parsing log line: " + err.Error())
 			}
 		}
 	}
--- a/plugins/inputs/mysql/mysql.go
+++ b/plugins/inputs/mysql/mysql.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"database/sql"
 	"fmt"
+	"log"
 	"strconv"
 	"strings"
 	"sync"
@@ -904,92 +905,98 @@ func (m *Mysql) gatherGlobalStatuses(db *sql.DB, serv string, acc telegraf.Accum
 	// gather connection metrics from processlist for each user
 	if m.GatherProcessList {
 		conn_rows, err := db.Query("SELECT user, sum(1) FROM INFORMATION_SCHEMA.PROCESSLIST GROUP BY user")
+		if err != nil {
+			log.Printf("E! MySQL Error gathering process list: %s", err)
+		} else {
+			for conn_rows.Next() {
+				var user string
+				var connections int64

-		for conn_rows.Next() {
-			var user string
-			var connections int64
+				err = conn_rows.Scan(&user, &connections)
+				if err != nil {
+					return err
+				}

-			err = conn_rows.Scan(&user, &connections)
-			if err != nil {
-				return err
+				tags := map[string]string{"server": servtag, "user": user}
+				fields := make(map[string]interface{})
+
+				if err != nil {
+					return err
+				}
+				fields["connections"] = connections
+				acc.AddFields("mysql_users", fields, tags)
 			}
-
-			tags := map[string]string{"server": servtag, "user": user}
-			fields := make(map[string]interface{})
-
-			if err != nil {
-				return err
-			}
-			fields["connections"] = connections
-			acc.AddFields("mysql_users", fields, tags)
 		}
 	}

 	// gather connection metrics from user_statistics for each user
 	if m.GatherUserStatistics {
 		conn_rows, err := db.Query("select user, total_connections, concurrent_connections, connected_time, busy_time, cpu_time, bytes_received, bytes_sent, binlog_bytes_written, rows_fetched, rows_updated, table_rows_read, select_commands, update_commands, other_commands, commit_transactions, rollback_transactions, denied_connections, lost_connections, access_denied, empty_queries, total_ssl_connections FROM INFORMATION_SCHEMA.USER_STATISTICS GROUP BY user")
+		if err != nil {
+			log.Printf("E! MySQL Error gathering user stats: %s", err)
+		} else {
+			for conn_rows.Next() {
+				var user string
+				var total_connections int64
+				var concurrent_connections int64
+				var connected_time int64
+				var busy_time int64
+				var cpu_time int64
+				var bytes_received int64
+				var bytes_sent int64
+				var binlog_bytes_written int64
+				var rows_fetched int64
+				var rows_updated int64
+				var table_rows_read int64
+				var select_commands int64
+				var update_commands int64
+				var other_commands int64
+				var commit_transactions int64
+				var rollback_transactions int64
+				var denied_connections int64
+				var lost_connections int64
+				var access_denied int64
+				var empty_queries int64
+				var total_ssl_connections int64

-		for conn_rows.Next() {
-			var user string
-			var total_connections int64
-			var concurrent_connections int64
-			var connected_time int64
-			var busy_time int64
-			var cpu_time int64
-			var bytes_received int64
-			var bytes_sent int64
-			var binlog_bytes_written int64
-			var rows_fetched int64
-			var rows_updated int64
-			var table_rows_read int64
-			var select_commands int64
-			var update_commands int64
-			var other_commands int64
-			var commit_transactions int64
-			var rollback_transactions int64
-			var denied_connections int64
-			var lost_connections int64
-			var access_denied int64
-			var empty_queries int64
-			var total_ssl_connections int64
+				err = conn_rows.Scan(&user, &total_connections, &concurrent_connections,
+					&connected_time, &busy_time, &cpu_time, &bytes_received, &bytes_sent, &binlog_bytes_written,
+					&rows_fetched, &rows_updated, &table_rows_read, &select_commands, &update_commands, &other_commands,
+					&commit_transactions, &rollback_transactions, &denied_connections, &lost_connections, &access_denied,
+					&empty_queries, &total_ssl_connections,
+				)

-			err = conn_rows.Scan(&user, &total_connections, &concurrent_connections,
-				&connected_time, &busy_time, &cpu_time, &bytes_received, &bytes_sent, &binlog_bytes_written,
-				&rows_fetched, &rows_updated, &table_rows_read, &select_commands, &update_commands, &other_commands,
-				&commit_transactions, &rollback_transactions, &denied_connections, &lost_connections, &access_denied,
-				&empty_queries, &total_ssl_connections,
-			)
+				if err != nil {
+					return err
+				}

-			if err != nil {
-				return err
+				tags := map[string]string{"server": servtag, "user": user}
+				fields := map[string]interface{}{
+					"total_connections":      total_connections,
+					"concurrent_connections": concurrent_connections,
+					"connected_time":         connected_time,
+					"busy_time":              busy_time,
+					"cpu_time":               cpu_time,
+					"bytes_received":         bytes_received,
+					"bytes_sent":             bytes_sent,
+					"binlog_bytes_written":   binlog_bytes_written,
+					"rows_fetched":           rows_fetched,
+					"rows_updated":           rows_updated,
+					"table_rows_read":        table_rows_read,
+					"select_commands":        select_commands,
+					"update_commands":        update_commands,
+					"other_commands":         other_commands,
+					"commit_transactions":    commit_transactions,
+					"rollback_transactions":  rollback_transactions,
+					"denied_connections":     denied_connections,
+					"lost_connections":       lost_connections,
+					"access_denied":          access_denied,
+					"empty_queries":          empty_queries,
+					"total_ssl_connections":  total_ssl_connections,
+				}
+
+				acc.AddFields("mysql_user_stats", fields, tags)
 			}
-
-			tags := map[string]string{"server": servtag, "user": user}
-			fields := map[string]interface{}{
-				"total_connections":      total_connections,
-				"concurrent_connections": concurrent_connections,
-				"connected_time":         connected_time,
-				"busy_time":              busy_time,
-				"cpu_time":               cpu_time,
-				"bytes_received":         bytes_received,
-				"bytes_sent":             bytes_sent,
-				"binlog_bytes_written":   binlog_bytes_written,
-				"rows_fetched":           rows_fetched,
-				"rows_updated":           rows_updated,
-				"table_rows_read":        table_rows_read,
-				"select_commands":        select_commands,
-				"update_commands":        update_commands,
-				"other_commands":         other_commands,
-				"commit_transactions":    commit_transactions,
-				"rollback_transactions":  rollback_transactions,
-				"denied_connections":     denied_connections,
-				"lost_connections":       lost_connections,
-				"access_denied":          access_denied,
-				"empty_queries":          empty_queries,
-				"total_ssl_connections":  total_ssl_connections,
-			}
-
-			acc.AddFields("mysql_user_stats", fields, tags)
 		}
 	}

--- a/plugins/inputs/ping/ping_windows.go
+++ b/plugins/inputs/ping/ping_windows.go
@@ -40,10 +40,10 @@ func (s *Ping) Description() string {
 const sampleConfig = `
 	## urls to ping
 	urls = ["www.google.com"] # required
-	
+
 	## number of pings to send per collection (ping -n <COUNT>)
 	count = 4 # required
-	
+
 	## Ping timeout, in seconds. 0 means default timeout (ping -w <TIMEOUT>)
 	Timeout = 0
 `
@@ -64,7 +64,7 @@ func hostPinger(timeout float64, args ...string) (string, error) {
 }

 // processPingOutput takes in a string output from the ping command
-// based on linux implementation but using regex ( multilanguage support ) ( shouldn't affect the performance of the program )
+// based on linux implementation but using regex ( multilanguage support )
 // It returns (<transmitted packets>, <received reply>, <received packet>, <average response>, <min response>, <max response>)
 func processPingOutput(out string) (int, int, int, int, int, int, error) {
 	// So find a line contain 3 numbers except reply lines
@@ -189,13 +189,13 @@ func (p *Ping) Gather(acc telegraf.Accumulator) error {
 				"percent_reply_loss":  lossReply,
 			}
 			if avg > 0 {
-				fields["average_response_ms"] = avg
+				fields["average_response_ms"] = float64(avg)
 			}
 			if min > 0 {
-				fields["minimum_response_ms"] = min
+				fields["minimum_response_ms"] = float64(min)
 			}
 			if max > 0 {
-				fields["maximum_response_ms"] = max
+				fields["maximum_response_ms"] = float64(max)
 			}
 			acc.AddFields("ping", fields, tags)
 		}(url)
--- a/plugins/inputs/ping/ping_windows_test.go
+++ b/plugins/inputs/ping/ping_windows_test.go
@@ -77,9 +77,9 @@ func TestPingGather(t *testing.T) {
 		"reply_received":      4,
 		"percent_packet_loss": 0.0,
 		"percent_reply_loss":  0.0,
-		"average_response_ms": 50,
-		"minimum_response_ms": 50,
-		"maximum_response_ms": 52,
+		"average_response_ms": 50.0,
+		"minimum_response_ms": 50.0,
+		"maximum_response_ms": 52.0,
 	}
 	acc.AssertContainsTaggedFields(t, "ping", fields, tags)

--- a/plugins/inputs/postgresql/README.md
+++ b/plugins/inputs/postgresql/README.md
@@ -29,3 +29,25 @@ _* value ignored and therefore not recorded._


 More information about the meaning of these metrics can be found in the [PostgreSQL Documentation](http://www.postgresql.org/docs/9.2/static/monitoring-stats.html#PG-STAT-DATABASE-VIEW)
+
+## Configruation
+Specify address via a url matching:
+
+  `postgres://[pqgotest[:password]]@localhost[/dbname]?sslmode=[disable|verify-ca|verify-full]`
+
+All connection parameters are optional. Without the dbname parameter, the driver will default to a database with the same name as the user. This dbname is just for instantiating a connection with the server and doesn't restrict the databases we are trying to grab metrics for.
+  
+A  list of databases to explicitly ignore.  If not specified, metrics for all databases are gathered.  Do NOT use with the 'databases' option.
+
+  `ignored_databases = ["postgres", "template0", "template1"]`
+  
+A list of databases to pull metrics about. If not specified, metrics for all databases are gathered.  Do NOT use with the 'ignored_databases' option.
+
+  `databases = ["app_production", "testing"]`
+  
+### Configuration example
+```
+[[inputs.postgresql]]
+  address = "postgres://telegraf@localhost/someDB"
+  ignored_databases = ["template0", "template1"]
+```
--- a/plugins/inputs/postgresql/postgresql.go
+++ b/plugins/inputs/postgresql/postgresql.go
@@ -43,7 +43,7 @@ var sampleConfig = `
  # ignored_databases = ["postgres", "template0", "template1"]

  ## A list of databases to pull metrics about. If not specified, metrics for all
-  ## databases are gathered.  Do NOT use with the 'ignore_databases' option.
+  ## databases are gathered.  Do NOT use with the 'ignored_databases' option.
  # databases = ["app_production", "testing"]
 `

--- a/plugins/inputs/procstat/procstat.go
+++ b/plugins/inputs/procstat/procstat.go
@@ -8,6 +8,8 @@ import (
 	"strconv"
 	"strings"

+	"github.com/shirou/gopsutil/process"
+
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
 )
@@ -21,12 +23,15 @@ type Procstat struct {
 	User        string
 	PidTag      bool

+	// pidmap maps a pid to a process object, so we don't recreate every gather
+	pidmap map[int32]*process.Process
 	// tagmap maps a pid to a map of tags for that pid
 	tagmap map[int32]map[string]string
 }

 func NewProcstat() *Procstat {
 	return &Procstat{
+		pidmap: make(map[int32]*process.Process),
 		tagmap: make(map[int32]map[string]string),
 	}
 }
@@ -62,26 +67,51 @@ func (_ *Procstat) Description() string {
 }

 func (p *Procstat) Gather(acc telegraf.Accumulator) error {
-	pids, err := p.getAllPids()
+	err := p.createProcesses()
 	if err != nil {
 		log.Printf("E! Error: procstat getting process, exe: [%s] pidfile: [%s] pattern: [%s] user: [%s] %s",
 			p.Exe, p.PidFile, p.Pattern, p.User, err.Error())
 	} else {
-		for _, pid := range pids {
+		for pid, proc := range p.pidmap {
 			if p.PidTag {
 				p.tagmap[pid]["pid"] = fmt.Sprint(pid)
 			}
-			p := NewSpecProcessor(p.ProcessName, p.Prefix, pid, acc, p.tagmap[pid])
-			err := p.pushMetrics()
-			if err != nil {
-				log.Printf("E! Error: procstat: %s", err.Error())
-			}
+			p := NewSpecProcessor(p.ProcessName, p.Prefix, pid, acc, proc, p.tagmap[pid])
+			p.pushMetrics()
 		}
 	}

 	return nil
 }

+func (p *Procstat) createProcesses() error {
+	var errstring string
+	var outerr error
+
+	pids, err := p.getAllPids()
+	if err != nil {
+		errstring += err.Error() + " "
+	}
+
+	for _, pid := range pids {
+		_, ok := p.pidmap[pid]
+		if !ok {
+			proc, err := process.NewProcess(pid)
+			if err == nil {
+				p.pidmap[pid] = proc
+			} else {
+				errstring += err.Error() + " "
+			}
+		}
+	}
+
+	if errstring != "" {
+		outerr = fmt.Errorf("%s", errstring)
+	}
+
+	return outerr
+}
+
 func (p *Procstat) getAllPids() ([]int32, error) {
 	var pids []int32
 	var err error
--- a/plugins/inputs/procstat/procstat_test.go
+++ b/plugins/inputs/procstat/procstat_test.go
@@ -6,6 +6,7 @@ import (
 	"strconv"
 	"testing"

+	"github.com/shirou/gopsutil/process"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

@@ -23,6 +24,7 @@ func TestGather(t *testing.T) {
 	p := Procstat{
 		PidFile: file.Name(),
 		Prefix:  "foo",
+		pidmap:  make(map[int32]*process.Process),
 		tagmap:  make(map[int32]map[string]string),
 	}
 	p.Gather(&acc)
--- a/plugins/inputs/procstat/spec_processor.go
+++ b/plugins/inputs/procstat/spec_processor.go
@@ -1,7 +1,6 @@
 package procstat

 import (
-	"fmt"
 	"time"

 	"github.com/shirou/gopsutil/process"
@@ -10,13 +9,12 @@ import (
 )

 type SpecProcessor struct {
-	ProcessName string
-	Prefix      string
-	pid         int32
-	tags        map[string]string
-	fields      map[string]interface{}
-	acc         telegraf.Accumulator
-	proc        *process.Process
+	Prefix string
+	pid    int32
+	tags   map[string]string
+	fields map[string]interface{}
+	acc    telegraf.Accumulator
+	proc   *process.Process
 }

 func NewSpecProcessor(
@@ -24,35 +22,29 @@ func NewSpecProcessor(
 	prefix string,
 	pid int32,
 	acc telegraf.Accumulator,
+	p *process.Process,
 	tags map[string]string,
 ) *SpecProcessor {
+	if processName != "" {
+		tags["process_name"] = processName
+	} else {
+		name, err := p.Name()
+		if err == nil {
+			tags["process_name"] = name
+		}
+	}
 	return &SpecProcessor{
-		ProcessName: processName,
-		Prefix:      prefix,
-		pid:         pid,
-		tags:        tags,
-		fields:      make(map[string]interface{}),
-		acc:         acc,
+		Prefix: prefix,
+		pid:    pid,
+		tags:   tags,
+		fields: make(map[string]interface{}),
+		acc:    acc,
+		proc:   p,
 	}
 }

-func (p *SpecProcessor) pushMetrics() error {
+func (p *SpecProcessor) pushMetrics() {
 	var prefix string
-	proc, err := process.NewProcess(p.pid)
-	if err != nil {
-		return fmt.Errorf("Failed to open process with pid '%d'. Error: '%s'",
-			p.pid, err)
-	}
-	p.proc = proc
-	if p.ProcessName != "" {
-		p.tags["process_name"] = p.ProcessName
-	} else {
-		name, err := p.proc.Name()
-		if err == nil {
-			p.tags["process_name"] = name
-		}
-	}
-
 	if p.Prefix != "" {
 		prefix = p.Prefix + "_"
 	}
@@ -115,5 +107,4 @@ func (p *SpecProcessor) pushMetrics() error {
 	}

 	p.acc.AddFields("procstat", fields, p.tags)
-	return nil
 }
--- a/plugins/inputs/prometheus/parser_test.go
+++ b/plugins/inputs/prometheus/parser_test.go
@@ -111,11 +111,9 @@ func TestParseValidPrometheus(t *testing.T) {
 		"gauge": float64(1),
 	}, metrics[0].Fields())
 	assert.Equal(t, map[string]string{
-		"osVersion":        "CentOS Linux 7 (Core)",
-		"dockerVersion":    "1.8.2",
-		"kernelVersion":    "3.10.0-229.20.1.el7.x86_64",
-		"cadvisorRevision": "",
-		"cadvisorVersion":  "",
+		"osVersion":     "CentOS Linux 7 (Core)",
+		"dockerVersion": "1.8.2",
+		"kernelVersion": "3.10.0-229.20.1.el7.x86_64",
 	}, metrics[0].Tags())

 	// Counter value
--- a/plugins/inputs/socket_listener/README.md
+++ b/plugins/inputs/socket_listener/README.md
@@ -0,0 +1,112 @@
+# socket listener service input plugin
+
+The Socket Listener is a service input plugin that listens for messages from
+streaming (tcp, unix) or datagram (udp, unixgram) protocols.
+
+The plugin expects messages in the
+[Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).
+
+### Configuration:
+
+This is a sample configuration for the plugin.
+
+```toml
+# Generic socket listener capable of handling multiple socket types.
+[[inputs.socket_listener]]
+  ## URL to listen on
+  # service_address = "tcp://:8094"
+  # service_address = "tcp://127.0.0.1:http"
+  # service_address = "tcp4://:8094"
+  # service_address = "tcp6://:8094"
+  # service_address = "tcp6://[2001:db8::1]:8094"
+  # service_address = "udp://:8094"
+  # service_address = "udp4://:8094"
+  # service_address = "udp6://:8094"
+  # service_address = "unix:///tmp/telegraf.sock"
+  # service_address = "unixgram:///tmp/telegraf.sock"
+
+  ## Maximum number of concurrent connections.
+  ## Only applies to stream sockets (e.g. TCP).
+  ## 0 (default) is unlimited.
+  # max_connections = 1024
+
+  ## Maximum socket buffer size in bytes.
+  ## For stream sockets, once the buffer fills up, the sender will start backing up.
+  ## For datagram sockets, once the buffer fills up, metrics will start dropping.
+  ## Defaults to the OS default.
+  # read_buffer_size = 65535
+
+  ## Data format to consume.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  # data_format = "influx"
+```
+
+## A Note on UDP OS Buffer Sizes
+
+The `read_buffer_size` config option can be used to adjust the size of the socket
+buffer, but this number is limited by OS settings. On Linux, `read_buffer_size`
+will default to `rmem_default` and will be capped by `rmem_max`. On BSD systems,
+`read_buffer_size` is capped by `maxsockbuf`, and there is no OS default
+setting.
+
+Instructions on how to adjust these OS settings are available below.
+
+Some OSes (most notably, Linux) place very restricive limits on the performance
+of UDP protocols. It is _highly_ recommended that you increase these OS limits to
+at least 8MB before trying to run large amounts of UDP traffic to your instance.
+8MB is just a recommendation, and can be adjusted higher.
+
+### Linux
+Check the current UDP/IP receive buffer limit & default by typing the following
+commands:
+
+```
+sysctl net.core.rmem_max
+sysctl net.core.rmem_default
+```
+
+If the values are less than 8388608 bytes you should add the following lines to
+the /etc/sysctl.conf file:
+
+```
+net.core.rmem_max=8388608
+net.core.rmem_default=8388608
+```
+
+Changes to /etc/sysctl.conf do not take effect until reboot.
+To update the values immediately, type the following commands as root:
+
+```
+sysctl -w net.core.rmem_max=8388608
+sysctl -w net.core.rmem_default=8388608
+```
+
+### BSD/Darwin
+
+On BSD/Darwin systems you need to add about a 15% padding to the kernel limit
+socket buffer. Meaning if you want an 8MB buffer (8388608 bytes) you need to set
+the kernel limit to `8388608*1.15 = 9646900`. This is not documented anywhere but
+happens
+[in the kernel here.](https://github.com/freebsd/freebsd/blob/master/sys/kern/uipc_sockbuf.c#L63-L64)
+
+Check the current UDP/IP buffer limit by typing the following command:
+
+```
+sysctl kern.ipc.maxsockbuf
+```
+
+If the value is less than 9646900 bytes you should add the following lines
+to the /etc/sysctl.conf file (create it if necessary):
+
+```
+kern.ipc.maxsockbuf=9646900
+```
+
+Changes to /etc/sysctl.conf do not take effect until reboot.
+To update the values immediately, type the following command as root:
+
+```
+sysctl -w kern.ipc.maxsockbuf=9646900
+```
--- a/plugins/inputs/tcp_listener/README.md
+++ b/plugins/inputs/tcp_listener/README.md
@@ -1,30 +1,4 @@
 # TCP listener service input plugin

-The TCP listener is a service input plugin that listens for messages on a TCP
-socket and adds those messages to InfluxDB.
-The plugin expects messages in the
-[Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).
-
-### Configuration:
-
-This is a sample configuration for the plugin.
-
-```toml
-# Generic TCP listener
-[[inputs.tcp_listener]]
-  ## Address and port to host TCP listener on
-  service_address = ":8094"
-
-  ## Number of TCP messages allowed to queue up. Once filled, the
-  ## TCP listener will start dropping packets.
-  allowed_pending_messages = 10000
-
-  ## Maximum number of concurrent TCP connections to allow
-  max_tcp_connections = 250
-
-  ## Data format to consume.
-  ## Each data format has it's own unique set of configuration options, read
-  ## more about them here:
-  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-  data_format = "influx"
-```
+> DEPRECATED: As of version 1.3 the TCP listener plugin has been deprecated in favor of the
+> [socket_listener plugin](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener)
--- a/plugins/inputs/tcp_listener/tcp_listener.go
+++ b/plugins/inputs/tcp_listener/tcp_listener.go
@@ -58,21 +58,9 @@ var malformedwarn = "E! tcp_listener has received %d malformed packets" +
 	" thus far."

 const sampleConfig = `
-  ## Address and port to host TCP listener on
-  # service_address = ":8094"
-
-  ## Number of TCP messages allowed to queue up. Once filled, the
-  ## TCP listener will start dropping packets.
-  # allowed_pending_messages = 10000
-
-  ## Maximum number of concurrent TCP connections to allow
-  # max_tcp_connections = 250
-
-  ## Data format to consume.
-  ## Each data format has it's own unique set of configuration options, read
-  ## more about them here:
-  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-  data_format = "influx"
+  # DEPRECATED: the TCP listener plugin has been deprecated in favor of the
+  # socket_listener plugin
+  # see https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener
 `

 func (t *TcpListener) SampleConfig() string {
@@ -98,6 +86,10 @@ func (t *TcpListener) Start(acc telegraf.Accumulator) error {
 	t.Lock()
 	defer t.Unlock()

+	log.Println("W! DEPRECATED: the TCP listener plugin has been deprecated " +
+		"in favor of the socket_listener plugin " +
+		"(https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener)")
+
 	tags := map[string]string{
 		"address": t.ServiceAddress,
 	}
--- a/plugins/inputs/udp_listener/README.md
+++ b/plugins/inputs/udp_listener/README.md
@@ -1,86 +1,4 @@
 # UDP listener service input plugin

-The UDP listener is a service input plugin that listens for messages on a UDP
-socket and adds those messages to InfluxDB.
-The plugin expects messages in the
-[Telegraf Input Data Formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md).
-
-### Configuration:
-
-This is a sample configuration for the plugin.
-
-```toml
-[[inputs.udp_listener]]
-  ## Address and port to host UDP listener on
-  service_address = ":8092"
-
-  ## Number of UDP messages allowed to queue up. Once filled, the
-  ## UDP listener will start dropping packets.
-  allowed_pending_messages = 10000
-
-  ## Data format to consume.
-  ## Each data format has it's own unique set of configuration options, read
-  ## more about them here:
-  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-  data_format = "influx"
-```
-
-## A Note on UDP OS Buffer Sizes
-
-Some OSes (most notably, Linux) place very restricive limits on the performance
-of UDP protocols. It is _highly_ recommended that you increase these OS limits to
-at least 8MB before trying to run large amounts of UDP traffic to your instance.
-8MB is just a recommendation, and can be adjusted higher.
-
-### Linux
-Check the current UDP/IP receive buffer limit & default by typing the following
-commands:
-
-```
-sysctl net.core.rmem_max
-sysctl net.core.rmem_default
-```
-
-If the values are less than 8388608 bytes you should add the following lines to
-the /etc/sysctl.conf file:
-
-```
-net.core.rmem_max=8388608
-net.core.rmem_default=8388608
-```
-
-Changes to /etc/sysctl.conf do not take effect until reboot.
-To update the values immediately, type the following commands as root:
-
-```
-sysctl -w net.core.rmem_max=8388608
-sysctl -w net.core.rmem_default=8388608
-```
-
-### BSD/Darwin
-
-On BSD/Darwin systems you need to add about a 15% padding to the kernel limit
-socket buffer. Meaning if you want an 8MB buffer (8388608 bytes) you need to set
-the kernel limit to `8388608*1.15 = 9646900`. This is not documented anywhere but
-happens
-[in the kernel here.](https://github.com/freebsd/freebsd/blob/master/sys/kern/uipc_sockbuf.c#L63-L64)
-
-Check the current UDP/IP buffer limit by typing the following command:
-
-```
-sysctl kern.ipc.maxsockbuf
-```
-
-If the value is less than 9646900 bytes you should add the following lines
-to the /etc/sysctl.conf file (create it if necessary):
-
-```
-kern.ipc.maxsockbuf=9646900
-```
-
-Changes to /etc/sysctl.conf do not take effect until reboot.
-To update the values immediately, type the following commands as root:
-
-```
-sysctl -w kern.ipc.maxsockbuf=9646900
-```
+> DEPRECATED: As of version 1.3 the UDP listener plugin has been deprecated in favor of the
+> [socket_listener plugin](https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener)
--- a/plugins/inputs/udp_listener/udp_listener.go
+++ b/plugins/inputs/udp_listener/udp_listener.go
@@ -66,22 +66,9 @@ var malformedwarn = "E! udp_listener has received %d malformed packets" +
 	" thus far."

 const sampleConfig = `
-  ## Address and port to host UDP listener on
-  # service_address = ":8092"
-
-  ## Number of UDP messages allowed to queue up. Once filled, the
-  ## UDP listener will start dropping packets.
-  # allowed_pending_messages = 10000
-
-  ## Set the buffer size of the UDP connection outside of OS default (in bytes)
-  ## If set to 0, take OS default
-  udp_buffer_size = 16777216
-
-  ## Data format to consume.
-  ## Each data format has it's own unique set of configuration options, read
-  ## more about them here:
-  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
-  data_format = "influx"
+  # DEPRECATED: the TCP listener plugin has been deprecated in favor of the
+  # socket_listener plugin
+  # see https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener
 `

 func (u *UdpListener) SampleConfig() string {
@@ -106,6 +93,10 @@ func (u *UdpListener) Start(acc telegraf.Accumulator) error {
 	u.Lock()
 	defer u.Unlock()

+	log.Println("W! DEPRECATED: the UDP listener plugin has been deprecated " +
+		"in favor of the socket_listener plugin " +
+		"(https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener)")
+
 	tags := map[string]string{
 		"address": u.ServiceAddress,
 	}
--- a/plugins/inputs/webhooks/github/github_webhooks.go
+++ b/plugins/inputs/webhooks/github/github_webhooks.go
@@ -34,9 +34,10 @@ func (gh *GithubWebhook) eventHandler(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-
-	p := e.NewMetric()
-	gh.acc.AddFields("github_webhooks", p.Fields(), p.Tags(), p.Time())
+	if e != nil {
+		p := e.NewMetric()
+		gh.acc.AddFields("github_webhooks", p.Fields(), p.Tags(), p.Time())
+	}

 	w.WriteHeader(http.StatusOK)
 }
@@ -84,6 +85,8 @@ func NewEvent(data []byte, name string) (Event, error) {
 		return generateEvent(data, &MembershipEvent{})
 	case "page_build":
 		return generateEvent(data, &PageBuildEvent{})
+	case "ping":
+		return nil, nil
 	case "public":
 		return generateEvent(data, &PublicEvent{})
 	case "pull_request":
--- a/plugins/inputs/webhooks/github/github_webhooks_test.go
+++ b/plugins/inputs/webhooks/github/github_webhooks_test.go
@@ -25,6 +25,10 @@ func TestCommitCommentEvent(t *testing.T) {
 	GithubWebhookRequest("commit_comment", CommitCommentEventJSON(), t)
 }

+func TestPingEvent(t *testing.T) {
+	GithubWebhookRequest("ping", "", t)
+}
+
 func TestDeleteEvent(t *testing.T) {
 	GithubWebhookRequest("delete", DeleteEventJSON(), t)
 }
--- a/plugins/inputs/win_perf_counters/pdh.go
+++ b/plugins/inputs/win_perf_counters/pdh.go
@@ -331,7 +331,7 @@ func PdhCollectQueryData(hQuery PDH_HQUERY) uint32 {
 func PdhGetFormattedCounterValueDouble(hCounter PDH_HCOUNTER, lpdwType *uint32, pValue *PDH_FMT_COUNTERVALUE_DOUBLE) uint32 {
 	ret, _, _ := pdh_GetFormattedCounterValue.Call(
 		uintptr(hCounter),
-		uintptr(PDH_FMT_DOUBLE),
+		uintptr(PDH_FMT_DOUBLE|PDH_FMT_NOCAP100),
 		uintptr(unsafe.Pointer(lpdwType)),
 		uintptr(unsafe.Pointer(pValue)))

@@ -378,7 +378,7 @@ func PdhGetFormattedCounterValueDouble(hCounter PDH_HCOUNTER, lpdwType *uint32,
 func PdhGetFormattedCounterArrayDouble(hCounter PDH_HCOUNTER, lpdwBufferSize *uint32, lpdwBufferCount *uint32, itemBuffer *PDH_FMT_COUNTERVALUE_ITEM_DOUBLE) uint32 {
 	ret, _, _ := pdh_GetFormattedCounterArrayW.Call(
 		uintptr(hCounter),
-		uintptr(PDH_FMT_DOUBLE),
+		uintptr(PDH_FMT_DOUBLE|PDH_FMT_NOCAP100),
 		uintptr(unsafe.Pointer(lpdwBufferSize)),
 		uintptr(unsafe.Pointer(lpdwBufferCount)),
 		uintptr(unsafe.Pointer(itemBuffer)))
--- a/plugins/outputs/amqp/README.md
+++ b/plugins/outputs/amqp/README.md
@@ -1,13 +1,18 @@
 # AMQP Output Plugin

-This plugin writes to a AMQP exchange using tag, defined in configuration file
-as RoutingTag, as a routing key.
+This plugin writes to a AMQP 0-9-1 Exchange, a promenent implementation of this protocol being [RabbitMQ](https://www.rabbitmq.com/).
+
+Metrics are written to a topic exchange using tag, defined in configuration file as RoutingTag, as a routing key.

 If RoutingTag is empty, then empty routing key will be used.
 Metrics are grouped in batches by RoutingTag.

 This plugin doesn't bind exchange to a queue, so it should be done by consumer.

+For an introduction to AMQP see:
+- https://www.rabbitmq.com/tutorials/amqp-concepts.html
+- https://www.rabbitmq.com/getstarted.html
+
 ### Configuration:

 ```
@@ -18,6 +23,8 @@ This plugin doesn't bind exchange to a queue, so it should be done by consumer.
  ## AMQP exchange
  exchange = "telegraf"
  ## Auth method. PLAIN and EXTERNAL are supported
+  ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as
+  ## described here: https://www.rabbitmq.com/plugins.html
  # auth_method = "PLAIN"
  ## Telegraf tag to use as a routing key
  ##  ie, if this tag exists, it's value will be used as the routing key
--- a/plugins/outputs/amqp/amqp.go
+++ b/plugins/outputs/amqp/amqp.go
@@ -40,6 +40,7 @@ type AMQP struct {
 	// Use SSL but skip chain & host verification
 	InsecureSkipVerify bool

+	conn    *amqp.Connection
 	channel *amqp.Channel
 	sync.Mutex
 	headers amqp.Table
@@ -68,6 +69,8 @@ var sampleConfig = `
  ## AMQP exchange
  exchange = "telegraf"
  ## Auth method. PLAIN and EXTERNAL are supported
+  ## Using EXTERNAL requires enabling the rabbitmq_auth_mechanism_ssl plugin as
+  ## described here: https://www.rabbitmq.com/plugins.html
  # auth_method = "PLAIN"
  ## Telegraf tag to use as a routing key
  ##  ie, if this tag exists, it's value will be used as the routing key
@@ -129,6 +132,8 @@ func (q *AMQP) Connect() error {
 	if err != nil {
 		return err
 	}
+	q.conn = connection
+
 	channel, err := connection.Channel()
 	if err != nil {
 		return fmt.Errorf("Failed to open a channel: %s", err)
@@ -148,7 +153,11 @@ func (q *AMQP) Connect() error {
 	}
 	q.channel = channel
 	go func() {
-		log.Printf("I! Closing: %s", <-connection.NotifyClose(make(chan *amqp.Error)))
+		err := <-connection.NotifyClose(make(chan *amqp.Error))
+		if err == nil {
+			return
+		}
+		log.Printf("I! Closing: %s", err)
 		log.Printf("I! Trying to reconnect")
 		for err := q.Connect(); err != nil; err = q.Connect() {
 			log.Println("E! ", err.Error())
@@ -160,7 +169,12 @@ func (q *AMQP) Connect() error {
 }

 func (q *AMQP) Close() error {
-	return q.channel.Close()
+	err := q.conn.Close()
+	if err != nil && err != amqp.ErrClosed {
+		log.Printf("E! Error closing AMQP connection: %s", err)
+		return err
+	}
+	return nil
 }

 func (q *AMQP) SampleConfig() string {
@@ -207,7 +221,7 @@ func (q *AMQP) Write(metrics []telegraf.Metric) error {
 				Body:        buf,
 			})
 		if err != nil {
-			return fmt.Errorf("FAILED to send amqp message: %s", err)
+			return fmt.Errorf("Failed to send AMQP message: %s", err)
 		}
 	}
 	return nil
--- a/plugins/outputs/file/README.md
+++ b/plugins/outputs/file/README.md
@@ -1 +1,16 @@
 # file Output Plugin
+
+This plugin writes telegraf metrics to files
+
+### Configuration
+```
+[[outputs.file]]
+  ## Files to write to, "stdout" is a specially handled file.
+  files = ["stdout", "/tmp/metrics.out"]
+
+  ## Data format to output.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
+```
--- a/plugins/outputs/influxdb/influxdb.go
+++ b/plugins/outputs/influxdb/influxdb.go
@@ -112,6 +112,8 @@ func (i *InfluxDB) Connect() error {
 				Timeout:   i.Timeout.Duration,
 				TLSConfig: tlsConfig,
 				UserAgent: i.UserAgent,
+				Username:  i.Username,
+				Password:  i.Password,
 			}
 			wp := client.WriteParams{
 				Database:        i.Database,
--- a/plugins/outputs/prometheus_client/prometheus_client.go
+++ b/plugins/outputs/prometheus_client/prometheus_client.go
@@ -1,6 +1,7 @@
 package prometheus_client

 import (
+	"context"
 	"fmt"
 	"log"
 	"net/http"
@@ -24,6 +25,7 @@ type MetricWithExpiration struct {
 type PrometheusClient struct {
 	Listen             string
 	ExpirationInterval internal.Duration `toml:"expiration_interval"`
+	server             *http.Server

 	metrics map[string]*MetricWithExpiration

@@ -41,30 +43,25 @@ var sampleConfig = `
 func (p *PrometheusClient) Start() error {
 	p.metrics = make(map[string]*MetricWithExpiration)
 	prometheus.Register(p)
-	defer func() {
-		if r := recover(); r != nil {
-			// recovering from panic here because there is no way to stop a
-			// running http go server except by a kill signal. Since the server
-			// does not stop on SIGHUP, Start() will panic when the process
-			// is reloaded.
-		}
-	}()
+
 	if p.Listen == "" {
 		p.Listen = "localhost:9126"
 	}

-	http.Handle("/metrics", prometheus.Handler())
-	server := &http.Server{
-		Addr: p.Listen,
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", prometheus.Handler())
+
+	p.server = &http.Server{
+		Addr:    p.Listen,
+		Handler: mux,
 	}

-	go server.ListenAndServe()
+	go p.server.ListenAndServe()
 	return nil
 }

 func (p *PrometheusClient) Stop() {
-	// TODO: Use a listener for http.Server that counts active connections
-	//       that can be stopped and closed gracefully
+	// plugin gets cleaned up in Close() already.
 }

 func (p *PrometheusClient) Connect() error {
@@ -73,8 +70,9 @@ func (p *PrometheusClient) Connect() error {
 }

 func (p *PrometheusClient) Close() error {
-	// This service output does not need to close any of its connections
-	return nil
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	return p.server.Shutdown(ctx)
 }

 func (p *PrometheusClient) SampleConfig() string {
--- a/plugins/outputs/riemann/riemann_test.go
+++ b/plugins/outputs/riemann/riemann_test.go
@@ -193,7 +193,16 @@ func TestConnectAndWrite(t *testing.T) {
 	err = r.Write(metrics)
 	require.NoError(t, err)

-	time.Sleep(200 * time.Millisecond)
+	start := time.Now()
+	for true {
+		events, _ := r.client.Query(`tagged "docker"`)
+		if len(events) > 0 {
+			break
+		}
+		if time.Since(start) > time.Second {
+			break
+		}
+	}

 	// are there any "docker" tagged events in Riemann?
 	events, err := r.client.Query(`tagged "docker"`)
--- a/plugins/outputs/socket_writer/README.md
+++ b/plugins/outputs/socket_writer/README.md
@@ -0,0 +1,27 @@
+# socket_writer Plugin
+
+The socket_writer plugin can write to a UDP, TCP, or unix socket.
+
+It can output data in any of the [supported output formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md).
+
+```toml
+# Generic socket writer capable of handling multiple socket types.
+[[outputs.socket_writer]]
+  ## URL to connect to
+  # address = "tcp://127.0.0.1:8094"
+  # address = "tcp://example.com:http"
+  # address = "tcp4://127.0.0.1:8094"
+  # address = "tcp6://127.0.0.1:8094"
+  # address = "tcp6://[2001:db8::1]:8094"
+  # address = "udp://127.0.0.1:8094"
+  # address = "udp4://127.0.0.1:8094"
+  # address = "udp6://127.0.0.1:8094"
+  # address = "unix:///tmp/telegraf.sock"
+  # address = "unixgram:///tmp/telegraf.sock"
+
+  ## Data format to generate.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  # data_format = "influx"
+```
--- a/plugins/parsers/nagios/parser_test.go
+++ b/plugins/parsers/nagios/parser_test.go
@@ -67,7 +67,7 @@ func TestParseValidOutput(t *testing.T) {
 	assert.Equal(t, map[string]interface{}{
 		"value": float64(0.008457),
 	}, metrics[0].Fields())
-	assert.Equal(t, map[string]string{"unit": ""}, metrics[0].Tags())
+	assert.Equal(t, map[string]string{}, metrics[0].Tags())
 }

 func TestParseInvalidOutput(t *testing.T) {
--- a/scripts/build.py
+++ b/scripts/build.py
@@ -22,6 +22,7 @@ INSTALL_ROOT_DIR = "/usr/bin"
 LOG_DIR = "/var/log/telegraf"
 SCRIPT_DIR = "/usr/lib/telegraf/scripts"
 CONFIG_DIR = "/etc/telegraf"
+CONFIG_DIR_D = "/etc/telegraf/telegraf.d"
 LOGROTATE_DIR = "/etc/logrotate.d"

 INIT_SCRIPT = "scripts/init.sh"
@@ -115,7 +116,7 @@ def create_package_fs(build_root):
    logging.debug("Creating a filesystem hierarchy from directory: {}".format(build_root))
    # Using [1:] for the path names due to them being absolute
    # (will overwrite previous paths, per 'os.path.join' documentation)
-    dirs = [ INSTALL_ROOT_DIR[1:], LOG_DIR[1:], SCRIPT_DIR[1:], CONFIG_DIR[1:], LOGROTATE_DIR[1:] ]
+    dirs = [ INSTALL_ROOT_DIR[1:], LOG_DIR[1:], SCRIPT_DIR[1:], CONFIG_DIR[1:], LOGROTATE_DIR[1:], CONFIG_DIR_D[1:] ]
    for d in dirs:
        os.makedirs(os.path.join(build_root, d))
        os.chmod(os.path.join(build_root, d), 0o755)
Author	SHA1	Message	Date
Cameron Sparr	5da28410e2	Load external .so plugins support for the Go 1.8 shared object feature of loading external plugins. this support relies on the developer defining a `Plugin` symbol in their .so file that is a telegraf plugin interface. So instead of the plugin developer "Adding" their own plugin to the telegraf registry, telegraf loads the .so, looks up the Plugin symbol, and then adds it if it finds it. The name of the plugin is determined by telegraf, and is namespaced based on the filename and path. see #1717	2017-03-09 11:49:16 +00:00
Cameron Sparr	e811e2600d	create telegraf.d directory in tarball closes #2513	2017-03-09 11:41:08 +00:00
Timothy	49c212337f	Update CONFIGURATION.md (#2516 ) Add information about default configuration file locations. Also mention that the config directory option is available.	2017-03-09 11:21:03 +00:00
Dennis Dryden	d243d69a09	Add configuration docs to Postgresql input plugin (#2515 ) * Add configuration docs to Postgresql input plugin Add configuration docs to PostgreSQL input plugin README (mostly from the source code) though I've not included the configuration example that seems to use all he connections on the database[1]. [1] https://github.com/influxdata/telegraf/issues/2410 * Fix typo in readme and sampleConfig string.	2017-03-09 11:19:03 +00:00
jeremydenoun	ae6a5d2255	Remove warning if parse empty content (#2500 ) closes #2448	2017-03-08 14:08:55 -08:00
Robpol86	56aa89e5c8	Exporting Ipmi.Path to be set by config. (#2498 ) * Exporting Ipmi.Path to be set by config. Currently "path" is not exported, giving this error when users try to override the variable via telegraf.conf as per the sample config: `field corresponding to `path' is not defined in `ipmi_sensor.Ipmi'` Exporting the variable solves the problem. Updating changelog.	2017-03-08 16:38:36 +00:00
vvvkamper	7513fcac4e	Fix part 2 of #1291 added PDH_FMT_NOCAP100 format option closes #2483	2017-03-08 13:39:03 +00:00
Cameron Sparr	9df2974a0f	update gopsutil for file close fixes hopefully this will fix #2472	2017-03-08 12:54:17 +00:00
Daniel Nelson	ceb36adac7	Update issue template	2017-03-06 11:20:53 -08:00
Cameron Sparr	7a8e821731	Revert "Procstat: don't cache PIDs" (#2479 )	2017-03-06 15:59:36 +00:00
François de Metz	76bcdecd21	Respond 200 when receiving a ping event. (#2492 )	2017-03-06 12:34:41 +00:00
Jack Zampolin	10744646db	AMQP Consumer plugin (#1678 )	2017-03-03 10:24:50 -08:00
Charles-Henri	1873abd248	Iptables input: document better the ignored rules behavior (#2482 ) During issue #2215 it was highlighted that the current behavior where rules without a comment are ignored is confusing for several users. This commit improves the documentation and adds a NOTE to the sample config to clarify the behavior for new users.	2017-03-02 09:58:26 +00:00
Chris Koehnke	9618515926	Disk counter array newline (#2481 ) Tweak formatting of `LogicalDisk` counter array to have one entry per line.	2017-03-02 08:43:33 +00:00
Cameron Sparr	a251adb838	Fix type conflict on windows ping plugin (#2462 ) closes #1433	2017-03-01 11:22:42 +00:00
Cameron Sparr	9e810ac463	Handle nil os.FileInfo in filepath.Walk closes #2466	2017-02-28 17:51:03 +00:00
Cameron Sparr	b9457a1092	log error message when invalid regex is used closes #2178	2017-02-28 12:48:14 +00:00
Cameron Sparr	6f2eeae498	Remove sleep from riemann test	2017-02-28 12:46:27 +00:00
Cameron Sparr	42a41d33cc	add cgroup plugin to README	2017-02-24 09:43:22 +00:00
Cameron Sparr	81408f9da7	switch out deprecated docker client library closes #2071	2017-02-22 10:55:00 +00:00
Rickard von Essen	c4212d69c9	Updated readme, now requires Go 1.8 (#2455 )	2017-02-21 22:13:22 +01:00
Carlos	e17164d3f0	Added default config to file output pugin's README (#2426 )	2017-02-20 11:50:39 +01:00
Cameron Sparr	e5349393f8	Check for errors in user stats & process list closes #2414	2017-02-17 15:38:33 +00:00
Cameron Sparr	06176ef410	Only set the buffer size once fixes #2380	2017-02-17 14:11:15 +00:00
Cameron Sparr	2a3448c8f3	socket_writer output plugin README	2017-02-16 23:13:14 +00:00
Leandro Piccilli	5da40d56ad	Check if tag value is empty before allocation closes #2390 closes #2404	2017-02-16 23:07:27 +00:00
Cameron Sparr	54c9a385d5	Fix prometheus_client reload behavior fixes #2282	2017-02-16 21:57:13 +00:00
Priyank Trivedi	25c55419df	Fix typo - Default from Defalt (#2417 )	2017-02-16 19:03:17 +00:00
Yaron de Leeuw	c19fb1535e	README: update golang requirement to 1.7 (#2412 ) The docker engine-api package we use needs golang 1.7+, see: https://github.com/docker/engine-api/pull/382#issuecomment-244512952 So telegraf won't compile without 1.7	2017-02-15 17:17:26 +00:00
François de Metz	45a168e425	Fix setting the username and the password to the influxdb output. (#2401 )	2017-02-13 15:30:30 +00:00
Cameron Sparr	22243a8354	Skip service input plugins in test mode	2017-02-13 10:40:38 +00:00
Cameron Sparr	ff9369f1a1	prepend 'inputs.' to --test output check	2017-02-13 10:33:51 +00:00
Cameron Sparr	21cf79163c	don't use influxdata/config, just use influxdata/toml	2017-02-10 17:27:18 +00:00
Cameron Sparr	f05fac74cb	update naoina/toml to do config validation	2017-02-10 17:05:13 +00:00
Cameron Sparr	c8cc01ba6a	deprecate udp_listener & tcp_listener	2017-02-06 10:41:44 +00:00