Use %q format when printing influxdb errors

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v1.8 [unreleased]
+
+### Features
+
+- [#4236](https://github.com/influxdata/telegraf/pull/4236): Add SSL/TLS support to redis input.
+
 ## v1.7 [unreleased]
 
 ### Release Notes
@@ -74,7 +80,7 @@
 - [#2468](https://github.com/influxdata/telegraf/issues/2468): Fix crash on 32-bit Windows in win_perf_counters.
 - [#4198](https://github.com/influxdata/telegraf/issues/4198): Fix win_perf_counters not collecting at every interval.
 
-## v1.6.4 [unreleased]
+## v1.6.4 [2018-06-05]
 
 ### Bugfixes
 

Godeps

@@ -28,7 +28,7 @@ github.com/golang/snappy 7db9049039a047d955fe8c19b83c8ff5abd765c7
 github.com/go-ole/go-ole be49f7c07711fcb603cff39e1de7c67926dc0ba7
 github.com/google/go-cmp f94e52cad91c65a63acc1e75d4be223ea22e99bc
 github.com/gorilla/mux 53c1911da2b537f792e7cafcb446b05ffe33b996
-github.com/go-redis/redis 73b70592cdaa9e6abdfcfbf97b4a90d80728c836
+github.com/go-redis/redis 83fb42932f6145ce52df09860384a4653d2d332a
 github.com/go-sql-driver/mysql 2e00b5cd70399450106cec6431c2e2ce3cae5034
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/hashicorp/consul 5174058f0d2bda63fa5198ab96c33d9a909c58ed

agent/agent.go

@@ -362,24 +362,6 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 	metricC := make(chan telegraf.Metric, 100)
 	aggC := make(chan telegraf.Metric, 100)
 
-	// Start all ServicePlugins
-	for _, input := range a.Config.Inputs {
-		input.SetDefaultTags(a.Config.Tags)
-		switch p := input.Input.(type) {
-		case telegraf.ServiceInput:
-			acc := NewAccumulator(input, metricC)
-			// Service input plugins should set their own precision of their
-			// metrics.
-			acc.SetPrecision(time.Nanosecond, 0)
-			if err := p.Start(acc); err != nil {
-				log.Printf("E! Service for input %s failed to start, exiting\n%s\n",
-					input.Name(), err.Error())
-				return err
-			}
-			defer p.Stop()
-		}
-	}
-
 	// Round collection to nearest interval by sleeping
 	if a.Config.Agent.RoundInterval {
 		i := int64(a.Config.Agent.Interval.Duration)
@@ -419,6 +401,25 @@ func (a *Agent) Run(shutdown chan struct{}) error {
 		}(input, interval)
 	}
 
+	// Start all ServicePlugins inputs after all other
+	// plugins are loaded so that no metrics get dropped
+	for _, input := range a.Config.Inputs {
+		input.SetDefaultTags(a.Config.Tags)
+		switch p := input.Input.(type) {
+		case telegraf.ServiceInput:
+			acc := NewAccumulator(input, metricC)
+			// Service input plugins should set their own precision of their
+			// metrics.
+			acc.SetPrecision(time.Nanosecond, 0)
+			if err := p.Start(acc); err != nil {
+				log.Printf("E! Service for input %s failed to start, exiting\n%s\n",
+					input.Name(), err.Error())
+				return err
+			}
+			defer p.Stop()
+		}
+	}
+
 	wg.Wait()
 	a.Close()
 	return nil

cmd/telegraf/telegraf.go

@@ -58,7 +58,7 @@ var fService = flag.String("service", "",
 var fRunAsConsole = flag.Bool("console", false, "run as console application (windows only)")
 
 var (
-	nextVersion = "1.7.0"
+	nextVersion = "1.8.0"
 	version     string
 	commit      string
 	branch      string

internal/models/running_aggregator.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"log"
 	"time"
 
 	"github.com/influxdata/telegraf"
@@ -153,6 +154,7 @@ func (r *RunningAggregator) Run(
 				m.Time().After(r.periodEnd.Add(truncation).Add(r.Config.Delay)) {
 				// the metric is outside the current aggregation period, so
 				// skip it.
+				log.Printf("D! aggregator: metric \"%s\" is not in the current timewindow, skipping", m.Name())
 				continue
 			}
 			r.add(m)

plugins/inputs/redis/README.md

@@ -14,6 +14,13 @@
   ## If no servers are specified, then localhost is used as the host.
   ## If no port is specified, 6379 is used
   servers = ["tcp://localhost:6379"]
+
+  ## Optional TLS Config
+  # tls_ca = "/etc/telegraf/ca.pem"
+  # tls_cert = "/etc/telegraf/cert.pem"
+  # tls_key = "/etc/telegraf/key.pem"
+  ## Use TLS but skip chain & host verification
+  # insecure_skip_verify = true
 ```
 
 ### Measurements & Fields:

plugins/inputs/redis/redis.go

@@ -13,11 +13,13 @@ import (
 
 	"github.com/go-redis/redis"
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Redis struct {
 	Servers []string
+	tls.ClientConfig
 
 	clients     []Client
 	initialized bool
@@ -56,6 +58,13 @@ var sampleConfig = `
   ## If no servers are specified, then localhost is used as the host.
   ## If no port is specified, 6379 is used
   servers = ["tcp://localhost:6379"]
+
+  ## Optional TLS Config
+  # tls_ca = "/etc/telegraf/ca.pem"
+  # tls_cert = "/etc/telegraf/cert.pem"
+  # tls_key = "/etc/telegraf/key.pem"
+  ## Use TLS but skip chain & host verification
+  # insecure_skip_verify = true
 `
 
 func (r *Redis) SampleConfig() string {
@@ -109,12 +118,18 @@ func (r *Redis) init(acc telegraf.Accumulator) error {
 			address = u.Host
 		}
 
+		tlsConfig, err := r.ClientConfig.TLSConfig()
+		if err != nil {
+			return err
+		}
+
 		client := redis.NewClient(
 			&redis.Options{
-				Addr:     address,
+				Addr:      address,
-				Password: password,
+				Password:  password,
-				Network:  u.Scheme,
+				Network:   u.Scheme,
-				PoolSize: 1,
+				PoolSize:  1,
+				TLSConfig: tlsConfig,
 			},
 		)
 

plugins/outputs/influxdb/influxdb.go

@@ -140,17 +140,17 @@ func (i *InfluxDB) Connect() error {
 		i.serializer.SetFieldTypeSupport(influx.UintSupport)
 	}
 
-	for _, u := range urls {
+	for _, loc := range urls {
-		u, err := url.Parse(u)
+		u, err := url.Parse(loc)
 		if err != nil {
-			return fmt.Errorf("error parsing url [%s]: %v", u, err)
+			return fmt.Errorf("error parsing url [%q]: %v", loc, err)
 		}
 
 		var proxy *url.URL
 		if len(i.HTTPProxy) > 0 {
 			proxy, err = url.Parse(i.HTTPProxy)
 			if err != nil {
-				return fmt.Errorf("error parsing proxy_url [%s]: %v", proxy, err)
+				return fmt.Errorf("error parsing proxy_url [%q]: %v", proxy, err)
 			}
 		}
 
@@ -170,7 +170,7 @@ func (i *InfluxDB) Connect() error {
 
 			i.clients = append(i.clients, c)
 		default:
-			return fmt.Errorf("unsupported scheme [%s]: %q", u, u.Scheme)
+			return fmt.Errorf("unsupported scheme [%q]: %q", u, u.Scheme)
 		}
 	}
 
@@ -209,14 +209,14 @@ func (i *InfluxDB) Write(metrics []telegraf.Metric) error {
 				if apiError.Type == DatabaseNotFound {
 					err := client.CreateDatabase(ctx)
 					if err != nil {
-						log.Printf("E! [outputs.influxdb] when writing to [%s]: database %q not found and failed to recreate",
+						log.Printf("E! [outputs.influxdb] when writing to [%q]: database %q not found and failed to recreate",
 							client.URL(), client.Database())
 					}
 				}
 			}
 		}
 
-		log.Printf("E! [outputs.influxdb]: when writing to [%s]: %v", client.URL(), err)
+		log.Printf("E! [outputs.influxdb]: when writing to [%q]: %v", client.URL(), err)
 	}
 
 	return errors.New("could not write any address")
@@ -231,7 +231,7 @@ func (i *InfluxDB) udpClient(url *url.URL) (Client, error) {
 
 	c, err := i.CreateUDPClientF(config)
 	if err != nil {
-		return nil, fmt.Errorf("error creating UDP client [%s]: %v", url, err)
+		return nil, fmt.Errorf("error creating UDP client [%q]: %v", url, err)
 	}
 
 	return c, nil
@@ -261,13 +261,13 @@ func (i *InfluxDB) httpClient(ctx context.Context, url *url.URL, proxy *url.URL)
 
 	c, err := i.CreateHTTPClientF(config)
 	if err != nil {
-		return nil, fmt.Errorf("error creating HTTP client [%s]: %v", url, err)
+		return nil, fmt.Errorf("error creating HTTP client [%q]: %v", url, err)
 	}
 
 	if !i.SkipDatabaseCreation {
 		err = c.CreateDatabase(ctx)
 		if err != nil {
-			log.Printf("W! [outputs.influxdb] when writing to [%s]: database %q creation failed: %v",
+			log.Printf("W! [outputs.influxdb] when writing to [%q]: database %q creation failed: %v",
 				c.URL(), c.Database(), err)
 		}
 	}

scripts/build.py

@@ -95,7 +95,7 @@ supported_packages = {
     "freebsd": [ "tar" ]
 }
 
-next_version = '1.7.0'
+next_version = '1.8.0'
 
 ################
 #### Telegraf Functions