Only split metrics if there is an udp output (#2799)

README.md

@@ -111,6 +111,7 @@ configuration options.
 * [couchbase](./plugins/inputs/couchbase)
 * [couchdb](./plugins/inputs/couchdb)
 * [disque](./plugins/inputs/disque)
+* [dmcache](./plugins/inputs/dmcache)
 * [dns query time](./plugins/inputs/dns_query)
 * [docker](./plugins/inputs/docker)
 * [dovecot](./plugins/inputs/dovecot)
@@ -127,6 +128,7 @@ configuration options.
 * [ipmi_sensor](./plugins/inputs/ipmi_sensor)
 * [iptables](./plugins/inputs/iptables)
 * [jolokia](./plugins/inputs/jolokia)
+* [kapacitor](./plugins/inputs/kapacitor)
 * [kubernetes](./plugins/inputs/kubernetes)
 * [leofs](./plugins/inputs/leofs)
 * [lustre2](./plugins/inputs/lustre2)
@@ -195,6 +197,7 @@ Telegraf can also collect metrics via the following service plugins:
   * [github](./plugins/inputs/webhooks/github)
   * [mandrill](./plugins/inputs/webhooks/mandrill)
   * [rollbar](./plugins/inputs/webhooks/rollbar)
+  * [papertrail](./plugins/inputs/webhooks/papertrail)
 
 Telegraf is able to parse the following input data formats into metrics, these
 formats may be used with input plugins supporting the `data_format` option:

plugins/outputs/influxdb/client/udp.go

@@ -25,6 +25,7 @@ type UDPConfig struct {
 	PayloadSize int
 }
 
+// NewUDP will return an instance of the telegraf UDP output plugin for influxdb
 func NewUDP(config UDPConfig) (Client, error) {
 	p, err := url.Parse(config.URL)
 	if err != nil {
@@ -55,20 +56,22 @@ type udpClient struct {
 	buffer []byte
 }
 
+// Query will send the provided query command to the client, returning an error if any issues arise
 func (c *udpClient) Query(command string) error {
 	return nil
 }
 
+// Write will send the byte stream to the given UDP client endpoint
 func (c *udpClient) Write(b []byte) (int, error) {
 	return c.WriteStream(bytes.NewReader(b), -1)
 }
 
-// write params are ignored by the UDP client
+// WriteWithParams are ignored by the UDP client, will forward to WriteStream
 func (c *udpClient) WriteWithParams(b []byte, wp WriteParams) (int, error) {
 	return c.WriteStream(bytes.NewReader(b), -1)
 }
 
-// contentLength is ignored by the UDP client.
+// WriteStream will send the provided data through to the client, contentLength is ignored by the UDP client
 func (c *udpClient) WriteStream(r io.Reader, contentLength int) (int, error) {
 	var totaln int
 	for {
@@ -88,12 +91,13 @@ func (c *udpClient) WriteStream(r io.Reader, contentLength int) (int, error) {
 	return totaln, nil
 }
 
-// contentLength is ignored by the UDP client.
+// WriteStreamWithParams will forward the stream to the client backend, contentLength is ignored by the UDP client
 // write params are ignored by the UDP client
 func (c *udpClient) WriteStreamWithParams(r io.Reader, contentLength int, wp WriteParams) (int, error) {
 	return c.WriteStream(r, -1)
 }
 
+// Close will terminate the provided client connection
 func (c *udpClient) Close() error {
 	return c.conn.Close()
 }

plugins/outputs/influxdb/influxdb.go

@@ -2,6 +2,7 @@ package influxdb
 
 import (
 	"fmt"
+	"io"
 	"log"
 	"math/rand"
 	"strings"
@@ -15,6 +16,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/outputs/influxdb/client"
 )
 
+// InfluxDB struct is the primary data structure for the plugin
 type InfluxDB struct {
 	// URL is only for backwards compatability
 	URL              string
@@ -40,7 +42,8 @@ type InfluxDB struct {
 	// Precision is only here for legacy support. It will be ignored.
 	Precision string
 
-	clients []client.Client
+	clients      []client.Client
+	splitPayload bool
 }
 
 var sampleConfig = `
@@ -79,11 +82,10 @@ var sampleConfig = `
   # insecure_skip_verify = false
 `
 
+// Connect initiates the primary connection to the range of provided URLs
 func (i *InfluxDB) Connect() error {
 	var urls []string
-	for _, u := range i.URLs {
-		urls = append(urls, u)
-	}
+	urls = append(urls, i.URLs...)
 
 	// Backward-compatability with single Influx URL config files
 	// This could eventually be removed in favor of specifying the urls as a list
@@ -109,6 +111,7 @@ func (i *InfluxDB) Connect() error {
 				return fmt.Errorf("Error creating UDP Client [%s]: %s", u, err)
 			}
 			i.clients = append(i.clients, c)
+			i.splitPayload = true
 		default:
 			// If URL doesn't start with "udp", assume HTTP client
 			config := client.HTTPConfig{
@@ -144,26 +147,41 @@ func (i *InfluxDB) Connect() error {
 	return nil
 }
 
+// Close will terminate the session to the backend, returning error if an issue arises
 func (i *InfluxDB) Close() error {
 	return nil
 }
 
+// SampleConfig returns the formatted sample configuration for the plugin
 func (i *InfluxDB) SampleConfig() string {
 	return sampleConfig
 }
 
+// Description returns the human-readable function definition of the plugin
 func (i *InfluxDB) Description() string {
 	return "Configuration for influxdb server to send metrics to"
 }
 
-// Choose a random server in the cluster to write to until a successful write
+func (i *InfluxDB) getReader(metrics []telegraf.Metric) io.Reader {
+	if !i.splitPayload {
+		return metric.NewReader(metrics)
+	}
+
+	splitData := make([]telegraf.Metric, 0)
+	for _, m := range metrics {
+		splitData = append(splitData, m.Split(i.UDPPayload)...)
+	}
+	return metric.NewReader(splitData)
+}
+
+// Write will choose a random server in the cluster to write to until a successful write
 // occurs, logging each unsuccessful. If all servers fail, return error.
 func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
 	bufsize := 0
 	for _, m := range metrics {
 		bufsize += m.Len()
 	}
-	r := metric.NewReader(metrics)
+	r := i.getReader(metrics)
 
 	// This will get set to nil if a successful write occurs
 	err := fmt.Errorf("Could not write to any InfluxDB server in cluster")

scripts/build.py

@@ -636,6 +636,10 @@ def package(build_output, pkg_name, version, nightly=False, iteration=1, static=
                     elif package_type not in ['zip', 'tar'] and static or "static_" in arch:
                         logging.info("Skipping package type '{}' for static builds.".format(package_type))
                     else:
+                        if package_type == 'rpm' and release and '~' in package_version:
+                            package_version, suffix = package_version.split('~', 1)
+                            # The ~ indicatees that this is a prerelease so we give it a leading 0.
+                            package_iteration = "0.%s" % suffix
                         fpm_command = "fpm {} --name {} -a {} -t {} --version {} --iteration {} -C {} -p {} ".format(
                             fpm_common_args,
                             name,
@@ -664,9 +668,6 @@ def package(build_output, pkg_name, version, nightly=False, iteration=1, static=
                                 if package_type == 'rpm':
                                     # rpm's convert any dashes to underscores
                                     package_version = package_version.replace("-", "_")
-                                new_outfile = outfile.replace("{}-{}".format(package_version, package_iteration), package_version)
-                                os.rename(outfile, new_outfile)
-                                outfile = new_outfile
                             outfiles.append(os.path.join(os.getcwd(), outfile))
         logging.debug("Produced package files: {}".format(outfiles))
         return outfiles

scripts/init.sh

@@ -135,7 +135,9 @@ case $1 in
         fi
 
         log_success_msg "Starting the process" "$name"
-        if which start-stop-daemon > /dev/null 2>&1; then
+        if command -v startproc >/dev/null; then
+            startproc -u "$USER" -g "$GROUP" -p "$pidfile" -q -- "$daemon" -pidfile "$pidfile" -config "$config" -config-directory "$confdir" $TELEGRAF_OPTS
+        elif which start-stop-daemon > /dev/null 2>&1; then
             start-stop-daemon --chuid $USER:$GROUP --start --quiet --pidfile $pidfile --exec $daemon -- -pidfile $pidfile -config $config -config-directory $confdir $TELEGRAF_OPTS >>$STDOUT 2>>$STDERR &
         else
             su -s /bin/sh -c "nohup $daemon -pidfile $pidfile -config $config -config-directory $confdir $TELEGRAF_OPTS >>$STDOUT 2>>$STDERR &" $USER

scripts/post-install.sh

@@ -11,7 +11,7 @@ function install_init {
 }
 
 function install_systemd {
-    cp -f $SCRIPT_DIR/telegraf.service /lib/systemd/system/telegraf.service
+    cp -f $SCRIPT_DIR/telegraf.service $1
     systemctl enable telegraf || true
     systemctl daemon-reload || true
 }
@@ -24,12 +24,12 @@ function install_chkconfig {
     chkconfig --add telegraf
 }
 
+if ! grep "^telegraf:" /etc/group &>/dev/null; then
+    groupadd -r telegraf
+fi
+
 if ! id telegraf &>/dev/null; then
-    if ! grep "^telegraf:" /etc/group &>/dev/null; then
-        useradd -r -K USERGROUPS_ENAB=yes -M telegraf -s /bin/false -d /etc/telegraf
-    else
-        useradd -r -K USERGROUPS_ENAB=yes -M telegraf -s /bin/false -d /etc/telegraf -g telegraf
-    fi
+    useradd -r -M telegraf -s /bin/false -d /etc/telegraf -g telegraf
 fi
 
 test -d $LOG_DIR || mkdir -p $LOG_DIR
@@ -56,10 +56,10 @@ if [[ ! -d /etc/telegraf/telegraf.d ]]; then
 fi
 
 # Distribution-specific logic
-if [[ -f /etc/redhat-release ]]; then
+if [[ -f /etc/redhat-release ]] || [[ -f /etc/SuSE-release ]]; then
     # RHEL-variant logic
     if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
-        install_systemd
+        install_systemd /usr/lib/systemd/system/telegraf.service
     else
         # Assuming SysVinit
         install_init
@@ -73,10 +73,10 @@ if [[ -f /etc/redhat-release ]]; then
 elif [[ -f /etc/debian_version ]]; then
     # Debian/Ubuntu logic
     if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
-        install_systemd
+        install_systemd /lib/systemd/system/telegraf.service
         systemctl restart telegraf || echo "WARNING: systemd not running."
     else
-	    # Assuming SysVinit
+        # Assuming SysVinit
         install_init
         # Run update-rc.d or fallback to chkconfig if not available
         if which update-rc.d &>/dev/null; then
@@ -89,7 +89,7 @@ elif [[ -f /etc/debian_version ]]; then
 elif [[ -f /etc/os-release ]]; then
     source /etc/os-release
     if [[ $ID = "amzn" ]]; then
-	    # Amazon Linux logic
+        # Amazon Linux logic
         install_init
         # Run update-rc.d or fallback to chkconfig if not available
         if which update-rc.d &>/dev/null; then
@@ -97,5 +97,6 @@ elif [[ -f /etc/os-release ]]; then
         else
             install_chkconfig
         fi
+        /etc/init.d/telegraf restart
     fi
 fi

scripts/post-remove.sh

@@ -2,7 +2,7 @@
 
 function disable_systemd {
     systemctl disable telegraf
-    rm -f /lib/systemd/system/telegraf.service
+    rm -f $1
 }
 
 function disable_update_rcd {
@@ -15,14 +15,14 @@ function disable_chkconfig {
     rm -f /etc/init.d/telegraf
 }
 
-if [[ -f /etc/redhat-release ]]; then
+if [[ -f /etc/redhat-release ]] || [[ -f /etc/SuSE-release ]]; then
     # RHEL-variant logic
     if [[ "$1" = "0" ]]; then
         # InfluxDB is no longer installed, remove from init system
         rm -f /etc/default/telegraf
 
         if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
-            disable_systemd
+            disable_systemd /usr/lib/systemd/system/telegraf.service
         else
             # Assuming sysv
             disable_chkconfig
@@ -35,7 +35,7 @@ elif [[ -f /etc/debian_version ]]; then
         rm -f /etc/default/telegraf
 
         if [[ "$(readlink /proc/1/exe)" == */systemd ]]; then
-            disable_systemd
+            disable_systemd /lib/systemd/system/telegraf.service
         else
             # Assuming sysv
             # Run update-rc.d or fallback to chkconfig if not available