SCFF-35 Added timer for sumo post minimum and Default values

Author: nmillalaf

main.go

@@ -19,19 +19,20 @@ import (
 )
 
 var (
-	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()
-	user              = "firehose_user"     //user created in CF, authorized to connect the firehose
-	password          = "firehose_password" // password created along with the firehose_user
-	skipSSLValidation = kingpin.Flag("skip-ssl-validation", "Please don't").Default("false").OverrideDefaultFromEnvar("SKIP_SSL_VALIDATION").Bool()
-	keepAlive, errK   = time.ParseDuration("25s") //default
-	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()
+	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()
+	user                 = "firehose_user"     //user created in CF, authorized to connect the firehose
+	password             = "firehose_password" // password created along with the firehose_user
+	skipSSLValidation    = kingpin.Flag("skip-ssl-validation", "Please don't").Default("false").OverrideDefaultFromEnvar("SKIP_SSL_VALIDATION").Bool()
+	keepAlive, errK      = time.ParseDuration("25s") //default
+	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").Default("10").Int()
+	sumoPostMinimumDelay = kingpin.Flag("sumo-Post-Minimum-Delay", "Sumo Post Minimum Delay").Default("10s").Duration()
 )
 
 var (
@@ -80,7 +81,7 @@ func main() {
 
 	logging.Info.Println("Creating queue")
 	queue := eventQueue.NewQueue(make([]*events.Event, 100))
-	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize)
+	loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize, *sumoPostMinimumDelay)
 	go loggingClientSumo.Start() //multi
 
 	logging.Info.Println("Creating Events")

sumoCFFirehose/sumoLogicAppender.go

@@ -18,9 +18,11 @@ type SumoLogicAppender struct {
 	eventsBatchSize          int
 	logEventsInCurrentBuffer int
 	logStringToSend          *bytes.Buffer
+	sumoPostMinimumDelay     time.Duration
+	timerPostMinimum         time.Time
 }
 
-func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int) *SumoLogicAppender {
+func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int, sumoPostMinimumDelay time.Duration) *SumoLogicAppender {
 	return &SumoLogicAppender{
 		url:                      urlValue,
 		connectionTimeout:        connectionTimeoutValue,
@@ -29,14 +31,16 @@ func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQue
 		eventsBatchSize:          eventsBatchSize,
 		logEventsInCurrentBuffer: 0,
 		logStringToSend:          bytes.NewBufferString(""),
+		sumoPostMinimumDelay:     sumoPostMinimumDelay,
 	}
 }
 
 func (s *SumoLogicAppender) Start() {
 	timer := time.Now()
+	s.timerPostMinimum = time.Now() //starting timer for sumo post minimum
 	logging.Info.Println("Starting Appender Worker")
 	for {
-		time.Sleep(300 * time.Millisecond)
+		time.Sleep(300 * time.Millisecond) //delay
 		// while queue is not empty && s.eventsBatchSize not completed, queue.POP (appendLogs)
 		for s.nozzleQueue.GetCount() != 0 && s.logEventsInCurrentBuffer <= s.eventsBatchSize {
 			s.AppendLogs()                                       //this method POP an event from queue
@@ -75,24 +79,28 @@ func (s *SumoLogicAppender) AppendLogs() {
 }
 
 func (s *SumoLogicAppender) SendToSumo(log *bytes.Buffer) {
-	logging.Trace.Println("Sending logs to Sumologic...")
-	request, err := http.NewRequest("POST", s.url, log)
-	if err != nil {
-		logging.Error.Printf("http.NewRequest() error: %v\n", err)
-		return
-	}
-	//request.Header.Add("content-type", "application/json")
-	//request.SetBasicAuth("admin", "admin")
-	response, err := s.httpClient.Do(request)
+	//wait period between posts to Sumo
+	if time.Since(s.timerPostMinimum) >= s.sumoPostMinimumDelay {
+		logging.Trace.Println("Sending logs to Sumologic...")
+		request, err := http.NewRequest("POST", s.url, log)
+		if err != nil {
+			logging.Error.Printf("http.NewRequest() error: %v\n", err)
+			return
+		}
+		//request.Header.Add("content-type", "application/json")
+		//request.SetBasicAuth("admin", "admin")
+		response, err := s.httpClient.Do(request)
 
-	if err != nil {
-		logging.Error.Printf("http.Do() error: %v\n", err)
-		return
-	} else {
-		logging.Trace.Println("Do(Request) successful")
+		if err != nil {
+			logging.Error.Printf("http.Do() error: %v\n", err)
+			return
+		} else {
+			logging.Trace.Println("Do(Request) successful")
+		}
+		s.logEventsInCurrentBuffer = 0                // reset counter
+		s.logStringToSend = bytes.NewBufferString("") //reset String
+		defer response.Body.Close()
+		s.timerPostMinimum = time.Now() //reset timer post minimum
 	}
-	s.logEventsInCurrentBuffer = 0                // reset counter
-	s.logStringToSend = bytes.NewBufferString("") //reset String
-	defer response.Body.Close()
 
 }