feat: onboard initial marketo builk upload implementation #5114
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #5114 +/- ##
==========================================
- Coverage 73.19% 72.97% -0.22%
==========================================
Files 417 423 +6
Lines 59085 59475 +390
==========================================
+ Hits 43246 43402 +156
- Misses 13426 13650 +224
- Partials 2413 2423 +10 ☔ View full report in Codecov by Sentry. |
…s/rudder-server into feat.marketo-bulk-uplaod-async
| func sendHTTPRequest(uploadURL, csvFilePath string, accessToken string, deduplicationField string) (*http.Response, error) { | ||
| file, err := os.Open(csvFilePath) | ||
| if err != nil { | ||
| return nil, err |
There was a problem hiding this comment.
Apart from returning error only, we can add oneliner as follows:
fmt.Errorf("error while opening the file: %v", err)
router/batchrouter/asyncdestinationmanager/marketo-bulk-upload/marketobulkupload.go
Outdated
Show resolved
Hide resolved
| return m.attemptImport(uploadURL, csvFilePath, deduplicationField, uploadTimeStat) | ||
| } | ||
|
|
||
| return "", apiError |
There was a problem hiding this comment.
Can we add a stat here to monitor how many times the retryable error occurred so that we can take steps if any unnecessary retries happen?
There was a problem hiding this comment.
But wont we have this in jobs db - and I believe router should have such metrics based on jobs db
| errorMessage := "" | ||
| if !marketoResponse.Success { | ||
| if len(marketoResponse.Errors) > 0 { | ||
| errorCode := marketoResponse.Errors[0].Code |
There was a problem hiding this comment.
Don't we need to consider other errors from errors array?
There was a problem hiding this comment.
We only get 1 error object form error array so far thats what we have seen
|
|
||
| // If the access token is nil or about to expire in 10 seconds, wait 10 seconds and fetch a new access token | ||
| } else if m.accessToken.FetchedAt+m.accessToken.ExpiresIn < 10 { | ||
| time.Sleep(11 * time.Second) |
There was a problem hiding this comment.
Why do we need to wait? Can't we generate a new accessToken when there is a valid accessToken?
There was a problem hiding this comment.
sudip, no if we call it within the valid window, same accessToken will be returned and that will eventually fail.
There was a problem hiding this comment.
I have removed this logic for now made it very simple, for now since marketo returns same code and since we wll call only after an hour I think if we really see need then we can optimse
| } | ||
|
|
||
| return nil | ||
| } |
There was a problem hiding this comment.
Do we need this separate file? Can't we move this function to util?
There was a problem hiding this comment.
This we will deprecate once we move to vdm hence for now kept separate (this is only needed to parse the ui mapping)
| StatusCode: 500, | ||
| HasFailed: true, | ||
| Error: asyncResponse.Error, | ||
| fmt.Println("File Upload Started for Dest Type ", destType) |
There was a problem hiding this comment.
is it intentional ? or planning to remove
There was a problem hiding this comment.
Using for debugging will remove
Description
This PR implements the Marketo Bulk upload Integration using the following API
We use the Async upload destination framework to onboard the destination
Linear Ticket
Linear Ticket
Security