Tweaks to test timings to put a dent in Travis failures.

Author: rafrombrc

.travis.yml

@@ -1,15 +1,18 @@
+sudo: false
 language: go
 go:
     - 1.4
 notifications:
     irc:
         channels:
             - "irc.mozilla.org#heka"
-before_install:
-    - sudo apt-get install protobuf-compiler cmake libgeoip-dev
-
+addons:
+    apt:
+        packages:
+        - protobuf-compiler
+        - cmake
+        - libgeoip-dev
 install:
     - . build.sh
-
 script:
     - make test

CHANGES.txt

@@ -31,6 +31,9 @@ Bug Handling
 * Fixed bug where a SandboxInput configured with a `ticker_interval` would
   get stuck in an infinite loop on shutdown (#1705).
 
+* Changes in StatAccumInput and FileOutput tests to minimize intermittent
+  Travis failures.
+
 0.10.0b1 (2015-08-07)
 =====================
 

pipeline/stat_accum_input_test.go

@@ -15,6 +15,7 @@
 package pipeline
 
 import (
+	"log"
 	"strconv"
 	"strings"
 	"sync"
@@ -94,7 +95,8 @@ func StatAccumInputSpec(c gs.Context) {
 			}
 
 			ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply)
-			ith.MockInputRunner.EXPECT().Name().Return("StatAccumInput").AnyTimes()
+			// Need one of these for every Inject
+			ith.MockInputRunner.EXPECT().Name().Return("StatAccumInput")
 
 			injectCall := ith.MockInputRunner.EXPECT().Inject(ith.Pack)
 			var injectCalled sync.WaitGroup
@@ -131,9 +133,26 @@ func StatAccumInputSpec(c gs.Context) {
 					}
 				}
 
+				drainStats := func() {
+					ok := true
+					for ok {
+						if len(statAccumInput.statChan) > 0 {
+							time.Sleep(100 * time.Millisecond)
+						} else {
+							ok = false
+						}
+					}
+				}
+
 				validateValueAtKey := func(msg *message.Message, key string, value interface{}) {
 					fieldValue, ok := msg.GetFieldValue(key)
+					if !ok {
+						log.Printf("%s field is missing from the message\n", key)
+					}
 					c.Expect(ok, gs.IsTrue)
+					if fieldValue != value {
+						log.Printf("%s should be %v is %v\n", key, value, fieldValue)
+					}
 					c.Expect(fieldValue, gs.Equals, value)
 				}
 
@@ -203,15 +222,17 @@ func StatAccumInputSpec(c gs.Context) {
 
 				c.Specify("emits proper idle stats", func() {
 					startInput()
+					inputStarted.Wait()
 					sendGauge("sample.gauge", 1, 2)
 					sendCounter("sample.cnt", 1, 2, 3, 4, 5)
 					sendTimer("sample.timer", 10, 10, 20, 20)
-					inputStarted.Wait()
+					drainStats()
 					tickChan <- time.Now()
 
 					injectCalled.Wait()
 					ith.Pack.Recycle(nil)
 					ith.PackSupply <- ith.Pack
+					ith.MockInputRunner.EXPECT().Name().Return("StatAccumInput")
 					ith.MockInputRunner.EXPECT().Inject(ith.Pack)
 
 					msg, err := finalizeSendingStats()
@@ -225,19 +246,22 @@ func StatAccumInputSpec(c gs.Context) {
 				c.Specify("omits idle stats", func() {
 					config.DeleteIdleStats = true
 					err := statAccumInput.Init(config)
-					c.Expect(err, gs.IsNil)
+					c.Assume(err, gs.IsNil)
 
 					startInput()
+					inputStarted.Wait() // Can't flush until the input has started.
 					sendGauge("sample.gauge", 1, 2)
 					sendCounter("sample.cnt", 1, 2, 3, 4, 5)
 					sendTimer("sample.timer", 10, 10, 20, 20)
-					inputStarted.Wait() // Can't flush until the input has started.
+					drainStats()
 					tickChan <- time.Now()
 					injectCalled.Wait()
 
 					sendTimer("sample2.timer", 10, 20)
+					drainStats()
 					ith.Pack.Recycle(nil)
 					ith.PackSupply <- ith.Pack
+					ith.MockInputRunner.EXPECT().Name().Return("StatAccumInput")
 					ith.MockInputRunner.EXPECT().Inject(ith.Pack)
 					msg, err := finalizeSendingStats()
 					c.Assume(err, gs.IsNil)
@@ -379,6 +403,7 @@ func StatAccumInputSpec(c gs.Context) {
 
 					// Prep pack and EXPECTS for the close.
 					ith.PackSupply <- ith.Pack
+					ith.MockInputRunner.EXPECT().Name().Return("StatAccumInput")
 					ith.MockInputRunner.EXPECT().Inject(ith.Pack)
 
 					close(statAccumInput.statChan)

plugins/file/file_output_test.go

@@ -107,34 +107,40 @@ func FileOutputSpec(c gs.Context) {
 			})
 		})
 
-		c.Specify("tests rotation of files", func() {
+		c.Specify("rotates files correctly", func() {
 			config.Path = "%Y-%m-%d"
+			config.RotationInterval = 24
 			rotateChan := make(chan time.Time)
 			closingChan := make(chan struct{})
 
 			err := fileOutput.Init(config)
-			defer fileOutput.file.Close()
-
 			c.Assume(err, gs.IsNil)
 
+			defer fileOutput.file.Close()
+
 			fileOutput.rotateChan = rotateChan
 			fileOutput.closing = closingChan
 
 			fileOutput.startRotateNotifier()
 
-			go fileOutput.committer(oth.MockOutputRunner, errChan)
+			committerChan := make(chan struct{})
+			go func() {
+				fileOutput.committer(oth.MockOutputRunner, errChan)
+				close(committerChan)
+			}()
 
 			c.Assume(fileOutput.path, gs.Equals, time.Now().Format("2006-01-02"))
 
 			futureDuration, _ := time.ParseDuration("24h")
 			futureNow := time.Now().Add(futureDuration)
 
 			rotateChan <- futureNow
+			close(inChan)
+			close(fileOutput.batchChan)
+			<-committerChan
 
 			c.Assume(fileOutput.path, gs.Equals, futureNow.Format("2006-01-02"))
 
-			close(inChan)
-			close(fileOutput.batchChan)
 		})
 
 		c.Specify("processes incoming messages", func() {
@@ -256,6 +262,7 @@ func FileOutputSpec(c gs.Context) {
 			timerChan := make(chan time.Time)
 
 			msg2 := pipeline_ts.GetTestMessage()
+			msg2.SetPayload("MESSAGE 2")
 			pack2 := NewPipelinePack(pConfig.InputRecycleChan())
 			pack2.Message = msg2
 
@@ -299,24 +306,26 @@ func FileOutputSpec(c gs.Context) {
 				defer cleanUp()
 				inChan <- pack
 
+				after := time.After(100 * time.Millisecond)
 				select {
 				case <-fileOutput.batchChan:
 					c.Expect("", gs.Equals, "fileOutput.batchChan should NOT have fired yet")
-				default:
+				case <-after:
 				}
 
 				timerChan <- time.Now()
+				after = time.After(100 * time.Millisecond)
 				select {
 				case <-fileOutput.batchChan:
 					c.Expect("", gs.Equals, "fileOutput.batchChan should NOT have fired yet")
-				default:
+				case <-after:
 				}
 
+				after = time.After(100 * time.Millisecond)
 				inChan <- pack2
-				runtime.Gosched()
 				select {
 				case <-fileOutput.batchChan:
-				default:
+				case <-after:
 					c.Expect("", gs.Equals, "fileOutput.batchChan SHOULD have fired by now")
 				}
 			})
@@ -329,18 +338,19 @@ func FileOutputSpec(c gs.Context) {
 				defer cleanUp()
 				inChan <- pack
 
+				after := time.After(100 * time.Millisecond)
 				select {
 				case <-fileOutput.batchChan:
 					c.Expect("", gs.Equals, "fileOutput.batchChan should NOT have fired yet")
-				default:
+				case <-after:
 				}
 
 				c.Specify("when interval triggers first", func() {
 					timerChan <- time.Now()
-					runtime.Gosched()
+					after = time.After(100 * time.Millisecond)
 					select {
 					case <-fileOutput.batchChan:
-					default:
+					case <-after:
 						c.Expect("", gs.Equals, "fileOutput.batchChan SHOULD have fired by now")
 					}
 				})
@@ -349,10 +359,10 @@ func FileOutputSpec(c gs.Context) {
 					out, err := encoder.Encode(pack2)
 					oth.MockOutputRunner.EXPECT().Encode(gomock.Any()).Return(out, err)
 					inChan <- pack2
-					runtime.Gosched()
+					after = time.After(100 * time.Millisecond)
 					select {
 					case <-fileOutput.batchChan:
-					default:
+					case <-after:
 						c.Expect("", gs.Equals, "fileOutput.batchChan SHOULD have fired by now")
 					}
 				})