Merged in feature/SCFF-31 (pull request #9)

Feature/SCFF-31

Author: sklava

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),
@@ -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(1)
 
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
@@ -66,10 +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)
@@ -99,17 +100,11 @@ func main() {
 		fmt.Printf("Connected to Server! Connecting to Firehose... \n")
 
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
-		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")
-		}
+		go firehoseClient.Start()
 
-	} 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

@@ -3,46 +3,54 @@ package sumoCFFirehose
 import (
 	"bytes"
 	"fmt"
-	"net"
 	"net/http"
 	"time"
 
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 )
 
 type SumoLogicAppender struct {
-	url               string
-	connectionTimeout int //10000
-	httpClient        http.Client
-	nozzleQueue       eventQueue.Queue
-	eventsBatchSize   int
+	url                      string
+	connectionTimeout        int //10000
+	httpClient               http.Client
+	nozzleQueue              *eventQueue.Queue
+	eventsBatchSize          int
+	logEventsInCurrentBuffer int
+	logStringToSend          *bytes.Buffer
 }
 
-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,
-		httpClient:        http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
-		nozzleQueue:       nozzleQueue,
-		eventsBatchSize:   eventsBatchSize,
+		url:                      urlValue,
+		connectionTimeout:        connectionTimeoutValue,
+		httpClient:               http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
+		nozzleQueue:              nozzleQueue,
+		eventsBatchSize:          eventsBatchSize,
+		logEventsInCurrentBuffer: 0,
+		logStringToSend:          bytes.NewBufferString(""),
 	}
 }
 
-func (s *SumoLogicAppender) Connect() bool {
-	success := false
-	if s.url != "" {
-		conn, err := net.Dial("tcp", s.url)
-		if err != nil {
-			fmt.Printf(fmt.Sprintf("Unable to connect to sumo server [%s]!\n", s.url), err.Error())
-		} else {
+func (s *SumoLogicAppender) Start() {
+	timer := time.Now()
+	fmt.Println("Starting Appender Worker")
+	for {
+		time.Sleep(300 * time.Millisecond)
+		if s.nozzleQueue.GetCount() != 0 { //if queue is not empty, AppendLogs
+			s.AppendLogs()
+		}
+		if s.logEventsInCurrentBuffer >= s.eventsBatchSize { //if buffer is full, send logs to sumo
+			fmt.Println("Buffer full, sending logs to sumo...")
+
+			s.SendToSumo(s.logStringToSend)
 
-			fmt.Printf(fmt.Sprintf("Connected to [%s]!\n", s.url), false)
-			success = true
-			defer conn.Close()
+		} else if time.Since(timer).Seconds() >= 10 { // else if timer is up, send existing logs to sumo
+			fmt.Println("timer finished, sending logs...")
+			s.SendToSumo(s.logStringToSend)
+			timer = time.Now() //reset timer
 		}
-	}
 
-	return success
+	}
 }
 
 func StringBuilder(node *eventQueue.Node) string {
@@ -61,17 +69,14 @@ func StringBuilder(node *eventQueue.Node) string {
 
 func (s *SumoLogicAppender) AppendLogs() {
 	// the appender calls for the next message in the queue and parse it to a string
-	logMessage := ""
-	if s.nozzleQueue.GetCount() > s.eventsBatchSize { //when the batch limit is met, call stringBuilder
-		logMessage = logMessage + StringBuilder(s.nozzleQueue.Pop())
-
-	}
-	s.SendToSumo(logMessage)
-
+	//timer := time.NewTimer(60 * time.Second)
+	s.logStringToSend.Write([]byte(StringBuilder(s.nozzleQueue.Pop())))
+	s.logEventsInCurrentBuffer++
 }
 
-func (s *SumoLogicAppender) SendToSumo(log string) {
-	request, err := http.NewRequest("POST", s.url, bytes.NewBufferString(log))
+func (s *SumoLogicAppender) SendToSumo(log *bytes.Buffer) {
+	fmt.Println(log)
+	request, err := http.NewRequest("POST", s.url, log)
 	if err != nil {
 		fmt.Printf("http.NewRequest() error: %v\n", err)
 		return
@@ -86,6 +91,8 @@ func (s *SumoLogicAppender) SendToSumo(log string) {
 	} else {
 		fmt.Println("Do(Request) successful")
 	}
+	s.logEventsInCurrentBuffer = 0                // reset counter
+	s.logStringToSend = bytes.NewBufferString("") //reset String
 	defer response.Body.Close()
 
 }

sumoCFFirehose/sumoLogicAppender_test.go

@@ -3,9 +3,10 @@ package sumoCFFirehose
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 	. "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
-	"github.com/stretchr/testify/assert"
 )
 
 func testAppenderStringBuilder(t *testing.T) {
@@ -55,5 +56,4 @@ func testAppenderStringBuilder(t *testing.T) {
 	assert.Equal(t, finalString, "2016-12-12 16:02:41.828366387 -0300 CLST"+"\t"+"OUT"+"\t"+"index [01]"+"\n"+
 		"2016-12-12 16:02:42.844737993 -0300 CLST"+"\t"+"OUT"+"\t"+"index [02]"+"\n"+
 		"2016-12-12 16:02:43.862436654 -0300 CLST"+"\t"+"OUT"+"\t"+"index [03]"+"\n", "")
-
 }