Fix https_batch deadlock due to golang timer changes

src/pkg/egress/syslog/https_batch.go

@@ -13,6 +13,8 @@ import (
 
 const BATCHSIZE = 256 * 1024
 
+var DefaultSendInterval = 1 * time.Second
+
 type HTTPSBatchWriter struct {
 	HTTPSWriter
 	msgs         chan []byte
@@ -40,7 +42,7 @@ func NewHTTPSBatchWriter(
 			syslogConverter: c,
 		},
 		batchSize:    BATCHSIZE,
-		sendInterval: 1 * time.Second,
+		sendInterval: DefaultSendInterval,
 		egrMsgCount:  0,
 		msgs:         make(chan []byte),
 	}
@@ -65,34 +67,37 @@ func (w *HTTPSBatchWriter) Write(env *loggregator_v2.Envelope) error {
 }
 
 func (w *HTTPSBatchWriter) startSender() {
-	t := time.NewTimer(w.sendInterval)
-
 	var msgBatch bytes.Buffer
 	var msgCount float64
-	reset := func() {
-		msgBatch.Reset()
-		msgCount = 0
-		t.Reset(w.sendInterval)
+	ticker := time.NewTicker(w.sendInterval)
+	defer ticker.Stop()
+
+	msgCount = 0
+
+	sendBatch := func() {
+		if msgBatch.Len() > 0 {
+			w.sendHttpRequest(msgBatch.Bytes(), msgCount) //nolint:errcheck
+			msgBatch.Reset()
+			msgCount = 0
+		}
 	}
+
 	for {
 		select {
 		case msg := <-w.msgs:
-			length, buffer_err := msgBatch.Write(msg)
+			_, buffer_err := msgBatch.Write(msg)
 			if buffer_err != nil {
-				log.Printf("Failed to write to buffer, dropping buffer of size %d , err: %s", length, buffer_err)
-				reset()
+				log.Printf("Failed to write to buffer, dropping buffer of size %d , err: %s", msgBatch.Len(), buffer_err)
+				msgBatch.Reset()
+				msgCount = 0
 			} else {
 				msgCount++
-				if length >= w.batchSize {
-					w.sendHttpRequest(msgBatch.Bytes(), msgCount) //nolint:errcheck
-					reset()
+				if msgBatch.Len() >= w.batchSize {
+					sendBatch()
 				}
 			}
-		case <-t.C:
-			if msgBatch.Len() > 0 {
-				w.sendHttpRequest(msgBatch.Bytes(), msgCount) //nolint:errcheck
-				reset()
-			}
+		case <-ticker.C:
+			sendBatch()
 		}
 	}
 }

src/pkg/egress/syslog/https_batch_test.go

@@ -19,6 +19,10 @@ import (
 
 var string_to_1024_chars = "saljdflajsdssdfsdfljkfkajafjajlköflkjöjaklgljksdjlakljkflkjweljklkwjejlkfekljwlkjefjklwjklsdajkljklwerlkaskldgjksakjekjwrjkljasdjkgfkljwejklrkjlklasdkjlsadjlfjlkadfljkajklsdfjklslkdfjkllkjasdjkflsdlakfjklasldfkjlasdjfkjlsadlfjklaljsafjlslkjawjklerkjljklasjkdfjklwerjljalsdjkflwerjlkwejlkarjklalkklfsdjlfhkjsdfkhsewhkjjasdjfkhwkejrkjahjefkhkasdjhfkashfkjwehfkksadfjaskfkhjdshjfhewkjhasdfjdajskfjwehkfajkankaskjdfasdjhfkkjhjjkasdfjhkjahksdf"
 
+func init() {
+	syslog.DefaultSendInterval = 100 * time.Millisecond // Modify behavior for tests
+}
+
 var _ = Describe("HTTPS_batch", func() {
 	var (
 		netConf          syslog.NetworkTimeoutConfig
@@ -86,21 +90,33 @@ var _ = Describe("HTTPS_batch", func() {
 		env1 := buildLogEnvelope("APP", "1", "string to get log to 1024 characters:"+string_to_1024_chars, loggregator_v2.Log_OUT)
 		for i := 0; i < 10; i++ {
 			Expect(writer.Write(env1)).To(Succeed())
-			time.Sleep(99 * time.Millisecond)
+			time.Sleep(5 * time.Millisecond)
 		}
 		Expect(drain.getMessagesSize()).Should(Equal(0))
 		time.Sleep(100 * time.Millisecond)
 		Expect(drain.getMessagesSize()).Should(Equal(10))
 	})
 
-	It("probabilistic test for race condition", func() {
+	It("test dispatching for batches before timewindow is finished", func() {
 		env1 := buildLogEnvelope("APP", "1", "string to get log to 1024 characters:"+string_to_1024_chars, loggregator_v2.Log_OUT)
-		for i := 0; i < 10; i++ {
+		for i := 0; i < 300; i++ {
 			Expect(writer.Write(env1)).To(Succeed())
-			time.Sleep(99 * time.Millisecond)
 		}
-		time.Sleep(100 * time.Millisecond)
-		Expect(drain.getMessagesSize()).Should(Equal(10))
+		Expect(drain.getMessagesSize()).Should(Equal(256))
+		time.Sleep(101 * time.Millisecond)
+		Expect(drain.getMessagesSize()).Should(Equal(300))
+	})
+
+	It("test for hanging after some ticks", func() {
+		// This test will not succeed on the timer based implementation,
+		// it works fine with a ticker based implementation.
+		env1 := buildLogEnvelope("APP", "1", "only a short test message", loggregator_v2.Log_OUT)
+		for i := 0; i < 5; i++ {
+			Expect(writer.Write(env1)).To(Succeed())
+			time.Sleep(300 * time.Millisecond)
+		}
+		time.Sleep(101 * time.Millisecond)
+		Expect(drain.getMessagesSize()).Should(Equal(5))
 	})
 })