remove sleep from tests (influxdata#2555)

plugins/inputs/cloudwatch/cloudwatch_test.go

@@ -207,14 +207,13 @@ func TestGenerateStatisticsInputParams(t *testing.T) {
 }
 
 func TestMetricsCacheTimeout(t *testing.T) {
-	ttl, _ := time.ParseDuration("5ms")
 	cache := &MetricCache{
 		Metrics: []*cloudwatch.Metric{},
 		Fetched: time.Now(),
-		TTL:     ttl,
+		TTL:     time.Minute,
 	}
 
 	assert.True(t, cache.IsValid())
-	time.Sleep(ttl)
+	cache.Fetched = time.Now().Add(-time.Minute)
 	assert.False(t, cache.IsValid())
 }

plugins/inputs/http_listener/http_listener_test.go

@@ -6,7 +6,6 @@ import (
 	"net/http"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/influxdata/telegraf/testutil"
 
@@ -43,14 +42,12 @@ func TestWriteHTTP(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post single message to listener
 	resp, err := http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(testMsg)))
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
+	acc.Wait(1)
 	acc.AssertContainsTaggedFields(t, "cpu_load_short",
 		map[string]interface{}{"value": float64(12)},
 		map[string]string{"host": "server01"},
@@ -61,7 +58,7 @@ func TestWriteHTTP(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
+	acc.Wait(2)
 	hostTags := []string{"server02", "server03",
 		"server04", "server05", "server06"}
 	for _, hostTag := range hostTags {
@@ -76,7 +73,7 @@ func TestWriteHTTP(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, 400, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
+	acc.Wait(3)
 	acc.AssertContainsTaggedFields(t, "cpu_load_short",
 		map[string]interface{}{"value": float64(12)},
 		map[string]string{"host": "server01"},
@@ -91,14 +88,12 @@ func TestWriteHTTPNoNewline(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post single message to listener
 	resp, err := http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(testMsgNoNewline)))
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
+	acc.Wait(1)
 	acc.AssertContainsTaggedFields(t, "cpu_load_short",
 		map[string]interface{}{"value": float64(12)},
 		map[string]string{"host": "server01"},
@@ -115,8 +110,6 @@ func TestWriteHTTPMaxLineSizeIncrease(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// Post a gigantic metric to the listener and verify that it writes OK this time:
 	resp, err := http.Post("http://localhost:8296/write?db=mydb", "", bytes.NewBuffer([]byte(hugeMetric)))
 	require.NoError(t, err)
@@ -133,8 +126,6 @@ func TestWriteHTTPVerySmallMaxBody(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	resp, err := http.Post("http://localhost:8297/write", "", bytes.NewBuffer([]byte(hugeMetric)))
 	require.NoError(t, err)
 	require.EqualValues(t, 413, resp.StatusCode)
@@ -150,15 +141,13 @@ func TestWriteHTTPVerySmallMaxLineSize(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	resp, err := http.Post("http://localhost:8298/write", "", bytes.NewBuffer([]byte(testMsgs)))
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
 	hostTags := []string{"server02", "server03",
 		"server04", "server05", "server06"}
+	acc.Wait(len(hostTags))
 	for _, hostTag := range hostTags {
 		acc.AssertContainsTaggedFields(t, "cpu_load_short",
 			map[string]interface{}{"value": float64(12)},
@@ -177,15 +166,13 @@ func TestWriteHTTPLargeLinesSkipped(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	resp, err := http.Post("http://localhost:8300/write", "", bytes.NewBuffer([]byte(hugeMetric+testMsgs)))
 	require.NoError(t, err)
 	require.EqualValues(t, 400, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 15)
 	hostTags := []string{"server02", "server03",
 		"server04", "server05", "server06"}
+	acc.Wait(len(hostTags))
 	for _, hostTag := range hostTags {
 		acc.AssertContainsTaggedFields(t, "cpu_load_short",
 			map[string]interface{}{"value": float64(12)},
@@ -204,8 +191,6 @@ func TestWriteHTTPGzippedData(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	data, err := ioutil.ReadFile("./testdata/testmsgs.gz")
 	require.NoError(t, err)
 
@@ -218,9 +203,9 @@ func TestWriteHTTPGzippedData(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 
-	time.Sleep(time.Millisecond * 50)
 	hostTags := []string{"server02", "server03",
 		"server04", "server05", "server06"}
+	acc.Wait(len(hostTags))
 	for _, hostTag := range hostTags {
 		acc.AssertContainsTaggedFields(t, "cpu_load_short",
 			map[string]interface{}{"value": float64(12)},
@@ -237,8 +222,6 @@ func TestWriteHTTPHighTraffic(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post many messages to listener
 	var wg sync.WaitGroup
 	for i := 0; i < 10; i++ {
@@ -254,9 +237,9 @@ func TestWriteHTTPHighTraffic(t *testing.T) {
 	}
 
 	wg.Wait()
-	time.Sleep(time.Millisecond * 250)
 	listener.Gather(acc)
 
+	acc.Wait(25000)
 	require.Equal(t, int64(25000), int64(acc.NMetrics()))
 }
 
@@ -267,59 +250,45 @@ func TestReceive404ForInvalidEndpoint(t *testing.T) {
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post single message to listener
 	resp, err := http.Post("http://localhost:8186/foobar", "", bytes.NewBuffer([]byte(testMsg)))
 	require.NoError(t, err)
 	require.EqualValues(t, 404, resp.StatusCode)
 }
 
 func TestWriteHTTPInvalid(t *testing.T) {
-	time.Sleep(time.Millisecond * 250)
-
 	listener := newTestHTTPListener()
 
 	acc := &testutil.Accumulator{}
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post single message to listener
 	resp, err := http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(badMsg)))
 	require.NoError(t, err)
 	require.EqualValues(t, 400, resp.StatusCode)
 }
 
 func TestWriteHTTPEmpty(t *testing.T) {
-	time.Sleep(time.Millisecond * 250)
-
 	listener := newTestHTTPListener()
 
 	acc := &testutil.Accumulator{}
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post single message to listener
 	resp, err := http.Post("http://localhost:8186/write?db=mydb", "", bytes.NewBuffer([]byte(emptyMsg)))
 	require.NoError(t, err)
 	require.EqualValues(t, 204, resp.StatusCode)
 }
 
 func TestQueryAndPingHTTP(t *testing.T) {
-	time.Sleep(time.Millisecond * 250)
-
 	listener := newTestHTTPListener()
 
 	acc := &testutil.Accumulator{}
 	require.NoError(t, listener.Start(acc))
 	defer listener.Stop()
 
-	time.Sleep(time.Millisecond * 25)
-
 	// post query to listener
 	resp, err := http.Post("http://localhost:8186/query?db=&q=CREATE+DATABASE+IF+NOT+EXISTS+%22mydb%22", "", nil)
 	require.NoError(t, err)

plugins/inputs/http_response/http_response_test.go

@@ -329,7 +329,7 @@ func TestTimeout(t *testing.T) {
 		Address:         ts.URL + "/twosecondnap",
 		Body:            "{ 'test': 'data'}",
 		Method:          "GET",
-		ResponseTimeout: internal.Duration{Duration: time.Second * 1},
+		ResponseTimeout: internal.Duration{Duration: time.Millisecond},
 		Headers: map[string]string{
 			"Content-Type": "application/json",
 		},

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -1,6 +1,7 @@
 package kafka_consumer
 
 import (
+	"fmt"
 	"log"
 	"strings"
 	"sync"
@@ -129,13 +130,13 @@ func (k *Kafka) receiver() {
 			return
 		case err := <-k.errs:
 			if err != nil {
-				log.Printf("E! Kafka Consumer Error: %s\n", err)
+				k.acc.AddError(fmt.Errorf("Kafka Consumer Error: %s\n", err))
 			}
 		case msg := <-k.in:
 			metrics, err := k.parser.Parse(msg.Value)
 			if err != nil {
-				log.Printf("E! Kafka Message Parse Error\nmessage: %s\nerror: %s",
-					string(msg.Value), err.Error())
+				k.acc.AddError(fmt.Errorf("E! Kafka Message Parse Error\nmessage: %s\nerror: %s",
+					string(msg.Value), err.Error()))
 			}
 
 			for _, metric := range metrics {
@@ -158,7 +159,7 @@ func (k *Kafka) Stop() {
 	defer k.Unlock()
 	close(k.done)
 	if err := k.Consumer.Close(); err != nil {
-		log.Printf("E! Error closing kafka consumer: %s\n", err.Error())
+		k.acc.AddError(fmt.Errorf("E! Error closing kafka consumer: %s\n", err.Error()))
 	}
 }
 

plugins/inputs/kafka_consumer/kafka_consumer_test.go

@@ -2,7 +2,6 @@ package kafka_consumer
 
 import (
 	"testing"
-	"time"
 
 	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/testutil"
@@ -43,7 +42,7 @@ func TestRunParser(t *testing.T) {
 	k.parser, _ = parsers.NewInfluxParser()
 	go k.receiver()
 	in <- saramaMsg(testMsg)
-	time.Sleep(time.Millisecond * 5)
+	acc.Wait(1)
 
 	assert.Equal(t, acc.NFields(), 1)
 }
@@ -58,7 +57,7 @@ func TestRunParserInvalidMsg(t *testing.T) {
 	k.parser, _ = parsers.NewInfluxParser()
 	go k.receiver()
 	in <- saramaMsg(invalidMsg)
-	time.Sleep(time.Millisecond * 5)
+	acc.WaitError(1)
 
 	assert.Equal(t, acc.NFields(), 0)
 }
@@ -73,7 +72,7 @@ func TestRunParserAndGather(t *testing.T) {
 	k.parser, _ = parsers.NewInfluxParser()
 	go k.receiver()
 	in <- saramaMsg(testMsg)
-	time.Sleep(time.Millisecond * 5)
+	acc.Wait(1)
 
 	k.Gather(&acc)
 
@@ -92,7 +91,7 @@ func TestRunParserAndGatherGraphite(t *testing.T) {
 	k.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil)
 	go k.receiver()
 	in <- saramaMsg(testMsgGraphite)
-	time.Sleep(time.Millisecond * 5)
+	acc.Wait(1)
 
 	k.Gather(&acc)
 
@@ -111,7 +110,7 @@ func TestRunParserAndGatherJSON(t *testing.T) {
 	k.parser, _ = parsers.NewJSONParser("kafka_json_test", []string{}, nil)
 	go k.receiver()
 	in <- saramaMsg(testMsgJSON)
-	time.Sleep(time.Millisecond * 5)
+	acc.Wait(1)
 
 	k.Gather(&acc)
 

plugins/inputs/logparser/logparser_test.go

@@ -6,7 +6,6 @@ import (
 	"runtime"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/influxdata/telegraf/testutil"
 
@@ -41,7 +40,6 @@ func TestGrokParseLogFilesNonExistPattern(t *testing.T) {
 	acc := testutil.Accumulator{}
 	assert.Error(t, logparser.Start(&acc))
 
-	time.Sleep(time.Millisecond * 500)
 	logparser.Stop()
 }
 
@@ -61,7 +59,8 @@ func TestGrokParseLogFiles(t *testing.T) {
 	acc := testutil.Accumulator{}
 	assert.NoError(t, logparser.Start(&acc))
 
-	time.Sleep(time.Millisecond * 500)
+	acc.Wait(2)
+
 	logparser.Stop()
 
 	acc.AssertContainsTaggedFields(t, "logparser_grok",
@@ -102,14 +101,13 @@ func TestGrokParseLogFilesAppearLater(t *testing.T) {
 	acc := testutil.Accumulator{}
 	assert.NoError(t, logparser.Start(&acc))
 
-	time.Sleep(time.Millisecond * 500)
 	assert.Equal(t, acc.NFields(), 0)
 
 	os.Symlink(
 		thisdir+"grok/testdata/test_a.log",
 		emptydir+"/test_a.log")
 	assert.NoError(t, logparser.Gather(&acc))
-	time.Sleep(time.Millisecond * 500)
+	acc.Wait(1)
 
 	logparser.Stop()
 
@@ -143,7 +141,7 @@ func TestGrokParseLogFilesOneBad(t *testing.T) {
 	acc.SetDebug(true)
 	assert.NoError(t, logparser.Start(&acc))
 
-	time.Sleep(time.Millisecond * 500)
+	acc.Wait(1)
 	logparser.Stop()
 
 	acc.AssertContainsTaggedFields(t, "logparser_grok",

plugins/inputs/mongodb/mongodb_server_test.go

@@ -4,7 +4,6 @@ package mongodb
 
 import (
 	"testing"
-	"time"
 
 	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/assert"
@@ -32,12 +31,11 @@ func TestAddDefaultStats(t *testing.T) {
 	err := server.gatherData(&acc, false)
 	require.NoError(t, err)
 
-	time.Sleep(time.Duration(1) * time.Second)
 	// need to call this twice so it can perform the diff
 	err = server.gatherData(&acc, false)
 	require.NoError(t, err)
 
 	for key, _ := range DefaultStats {
-		assert.True(t, acc.HasIntValue(key))
+		assert.True(t, acc.HasIntField("mongodb", key))
 	}
 }