Skip to content

Commit 0201a8c

Browse files
nmillalafnmillalaf
committed
Merge branch 'feature/SCFF-35' into develop
2 parents 87bc0c0 + 87b181b commit 0201a8c

2 files changed

Lines changed: 95 additions & 53 deletions

File tree

‎main.go‎

Lines changed: 15 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -19,19 +19,20 @@ import (
1919
)
2020

2121
var (
22-
debug = true //debug", "Enable debug mode, print in console
23-
apiEndpoint = kingpin.Flag("api-endpoint", "Sumo Endpoint").String() //"https://api.bosh-lite.com"
24-
sumoEndpoint = kingpin.Flag("sumo-endpoint", "Sumo Endpoint").String()
25-
dopplerEndpoint = kingpin.Flag("doppler-endpoint", "Overwrite default doppler endpoint return by /v2/info").OverrideDefaultFromEnvar("DOPPLER_ENDPOINT").String()
26-
subscriptionId = kingpin.Flag("subscription-id", "Id for the subscription.").Default("firehose").OverrideDefaultFromEnvar("FIREHOSE_SUBSCRIPTION_ID").String()
27-
user = "firehose_user" //user created in CF, authorized to connect the firehose
28-
password = "firehose_password" // password created along with the firehose_user
29-
skipSSLValidation = kingpin.Flag("skip-ssl-validation", "Please don't").Default("false").OverrideDefaultFromEnvar("SKIP_SSL_VALIDATION").Bool()
30-
keepAlive, errK = time.ParseDuration("25s") //default
31-
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()
32-
boltDatabasePath = "my.db" //default
33-
tickerTime, errT = time.ParseDuration("60s") //Default
34-
eventsBatchSize = kingpin.Flag("log-events-batch-size", "Log Events Batch Size").Int()
22+
debug = true //debug", "Enable debug mode, print in console
23+
apiEndpoint = kingpin.Flag("api-endpoint", "Sumo Endpoint").String() //"https://api.bosh-lite.com"
24+
sumoEndpoint = kingpin.Flag("sumo-endpoint", "Sumo Endpoint").String()
25+
dopplerEndpoint = kingpin.Flag("doppler-endpoint", "Overwrite default doppler endpoint return by /v2/info").OverrideDefaultFromEnvar("DOPPLER_ENDPOINT").String()
26+
subscriptionId = kingpin.Flag("subscription-id", "Id for the subscription.").Default("firehose").OverrideDefaultFromEnvar("FIREHOSE_SUBSCRIPTION_ID").String()
27+
user = "firehose_user" //user created in CF, authorized to connect the firehose
28+
password = "firehose_password" // password created along with the firehose_user
29+
skipSSLValidation = kingpin.Flag("skip-ssl-validation", "Please don't").Default("false").OverrideDefaultFromEnvar("SKIP_SSL_VALIDATION").Bool()
30+
keepAlive, errK = time.ParseDuration("25s") //default
31+
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()
32+
boltDatabasePath = "my.db" //default
33+
tickerTime, errT = time.ParseDuration("60s") //Default
34+
eventsBatchSize = kingpin.Flag("log-events-batch-size", "Log Events Batch Size").Default("10").Int()
35+
sumoPostMinimumDelay = kingpin.Flag("sumo-Post-Minimum-Delay", "Sumo Post Minimum Delay").Default("500ms").Duration()
3536
)
3637

3738
var (
@@ -80,7 +81,7 @@ func main() {
8081

8182
logging.Info.Println("Creating queue")
8283
queue := eventQueue.NewQueue(make([]*events.Event, 100))
83-
loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize)
84+
loggingClientSumo := sumoCFFirehose.NewSumoLogicAppender(*sumoEndpoint, 1000, &queue, *eventsBatchSize, *sumoPostMinimumDelay)
8485
go loggingClientSumo.Start() //multi
8586

8687
logging.Info.Println("Creating Events")

‎sumoCFFirehose/sumoLogicAppender.go‎

Lines changed: 80 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package sumoCFFirehose
33
import (
44
"bytes"
55
"net/http"
6+
"runtime"
67
"time"
78

89
"bitbucket.org/mcplusa-ondemand/firehose-to-sumologic/eventQueue"
@@ -11,47 +12,84 @@ import (
1112
)
1213

1314
type SumoLogicAppender struct {
14-
url string
15-
connectionTimeout int //10000
16-
httpClient http.Client
17-
nozzleQueue *eventQueue.Queue
18-
eventsBatchSize int
19-
logEventsInCurrentBuffer int
15+
url string
16+
connectionTimeout int //10000
17+
httpClient http.Client
18+
nozzleQueue *eventQueue.Queue
19+
eventsBatchSize int
20+
sumoPostMinimumDelay time.Duration
21+
timerBetweenPost time.Time
22+
}
23+
24+
type SumoBuffer struct {
2025
logStringToSend *bytes.Buffer
26+
logEventsInCurrentBuffer int
27+
timerIdlebuffer time.Time
2128
}
2229

23-
func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int) *SumoLogicAppender {
30+
func NewSumoLogicAppender(urlValue string, connectionTimeoutValue int, nozzleQueue *eventQueue.Queue, eventsBatchSize int, sumoPostMinimumDelay time.Duration) *SumoLogicAppender {
2431
return &SumoLogicAppender{
25-
url: urlValue,
26-
connectionTimeout: connectionTimeoutValue,
27-
httpClient: http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
28-
nozzleQueue: nozzleQueue,
29-
eventsBatchSize: eventsBatchSize,
30-
logEventsInCurrentBuffer: 0,
32+
url: urlValue,
33+
connectionTimeout: connectionTimeoutValue,
34+
httpClient: http.Client{Timeout: time.Duration(connectionTimeoutValue * int(time.Millisecond))},
35+
nozzleQueue: nozzleQueue,
36+
eventsBatchSize: eventsBatchSize,
37+
sumoPostMinimumDelay: sumoPostMinimumDelay,
38+
}
39+
}
40+
41+
func newBuffer() SumoBuffer {
42+
return SumoBuffer{
3143
logStringToSend: bytes.NewBufferString(""),
44+
logEventsInCurrentBuffer: 0,
3245
}
3346
}
3447

3548
func (s *SumoLogicAppender) Start() {
36-
timer := time.Now()
49+
s.timerBetweenPost = time.Now()
50+
runtime.GOMAXPROCS(1)
51+
Buffer := newBuffer()
52+
Buffer.timerIdlebuffer = time.Now()
3753
logging.Info.Println("Starting Appender Worker")
3854
for {
39-
time.Sleep(300 * time.Millisecond)
40-
// while queue is not empty && s.eventsBatchSize not completed, queue.POP (appendLogs)
41-
for s.nozzleQueue.GetCount() != 0 && s.logEventsInCurrentBuffer <= s.eventsBatchSize {
42-
s.AppendLogs() //this method POP an event from queue
43-
timer = time.Now() //reset timer
44-
if s.logEventsInCurrentBuffer == s.eventsBatchSize { //if buffer is full, send logs to sumo
45-
logging.Trace.Println("Batch Size complete")
46-
break
47-
} else if time.Since(timer).Seconds() >= 10 { // else if timer is up, send existing logs to sumo
48-
logging.Trace.Println("Sending current batch of logs after timer exceeded limit")
49-
break
55+
logging.Info.Println("Log queue size: ")
56+
logging.Info.Println(s.nozzleQueue.GetCount())
57+
if s.nozzleQueue.GetCount() == 0 {
58+
logging.Trace.Println("Waiting for 300 ms")
59+
time.Sleep(300 * time.Millisecond)
60+
}
61+
62+
if time.Since(Buffer.timerIdlebuffer).Seconds() >= 10 && Buffer.logEventsInCurrentBuffer > 0 {
63+
logging.Info.Println("Sending current batch of logs after timer exceeded limit")
64+
go s.SendToSumo(&Buffer)
65+
Buffer = newBuffer()
66+
Buffer.timerIdlebuffer = time.Now()
67+
continue
68+
}
69+
70+
if s.nozzleQueue.GetCount() != 0 {
71+
queueCount := s.nozzleQueue.GetCount()
72+
remainingBufferCount := s.eventsBatchSize - Buffer.logEventsInCurrentBuffer
73+
if queueCount >= remainingBufferCount {
74+
logging.Trace.Println("Pushing Logs to Sumo: ")
75+
logging.Trace.Println(remainingBufferCount)
76+
for i := 0; i < remainingBufferCount; i++ {
77+
s.AppendLogs(&Buffer)
78+
Buffer.timerIdlebuffer = time.Now()
79+
}
80+
go s.SendToSumo(&Buffer)
81+
Buffer = newBuffer()
82+
} else {
83+
logging.Trace.Println("Pushing Logs to Buffer: ")
84+
logging.Trace.Println(queueCount)
85+
for i := 0; i < queueCount; i++ {
86+
s.AppendLogs(&Buffer)
87+
Buffer.timerIdlebuffer = time.Now()
88+
}
5089
}
5190
}
52-
s.SendToSumo(s.logStringToSend)
53-
timer = time.Now() //reset timer
5491
}
92+
5593
}
5694

5795
func StringBuilder(event *events.Event) string {
@@ -68,31 +106,34 @@ func StringBuilder(event *events.Event) string {
68106
return buf.String()
69107
}
70108

71-
func (s *SumoLogicAppender) AppendLogs() {
72-
// the appender calls for the next message in the queue and parse it to a string
73-
s.logStringToSend.Write([]byte(StringBuilder(s.nozzleQueue.Pop())))
74-
s.logEventsInCurrentBuffer++
109+
func (s *SumoLogicAppender) AppendLogs(buffer *SumoBuffer) {
110+
buffer.logStringToSend.Write([]byte(StringBuilder(s.nozzleQueue.Pop())))
111+
buffer.logEventsInCurrentBuffer++
112+
75113
}
76114

77-
func (s *SumoLogicAppender) SendToSumo(log *bytes.Buffer) {
78-
logging.Trace.Println("Sending logs to Sumologic...")
79-
request, err := http.NewRequest("POST", s.url, log)
115+
func (s *SumoLogicAppender) SendToSumo(buffer *SumoBuffer) {
116+
for time.Since(s.timerBetweenPost) < s.sumoPostMinimumDelay {
117+
logging.Trace.Println("Delaying post to honor minimum post delay")
118+
time.Sleep(100 * time.Millisecond)
119+
}
120+
121+
logging.Info.Println("Sending logs to Sumologic...")
122+
request, err := http.NewRequest("POST", s.url, buffer.logStringToSend)
80123
if err != nil {
81124
logging.Error.Printf("http.NewRequest() error: %v\n", err)
82125
return
83126
}
84127
//request.Header.Add("content-type", "application/json")
85128
//request.SetBasicAuth("admin", "admin")
86129
response, err := s.httpClient.Do(request)
87-
88130
if err != nil {
89131
logging.Error.Printf("http.Do() error: %v\n", err)
90132
return
91133
} else {
92-
logging.Trace.Println("Do(Request) successful")
134+
logging.Trace.Println("Post of logs successful")
135+
s.timerBetweenPost = time.Now()
93136
}
94-
s.logEventsInCurrentBuffer = 0 // reset counter
95-
s.logStringToSend = bytes.NewBufferString("") //reset String
96-
defer response.Body.Close()
97137

138+
defer response.Body.Close()
98139
}

0 commit comments

Comments
 (0)