Skip to content

Feature/add debug messages #67

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sl-kenlee wants to merge 7 commits into
base: master
Choose a base branch
from
Open

Feature/add debug messages #67

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
221ab87
chore: add logs for bug trace
wood-shopline Apr 14, 2022
53405a4
chore: log delay send event time remain
wood-shopline Apr 14, 2022
aff7521
chore(dispatcher): add log
terencewu-shopline Apr 14, 2022
ecdf1ea
feat(add debug messages): add log for error cases
sl-kenlee May 16, 2022
272aee9
feat(add debug messages): add log before error and time remaining bef…
sl-kenlee May 20, 2022
3f8b0f3
feat(add debug messages): update logging level
sl-kenlee May 20, 2022
b59a18d
Merge remote-tracking branch 'origin/master' into feature/add-debug-m…
Sep 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 36 additions & 1 deletion internal/outgoing/dispatcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,17 @@ func (d *Dispatcher) processDelayedEvent(e models.IncomingEvent, timeRemain time
// Check if store have payload
dataKey := getEventDataKey(store, e, dest)
storedData, dataExists, _, storeErr := store.Get(dataKey)

dLogger.WithFields(log.Fields{
"event": e.GetTraceInfo(),
"eventDataKey": dataKey,
}).Info("Data exists: " + strconv.FormatBool(dataExists) + ", stored data: " + storedData)

if storeErr != nil {
dLogger.WithFields(log.Fields{
"event": e.GetTraceInfo(),
"eventDataKey": dataKey,
}).Error("Get stored data error: ", storeErr)
panic(storeErr)
}

Expand All @@ -268,6 +278,10 @@ func (d *Dispatcher) processDelayedEvent(e models.IncomingEvent, timeRemain time
queueKey := getEventThrottledPayloadsKey(store, e, dest)
jsonString, jsonErr := json.Marshal(customizedPayload)
if jsonErr != nil {
dLogger.WithFields(log.Fields{
"queueKey": queueKey,
"event": e.GetTraceInfo(),
}).Error("Marshal throttled payload error")
panic(jsonErr)
}
dLogger.WithFields(log.Fields{
Expand All @@ -283,6 +297,10 @@ func (d *Dispatcher) processDelayedEvent(e models.IncomingEvent, timeRemain time
queueKey := getEventThrottledDocumentsKey(store, e, dest)
jsonString, jsonErr := json.Marshal(customizedDocument)
if jsonErr != nil {
dLogger.WithFields(log.Fields{
"queueKey": queueKey,
"event": e.GetTraceInfo(),
}).Error("Marshal throttled document error")
panic(jsonErr)
}
dLogger.WithFields(log.Fields{
Expand All @@ -295,27 +313,42 @@ func (d *Dispatcher) processDelayedEvent(e models.IncomingEvent, timeRemain time

jsonString, jsonErr := json.Marshal(e)
if jsonErr != nil {
dLogger.WithFields(log.Fields{
"event": e.GetTraceInfo(),
"enqueuePayload": jsonString,
}).Error("Marshal event error")
panic(jsonErr)
}

if dataExists {
// Update Value
_, updateErr := store.Update(dataKey, string(jsonString))
if updateErr != nil {
dLogger.WithFields(log.Fields{
"event": e.GetTraceInfo(),
"dataKey": dataKey,
}).Error("Update payload error: ", updateErr)
panic(updateErr)
}
} else {
// Create Value
config := dest.Config
dLogger.WithFields(log.Fields{"key": dataKey, "event": e.GetTraceInfo(), "payload": string(jsonString)}).Info("Store data for delayed event")
_, saveErr := store.Set(dataKey, string(jsonString), config.GetThrottleValue()*2)
if saveErr != nil {
dLogger.WithFields(log.Fields{
"event": e.GetTraceInfo(),
"dataKey": dataKey,
}).Error("Save config error: ", saveErr)
panic(saveErr)
}

// Schedule send event later
dLogger.WithFields(log.Fields{"key": dataKey, "event": e.GetTraceInfo(), "timeRemain": timeRemain}).Info("AfterFunc")
dispatcher.TrackAfterFuncJob(timeRemain, func() {
dLogger.WithFields(log.Fields{"key": dataKey}).Debug("After event callback")
payload, _, _, _ := store.Get(dataKey)
payload, exists, ttl, err := store.Get(dataKey)
dLogger.WithFields(log.Fields{"key": dataKey, "event": e.GetTraceInfo(), "payload": payload, "key_exists": exists, "ttl": ttl, "err": err}).Info("Fetch data for delayed event")
event := models.IncomingEvent{}
json.Unmarshal([]byte(payload), &event)
d.sendEvent(event, dest, store, documentStore)
Expand Down Expand Up @@ -422,6 +455,7 @@ func (d *Dispatcher) sendEvent(evt models.IncomingEvent, destination models.Dest
_sendEvent := func() {
defer func() {
if err := recover(); err != nil {
callbackLogger.Error(fmt.Sprintf("Dispatcher Error: %s", err))
d.OnError(evt, &captin_errors.DispatcherError{
Msg: err.(error).Error(),
Destination: destination,
Expand All @@ -434,6 +468,7 @@ func (d *Dispatcher) sendEvent(evt models.IncomingEvent, destination models.Dest
event := deepcopy.Copy(evt).(models.IncomingEvent)
err := sender.SendEvent(event, destination)
if err != nil {
callbackLogger.Error(fmt.Sprintf("Send event failed, %s", err))
panic(err)
}
callbackLogger.Info(fmt.Sprintf("Event successfully sent to %s [%s]", config.GetName(), destination.GetCallbackURL()))
Expand Down