SCFF-35 better loop handling, added some logging regarding buffer and queue

Author: nmillalaf

main.go

@@ -32,7 +32,7 @@ var (
 	boltDatabasePath     = "my.db"                   //default
 	tickerTime, errT     = time.ParseDuration("60s") //Default
 	eventsBatchSize      = kingpin.Flag("log-events-batch-size", "Log Events Batch Size").Default("10").Int()
-	sumoPostMinimumDelay = kingpin.Flag("sumo-Post-Minimum-Delay", "Sumo Post Minimum Delay").Default("10s").Duration()
+	sumoPostMinimumDelay = kingpin.Flag("sumo-Post-Minimum-Delay", "Sumo Post Minimum Delay").Default("500ms").Duration()
 )
 
 var (

sumoCFFirehose/sumoLogicAppender.go

@@ -25,7 +25,6 @@ type SumoBuffer struct {
 	logStringToSend          *bytes.Buffer
 	logEventsInCurrentBuffer int
 	timerIdlebuffer          time.Time
-	channelMessage           chan string
 }
 
 func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int, sumoPostMinimumDelay time.Duration) *SumoLogicAppender {
@@ -43,7 +42,6 @@ func newBuffer() SumoBuffer {
 	return SumoBuffer{
 		logStringToSend:          bytes.NewBufferString(""),
 		logEventsInCurrentBuffer: 0,
-		channelMessage:           make(chan string),
 	}
 }
 
@@ -54,29 +52,37 @@ func (s *SumoLogicAppender) Start() {
 	Buffer.timerIdlebuffer = time.Now()
 	logging.Info.Println("Starting Appender Worker")
 	for {
-		for s.nozzleQueue.GetCount() == 0 {
+		logging.Info.Println("Log queue size: ")
+		logging.Info.Println(s.nozzleQueue.GetCount())
+		if s.nozzleQueue.GetCount() == 0 {
+			logging.Trace.Println("Waiting for 300 ms")
 			time.Sleep(300 * time.Millisecond)
 		}
 
 		if time.Since(Buffer.timerIdlebuffer).Seconds() >= 10 && Buffer.logEventsInCurrentBuffer > 0 {
 			logging.Info.Println("Sending current batch of logs after timer exceeded limit")
 			go s.SendToSumo(&Buffer)
-			<-Buffer.channelMessage
 			Buffer = newBuffer()
+			Buffer.timerIdlebuffer = time.Now()
+			continue
 		}
-		for s.nozzleQueue.GetCount() != 0 {
-			if s.nozzleQueue.GetCount() >= s.eventsBatchSize-Buffer.logEventsInCurrentBuffer {
-				for Buffer.logEventsInCurrentBuffer < s.eventsBatchSize {
+
+		if s.nozzleQueue.GetCount() != 0 {
+			queueCount := s.nozzleQueue.GetCount()
+			remainingBufferCount := s.eventsBatchSize - Buffer.logEventsInCurrentBuffer
+			if queueCount >= remainingBufferCount {
+				logging.Trace.Println("Pushing Logs to Sumo: ")
+				logging.Trace.Println(remainingBufferCount)
+				for i := 0; i < remainingBufferCount; i++ {
 					s.AppendLogs(&Buffer)
 					Buffer.timerIdlebuffer = time.Now()
 				}
-				logging.Trace.Println("Batch Size complete")
 				go s.SendToSumo(&Buffer)
-				<-Buffer.channelMessage
 				Buffer = newBuffer()
 			} else {
-				for s.nozzleQueue.GetCount() > 0 && Buffer.logEventsInCurrentBuffer < s.eventsBatchSize {
-					//TODO fill the buffer with whatever is in the queue without sending to sumo
+				logging.Trace.Println("Pushing Logs to Buffer: ")
+				logging.Trace.Println(queueCount)
+				for i := 0; i < queueCount; i++ {
 					s.AppendLogs(&Buffer)
 					Buffer.timerIdlebuffer = time.Now()
 				}
@@ -107,15 +113,12 @@ func (s *SumoLogicAppender) AppendLogs(buffer *SumoBuffer) {
 }
 
 func (s *SumoLogicAppender) SendToSumo(buffer *SumoBuffer) {
-	//wait period between posts to Sumo
 	for time.Since(s.timerBetweenPost) < s.sumoPostMinimumDelay {
-		time.Sleep(100 * time.Millisecond) // wait to retry
+		logging.Trace.Println("Delaying post to honor minimum post delay")
+		time.Sleep(100 * time.Millisecond)
 	}
-	/*fmt.Println(buffer.logStringToSend.String())
-	fmt.Println("........")*/
-	buffer.channelMessage <- "Buffer being sent"
 
-	logging.Trace.Println("Sending logs to Sumologic...")
+	logging.Info.Println("Sending logs to Sumologic...")
 	request, err := http.NewRequest("POST", s.url, buffer.logStringToSend)
 	if err != nil {
 		logging.Error.Printf("http.NewRequest() error: %v\n", err)
@@ -128,8 +131,8 @@ func (s *SumoLogicAppender) SendToSumo(buffer *SumoBuffer) {
 		logging.Error.Printf("http.Do() error: %v\n", err)
 		return
 	} else {
-		logging.Trace.Println("Do(Request) successful")
-		s.timerBetweenPost = time.Now() //reset timer post minimum
+		logging.Trace.Println("Post of logs successful")
+		s.timerBetweenPost = time.Now()
 	}
 
 	defer response.Body.Close()