Skip to content

Commit 0887197

Browse files
duchatranduchatran
authored
Merge pull request #66 from SumoLogic/hpal_SUMO-94622_addingretries
Sumo-94622: Adding retry logic in lambda functions
2 parents 3fa5da6 + 997ef98 commit 0887197

File tree

2 files changed

+133
-64
lines changed

2 files changed

+133
-64
lines changed

‎cloudwatchevents/cloudwatchevents.js‎

Lines changed: 65 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -19,23 +19,48 @@ var sourceCategoryOverride = process.env.SOURCE_CATEGORY_OVERRIDE || ''; // If
1919
var sourceHostOverride = process.env.SOURCE_HOST_OVERRIDE || ''; // If empty sourceHostOverride will not be set to the name of the logGroup
2020
var sourceNameOverride = process.env.SOURCE_NAME_OVERRIDE || ''; // If empty sourceNameOverride will not be set to the name of the logStream
2121

22+
var retryInterval = process.env.RETRY_INTERVAL || 5000; // the interval in millisecs between retries
23+
var numOfRetries = process.env.NUMBER_OF_RETRIES || 3; // the number of retries
24+
2225
var https = require('https');
2326
var zlib = require('zlib');
2427
var url = require('url');
2528

29+
Promise.retryMax = function(fn,retry,interval,fnParams) {
30+
return fn.apply(this,fnParams).catch( err => {
31+
var waitTime = typeof interval === 'function' ? interval() : interval;
32+
console.log("Retries left " + (retry-1) + " delay(in ms) " + waitTime);
33+
return (retry>1? Promise.wait(waitTime).then(()=> Promise.retryMax(fn,retry-1,interval, fnParams)):Promise.reject(err));
34+
});
35+
}
36+
37+
Promise.wait = function(delay) {
38+
return new Promise((fulfill,reject)=> {
39+
//console.log(Date.now());
40+
setTimeout(fulfill,delay||0);
41+
});
42+
};
43+
44+
function exponentialBackoff(seed) {
45+
var count = 0;
46+
return function() {
47+
count++;
48+
return count*seed;
49+
}
50+
}
2651

2752
function postToSumo(context, messages) {
2853
var messagesTotal = Object.keys(messages).length;
2954
var messagesSent = 0;
3055
var messageErrors = [];
31-
56+
3257
var urlObject = url.parse(SumoURL);
3358
var options = {
3459
'hostname': urlObject.hostname,
3560
'path': urlObject.pathname,
3661
'method': 'POST'
3762
};
38-
63+
3964
var finalizeContext = function () {
4065
var total = messagesSent + messageErrors.length;
4166
if (total == messagesTotal) {
@@ -48,45 +73,54 @@ function postToSumo(context, messages) {
4873
}
4974
};
5075

51-
76+
function httpSend(options, headers, data) {
77+
return new Promise( (resolve,reject) => {
78+
var curOptions = options;
79+
curOptions.headers = headers;
80+
var req = https.request(curOptions, function (res) {
81+
var body = '';
82+
res.setEncoding('utf8');
83+
res.on('data', function (chunk) {
84+
body += chunk; // don't really do anything with body
85+
});
86+
res.on('end', function () {
87+
if (res.statusCode == 200) {
88+
resolve(body);
89+
} else {
90+
reject({'error':'HTTP Return code ' + res.statusCode,'res':res});
91+
}
92+
});
93+
});
94+
req.on('error', function (e) {
95+
reject({'error':e,'res':null});
96+
});
97+
for (var i = 0; i < data.length; i++) {
98+
req.write(JSON.stringify(data[i]) + '\n');
99+
}
100+
console.log("sending to Sumo...")
101+
req.end();
102+
});
103+
}
52104
Object.keys(messages).forEach(function (key, index) {
53105
var headerArray = key.split(':');
54-
options.headers = {
106+
var headers = {
55107
'X-Sumo-Name': headerArray[0],
56108
'X-Sumo-Category': headerArray[1],
57109
'X-Sumo-Host': headerArray[2],
58-
'X-Sumo-Client': 'cloudwatchevents-aws-lambda'
110+
'X-Sumo-Client': 'kinesis-aws-lambda'
59111
};
60-
61-
var req = https.request(options, function (res) {
62-
res.setEncoding('utf8');
63-
res.on('data', function (chunk) {});
64-
res.on('end', function () {
65-
console.log("Got response code: "+ res.statusCode);
66-
if (res.statusCode == 200) {
67-
messagesSent++;
68-
} else {
69-
messageErrors.push('HTTP Return code ' + res.statusCode);
70-
}
71-
finalizeContext();
72-
});
73-
});
74-
75-
req.on('error', function (e) {
76-
messageErrors.push(e.message);
112+
Promise.retryMax(httpSend, numOfRetries, retryInterval, [options, headers, messages[key]]).then((body)=> {
113+
messagesSent++;
114+
finalizeContext()
115+
}).catch((e) => {
116+
messageErrors.push(e.error);
77117
finalizeContext();
78118
});
79-
80-
for (var i = 0; i < messages[key].length; i++) {
81-
req.write(JSON.stringify(messages[key][i]) + '\n');
82-
}
83-
req.end();
84119
});
85120
}
86121

87-
88122
exports.handler = function (event, context) {
89-
123+
90124
// Used to hold chunks of messages to post to SumoLogic
91125
var messageList = {};
92126
var final_event;
@@ -95,13 +129,13 @@ exports.handler = function (event, context) {
95129
if (urlObject.protocol != 'https:' || urlObject.host === null || urlObject.path === null) {
96130
context.fail('Invalid SUMO_ENDPOINT environment variable: ' + SumoURL);
97131
}
98-
132+
99133
//console.log(event);
100134
if ((event.source==="aws.guardduty") || (removeOuterFields)) {
101135
final_event =event.detail;
102136
} else {
103137
final_event = event;
104-
}
138+
}
105139
messageList[sourceNameOverride+':'+sourceCategoryOverride+':'+sourceHostOverride]=[final_event];
106140
postToSumo(context, messageList);
107141
};

‎kinesis/node.js/k2sl_lambda.js‎

Lines changed: 68 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,9 @@ var sourceCategoryOverride = process.env.SOURCE_CATEGORY_OVERRIDE || 'none'; //
1212
var sourceHostOverride = process.env.SOURCE_HOST_OVERRIDE || 'none'; // If none sourceHostOverride will not be set to the name of the logGroup
1313
var sourceNameOverride = process.env.SOURCE_NAME_OVERRIDE || 'none'; // If none sourceNameOverride will not be set to the name of the logStream
1414

15+
var retryInterval = process.env.RETRY_INTERVAL || 5000; // the interval in millisecs between retries
16+
var numOfRetries = process.env.NUMBER_OF_RETRIES || 3; // the number of retries
17+
1518
// CloudWatch logs encoding
1619
var encoding = process.env.ENCODING || 'utf-8'; // default is utf-8
1720

@@ -20,7 +23,7 @@ var includeLogInfo = true; // default is true
2023

2124
// Regex used to detect logs coming from lambda functions.
2225
// The regex will parse out the requestID and strip the timestamp
23-
// Example: 2016-11-10T23:11:54.523Z 108af3bb-a79b-11e6-8bd7-91c363cc05d9 some message
26+
// Example: 2016-11-10T23:11:54.523Z 108af3bb-a79b-11e6-8bd7-91c363cc05d9 some message
2427
var consoleFormatRegex = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}Z\t(\w+?-\w+?-\w+?-\w+?-\w+)\t/;
2528

2629
// Used to extract RequestID
@@ -31,7 +34,30 @@ var zlib = require('zlib');
3134
var url = require('url');
3235

3336

34-
function sumoMetaKey(awslogsData, message) {
37+
Promise.retryMax = function(fn,retry,interval,fnParams) {
38+
return fn.apply(this,fnParams).catch( err => {
39+
var waitTime = typeof interval === 'function' ? interval() : interval;
40+
console.log("Retries left " + (retry-1) + " delay(in ms) " + waitTime);
41+
return (retry>1? Promise.wait(waitTime).then(()=> Promise.retryMax(fn,retry-1,interval, fnParams)):Promise.reject(err));
42+
});
43+
}
44+
45+
Promise.wait = function(delay) {
46+
return new Promise((fulfill,reject)=> {
47+
//console.log(Date.now());
48+
setTimeout(fulfill,delay||0);
49+
});
50+
};
51+
52+
function exponentialBackoff(seed) {
53+
var count = 0;
54+
return function() {
55+
count++;
56+
return count*seed;
57+
}
58+
}
59+
60+
function sumoMetaKey(awslogsData, message, data) {
3561
var sourceCategory = '';
3662
var sourceName = '';
3763
var sourceHost = '';
@@ -95,43 +121,52 @@ function postToSumo(context, messages) {
95121
}
96122
};
97123

98-
124+
function httpSend(options, headers, data) {
125+
return new Promise( (resolve,reject) => {
126+
var curOptions = options;
127+
curOptions.headers = headers;
128+
var req = https.request(curOptions, function (res) {
129+
var body = '';
130+
res.setEncoding('utf8');
131+
res.on('data', function (chunk) {
132+
body += chunk; // don't really do anything with body
133+
});
134+
res.on('end', function () {
135+
if (res.statusCode == 200) {
136+
resolve(body);
137+
} else {
138+
reject({'error':'HTTP Return code ' + res.statusCode,'res':res});
139+
}
140+
});
141+
});
142+
req.on('error', function (e) {
143+
reject({'error':e,'res':null});
144+
});
145+
for (var i = 0; i < data.length; i++) {
146+
req.write(JSON.stringify(data[i]) + '\n');
147+
}
148+
console.log("sending to Sumo...")
149+
req.end();
150+
});
151+
}
99152
Object.keys(messages).forEach(function (key, index) {
100153
var headerArray = key.split(':');
101-
102-
options.headers = {
154+
var headers = {
103155
'X-Sumo-Name': headerArray[0],
104156
'X-Sumo-Category': headerArray[1],
105157
'X-Sumo-Host': headerArray[2],
106158
'X-Sumo-Client': 'kinesis-aws-lambda'
107159
};
108-
109-
var req = https.request(options, function (res) {
110-
res.setEncoding('utf8');
111-
res.on('data', function (chunk) {});
112-
res.on('end', function () {
113-
if (res.statusCode == 200) {
114-
messagesSent++;
115-
} else {
116-
messageErrors.push('HTTP Return code ' + res.statusCode);
117-
}
118-
finalizeContext();
119-
});
120-
});
121-
122-
req.on('error', function (e) {
123-
messageErrors.push(e.message);
160+
Promise.retryMax(httpSend, numOfRetries, retryInterval, [options, headers, messages[key]]).then((body)=> {
161+
messagesSent++;
162+
finalizeContext()
163+
}).catch((e) => {
164+
messageErrors.push(e.error);
124165
finalizeContext();
125166
});
126-
127-
for (var i = 0; i < messages[key].length; i++) {
128-
req.write(JSON.stringify(messages[key][i]) + '\n');
129-
}
130-
req.end();
131167
});
132168
}
133169

134-
135170
exports.handler = function (event, context) {
136171

137172
// console.log(JSON.stringify(event));
@@ -144,15 +179,14 @@ exports.handler = function (event, context) {
144179
if (urlObject.protocol != 'https:' || urlObject.host === null || urlObject.path === null) {
145180
context.fail('Invalid SUMO_ENDPOINT environment variable: ' + SumoURL);
146181
}
147-
148-
event.Records.forEach(function(record) {
182+
var numOfRecords = event.Records.length;
183+
event.Records.forEach(function(record, index) {
149184
var zippedInput = new Buffer(record.kinesis.data, 'base64');
150185

151186
zlib.gunzip(zippedInput, function (e, buffer) {
152187
if (e) {
153188
context.fail(e);
154189
}
155-
156190
var awslogsData = JSON.parse(buffer.toString(encoding));
157191

158192
if (awslogsData.messageType === 'CONTROL_MESSAGE') {
@@ -211,10 +245,11 @@ exports.handler = function (event, context) {
211245
messageList[metadataKey] = [log];
212246
}
213247
});
214-
215248
// Push messages to Sumo
216-
postToSumo(context, messageList);
249+
if (index === numOfRecords-1) {
250+
postToSumo(context, messageList);
251+
}
217252
})
218253

219254
});
220-
};
255+
};

0 commit comments

Comments
 (0)