SCFF-31 filling the queue, buffering, and sending to sumo

Author: nmillalaf

eventQueue/eventQueue.go

@@ -1,8 +1,6 @@
 package eventQueue
 
-import (
-	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
-)
+import . "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
 
 //Node to put in queue
 type Node struct {
@@ -23,8 +21,8 @@ func NewNode(event Event) *Node {
 	}
 }
 
-func NewQueue(n []*Node) *Queue {
-	return &Queue{
+func NewQueue(n []*Node) Queue {
+	return Queue{
 		Nodes: n,
 	}
 }

eventRouting/eventrouting.go

@@ -21,10 +21,10 @@ type EventRouting struct {
 	selectedEventsCount map[string]uint64
 	mutex               *sync.Mutex
 	sLAppender          sumoCFFirehose.SumoLogicAppender //**
-	queue               eventQueue.Queue
+	queue               *eventQueue.Queue
 }
 
-func NewEventRouting(caching caching.Caching, sLAppender sumoCFFirehose.SumoLogicAppender, queue eventQueue.Queue) *EventRouting {
+func NewEventRouting(caching caching.Caching, sLAppender sumoCFFirehose.SumoLogicAppender, queue *eventQueue.Queue) *EventRouting {
 	return &EventRouting{
 		CachingClient:       caching,
 		selectedEvents:      make(map[string]bool),

main.go

@@ -40,7 +40,7 @@ var (
 func main() {
 	kingpin.Version(version)
 	kingpin.Parse()
-	runtime.GOMAXPROCS(2)
+	runtime.GOMAXPROCS(1)
 
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
@@ -66,11 +66,11 @@ func main() {
 
 	//Creating queue
 	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue, *eventsBatchSize)
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize)
 	go loggingClientSumo.Start() //multi
 
 	//Creating Events
-	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
+	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, &queue)
 	err := events.SetupEventRouting(*wantedEvents)
 	if err != nil {
 		log.Fatal("Error setting up event routing: ", err)
@@ -100,20 +100,11 @@ func main() {
 		fmt.Printf("Connected to Server! Connecting to Firehose... \n")
 
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
+		go firehoseClient.Start()
 
-		//go firehoseClient.Start()
-		err = firehoseClient.Start()
-		if err != nil {
-			fmt.Printf("Failed connecting to Firehose...Please check settings and try again! \n") //Log error
-
-		} else {
-			fmt.Printf("Firehose Subscription Succesfull! Routing events... \n")
-		}
-
-	} else {
-		fmt.Printf("Failed connecting to the Fluentd Server...Please check settings and try again! \n") //Log error
-
+		defer firehoseClient.Start()
 	}
 
 	defer cachingClient.Close()
+
 }

sumoCFFirehose/sumoLogicAppender.go

@@ -14,13 +14,13 @@ type SumoLogicAppender struct {
 	url                      string
 	connectionTimeout        int //10000
 	httpClient               http.Client
-	nozzleQueue              eventQueue.Queue
+	nozzleQueue              *eventQueue.Queue
 	eventsBatchSize          int
 	logEventsInCurrentBuffer int
 	logStringToSend          string
 }
 
-func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue, eventsBatchSize int) *SumoLogicAppender {
+func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int) *SumoLogicAppender {
 	return &SumoLogicAppender{
 		url:                      urlValue,
 		connectionTimeout:        connectionTimeoutValue,
@@ -50,30 +50,29 @@ func (s *SumoLogicAppender) Connect() bool {
 }
 
 func (s *SumoLogicAppender) Start() {
-	timer := time.NewTimer(60 * time.Second)
+	timer := time.Now()
 	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 {
-			fmt.Println("i'm on the if")
 			s.AppendLogs()
+			fmt.Println(s.logStringToSend)
 		}
-		fmt.Println("passed first if")
 		if s.logEventsInCurrentBuffer >= s.eventsBatchSize {
+			fmt.Println("Buffer full")
+			fmt.Println(s.logStringToSend)
 			s.SendToSumo(s.logStringToSend)
 			s.logEventsInCurrentBuffer = 0 // reset counter
 			s.logStringToSend = ""         //reset String
-		} else if (<-timer.C).Second() == 0 {
+		} else if time.Since(timer).Seconds() >= 10 {
+			fmt.Println("timer finished, sending logs...")
+			fmt.Println(s.logStringToSend)
 			s.SendToSumo(s.logStringToSend)
 			s.logEventsInCurrentBuffer = 0 // reset counter
 			s.logStringToSend = ""         //reset String
+			timer = time.Now()             //reset timer
 		}
-		fmt.Println("passed second if")
-		fmt.Println("end of bucle")
 
 	}
 }
@@ -97,7 +96,6 @@ func (s *SumoLogicAppender) AppendLogs() {
 	//timer := time.NewTimer(60 * time.Second)
 	s.logStringToSend = s.logStringToSend + StringBuilder(s.nozzleQueue.Pop())
 	s.logEventsInCurrentBuffer++
-
 }
 
 func (s *SumoLogicAppender) SendToSumo(log string) {