Merged in feature/SCFF-30 (pull request #8)

Feature/SCFF-30

Author: sklava

eventQueue/eventQueue.go

@@ -6,53 +6,53 @@ import (
 
 //Node to put in queue
 type Node struct {
-	event Event
+	Event Event
 }
 
 // Queue is a basic FIFO queue based on a circular list that resizes as needed.
 type Queue struct {
-	nodes []*Node
+	Nodes []*Node
 	head  int
 	tail  int
 	count int
 }
 
 func NewNode(event Event) *Node {
 	return &Node{
-		event: event,
+		Event: event,
 	}
 }
 
 func NewQueue(n []*Node) *Queue {
 	return &Queue{
-		nodes: n,
+		Nodes: n,
 	}
 }
 
 func (q *Queue) GetNode() []*Node {
-	return q.nodes
+	return q.Nodes
 }
 
 func (n *Queue) GetCount() int {
 	return n.count
 }
 
 func (n *Node) GetNodeEvent() Event {
-	return n.event
+	return n.Event
 }
 
 // Push adds a node to the queue.
 func (q *Queue) Push(n *Node) {
 	if q.head == q.tail && q.count > 0 {
-		nodes := make([]*Node, len(q.nodes)*2)
-		copy(nodes, q.nodes[q.head:])
-		copy(nodes[len(q.nodes)-q.head:], q.nodes[:q.head])
+		nodes := make([]*Node, len(q.Nodes)*2)
+		copy(nodes, q.Nodes[q.head:])
+		copy(nodes[len(q.Nodes)-q.head:], q.Nodes[:q.head])
 		q.head = 0
-		q.tail = len(q.nodes)
-		q.nodes = nodes
+		q.tail = len(q.Nodes)
+		q.Nodes = nodes
 	}
-	q.nodes[q.tail] = n
-	q.tail = (q.tail + 1) % len(q.nodes)
+	q.Nodes[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.Nodes)
 	q.count++
 }
 
@@ -61,8 +61,8 @@ func (q *Queue) Pop() *Node {
 	if q.count == 0 {
 		return nil
 	}
-	node := q.nodes[q.head]
-	q.head = (q.head + 1) % len(q.nodes)
+	node := q.Nodes[q.head]
+	q.head = (q.head + 1) % len(q.Nodes)
 	q.count--
 	return node
 }

eventQueue/eventQueue_test.go

@@ -8,9 +8,9 @@ import (
 )
 
 func TestQueueFIFO(t *testing.T) {
-
+	assert := assert.New(t)
 	node1 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "011",
@@ -19,7 +19,7 @@ func TestQueueFIFO(t *testing.T) {
 		},
 	}
 	node2 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "022",
@@ -28,7 +28,7 @@ func TestQueueFIFO(t *testing.T) {
 		},
 	}
 	node3 := Node{
-		event: Event{
+		Event: Event{
 			Fields: map[string]interface{}{
 				"message_type": "OUT",
 				"cf_app_id":    "033",
@@ -38,14 +38,14 @@ func TestQueueFIFO(t *testing.T) {
 	}
 
 	queue := Queue{
-		nodes: make([]*Node, 3),
+		Nodes: make([]*Node, 3),
 	}
 
 	queue.Push(&node1)
 	queue.Push(&node2)
 	queue.Push(&node3)
 
-	assert.Equal(t, queue.Pop().event.Msg, "index [01]", "")
-	assert.Equal(t, queue.Pop().event.Msg, "index [02]", "")
-	assert.Equal(t, queue.Pop().event.Msg, "index [03]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [01]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [02]", "")
+	assert.Equal(queue.Pop().Event.Msg, "index [03]", "")
 }

main.go

@@ -28,6 +28,7 @@ var (
 	wantedEvents      = kingpin.Flag("events", fmt.Sprintf("Comma separated list of events you would like. Valid options are %s", eventRouting.GetListAuthorizedEventEvents())).Default("LogMessage").OverrideDefaultFromEnvar("EVENTS").String()
 	boltDatabasePath  = "my.db"                   //default
 	tickerTime, errT  = time.ParseDuration("60s") //Default
+	eventsBatchSize   = kingpin.Flag("log-events-batch-size", "Log Events Batch Size").Int()
 )
 
 var (
@@ -41,10 +42,6 @@ func main() {
 	fmt.Println("this is the sumo endpoint")
 	fmt.Println(sumoEndpoint)
 
-	//Creating queue
-	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue)
-
 	fmt.Printf("Starting firehose-to-sumo %s \n", version)
 
 	c := cfclient.Config{
@@ -67,6 +64,10 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 
+	//Creating queue
+	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, *queue, *eventsBatchSize)
+
 	//Creating Events
 	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, *queue)
 	err := events.SetupEventRouting(*wantedEvents)
@@ -99,7 +100,6 @@ func main() {
 
 		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
 		err = firehoseClient.Start()
-		fmt.Printf("I created the Firehose... \n")
 		if err != nil {
 			fmt.Printf("Failed connecting to Firehose...Please check settings and try again! \n") //Log error
 

sumoCFFirehose/sumoLogicAppender.go

@@ -15,14 +15,16 @@ type SumoLogicAppender struct {
 	connectionTimeout int //10000
 	httpClient        http.Client
 	nozzleQueue       eventQueue.Queue
+	eventsBatchSize   int
 }
 
-func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue eventQueue.Queue) *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,
 	}
 }
 
@@ -43,26 +45,29 @@ func (s *SumoLogicAppender) Connect() bool {
 	return success
 }
 
+func StringBuilder(node *eventQueue.Node) string {
+	buf := new(bytes.Buffer)
+	if node.Event.Fields["message_type"] == nil {
+		return ""
+	}
+	if node.Event.Fields["message_type"] == "" {
+		return ""
+	}
+	message := time.Unix(0, node.Event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + node.Event.Fields["message_type"].(string) + "\t" + node.Event.Msg + "\n"
+	buf.WriteString(message)
+
+	return buf.String()
+}
+
 func (s *SumoLogicAppender) AppendLogs() {
 	// the appender calls for the next message in the queue and parse it to a string
-	fmt.Println("i'm in appendLogs")
-	event := s.nozzleQueue.Pop().GetNodeEvent()
-	/*
-		if event == nil {
-			return
-		}*/
+	logMessage := ""
+	if s.nozzleQueue.GetCount() > s.eventsBatchSize { //when the batch limit is met, call stringBuilder
+		logMessage = logMessage + StringBuilder(s.nozzleQueue.Pop())
 
-	if event.Fields["message_type"] == nil {
-		return
-	}
-
-	if event.Fields["message_type"] == "" {
-		return
 	}
+	s.SendToSumo(logMessage)
 
-	Message := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String() + "\t" + event.Fields["message_type"].(string) + "\t" + event.Msg + "\n"
-	fmt.Println(Message)
-	s.SendToSumo(Message)
 }
 
 func (s *SumoLogicAppender) SendToSumo(log string) {

sumoCFFirehose/sumoLogicAppender_test.go

@@ -0,0 +1,59 @@
+package sumoCFFirehose
+
+import (
+	"testing"
+
+	. "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) {
+
+	node1 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569361828366387",
+				"message_type": "OUT",
+				"cf_app_id":    "011",
+			},
+			Msg: "index [01]",
+		},
+	}
+	node2 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569362844737993",
+				"message_type": "OUT",
+				"cf_app_id":    "022",
+			},
+			Msg: "index [02]",
+		},
+	}
+	node3 := Node{
+		Event: Event{
+			Fields: map[string]interface{}{
+				"timestamp":    "1481569363862436654",
+				"message_type": "OUT",
+				"cf_app_id":    "033",
+			},
+			Msg: "index [03]",
+		},
+	}
+
+	queue := Queue{
+		Nodes: make([]*Node, 3),
+	}
+	queue.Push(&node1)
+	queue.Push(&node2)
+	queue.Push(&node3)
+
+	finalString := ""
+	for queue.GetCount() > 0 {
+		finalString = finalString + StringBuilder(queue.Pop())
+	}
+	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", "")
+
+}