SCFF-31 appender multithreading

Author: nmillalaf

eventRouting/eventrouting.go

@@ -148,7 +148,6 @@ func (e *EventRouting) LogEventTotals(logTotalsTime time.Duration) {
 
 			//Push the event to the queue
 			e.queue.Push(eventQueue.NewNode(*event))
-			//e.sLAppender.AppendLogs(*event)
 		}
 	}()
 }

main.go

@@ -6,6 +6,8 @@ import (
 	"os"
 	"time"
 
+	"runtime"
+
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/caching"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventRouting"
@@ -38,9 +40,7 @@ var (
 func main() {
 	kingpin.Version(version)
 	kingpin.Parse()
-
-	fmt.Println("this is the sumo endpoint")
-	fmt.Println(sumoEndpoint)
+	runtime.GOMAXPROCS(2)
 
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
@@ -100,6 +100,8 @@ func main() {
 		fmt.Printf("Connected to Server! Connecting to Firehose... \n")
 
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
+
+		//go firehoseClient.Start()
 		err = firehoseClient.Start()
 		if err != nil {
 			fmt.Printf("Failed connecting to Firehose...Please check settings and try again! \n") //Log error
@@ -110,6 +112,7 @@ func main() {
 
 	} else {
 		fmt.Printf("Failed connecting to the Fluentd Server...Please check settings and try again! \n") //Log error
+
 	}
 
 	defer cachingClient.Close()

sumoCFFirehose/sumoLogicAppender.go

@@ -52,13 +52,18 @@ func (s *SumoLogicAppender) Connect() bool {
 func (s *SumoLogicAppender) Start() {
 	timer := time.NewTimer(60 * time.Second)
 	fmt.Println("Starting Appender Worker")
+
 	s.logStringToSend = ""
 	for {
+		fmt.Println("nozzle queue")
+		fmt.Println(s.nozzleQueue.GetCount())
 		time.Sleep(300 * time.Millisecond)
-		if s.nozzleQueue.GetCount() > 0 {
+		if s.nozzleQueue.GetCount() != 0 {
+			fmt.Println("i'm on the if")
 			s.AppendLogs()
 		}
-		if s.eventsBatchSize >= s.logEventsInCurrentBuffer {
+		fmt.Println("passed first if")
+		if s.logEventsInCurrentBuffer >= s.eventsBatchSize {
 			s.SendToSumo(s.logStringToSend)
 			s.logEventsInCurrentBuffer = 0 // reset counter
 			s.logStringToSend = ""         //reset String
@@ -67,6 +72,8 @@ func (s *SumoLogicAppender) Start() {
 			s.logEventsInCurrentBuffer = 0 // reset counter
 			s.logStringToSend = ""         //reset String
 		}
+		fmt.Println("passed second if")
+		fmt.Println("end of bucle")
 
 	}
 }