Connection Loss Improvement

CHANGELOG.md

@@ -41,6 +41,7 @@ should now look like:
 - [#1405](https://github.com/influxdata/telegraf/issues/1405): Fix memory/connection leak in prometheus input plugin.
 - [#1378](https://github.com/influxdata/telegraf/issues/1378): Trim BOM from config file for Windows support.
 - [#1339](https://github.com/influxdata/telegraf/issues/1339): Prometheus client output panic on service reload.
+- [#1412](https://github.com/influxdata/telegraf/pull/1412): Instrumental output has better reconnect behavior
 
 ## v1.0 beta 2 [2016-06-21]
 

plugins/outputs/instrumental/instrumental.go

@@ -28,8 +28,10 @@ type Instrumental struct {
 }
 
 const (
-	DefaultHost = "collector.instrumentalapp.com"
-	AuthFormat  = "hello version go/telegraf/1.0\nauthenticate %s\n"
+	DefaultHost     = "collector.instrumentalapp.com"
+	HelloMessage    = "hello version go/telegraf/1.1\n"
+	AuthFormat      = "authenticate %s\n"
+	HandshakeFormat = HelloMessage + AuthFormat
 )
 
 var (
@@ -52,6 +54,7 @@ var sampleConfig = `
 
 func (i *Instrumental) Connect() error {
 	connection, err := net.DialTimeout("tcp", i.Host+":8000", i.Timeout.Duration)
+
 	if err != nil {
 		i.conn = nil
 		return err
@@ -151,6 +154,11 @@ func (i *Instrumental) Write(metrics []telegraf.Metric) error {
 		return err
 	}
 
+	// force the connection closed after sending data
+	// to deal with various disconnection scenarios and eschew holding
+	// open idle connections en masse
+	i.Close()
+
 	return nil
 }
 
@@ -163,7 +171,7 @@ func (i *Instrumental) SampleConfig() string {
 }
 
 func (i *Instrumental) authenticate(conn net.Conn) error {
-	_, err := fmt.Fprintf(conn, AuthFormat, i.ApiToken)
+	_, err := fmt.Fprintf(conn, HandshakeFormat, i.ApiToken)
 	if err != nil {
 		return err
 	}

plugins/outputs/instrumental/instrumental_test.go

@@ -24,7 +24,6 @@ func TestWrite(t *testing.T) {
 		ApiToken: "abc123token",
 		Prefix:   "my.prefix",
 	}
-	i.Connect()
 
 	// Default to gauge
 	m1, _ := telegraf.NewMetric(
@@ -40,10 +39,8 @@ func TestWrite(t *testing.T) {
 		time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
 	)
 
-	// Simulate a connection close and reconnect.
 	metrics := []telegraf.Metric{m1, m2}
 	i.Write(metrics)
-	i.Close()
 
 	// Counter and Histogram are increments
 	m3, _ := telegraf.NewMetric(
@@ -70,7 +67,6 @@ func TestWrite(t *testing.T) {
 	i.Write(metrics)
 
 	wg.Wait()
-	i.Close()
 }
 
 func TCPServer(t *testing.T, wg *sync.WaitGroup) {
@@ -82,10 +78,9 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
 	tp := textproto.NewReader(reader)
 
 	hello, _ := tp.ReadLine()
-	assert.Equal(t, "hello version go/telegraf/1.0", hello)
+	assert.Equal(t, "hello version go/telegraf/1.1", hello)
 	auth, _ := tp.ReadLine()
 	assert.Equal(t, "authenticate abc123token", auth)
-
 	conn.Write([]byte("ok\nok\n"))
 
 	data1, _ := tp.ReadLine()
@@ -99,10 +94,9 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
 	tp = textproto.NewReader(reader)
 
 	hello, _ = tp.ReadLine()
-	assert.Equal(t, "hello version go/telegraf/1.0", hello)
+	assert.Equal(t, "hello version go/telegraf/1.1", hello)
 	auth, _ = tp.ReadLine()
 	assert.Equal(t, "authenticate abc123token", auth)
-
 	conn.Write([]byte("ok\nok\n"))
 
 	data3, _ := tp.ReadLine()