Skip to content

Commit fd600bb

Browse files
nmillalafnmillalaf
committed
Merge branch 'hotfix/timestamp'
2 parents 68cf70c + 6a3bf8b commit fd600bb

2 files changed

Lines changed: 143 additions & 15 deletions

File tree

‎sumoCFFirehose/sumoLogicAppender.go‎

Lines changed: 33 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@ import (
77
"errors"
88
"fmt"
99
"net/http"
10+
"reflect"
11+
"strconv"
1012
"strings"
1113
"time"
1214

@@ -79,7 +81,6 @@ func (s *SumoLogicAppender) Start() {
7981

8082
if time.Since(Buffer.timerIdlebuffer).Seconds() >= 10 && Buffer.logEventsInCurrentBuffer > 0 {
8183
logging.Info.Println("Sending current batch of logs after timer exceeded limit")
82-
//wg.Add(1)
8384
go s.SendToSumo(Buffer.logStringToSend.String())
8485
Buffer = newBuffer()
8586
Buffer.timerIdlebuffer = time.Now()
@@ -147,6 +148,30 @@ func WantedEvent(event string, includeOnlyMatchingFilter string, excludeAlwaysMa
147148
}
148149
return true
149150

151+
}
152+
func FormatTimestamp(event *events.Event, timestamp string) {
153+
defer func() {
154+
if r := recover(); r != nil {
155+
fmt.Println("Recovered in f", r)
156+
}
157+
}()
158+
159+
if reflect.TypeOf(event.Fields[timestamp]).Kind() != reflect.Int64 {
160+
if reflect.TypeOf(event.Fields[timestamp]).Kind() == reflect.String {
161+
event.Fields[timestamp] = event.Fields[timestamp].(string)
162+
} else {
163+
event.Fields[timestamp] = ""
164+
}
165+
166+
}
167+
if reflect.TypeOf(event.Fields[timestamp]).Kind() == reflect.Int64 {
168+
if len(strconv.FormatInt(event.Fields[timestamp].(int64), 10)) == 19 {
169+
event.Fields[timestamp] = time.Unix(0, event.Fields[timestamp].(int64)*int64(time.Nanosecond)).String()
170+
} else if len(strconv.FormatInt(event.Fields[timestamp].(int64), 10)) < 13 {
171+
event.Fields[timestamp] = ""
172+
}
173+
}
174+
150175
}
151176

152177
func StringBuilder(event *events.Event, verboseLogMessages bool, includeOnlyMatchingFilter string, excludeAlwaysMatchingFilter string, customMetadata string) string {
@@ -160,31 +185,26 @@ func StringBuilder(event *events.Event, verboseLogMessages bool, includeOnlyMatc
160185
var msg []byte
161186
switch eventType {
162187
case "HttpStart":
163-
timestamp := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String()
164-
event.Fields["timestamp"] = timestamp
188+
FormatTimestamp(event, "timestamp")
165189
message, err := json.Marshal(event)
166190
if err == nil {
167191
msg = message
168192
}
169193
case "HttpStop":
170-
timestamp := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String()
171-
event.Fields["timestamp"] = timestamp
194+
FormatTimestamp(event, "timestamp")
172195
message, err := json.Marshal(event)
173196
if err == nil {
174197
msg = message
175198
}
176199
case "HttpStartStop":
177-
start_timestamp := time.Unix(0, event.Fields["start_timestamp"].(int64)*int64(time.Nanosecond)).String()
178-
event.Fields["start_timestamp"] = start_timestamp
179-
stop_timestamp := time.Unix(0, event.Fields["stop_timestamp"].(int64)*int64(time.Nanosecond)).String()
180-
event.Fields["stop_timestamp"] = stop_timestamp
200+
FormatTimestamp(event, "start_timestamp")
201+
FormatTimestamp(event, "stop_timestamp")
181202
message, err := json.Marshal(event)
182203
if err == nil {
183204
msg = message
184205
}
185206
case "LogMessage":
186-
timestamp := time.Unix(0, event.Fields["timestamp"].(int64)*int64(time.Nanosecond)).String()
187-
event.Fields["timestamp"] = timestamp
207+
FormatTimestamp(event, "timestamp")
188208
if verboseLogMessages == true {
189209
message, err := json.Marshal(event)
190210
if err == nil {
@@ -279,7 +299,6 @@ func (s *SumoLogicAppender) SendToSumo(logStringToSend string) {
279299
if s.sumoCategory != "" {
280300
request.Header.Add("X-Sumo-Category", s.sumoCategory)
281301
}
282-
/*REMOVE*/ fmt.Println(time.Since(s.timerBetweenPost))
283302
//checking the timer before first POST intent
284303
for time.Since(s.timerBetweenPost) < s.sumoPostMinimumDelay {
285304
logging.Trace.Println("Delaying Post because minimum post timer not expired")
@@ -310,7 +329,6 @@ func (s *SumoLogicAppender) SendToSumo(logStringToSend string) {
310329
if s.sumoCategory != "" {
311330
request.Header.Add("X-Sumo-Category", s.sumoCategory)
312331
}
313-
314332
//checking the timer before POST (retry intent)
315333
for time.Since(s.timerBetweenPost) < s.sumoPostMinimumDelay {
316334
logging.Trace.Println("Delaying Post because minimum post timer not expired")
@@ -329,7 +347,7 @@ func (s *SumoLogicAppender) SendToSumo(logStringToSend string) {
329347
time.Sleep(300 * time.Millisecond)
330348
return attempt < 5, errRetry
331349
} else if response.StatusCode == 200 {
332-
logging.Info. /*Trace.*/ Println("Post of logs successful after retry...")
350+
logging.Trace.Println("Post of logs successful after retry...")
333351
s.timerBetweenPost = time.Now()
334352
statusCode = response.StatusCode
335353
return true, err
@@ -344,7 +362,7 @@ func (s *SumoLogicAppender) SendToSumo(logStringToSend string) {
344362
logging.Error.Printf("Not able to post after retry, with status code: %d", statusCode)
345363
}
346364
} else if response.StatusCode == 200 {
347-
logging.Info. /*Trace.*/ Println("Post of logs successful")
365+
logging.Trace.Println("Post of logs successful")
348366
s.timerBetweenPost = time.Now()
349367
}
350368

‎sumoCFFirehose/sumoLogicAppender_test.go‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -244,3 +244,113 @@ func TestSendNoFilter(t *testing.T) {
244244
buf.Write(msg)
245245
assert.True(t, WantedEvent(buf.String(), "", ""), "This Event should be included")
246246
}
247+
248+
func TestSendStringTimestamp(t *testing.T) {
249+
eventStringTimestamp := Event{
250+
Fields: map[string]interface{}{
251+
"message_type": "OUT",
252+
"source_instance": 0,
253+
"deployment": "cf",
254+
"ip": "10.193.166.47",
255+
"job": "diego_cell",
256+
"job_index": "c62aebe5-16b8-43f5-a589-1267e09b9537",
257+
"cf_ignored_app": "false",
258+
"timestamp": "1483629662001580569",
259+
"source_type": "APP",
260+
"origin": "rep",
261+
"cf_app_id": "7833dc75-4484-409c-9b74-90b6454906c6",
262+
},
263+
Msg: "Triggering 'app usage events fetcher'",
264+
Type: "LogMessage",
265+
}
266+
FormatTimestamp(&eventStringTimestamp, "timestamp")
267+
timestamp := eventStringTimestamp.Fields["timestamp"]
268+
assert.Equal(t, timestamp, "1483629662001580569", "This timestamp should be in the string")
269+
}
270+
271+
func TestSendInt64Timestamp19(t *testing.T) {
272+
eventStringTimestamp := Event{
273+
Fields: map[string]interface{}{
274+
"message_type": "OUT",
275+
"source_instance": 0,
276+
"deployment": "cf",
277+
"ip": "10.193.166.47",
278+
"job": "diego_cell",
279+
"job_index": "c62aebe5-16b8-43f5-a589-1267e09b9537",
280+
"cf_ignored_app": "false",
281+
"timestamp": int64(1483629662001580569),
282+
"source_type": "APP",
283+
"origin": "rep",
284+
"cf_app_id": "7833dc75-4484-409c-9b74-90b6454906c6",
285+
},
286+
Msg: "Triggering 'app usage events fetcher'",
287+
Type: "LogMessage",
288+
}
289+
FormatTimestamp(&eventStringTimestamp, "timestamp")
290+
timestamp := eventStringTimestamp.Fields["timestamp"]
291+
assert.Equal(t, timestamp, "2017-01-05 12:21:02.001580569 -0300 CLST", "This timestamp should be in the string")
292+
}
293+
func TestSendInt64Timestamp14(t *testing.T) {
294+
eventStringTimestamp := Event{
295+
Fields: map[string]interface{}{
296+
"message_type": "OUT",
297+
"source_instance": 0,
298+
"deployment": "cf",
299+
"ip": "10.193.166.47",
300+
"job": "diego_cell",
301+
"job_index": "c62aebe5-16b8-43f5-a589-1267e09b9537",
302+
"cf_ignored_app": "false",
303+
"timestamp": int64(148362966200),
304+
"source_type": "APP",
305+
"origin": "rep",
306+
"cf_app_id": "7833dc75-4484-409c-9b74-90b6454906c6",
307+
},
308+
Msg: "Triggering 'app usage events fetcher'",
309+
Type: "LogMessage",
310+
}
311+
FormatTimestamp(&eventStringTimestamp, "timestamp")
312+
timestamp := eventStringTimestamp.Fields["timestamp"]
313+
assert.Equal(t, timestamp, "", "This timestamp should be in the string")
314+
}
315+
func TestSendWrongTimestampField(t *testing.T) {
316+
eventStringTimestamp := Event{
317+
Fields: map[string]interface{}{
318+
"message_type": "OUT",
319+
"source_instance": 0,
320+
"deployment": "cf",
321+
"ip": "10.193.166.47",
322+
"job": "diego_cell",
323+
"job_index": "c62aebe5-16b8-43f5-a589-1267e09b9537",
324+
"cf_ignored_app": "false",
325+
"timestamp": int64(148362966200),
326+
"source_type": "APP",
327+
"origin": "rep",
328+
"cf_app_id": "7833dc75-4484-409c-9b74-90b6454906c6",
329+
},
330+
Msg: "Triggering 'app usage events fetcher'",
331+
Type: "LogMessage",
332+
}
333+
assert.NotPanics(t, func() { FormatTimestamp(&eventStringTimestamp, "timestamp2") }, "msgAndArgs")
334+
}
335+
func TestSendNotIntNotStringTimestampField(t *testing.T) {
336+
eventStringTimestamp := Event{
337+
Fields: map[string]interface{}{
338+
"message_type": "OUT",
339+
"source_instance": 0,
340+
"deployment": "cf",
341+
"ip": "10.193.166.47",
342+
"job": "diego_cell",
343+
"job_index": "c62aebe5-16b8-43f5-a589-1267e09b9537",
344+
"cf_ignored_app": "false",
345+
"timestamp": float32(148362966200),
346+
"source_type": "APP",
347+
"origin": "rep",
348+
"cf_app_id": "7833dc75-4484-409c-9b74-90b6454906c6",
349+
},
350+
Msg: "Triggering 'app usage events fetcher'",
351+
Type: "LogMessage",
352+
}
353+
FormatTimestamp(&eventStringTimestamp, "timestamp")
354+
timestamp := eventStringTimestamp.Fields["timestamp"]
355+
assert.Equal(t, timestamp, "", "This timestamp should be in the string")
356+
}

0 commit comments

Comments
 (0)