-
Notifications
You must be signed in to change notification settings - Fork 134
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #172 from SumoLogic/python-js-update
Updating runtimes of python and nodeJS
- Loading branch information
Showing
15 changed files
with
288 additions
and
266 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,77 +1,82 @@ | ||
| var AWS = require("aws-sdk"); | ||
| var processLogsHandler = require('./cloudwatchlogs_lambda').processLogs; | ||
| var getEndpointURL = require('./cloudwatchlogs_lambda').getEndpointURL; | ||
| var DLQUtils = require("./sumo-dlq-function-utils").DLQUtils; | ||
| var Messages = DLQUtils.Messages; | ||
| var invokeLambdas = DLQUtils.invokeLambdas; | ||
| const { processLogs: processLogsHandler, getEndpointURL } = require('./cloudwatchlogs_lambda'); | ||
| const { DLQUtils } = require("./sumo-dlq-function-utils"); | ||
|
|
||
| const { Messages, invokeLambdas } = DLQUtils; | ||
|
|
||
| exports.consumeMessages = async function (env, context, callback) { | ||
| var sqs = new AWS.SQS({region: env.AWS_REGION}); | ||
| var MessagesObj = new Messages(env); | ||
| env.SUMO_CLIENT_HEADER="dlq-aws-lambda"; | ||
| const MessagesObj = new Messages(env); | ||
| env.SUMO_CLIENT_HEADER = "dlq-aws-lambda"; | ||
|
|
||
| if (!env.SUMO_ENDPOINT) { | ||
| let SUMO_ENDPOINT = await getEndpointURL(); | ||
| if (SUMO_ENDPOINT instanceof Error) { | ||
| console.log("Error in getEndpointURL: ", SUMO_ENDPOINT); | ||
| callback(SUMO_ENDPOINT, null); | ||
| try { | ||
| let SUMO_ENDPOINT = await getEndpointURL(); | ||
| env.SUMO_ENDPOINT = SUMO_ENDPOINT; | ||
| } catch (error) { | ||
| console.log("Error in getEndpointURL: ", error); | ||
| callback(error, null); | ||
| return; | ||
| } | ||
| env.SUMO_ENDPOINT = SUMO_ENDPOINT; | ||
| } else { | ||
| console.log("consumeMessages: Getting SUMO_ENDPOINT from env"); | ||
| } | ||
| MessagesObj.receiveMessages(10, function (err, data) { | ||
| var messages = (data)? data.Messages: null; | ||
| if (err) { | ||
| callback(err); | ||
| } else if (messages && messages.length > 0) { | ||
| var fail_cnt = 0, msgCount = 0; | ||
|
|
||
| try { | ||
| const messages = await MessagesObj.receiveMessages(10); | ||
|
|
||
|
|
||
| if (messages && messages.length > 0) { | ||
| let fail_cnt = 0, msgCount = 0; | ||
| console.log("Messages Received", messages.length); | ||
| for (var i = 0; i < messages.length; i++) { | ||
| (function(idx) { | ||
| var payload = JSON.parse(messages[idx].Body); | ||
| var receiptHandle = messages[idx].ReceiptHandle; | ||
|
|
||
| for (let i = 0; i < messages.length; i++) { | ||
| (function (idx) { | ||
| const payload = JSON.parse(messages[idx].Body); | ||
| const receiptHandle = messages[idx].ReceiptHandle; | ||
|
|
||
| if (!(payload.awslogs && payload.awslogs.data)) { | ||
| console.log("Message does not contain awslogs or awslogs.data attributes", payload); | ||
| //deleting msg in DLQ after injesting in sumo | ||
| MessagesObj.deleteMessage(receiptHandle, function (err, data) { | ||
| if (err) console.log(err, err.stack); | ||
| }); | ||
|
|
||
| MessagesObj.deleteMessage(receiptHandle) | ||
| .catch((err) => console.log(err, err.stack)); | ||
|
|
||
| return; | ||
| } | ||
| var logdata = payload.awslogs.data; | ||
|
|
||
| const logdata = payload.awslogs.data; | ||
|
|
||
| processLogsHandler(env, logdata, function (err, msg) { | ||
| msgCount++; | ||
|
|
||
| if (err) { | ||
| console.log(err, msg); | ||
| fail_cnt++; | ||
| } else { | ||
| //deleting msg in DLQ after injesting in sumo | ||
| MessagesObj.deleteMessage(receiptHandle, function (err, data) { | ||
| if (err) console.log(err, err.stack); | ||
| }); | ||
| MessagesObj.deleteMessage(receiptHandle) | ||
| .catch((err) => console.log(err, err.stack)); | ||
| } | ||
| if (msgCount == messages.length) { | ||
| if (fail_cnt == 0 && (parseInt(env.is_worker) === 0)) { | ||
|
|
||
| if (msgCount === messages.length) { | ||
| if (fail_cnt === 0 && parseInt(env.is_worker) === 0) { | ||
| invokeLambdas(env.AWS_REGION, parseInt(env.NUM_OF_WORKERS), | ||
| context.functionName, '{"is_worker": "1"}', context); | ||
| context.functionName, '{"is_worker": "1"}', context); | ||
| } | ||
| callback(null, (messages.length-fail_cnt) + ' success'); | ||
|
|
||
| callback(null, `${messages.length - fail_cnt} success`); | ||
| } | ||
| }); | ||
| })(i); | ||
| } | ||
|
|
||
| } else { | ||
|
|
||
| callback(null, 'success'); | ||
| } | ||
| }); | ||
| } catch (error) { | ||
| callback(error); | ||
| } | ||
| }; | ||
|
|
||
| exports.handler = function (event, context, callback) { | ||
|
|
||
| var env = Object.assign({}, process.env); | ||
| env['is_worker'] = event.is_worker || 0; | ||
| const env = Object.assign({}, process.env); | ||
| env.is_worker = event.is_worker || 0; | ||
| exports.consumeMessages(env, context, callback); | ||
| }; | ||
|
|
||
| }; |
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.