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

.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.

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:
@@ -48,18 +54,30 @@ be deprecated eventually.
 - [#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]
 

Godeps

@@ -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

Makefile

@@ -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 \

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)

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
 }

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:

cmd/telegraf/telegraf.go

@@ -13,12 +13,14 @@ 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/registry"
-	"github.com/influxdata/telegraf/registry/inputs"
-	"github.com/influxdata/telegraf/registry/outputs"
+	"github.com/influxdata/telegraf/plugins/aggregators"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	"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"
@@ -56,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.
@@ -91,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'
@@ -196,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())
@@ -280,33 +290,62 @@ func loadExternalPlugins(rootDir string) error {
 		//     will be "group1/foo"
 		name := strings.TrimPrefix(strings.TrimPrefix(pth, rootDir), string(os.PathSeparator))
 		name = strings.TrimSuffix(name, filepath.Ext(pth))
-		registry.SetName("external" + string(os.PathSeparator) + name)
-		defer registry.SetName("")
+		name = "external" + string(os.PathSeparator) + name
 
 		// Load plugin.
-		_, err = plugin.Open(pth)
+		p, err := plugin.Open(pth)
 		if err != nil {
 			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())
 		}
 	}
 
@@ -329,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(
@@ -357,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(

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"]
-```
+```

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",

internal/config/config.go

@@ -18,14 +18,13 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/models"
+	"github.com/influxdata/telegraf/plugins/aggregators"
+	"github.com/influxdata/telegraf/plugins/inputs"
+	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/processors"
 	"github.com/influxdata/telegraf/plugins/serializers"
-	"github.com/influxdata/telegraf/registry/aggregators"
-	"github.com/influxdata/telegraf/registry/inputs"
-	"github.com/influxdata/telegraf/registry/outputs"
-	"github.com/influxdata/telegraf/registry/processors"
 
-	"github.com/influxdata/config"
 	"github.com/influxdata/toml"
 	"github.com/influxdata/toml/ast"
 )
@@ -514,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
 		}
@@ -574,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)
 			}
@@ -587,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)
 		}
@@ -727,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
 	}
 
@@ -747,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
 	}
 
@@ -787,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
 	}
 
@@ -828,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
 	}
 
@@ -920,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)
 			}
 		}
@@ -1157,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)
 			}
 		}

internal/config/config_test.go

@@ -6,11 +6,11 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf/internal/models"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/inputs/exec"
 	"github.com/influxdata/telegraf/plugins/inputs/memcached"
 	"github.com/influxdata/telegraf/plugins/inputs/procstat"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 
 	"github.com/stretchr/testify/assert"
 )

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

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"))

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)
+
+}

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)
 	}

plugins/aggregators/minmax/minmax.go

@@ -2,7 +2,7 @@ package minmax
 
 import (
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/aggregators"
+	"github.com/influxdata/telegraf/plugins/aggregators"
 )
 
 type MinMax struct {

plugins/aggregators/registry.go

@@ -0,0 +1,11 @@
+package aggregators
+
+import "github.com/influxdata/telegraf"
+
+type Creator func() telegraf.Aggregator
+
+var Aggregators = map[string]Creator{}
+
+func Add(name string, creator Creator) {
+	Aggregators[name] = creator
+}

plugins/inputs/aerospike/aerospike.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 
 	as "github.com/aerospike/aerospike-client-go"
 )

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"

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"
+```

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,
+		}
+	})
+}

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
 

plugins/inputs/apache/apache.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Apache struct {

plugins/inputs/bcache/bcache.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Bcache struct {

plugins/inputs/cassandra/cassandra.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"io/ioutil"
 	"log"
 	"net/http"

plugins/inputs/ceph/ceph.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 const (

plugins/inputs/cgroup/cgroup.go

@@ -2,7 +2,7 @@ package cgroup
 
 import (
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type CGroup struct {

plugins/inputs/chrony/chrony.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 var (

plugins/inputs/cloudwatch/cloudwatch.go

@@ -15,7 +15,7 @@ import (
 	internalaws "github.com/influxdata/telegraf/internal/config/aws"
 	"github.com/influxdata/telegraf/internal/errchan"
 	"github.com/influxdata/telegraf/internal/limiter"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type (

plugins/inputs/conntrack/conntrack.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"log"
 	"path/filepath"
 )

plugins/inputs/consul/consul.go

@@ -6,7 +6,7 @@ import (
 	"github.com/hashicorp/consul/api"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Consul struct {

plugins/inputs/couchbase/couchbase.go

@@ -3,7 +3,7 @@ package couchbase
 import (
 	couchbase "github.com/couchbase/go-couchbase"
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"sync"
 )
 

plugins/inputs/couchdb/couchdb.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"net/http"
 	"reflect"
 	"strings"

plugins/inputs/disque/disque.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Disque struct {

plugins/inputs/dns_query/dns_query.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type DnsQuery struct {

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:

plugins/inputs/docker/docker.go

@@ -1,6 +1,7 @@
-package system
+package docker
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -11,13 +12,12 @@ 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/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Docker object
@@ -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

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,

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
+}

plugins/inputs/dovecot/dovecot.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Dovecot struct {

plugins/inputs/elasticsearch/elasticsearch.go

@@ -11,8 +11,8 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/errchan"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	jsonparser "github.com/influxdata/telegraf/plugins/parsers/json"
-	"github.com/influxdata/telegraf/registry/inputs"
 	"io/ioutil"
 	"strings"
 )

plugins/inputs/exec/exec.go

@@ -16,9 +16,9 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/errchan"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/plugins/parsers/nagios"
-	"github.com/influxdata/telegraf/registry/inputs"
 )
 
 const sampleConfig = `

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{

plugins/inputs/filestat/filestat.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/globpath"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 const sampleConfig = `

plugins/inputs/graylog/graylog.go

@@ -16,7 +16,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type ResponseMetrics struct {

plugins/inputs/haproxy/haproxy.go

@@ -14,7 +14,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 //CSV format: https://cbonte.github.io/haproxy-dconv/1.5/configuration.html#9.1

plugins/inputs/hddtemp/hddtemp.go

@@ -4,8 +4,8 @@ package hddtemp
 
 import (
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	gohddtemp "github.com/influxdata/telegraf/plugins/inputs/hddtemp/go-hddtemp"
-	"github.com/influxdata/telegraf/registry/inputs"
 )
 
 const defaultAddress = "127.0.0.1:7634"

plugins/inputs/http_listener/http_listener.go

@@ -12,8 +12,8 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers/influx"
-	"github.com/influxdata/telegraf/registry/inputs"
 	"github.com/influxdata/telegraf/selfstat"
 )
 

plugins/inputs/http_response/http_response.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // HTTPResponse struct

plugins/inputs/httpjson/httpjson.go

@@ -12,8 +12,8 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 )
 
 // HttpJson struct

plugins/inputs/influxdb/influxdb.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type InfluxDB struct {

plugins/inputs/internal/internal.go

@@ -4,7 +4,7 @@ import (
 	"runtime"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/selfstat"
 )
 

plugins/inputs/ipmi_sensor/ipmi.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 var (
@@ -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

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)

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:
 

plugins/inputs/iptables/iptables.go

@@ -10,7 +10,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Iptables is a telegraf plugin to gather packets and bytes throughput from Linux's iptables packet filter.
@@ -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" ]
 `
 }

plugins/inputs/jolokia/jolokia.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Default http timeouts

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -6,8 +6,8 @@ import (
 	"sync"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 
 	"github.com/Shopify/sarama"
 	"github.com/wvanbergen/kafka/consumergroup"

plugins/inputs/kubernetes/kubernetes.go

@@ -12,7 +12,7 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Kubernetes represents the config object for the plugin

plugins/inputs/leofs/leofs.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 const oid = ".1.3.6.1.4.1.35450"

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",

plugins/inputs/logparser/logparser.go

@@ -11,7 +11,7 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
 	"github.com/influxdata/telegraf/internal/globpath"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 
 	// Parsers
 	"github.com/influxdata/telegraf/plugins/inputs/logparser/grok"
@@ -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())
 			}
 		}
 	}

plugins/inputs/lustre2/lustre2.go

@@ -15,7 +15,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Lustre proc files can change between versions, so we want to future-proof

plugins/inputs/mailchimp/mailchimp.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type MailChimp struct {

plugins/inputs/memcached/memcached.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Memcached is a memcached plugin

plugins/inputs/mesos/mesos.go

@@ -13,8 +13,8 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	jsonparser "github.com/influxdata/telegraf/plugins/parsers/json"
-	"github.com/influxdata/telegraf/registry/inputs"
 )
 
 type Role string

plugins/inputs/mongodb/mongodb.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"gopkg.in/mgo.v2"
 )
 

plugins/inputs/mqtt_consumer/mqtt_consumer.go

@@ -9,8 +9,8 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 
 	"github.com/eclipse/paho.mqtt.golang"
 )

plugins/inputs/mysql/mysql.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"database/sql"
 	"fmt"
+	"log"
 	"strconv"
 	"strings"
 	"sync"
@@ -11,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 
 	"github.com/go-sql-driver/mysql"
 )
@@ -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)
 		}
 	}
 

plugins/inputs/nats_consumer/nats_consumer.go

@@ -6,8 +6,8 @@ import (
 	"sync"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 	"github.com/nats-io/nats"
 )
 

plugins/inputs/net_response/net_response.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // NetResponses struct

plugins/inputs/nginx/nginx.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Nginx struct {

plugins/inputs/nsq/nsq.go

@@ -33,7 +33,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Might add Lookupd endpoints for cluster discovery

plugins/inputs/nsq_consumer/nsq_consumer.go

@@ -4,8 +4,8 @@ import (
 	"log"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
-	"github.com/influxdata/telegraf/registry/inputs"
 	"github.com/nsqio/go-nsq"
 )
 

plugins/inputs/nstat/nstat.go

@@ -7,7 +7,7 @@ import (
 	"strconv"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 var (

plugins/inputs/ntpq/ntpq.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Mapping of ntpq header names to tag keys

plugins/inputs/passenger/passenger.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"golang.org/x/net/html/charset"
 )
 

plugins/inputs/phpfpm/phpfpm.go

@@ -13,7 +13,7 @@ import (
 	"sync"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 const (

plugins/inputs/ping/ping.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // HostPinger is a function that runs the "ping" function using a list of

plugins/inputs/ping/ping_windows.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"os/exec"
 	"regexp"
 	"strconv"
@@ -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)

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)
 

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"]
+```

plugins/inputs/postgresql/postgresql.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Postgresql struct {
@@ -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"]
 `
 

plugins/inputs/postgresql_extensible/postgresql_extensible.go

@@ -8,8 +8,8 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/inputs/postgresql"
-	"github.com/influxdata/telegraf/registry/inputs"
 )
 
 type Postgresql struct {

plugins/inputs/powerdns/powerdns.go

@@ -11,7 +11,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Powerdns struct {

plugins/inputs/procstat/procstat.go

@@ -8,8 +8,10 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/shirou/gopsutil/process"
+
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Procstat struct {
@@ -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

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)

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
 }

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

plugins/inputs/prometheus/prometheus.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 	"io/ioutil"
 	"net"
 	"net/http"

plugins/inputs/puppetagent/puppetagent.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // PuppetAgent is a PuppetAgent plugin

plugins/inputs/rabbitmq/rabbitmq.go

@@ -11,7 +11,7 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // DefaultUsername will set a default value that corrasponds to the default

plugins/inputs/raindrops/raindrops.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Raindrops struct {

plugins/inputs/redis/redis.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/errchan"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Redis struct {

plugins/inputs/registry.go

@@ -0,0 +1,11 @@
+package inputs
+
+import "github.com/influxdata/telegraf"
+
+type Creator func() telegraf.Input
+
+var Inputs = map[string]Creator{}
+
+func Add(name string, creator Creator) {
+	Inputs[name] = creator
+}

plugins/inputs/rethinkdb/rethinkdb.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 
 	"gopkg.in/dancannon/gorethink.v1"
 )

plugins/inputs/riak/riak.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 // Type Riak gathers statistics from one or more Riak instances

plugins/inputs/sensors/sensors.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 var (

plugins/inputs/snmp/snmp.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
-	"github.com/influxdata/telegraf/registry/inputs"
+	"github.com/influxdata/telegraf/plugins/inputs"
 
 	"github.com/soniah/gosnmp"
 )