Merged in feature/SCFF-34 (pull request #10)

SCFF-34 added logging and using Events instead Node in queue

Author: sklava

eventQueue/eventQueue.go

@@ -2,65 +2,55 @@ package eventQueue
 
 import . "bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
 
-//Node to put in queue
-type Node struct {
-	Event Event
-}
-
 // Queue is a basic FIFO queue based on a circular list that resizes as needed.
 type Queue struct {
-	Nodes []*Node
-	head  int
-	tail  int
-	count int
-}
-
-func NewNode(event Event) *Node {
-	return &Node{
-		Event: event,
-	}
+	Events []*Event
+	head   int
+	tail   int
+	count  int
 }
 
-func NewQueue(n []*Node) Queue {
+func NewQueue(n []*Event) Queue {
 	return Queue{
-		Nodes: n,
+		Events: n,
 	}
 }
 
-func (q *Queue) GetNode() []*Node {
-	return q.Nodes
+func (q *Queue) GetNode() []*Event {
+	return q.Events
 }
 
-func (n *Queue) GetCount() int {
-	return n.count
+func (q *Queue) GetCount() int {
+	return q.count
 }
 
-func (n *Node) GetNodeEvent() Event {
-	return n.Event
-}
+/*
+func (q *Queue) GetEvents() Event {
+	return q.Events
+}*/
 
 // Push adds a node to the queue.
-func (q *Queue) Push(n *Node) {
+func (q *Queue) Push(n *Event) {
 	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])
+		events := make([]*Event, len(q.Events)*2)
+		copy(events, q.Events[q.head:])
+		copy(events[len(q.Events)-q.head:], q.Events[:q.head])
 		q.head = 0
-		q.tail = len(q.Nodes)
-		q.Nodes = nodes
+		q.tail = len(q.Events)
+		q.Events = events
 	}
-	q.Nodes[q.tail] = n
-	q.tail = (q.tail + 1) % len(q.Nodes)
+	q.Events[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.Events)
 	q.count++
 }
 
 // Pop removes and returns a node from the queue in first to last order.
-func (q *Queue) Pop() *Node {
+func (q *Queue) Pop() *Event {
 	if q.count == 0 {
 		return nil
 	}
-	node := q.Nodes[q.head]
-	q.head = (q.head + 1) % len(q.Nodes)
+	node := q.Events[q.head]
+	q.head = (q.head + 1) % len(q.Events)
 	q.count--
 	return node
 }

eventQueue/eventQueue_test.go

@@ -9,43 +9,39 @@ import (
 
 func TestQueueFIFO(t *testing.T) {
 	assert := assert.New(t)
-	node1 := Node{
-		Event: Event{
-			Fields: map[string]interface{}{
-				"message_type": "OUT",
-				"cf_app_id":    "011",
-			},
-			Msg: "index [01]",
+	event1 := Event{
+		Fields: map[string]interface{}{
+			"message_type": "OUT",
+			"cf_app_id":    "011",
 		},
+		Msg: "index [01]",
 	}
-	node2 := Node{
-		Event: Event{
-			Fields: map[string]interface{}{
-				"message_type": "OUT",
-				"cf_app_id":    "022",
-			},
-			Msg: "index [02]",
+
+	event2 := Event{
+		Fields: map[string]interface{}{
+			"message_type": "OUT",
+			"cf_app_id":    "022",
 		},
+		Msg: "index [02]",
 	}
-	node3 := Node{
-		Event: Event{
-			Fields: map[string]interface{}{
-				"message_type": "OUT",
-				"cf_app_id":    "033",
-			},
-			Msg: "index [03]",
+
+	event3 := Event{
+		Fields: map[string]interface{}{
+			"message_type": "OUT",
+			"cf_app_id":    "033",
 		},
+		Msg: "index [03]",
 	}
 
 	queue := Queue{
-		Nodes: make([]*Node, 3),
+		Events: make([]*Event, 3),
 	}
 
-	queue.Push(&node1)
-	queue.Push(&node2)
-	queue.Push(&node3)
+	queue.Push(&event1)
+	queue.Push(&event2)
+	queue.Push(&event3)
 
-	assert.Equal(queue.Pop().Event.Msg, "index [01]", "")
-	assert.Equal(queue.Pop().Event.Msg, "index [02]", "")
-	assert.Equal(queue.Pop().Event.Msg, "index [03]", "")
+	assert.Equal(queue.Pop().Msg, "index [01]", "")
+	assert.Equal(queue.Pop().Msg, "index [02]", "")
+	assert.Equal(queue.Pop().Msg, "index [03]", "")
 }

eventRouting/eventrouting.go

@@ -76,10 +76,8 @@ func (e *EventRouting) RouteEvent(msg *events.Envelope) {
 			e.selectedEventsCount["ignored_app_message"]++
 		} else {
 			//Push the event to the queue
-			fmt.Println("pushing event to queue")
-			e.queue.Push(eventQueue.NewNode(*event))
+			e.queue.Push(event)
 			e.selectedEventsCount[eventType.String()]++
-
 		}
 		e.mutex.Unlock()
 	}
@@ -147,7 +145,7 @@ func (e *EventRouting) LogEventTotals(logTotalsTime time.Duration) {
 			count = lastCount
 
 			//Push the event to the queue
-			e.queue.Push(eventQueue.NewNode(*event))
+			e.queue.Push(event)
 		}
 	}()
 }

firehoseclient/firehoseclient.go

@@ -2,10 +2,10 @@ package firehoseclient
 
 import (
 	"crypto/tls"
-	"fmt"
 	"time"
 
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventRouting"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/logging"
 	"github.com/cloudfoundry-community/go-cfclient"
 	"github.com/cloudfoundry/noaa/consumer"
 	"github.com/cloudfoundry/sonde-go/events"
@@ -43,11 +43,10 @@ func NewFirehoseNozzle(cfClient *cfclient.Client, eventRouting *eventRouting.Eve
 }
 
 func (f *FirehoseNozzle) Start() error {
-	fmt.Printf("Started the Nozzle... \n")
+	logging.Info.Printf("Started the Nozzle... \n")
 	f.consumeFirehose()
-	fmt.Printf("consume the firehose... \n")
+	logging.Info.Printf("consume the firehose... \n")
 	err := f.routeEvent()
-	fmt.Printf("route event... \n")
 	return err
 }
 

logging/logging.go

@@ -0,0 +1,45 @@
+package logging
+
+import (
+	"io"
+	"log"
+)
+
+var (
+	Trace   *log.Logger
+	Info    *log.Logger
+	Warning *log.Logger
+	Error   *log.Logger
+)
+
+func Init(
+	traceHandle io.Writer,
+	infoHandle io.Writer,
+	warningHandle io.Writer,
+	errorHandle io.Writer) {
+
+	Trace = log.New(traceHandle,
+		"TRACE: ",
+		log.Ldate|log.Ltime|log.Lshortfile)
+
+	Info = log.New(infoHandle,
+		"INFO: ",
+		log.Ldate|log.Ltime|log.Lshortfile)
+
+	Warning = log.New(warningHandle,
+		"WARNING: ",
+		log.Ldate|log.Ltime|log.Lshortfile)
+
+	Error = log.New(errorHandle,
+		"ERROR: ",
+		log.Ldate|log.Ltime|log.Lshortfile)
+}
+
+/*func main() {
+    Init(ioutil.Discard, os.Stdout, os.Stdout, os.Stderr)
+
+    Trace.Println("I have something standard to say")
+    Info.Println("Special Information")
+    Warning.Println("There is something you need to know about")
+    Error.Println("Something has failed")
+}*/

main.go

@@ -2,24 +2,25 @@ package main
 
 import (
 	"fmt"
-	"log"
+	"io/ioutil"
 	"os"
-	"time"
-
 	"runtime"
+	"time"
 
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/caching"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventRouting"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/events"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/firehoseclient"
+	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/logging"
 	"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/sumoCFFirehose"
 	"github.com/cloudfoundry-community/go-cfclient"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
 
 var (
-	debug             = true //debug", "Enable debug mode, print in console
-	apiEndpoint       = "https://api.bosh-lite.com"
+	debug             = true                                                   //debug", "Enable debug mode, print in console
+	apiEndpoint       = kingpin.Flag("api-endpoint", "Sumo Endpoint").String() //"https://api.bosh-lite.com"
 	sumoEndpoint      = kingpin.Flag("sumo-endpoint", "Sumo Endpoint").String()
 	dopplerEndpoint   = kingpin.Flag("doppler-endpoint", "Overwrite default doppler endpoint return by /v2/info").OverrideDefaultFromEnvar("DOPPLER_ENDPOINT").String()
 	subscriptionId    = kingpin.Flag("subscription-id", "Id for the subscription.").Default("firehose").OverrideDefaultFromEnvar("FIREHOSE_SUBSCRIPTION_ID").String()
@@ -38,14 +39,27 @@ var (
 )
 
 func main() {
+	//logging init
+	logging.Init(ioutil.Discard, os.Stdout, os.Stdout, os.Stderr)
+
 	kingpin.Version(version)
 	kingpin.Parse()
+
 	runtime.GOMAXPROCS(1)
 
-	fmt.Printf("Starting firehose-to-sumo %s \n", version)
+	logging.Info.Println("Configurations set:")
+	logging.Info.Println("Api Endpoint: " + *apiEndpoint)
+	logging.Info.Println("Sumo Endpoint: " + *sumoEndpoint)
+	logging.Info.Println("Cloudfoundry Doppler Endpoint: " + *dopplerEndpoint)
+	logging.Info.Println("Cloudfoundry Nozzle Subscription ID: " + *subscriptionId)
+	logging.Info.Println("Cloudfoundry User: " + user)
+
+	logging.Info.Printf("Events Batch Size: [%d]\n", eventsBatchSize)
+
+	logging.Info.Println("Starting firehose-to-sumo " + version)
 
 	c := cfclient.Config{
-		ApiAddress:        apiEndpoint,
+		ApiAddress:        *apiEndpoint,
 		Username:          user,
 		Password:          password,
 		SkipSslValidation: *skipSSLValidation,
@@ -64,26 +78,26 @@ func main() {
 		cachingClient = caching.NewCachingEmpty()
 	}
 
-	//Creating queue
-	queue := eventQueue.NewQueue(make([]*eventQueue.Node, 100))
+	logging.Info.Println("Creating queue")
+	queue := eventQueue.NewQueue(make([]*events.Event, 100))
 	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize)
 	go loggingClientSumo.Start() //multi
 
-	//Creating Events
+	logging.Info.Println("Creating Events")
 	events := eventRouting.NewEventRouting(cachingClient, *loggingClientSumo, &queue)
 	err := events.SetupEventRouting(*wantedEvents)
 	if err != nil {
-		log.Fatal("Error setting up event routing: ", err)
+		logging.Error.Fatal("Error setting up event routing: ", err)
 		os.Exit(1)
 
 	}
 
 	// Parse extra fields from cmd call
 	cachingClient.CreateBucket()
 	//Let's Update the database the first time
-	fmt.Printf("Start filling app/space/org cache.\n")
+	logging.Info.Printf("Start filling app/space/org cache.\n")
 	apps := cachingClient.GetAllApp()
-	fmt.Printf("Done filling cache! Found [%d] Apps \n", len(apps))
+	logging.Info.Printf("Done filling cache! Found [%d] Apps \n", len(apps))
 
 	//Let's start the goRoutine
 	cachingClient.PerformPoollingCaching(tickerTime)
@@ -95,15 +109,12 @@ func main() {
 		FirehoseSubscriptionID: *subscriptionId,
 	}
 
-	if /*loggingClientSumo.Connect() ||*/ debug {
+	logging.Info.Printf("Connecting to Firehose... \n")
 
-		fmt.Printf("Connected to Server! Connecting to Firehose... \n")
+	firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
+	go firehoseClient.Start()
 
-		firehoseClient := firehoseclient.NewFirehoseNozzle(cfClient, events, firehoseConfig)
-		go firehoseClient.Start()
-
-		defer firehoseClient.Start()
-	}
+	defer firehoseClient.Start()
 
 	defer cachingClient.Close()