processes: add 'unknown' procs (?)

closes #1156

CHANGELOG.md

@@ -48,7 +48,15 @@ based on _prefix_ in addition to globs. This means that a filter like
   - disque: `host -> disque_host`
   - rethinkdb: `host -> rethinkdb_host`
 
-- **Breaking Change**: The `win_perf_counters` input has been changed to sanitize field names, replacing `/Sec` and `/sec` with `_persec`, as well as spaces with underscores. This is needed because Graphite doesn't like slashes and spaces, and was failing to accept metrics that had them. The `/[sS]ec` -> `_persec` is just to make things clearer and uniform.
+- **Breaking Change**: The `win_perf_counters` input has been changed to
+sanitize field names, replacing `/Sec` and `/sec` with `_persec`, as well as
+spaces with underscores. This is needed because Graphite doesn't like slashes
+and spaces, and was failing to accept metrics that had them.
+The `/[sS]ec` -> `_persec` is just to make things clearer and uniform.
+
+- **Breaking Change**: snmp plugin. The `host` tag of the snmp plugin has been
+changed to the `snmp_host` tag.
+
 - The `disk` input plugin can now be configured with the `HOST_MOUNT_PREFIX` environment variable.
 This value is prepended to any mountpaths discovered before retrieving stats.
 It is not included on the report path. This is necessary for reporting host disk stats when running from within a container.

etc/telegraf.conf

@@ -638,8 +638,8 @@
 # 
 #   ## If no servers are specified, then default to 127.0.0.1:1936
 #   servers = ["http://myhaproxy.com:1936", "http://anotherhaproxy.com:1936"]
-#   ## Or you can also use local socket(not work yet)
-#   ## servers = ["socket://run/haproxy/admin.sock"]
+#   ## Or you can also use local socket
+#   ## servers = ["socket:/run/haproxy/admin.sock"]
 
 
 # # HTTP/HTTPS request given an address a method and a timeout

plugins/inputs/docker/docker.go

@@ -307,7 +307,11 @@ func gatherContainerStats(
 	for i, percpu := range stat.CPUStats.CPUUsage.PercpuUsage {
 		percputags := copyTags(tags)
 		percputags["cpu"] = fmt.Sprintf("cpu%d", i)
-		acc.AddFields("docker_container_cpu", map[string]interface{}{"usage_total": percpu}, percputags, now)
+		fields := map[string]interface{}{
+			"usage_total":  percpu,
+			"container_id": id,
+		}
+		acc.AddFields("docker_container_cpu", fields, percputags, now)
 	}
 
 	for network, netstats := range stat.Networks {
@@ -466,6 +470,8 @@ func parseSize(sizeStr string) (int64, error) {
 
 func init() {
 	inputs.Add("docker", func() telegraf.Input {
-		return &Docker{}
+		return &Docker{
+			Timeout: internal.Duration{Duration: time.Second * 5},
+		}
 	})
 }

plugins/inputs/docker/docker_test.go

@@ -111,13 +111,15 @@ func TestDockerGatherContainerStats(t *testing.T) {
 
 	cputags["cpu"] = "cpu0"
 	cpu0fields := map[string]interface{}{
-		"usage_total": uint64(1),
+		"usage_total":  uint64(1),
+		"container_id": "123456789",
 	}
 	acc.AssertContainsTaggedFields(t, "docker_container_cpu", cpu0fields, cputags)
 
 	cputags["cpu"] = "cpu1"
 	cpu1fields := map[string]interface{}{
-		"usage_total": uint64(1002),
+		"usage_total":  uint64(1002),
+		"container_id": "123456789",
 	}
 	acc.AssertContainsTaggedFields(t, "docker_container_cpu", cpu1fields, cputags)
 }
@@ -372,7 +374,8 @@ func TestDockerGatherInfo(t *testing.T) {
 	acc.AssertContainsTaggedFields(t,
 		"docker_container_cpu",
 		map[string]interface{}{
-			"usage_total": uint64(1231652),
+			"usage_total":  uint64(1231652),
+			"container_id": "b7dfbb9478a6ae55e237d4d74f8bbb753f0817192b5081334dc78476296e2173",
 		},
 		map[string]string{
 			"container_name":  "etcd2",

plugins/inputs/mqtt_consumer/README.md

@@ -21,8 +21,12 @@ The plugin expects messages in the
     "sensors/#",
   ]
 
-  ## Maximum number of metrics to buffer between collection intervals
-  metric_buffer = 100000
+  # if true, messages that can't be delivered while the subscriber is offline
+  # will be delivered when it comes back (such as on service restart).
+  # NOTE: if true, client_id MUST be set
+  persistent_session = false
+  # If empty, a random client ID will be generated.
+  client_id = ""
 
   ## username and password to connect MQTT server.
   # username = "telegraf"

plugins/inputs/snmp/snmp.go

@@ -26,9 +26,6 @@ type Snmp struct {
 	nameToOid   map[string]string
 	initNode    Node
 	subTableMap map[string]Subtable
-
-	// TODO change as unexportable
-	//OidInstanceMapping map[string]map[string]string
 }
 
 type Host struct {
@@ -53,6 +50,8 @@ type Host struct {
 	// array of processed oids
 	// to skip oid duplication
 	processedOids []string
+
+	OidInstanceMapping map[string]map[string]string
 }
 
 type Table struct {
@@ -116,9 +115,6 @@ type Node struct {
 	subnodes map[string]Node
 }
 
-// TODO move this var to snmp struct
-var OidInstanceMapping = make(map[string]map[string]string)
-
 var sampleConfig = `
   ## Use 'oids.txt' file to translate oids to names
   ## To generate 'oids.txt' you need to run:
@@ -396,7 +392,7 @@ func (s *Snmp) Gather(acc telegraf.Accumulator) error {
 		// TODO save mapping and computed oids
 		// to do it only the first time
 		// only if len(s.OidInstanceMapping) == 0
-		if len(OidInstanceMapping) >= 0 {
+		if len(host.OidInstanceMapping) >= 0 {
 			if err := host.SNMPMap(acc, s.nameToOid, s.subTableMap); err != nil {
 				log.Printf("SNMP Mapping error for host '%s': %s", host.Address, err)
 				continue
@@ -413,7 +409,14 @@ func (s *Snmp) Gather(acc telegraf.Accumulator) error {
 	return nil
 }
 
-func (h *Host) SNMPMap(acc telegraf.Accumulator, nameToOid map[string]string, subTableMap map[string]Subtable) error {
+func (h *Host) SNMPMap(
+	acc telegraf.Accumulator,
+	nameToOid map[string]string,
+	subTableMap map[string]Subtable,
+) error {
+	if h.OidInstanceMapping == nil {
+		h.OidInstanceMapping = make(map[string]map[string]string)
+	}
 	// Get snmp client
 	snmpClient, err := h.GetSNMPClient()
 	if err != nil {
@@ -523,11 +526,11 @@ func (h *Host) SNMPMap(acc telegraf.Accumulator, nameToOid map[string]string, su
 
 								// Building mapping table
 								mapping := map[string]string{strings.Trim(key, "."): string(variable.Value.([]byte))}
-								_, exists := OidInstanceMapping[table.oid]
+								_, exists := h.OidInstanceMapping[table.oid]
 								if exists {
-									OidInstanceMapping[table.oid][strings.Trim(key, ".")] = string(variable.Value.([]byte))
+									h.OidInstanceMapping[table.oid][strings.Trim(key, ".")] = string(variable.Value.([]byte))
 								} else {
-									OidInstanceMapping[table.oid] = mapping
+									h.OidInstanceMapping[table.oid] = mapping
 								}
 
 								// Add table oid in bulk oid list
@@ -720,7 +723,12 @@ func (h *Host) GetSNMPClient() (*gosnmp.GoSNMP, error) {
 	return snmpClient, nil
 }
 
-func (h *Host) HandleResponse(oids map[string]Data, result *gosnmp.SnmpPacket, acc telegraf.Accumulator, initNode Node) (string, error) {
+func (h *Host) HandleResponse(
+	oids map[string]Data,
+	result *gosnmp.SnmpPacket,
+	acc telegraf.Accumulator,
+	initNode Node,
+) (string, error) {
 	var lastOid string
 	for _, variable := range result.Variables {
 		lastOid = variable.Name
@@ -755,7 +763,7 @@ func (h *Host) HandleResponse(oids map[string]Data, result *gosnmp.SnmpPacket, a
 						strings.Split(string(variable.Name[1:]), "."))
 					// Set instance tag
 					// From mapping table
-					mapping, inMappingNoSubTable := OidInstanceMapping[oid_key]
+					mapping, inMappingNoSubTable := h.OidInstanceMapping[oid_key]
 					if inMappingNoSubTable {
 						// filter if the instance in not in
 						// OidInstanceMapping mapping map
@@ -784,7 +792,7 @@ func (h *Host) HandleResponse(oids map[string]Data, result *gosnmp.SnmpPacket, a
 						// Because the result oid is equal to inputs.snmp.get section
 						field_name = oid.Name
 					}
-					tags["host"], _, _ = net.SplitHostPort(h.Address)
+					tags["snmp_host"], _, _ = net.SplitHostPort(h.Address)
 					fields := make(map[string]interface{})
 					fields[string(field_name)] = variable.Value
 

plugins/inputs/snmp/snmp_test.go

@@ -102,8 +102,8 @@ func TestSNMPGet1(t *testing.T) {
 			"oid1": uint(543846),
 		},
 		map[string]string{
-			"unit": "octets",
-			"host": testutil.GetLocalHost(),
+			"unit":      "octets",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -140,8 +140,8 @@ func TestSNMPGet2(t *testing.T) {
 			"ifNumber": int(4),
 		},
 		map[string]string{
-			"instance": "0",
-			"host":     testutil.GetLocalHost(),
+			"instance":  "0",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -180,9 +180,9 @@ func TestSNMPGet3(t *testing.T) {
 			"ifSpeed": uint(10000000),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "1",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "1",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -222,9 +222,9 @@ func TestSNMPEasyGet4(t *testing.T) {
 			"ifSpeed": uint(10000000),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "1",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "1",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -234,8 +234,8 @@ func TestSNMPEasyGet4(t *testing.T) {
 			"ifNumber": int(4),
 		},
 		map[string]string{
-			"instance": "0",
-			"host":     testutil.GetLocalHost(),
+			"instance":  "0",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -275,9 +275,9 @@ func TestSNMPEasyGet5(t *testing.T) {
 			"ifSpeed": uint(10000000),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "1",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "1",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -287,8 +287,8 @@ func TestSNMPEasyGet5(t *testing.T) {
 			"ifNumber": int(4),
 		},
 		map[string]string{
-			"instance": "0",
-			"host":     testutil.GetLocalHost(),
+			"instance":  "0",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -320,8 +320,8 @@ func TestSNMPEasyGet6(t *testing.T) {
 			"ifNumber": int(4),
 		},
 		map[string]string{
-			"instance": "0",
-			"host":     testutil.GetLocalHost(),
+			"instance":  "0",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -360,9 +360,9 @@ func TestSNMPBulk1(t *testing.T) {
 			"ifOutOctets": uint(543846),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "1",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "1",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -372,9 +372,9 @@ func TestSNMPBulk1(t *testing.T) {
 			"ifOutOctets": uint(26475179),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "2",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "2",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -384,9 +384,9 @@ func TestSNMPBulk1(t *testing.T) {
 			"ifOutOctets": uint(108963968),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "3",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "3",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -396,9 +396,9 @@ func TestSNMPBulk1(t *testing.T) {
 			"ifOutOctets": uint(12991453),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "36",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "36",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }
@@ -438,9 +438,9 @@ func dTestSNMPBulk2(t *testing.T) {
 			"ifOutOctets": uint(543846),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "1",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "1",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -450,9 +450,9 @@ func dTestSNMPBulk2(t *testing.T) {
 			"ifOutOctets": uint(26475179),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "2",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "2",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -462,9 +462,9 @@ func dTestSNMPBulk2(t *testing.T) {
 			"ifOutOctets": uint(108963968),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "3",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "3",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 
@@ -474,9 +474,9 @@ func dTestSNMPBulk2(t *testing.T) {
 			"ifOutOctets": uint(12991453),
 		},
 		map[string]string{
-			"unit":     "octets",
-			"instance": "36",
-			"host":     testutil.GetLocalHost(),
+			"unit":      "octets",
+			"instance":  "36",
+			"snmp_host": testutil.GetLocalHost(),
 		},
 	)
 }

plugins/inputs/system/processes.go

@@ -70,6 +70,7 @@ func getEmptyFields() map[string]interface{} {
 		"running":  int64(0),
 		"sleeping": int64(0),
 		"total":    int64(0),
+		"unknown":  int64(0),
 	}
 	switch runtime.GOOS {
 	case "freebsd":
@@ -114,6 +115,8 @@ func (p *Processes) gatherFromPS(fields map[string]interface{}) error {
 			fields["sleeping"] = fields["sleeping"].(int64) + int64(1)
 		case 'I':
 			fields["idle"] = fields["idle"].(int64) + int64(1)
+		case '?':
+			fields["unknown"] = fields["unknown"].(int64) + int64(1)
 		default:
 			log.Printf("processes: Unknown state [ %s ] from ps",
 				string(status[0]))

scripts/build.py

@@ -132,13 +132,16 @@ def create_package_fs(build_root):
         os.makedirs(os.path.join(build_root, d))
         os.chmod(os.path.join(build_root, d), 0o755)
 
-def package_scripts(build_root, windows=False):
+def package_scripts(build_root, config_only=False, windows=False):
     """Copy the necessary scripts and configuration files to the package
     filesystem.
     """
-    if windows:
-        logging.info("Copying configuration to build directory.")
-        shutil.copyfile(DEFAULT_WINDOWS_CONFIG, os.path.join(build_root, "telegraf.conf"))
+    if config_only or windows:
+        logging.info("Copying configuration to build directory")
+        if windows:
+            shutil.copyfile(DEFAULT_WINDOWS_CONFIG, os.path.join(build_root, "telegraf.conf"))
+        else:
+            shutil.copyfile(DEFAULT_CONFIG, os.path.join(build_root, "telegraf.conf"))
         os.chmod(os.path.join(build_root, "telegraf.conf"), 0o644)
     else:
         logging.info("Copying scripts and configuration to build directory")
@@ -239,21 +242,15 @@ def get_current_version():
     """Parse version information from git tag output.
     """
     version_tag = get_current_version_tag()
-    # Remove leading 'v' and possible '-rc\d+'
+    # Remove leading 'v'
     if version_tag[0] == 'v':
         version_tag = version_tag[1:]
-    version = re.sub(r'-rc\d+', '', str(version_tag))
-    return version
-
-def get_current_rc():
-    """Parse release candidate from git tag output.
-    """
-    rc = None
-    version_tag = get_current_version_tag()
-    matches = re.match(r'.*-rc(\d+)', str(version_tag))
-    if matches:
-        rc, = matches.groups(1)
-    return rc
+    # Replace any '-'/'_' with '~'
+    if '-' in version_tag:
+        version_tag = version_tag.replace("-","~")
+    if '_' in version_tag:
+        version_tag = version_tag.replace("_","~")
+    return version_tag
 
 def get_current_commit(short=False):
     """Retrieve the current git commit.
@@ -418,7 +415,6 @@ def build(version=None,
           platform=None,
           arch=None,
           nightly=False,
-          rc=None,
           race=False,
           clean=False,
           outdir=".",
@@ -445,9 +441,6 @@ def build(version=None,
         shutil.rmtree(outdir)
         os.makedirs(outdir)
 
-    if rc:
-        # If a release candidate, update the version information accordingly
-        version = "{}rc{}".format(version, rc)
     logging.info("Using version '{}' for build.".format(version))
 
     tmp_build_dir = create_temp_dir()
@@ -540,7 +533,7 @@ def generate_sig_from_file(path):
         run('gpg --armor --detach-sign --yes {}'.format(path))
     return True
 
-def package(build_output, version, nightly=False, rc=None, iteration=1, static=False, release=False):
+def package(build_output, pkg_name, version, nightly=False, iteration=1, static=False, release=False):
     """Package the output of the build process.
     """
     outfiles = []
@@ -564,11 +557,13 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
                 os.makedirs(build_root)
 
                 # Copy packaging scripts to build directory
-                if platform == "windows" or static or "static_" in arch:
+                if platform == "windows":
                     # For windows and static builds, just copy
                     # binaries to root of package (no other scripts or
                     # directories)
-                    package_scripts(build_root, windows=True)
+                    package_scripts(build_root, config_only=True, windows=True)
+                elif static or "static_" in arch:
+                    package_scripts(build_root, config_only=True)
                 else:
                     create_package_fs(build_root)
                     package_scripts(build_root)
@@ -592,7 +587,7 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
                 for package_type in supported_packages[platform]:
                     # Package the directory structure for each package type for the platform
                     logging.debug("Packaging directory '{}' as '{}'.".format(build_root, package_type))
-                    name = PACKAGE_NAME
+                    name = pkg_name
                     # Reset version, iteration, and current location on each run
                     # since they may be modified below.
                     package_version = version
@@ -604,17 +599,12 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
                         package_arch = arch
                     if not release and not nightly:
                         # For non-release builds, just use the commit hash as the version
-                        package_version = "{}~{}.{}".format(version,
-                                                            get_current_branch(),
-                                                            get_current_commit(short=True))
+                        package_version = "{}~{}".format(version,
+                                                         get_current_commit(short=True))
                         package_iteration = "0"
                     package_build_root = build_root
                     current_location = build_output[platform][arch]
 
-                    if rc is not None and release:
-                        # Set iteration to 0 since it's a release candidate
-                        package_iteration = "0.rc{}".format(rc)
-
                     if package_type in ['zip', 'tar']:
                         # For tars and zips, start the packaging one folder above
                         # the build root (to include the package name)
@@ -639,18 +629,17 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
                                                             package_version,
                                                             platform,
                                                             package_arch)
-
                         current_location = os.path.join(os.getcwd(), current_location)
                         if package_type == 'tar':
-                            tar_command = "cd {} && tar -cvzf {}.tar.gz ./*".format(build_root, name)
+                            tar_command = "cd {} && tar -cvzf {}.tar.gz ./*".format(package_build_root, name)
                             run(tar_command, shell=True)
-                            run("mv {}.tar.gz {}".format(os.path.join(build_root, name), current_location), shell=True)
+                            run("mv {}.tar.gz {}".format(os.path.join(package_build_root, name), current_location), shell=True)
                             outfile = os.path.join(current_location, name + ".tar.gz")
                             outfiles.append(outfile)
                         elif package_type == 'zip':
-                            zip_command = "cd {} && zip -r {}.zip ./*".format(build_root, name)
+                            zip_command = "cd {} && zip -r {}.zip ./*".format(package_build_root, name)
                             run(zip_command, shell=True)
-                            run("mv {}.zip {}".format(os.path.join(build_root, name), current_location), shell=True)
+                            run("mv {}.zip {}".format(os.path.join(package_build_root, name), current_location), shell=True)
                             outfile = os.path.join(current_location, name + ".zip")
                             outfiles.append(outfile)
                     elif package_type not in ['zip', 'tar'] and static or "static_" in arch:
@@ -681,7 +670,6 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
                                 os.rename(outfile, new_outfile)
                                 outfile = new_outfile
                             else:
-                                # Strip iteration from package name
                                 if package_type == 'rpm':
                                     # rpm's convert any dashes to underscores
                                     package_version = package_version.replace("-", "_")
@@ -698,9 +686,6 @@ def package(build_output, version, nightly=False, rc=None, iteration=1, static=F
 def main(args):
     global PACKAGE_NAME
 
-    if args.nightly and args.rc:
-        logging.error("Cannot be both a nightly and a release candidate.")
-        return 1
     if args.release and args.nightly:
         logging.error("Cannot be both a nightly and a release.")
         return 1
@@ -710,8 +695,6 @@ def main(args):
         args.version = "{}~n{}".format(args.version,
                                        datetime.utcnow().strftime("%Y%m%d%H%M"))
         args.iteration = 0
-    elif args.rc:
-        args.iteration = 0
 
     # Pre-build checks
     check_environ()
@@ -778,7 +761,6 @@ def main(args):
                          platform=platform,
                          arch=arch,
                          nightly=args.nightly,
-                         rc=args.rc,
                          race=args.race,
                          clean=args.clean,
                          outdir=od,
@@ -793,9 +775,9 @@ def main(args):
             logging.error("FPM ruby gem required for packaging. Stopping.")
             return 1
         packages = package(build_output,
+                           args.name,
                            args.version,
                            nightly=args.nightly,
-                           rc=args.rc,
                            iteration=args.iteration,
                            static=args.static,
                            release=args.release)
@@ -844,6 +826,7 @@ if __name__ == '__main__':
                         help='Output directory')
     parser.add_argument('--name', '-n',
                         metavar='<name>',
+                        default=PACKAGE_NAME,
                         type=str,
                         help='Name to use for package name (when package is specified)')
     parser.add_argument('--arch',
@@ -871,14 +854,10 @@ if __name__ == '__main__':
                         type=str,
                         default=get_current_version(),
                         help='Version information to apply to build output (ex: 0.12.0)')
-    parser.add_argument('--rc',
-                        metavar='<release candidate>',
-                        type=int,
-                        help='Release Candidate (RC) version to apply to build output')
     parser.add_argument('--iteration',
                         metavar='<package iteration>',
-                        type=int,
-                        default=1,
+                        type=str,
+                        default="1",
                         help='Package iteration to apply to build output (defaults to 1)')
     parser.add_argument('--stats',
                         action='store_true',