Repository: alexcasalboni/aws-lambda-power-tuning (js)

Issue:
Bug: log processed payload in case of execution error
Hey @alexcasalboni, thanks for the great project. I wrapped a Lambda with it today and it gave some great insight on the ideal configuration.

One small thing that stumped me for a while that I thought I'd mention is that when you make use of a pre-processor function and it goes wrong, the payload reported in the thrown error is the original payload rather than the post-processed payload that was actually sent to the target Lambda: https://github.com/alexcasalboni/aws-lambda-power-tuning/blob/master/lambda/executor.js#L84

Cheers

Pull request:
Bugfix: include processed payload log in case of failed invocation
Fix #102 

Patch:
diff --git a/lambda/executor.js b/lambda/executor.js
index cf6c660..fea4e81 100644
--- a/lambda/executor.js
+++ b/lambda/executor.js
@@ -78,12 +78,12 @@ const runInParallel = async(num, lambdaARN, lambdaAlias, payloads, preARN, postA
     const results = [];
     // run all invocations in parallel ...
     const invocations = utils.range(num).map(async(_, i) => {
-        const data = await utils.invokeLambdaWithProcessors(lambdaARN, lambdaAlias, payloads[i], preARN, postARN);
+        const {invocationResults, actualPayload} = await utils.invokeLambdaWithProcessors(lambdaARN, lambdaAlias, payloads[i], preARN, postARN);
         // invocation errors return 200 and contain FunctionError and Payload
-        if (data.FunctionError) {
-            throw new Error(`Invocation error (running in parallel): ${data.Payload} with payload ${JSON.stringify(payloads[i])}`);
+        if (invocationResults.FunctionError) {
+            throw new Error(`Invocation error (running in parallel): ${invocationResults.Payload} with payload ${JSON.stringify(actualPayload)}`);
         }
-        results.push(data);
+        results.push(invocationResults);
     });
     // ... and wait for results
     await Promise.all(invocations);
@@ -94,12 +94,12 @@ const runInSeries = async(num, lambdaARN, lambdaAlias, payloads, preARN, postARN
     const results = [];
     for (let i = 0; i < num; i++) {
         // run invocations in series
-        const data = await utils.invokeLambdaWithProcessors(lambdaARN, lambdaAlias, payloads[i], preARN, postARN);
+        const {invocationResults, actualPayload} = await utils.invokeLambdaWithProcessors(lambdaARN, lambdaAlias, payloads[i], preARN, postARN);
         // invocation errors return 200 and contain FunctionError and Payload
-        if (data.FunctionError) {
-            throw new Error(`Invocation error (running in series): ${data.Payload} with payload ${JSON.stringify(payloads[i])}`);
+        if (invocationResults.FunctionError) {
+            throw new Error(`Invocation error (running in series): ${invocationResults.Payload} with payload ${JSON.stringify(actualPayload)}`);
         }
-        results.push(data);
+        results.push(invocationResults);
     }
     return results;
 };
diff --git a/lambda/utils.js b/lambda/utils.js
index 0bd15c4..1e0a038 100644
--- a/lambda/utils.js
+++ b/lambda/utils.js
@@ -175,10 +175,10 @@ module.exports.deleteLambdaAlias = (lambdaARN, alias) => {
 /**
  * Invoke a (pre/post-)processor Lambda function and return its output (data.Payload).
  */
-module.exports.invokeLambdaProcessor = async(processorARN, payload) => {
+module.exports.invokeLambdaProcessor = async(processorARN, payload, preOrPost = 'Pre') => {
     const processorData = await utils.invokeLambda(processorARN, null, payload);
     if (processorData.FunctionError) {
-        throw new Error(`Processor ${processorARN} failed with error ${processorData.Payload} and payload ${JSON.stringify(payload)}`);
+        throw new Error(`${preOrPost}Processor ${processorARN} failed with error ${processorData.Payload} and payload ${JSON.stringify(payload)}`);
     }
     return processorData.Payload;
 };
@@ -187,27 +187,33 @@ module.exports.invokeLambdaProcessor = async(processorARN, payload) => {
  * Wrapper around Lambda function invocation with pre/post-processor functions.
  */
 module.exports.invokeLambdaWithProcessors = async(lambdaARN, alias, payload, preARN, postARN) => {
+
+    var actualPayload = payload; // might change based on pre-processor
+
     // first invoke pre-processor, if provided
     if (preARN) {
         console.log('Invoking pre-processor');
         // overwrite payload with pre-processor's output (only if not empty)
-        const preProcessorOutput = await utils.invokeLambdaProcessor(preARN, payload);
+        const preProcessorOutput = await utils.invokeLambdaProcessor(preARN, payload, 'Pre');
         if (preProcessorOutput) {
-            payload = preProcessorOutput;
+            actualPayload = preProcessorOutput;
         }
     }
 
     // invoke function to be power-tuned
-    const data = await utils.invokeLambda(lambdaARN, alias, payload);
+    const invocationResults = await utils.invokeLambda(lambdaARN, alias, actualPayload);
 
     // then invoke post-processor, if provided
     if (postARN) {
         console.log('Invoking post-processor');
-        // note: invocation may have failed (data.FunctionError)
-        await utils.invokeLambdaProcessor(postARN, data.Payload);
+        // note: invocation may have failed (invocationResults.FunctionError)
+        await utils.invokeLambdaProcessor(postARN, invocationResults.Payload, 'Post');
     }
 
-    return data;
+    return {
+        actualPayload,
+        invocationResults,
+    };
 };
 
 /**


Test patch:
diff --git a/test/unit/test-lambda.js b/test/unit/test-lambda.js
index c50f5f1..c11221d 100644
--- a/test/unit/test-lambda.js
+++ b/test/unit/test-lambda.js
@@ -364,7 +364,7 @@ describe('Lambda Functions', async() => {
 
             invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
             invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
-                .callsFake(async(_arn, _alias, payload) => {
+                .callsFake(async(_arn, _payload, _preOrPost) => {
                     invokeProcessorCounter++;
                     invokeLambdaCounter++;
                     return '{"Processed": true}';
@@ -862,7 +862,7 @@ describe('Lambda Functions', async() => {
 
             invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
             invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
-                .callsFake(async(_arn, _alias, payload) => {
+                .callsFake(async(_arn, _payload, _preOrPost) => {
                     invokeProcessorCounter++;
                     invokeLambdaCounter++;
                     return null; // empty output from pre-processor
@@ -885,8 +885,8 @@ describe('Lambda Functions', async() => {
 
             invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
             invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
-                .callsFake(async(_arn, _alias, _payload) => {
-                    throw new Error('Processor XXX failed with error YYY and payload ZZZ');
+                .callsFake(async(_arn, _payload, _preOrPost) => {
+                    throw new Error('PreProcessor XXX failed with error YYY and payload ZZZ');
                 });
 
             await invokeForFailure(handler, {
@@ -905,8 +905,8 @@ describe('Lambda Functions', async() => {
 
             invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
             invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
-                .callsFake(async(_arn, _alias, _payload) => {
-                    throw new Error('Processor XXX failed with error YYY and payload ZZZ');
+                .callsFake(async(_arn, _payload, _preOrPost) => {
+                    throw new Error('PostProcessor XXX failed with error YYY and payload ZZZ');
                 });
 
             await invokeForFailure(handler, {
@@ -921,6 +921,75 @@ describe('Lambda Functions', async() => {
 
         });
 
+        it('should explode with processed payload in case of execution error (series)', async() => {
+
+            invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
+            invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
+                .callsFake(async(_arn, _payload, _preOrPost) => {
+                    invokeProcessorCounter++;
+                    invokeLambdaCounter++;
+                    return {Processed: true};
+                });
+
+            invokeLambdaStub && invokeLambdaStub.restore();
+            invokeLambdaStub = sandBox.stub(utils, 'invokeLambda')
+                .callsFake(async(_arn, _alias, payload) => {
+                    return {
+                        FunctionError: 'Unhandled',
+                        Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
+                    };
+                });
+
+            const error = await invokeForFailure(handler, {
+                value: '128',
+                input: {
+                    lambdaARN: 'arnOK',
+                    num: 10,
+                    payload: {Original: true},
+                    preProcessorARN: 'postArnOK',
+                },
+            });
+
+            expect(error.message).to.contain('in series');
+            expect(error.message).to.contain(JSON.stringify({Processed: true}));
+
+        });
+
+        it('should explode with processed payload in case of execution error (parallel)', async() => {
+
+            invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
+            invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
+                .callsFake(async(_arn, _payload, _preOrPost) => {
+                    invokeProcessorCounter++;
+                    invokeLambdaCounter++;
+                    return {Processed: true};
+                });
+
+            invokeLambdaStub && invokeLambdaStub.restore();
+            invokeLambdaStub = sandBox.stub(utils, 'invokeLambda')
+                .callsFake(async(_arn, _alias, payload) => {
+                    return {
+                        FunctionError: 'Unhandled',
+                        Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
+                    };
+                });
+
+            const error = await invokeForFailure(handler, {
+                value: '128',
+                input: {
+                    lambdaARN: 'arnOK',
+                    num: 10,
+                    payload: {Original: true},
+                    parallelInvocation: true,
+                    preProcessorARN: 'postArnOK',
+                },
+            });
+
+            expect(error.message).to.contain('in parallel');
+            expect(error.message).to.contain(JSON.stringify({Processed: true}));
+
+        });
+
     });
 
     describe('analyzer', () => {