Repository: alexcasalboni/aws-lambda-power-tuning
Pull request #88 (2020-05-21 17:11:58)

Implement pre/post-processing functions logic
Implements #87 

TODO:
- [x] Basic implementation
- [x] Basic tests
- [x] Add more tests (wait for #79 to be merged)
- [x] Update documentation


Patch:
diff --git a/README-SAR.md b/README-SAR.md
index dea6b2d..50fda27 100644
--- a/README-SAR.md
+++ b/README-SAR.md
@@ -47,6 +47,8 @@ The AWS Step Functions state machine accepts the following parameters:
 * **autoOptimize** (false by default): if `true`, the state machine will apply the optimal configuration at the end of its execution
 * **autoOptimizeAlias** (string): if provided - and only if `autoOptimize` if `true`, the state machine will create or update this alias with the new optimal power value
 * **dryRun** (false by default): if true, the state machine will execute the input function only once and it will disable every functionality related to logs analysis, auto-tuning, and visualization; the dry-run mode is intended for testing purposes, for example to verify that IAM permissions are set up correctly
+* **preProcessorARN** (string): it must be the ARN of a Lambda function; if provided, the function will be invoked before every invocation of `lambdaARN`; more details below in the Pre/Post-processing functions section
+* **postProcessorARN** (string): it must be the ARN of a Lambda function; if provided, the function will be invoked after every invocation of `lambdaARN`; more details below in the Pre/Post-processing functions section
 
 
 Additionally, you can specify a list of power values at deploy-time in the `PowerValues` CloudFormation parameter. These power values will be used as the default in case no `powerValues` input parameter is provided.
@@ -97,6 +99,33 @@ To simplify these calculations, you could use weights that sum up to 100.
 
 Note: the number of weighted payloads must always be smaller or equal than `num` (or `num >= count(payloads)`). For example, if you have 50 weighted payloads, you'll need to set at least `num: 50` so that each payload will be used at least once.
 
+
+### Pre/Post-processing functions
+
+Sometimes you need to power-tune Lambda functions that have side effects such as creating or deleting records in a database. In these cases, you may need to execute some pre-processing or post-processing logic before and/or after each function invocation.
+
+For example, imagine that you are power-tuning a function that deletes one record from a downstream database. Since you want to execute this function `num` times you'd need to insert some records in advance and then find a way to delete all of them with a dynamic payload. Or you could simply configure a pre-processing function (using the `preProcessorARN` input parameter) that will create a brand new record before the actual function is executed.
+
+Here's the flow in pseudo-code:
+
+```
+function Executor:
+  iterate from 0 to num:
+    [payload = execute Pre-processor (payload)]
+    results = execute Main Function (payload)
+    [execute Post-processor (results)]
+```
+
+A few things to note:
+
+* You can configure a pre-processor and/or a post-processor independently
+* The pre-processor will receive the original payload
+* If the pre-processor returns a non-empty output, it will overwrite the original payload
+* The post-processor will receive the main function's output as payload
+* If a pre-processor or post-processor fails, the whole power-tuning state machine will fail
+* Pre/post-processors don't have to be in the same region of the main function
+* Pre/post-processors don't alter the statistics related to cost and performance
+
 ## State Machine Output
 
 The state machine will return the following output:
diff --git a/README.md b/README.md
index 28791e9..c1338c1 100644
--- a/README.md
+++ b/README.md
@@ -107,7 +107,7 @@ cd the-lambda-power-tuner
 npm run deploy
 ```
 
-For Python deployment, see the instructions [here](https://github.com/cdk-patterns/serverless#2-download-pattern-in-python-or-typescript-cdk)
+For Python deployment, see the instructions [here](https://github.com/cdk-patterns/serverless#2-download-pattern-in-python-or-typescript-cdk).
 
 ## How to execute the state machine (programmatically)
 
@@ -146,13 +146,15 @@ The AWS Step Functions state machine accepts the following parameters:
 * **lambdaARN** (required, string): unique identifier of the Lambda function you want to optimize
 * **powerValues** (optional, string or list of integers): the list of power values to be tested; if not provided, the default values configured at deploy-time are used (by default: 128MB, 256MB, 512MB, 1024MB, 1536MB, and 3008MB); you can provide any power values between 128MB and 3,008MB in 64 MB increments; if you provide the string `"ALL"` instead of a list, all possible power configurations will be tested
 * **num** (required, integer): the # of invocations for each power configuration (minimum 5, recommended: between 10 and 100)
-* **payload** (string, object, or list): the static payload that will be used for every invocation (object or string); when using a list, a weighted payload is expected in the shape of `[{"payload": {...}, "weight": X }, {"payload": {...}, "weight": Y }, {"payload": {...}, "weight": Z }]`, where the weights `X`, `Y`, and `Z` are treated as relative weights (not perentages); more details below in the Weighted Payloads section
+* **payload** (string, object, or list): the static payload that will be used for every invocation (object or string); when using a list, a weighted payload is expected in the shape of `[{"payload": {...}, "weight": X }, {"payload": {...}, "weight": Y }, {"payload": {...}, "weight": Z }]`, where the weights `X`, `Y`, and `Z` are treated as relative weights (not perentages); more details below in the [Weighted Payloads section](#user-content-weighted-payloads)
 * **parallelInvocation** (false by default): if true, all the invocations will be executed in parallel (note: depending on the value of `num`, you may experience throttling when setting `parallelInvocation` to true)
 * **strategy** (string): it can be `"cost"` or `"speed"` or `"balanced"` (the default value is `"cost"`); if you use `"cost"` the state machine will suggest the cheapest option (disregarding its performance), while if you use `"speed"` the state machine will suggest the fastest option (disregarding its cost). When using `"balanced"` the state machine will choose a compromise between `"cost"` and `"speed"` according to the parameter `"balancedWeight"`
 * **balancedWeight** (number between 0.0 and 1.0, by default is 0.5): parameter that express the trade-off between cost and time, 0.0 is equivalent to `"speed"` strategy, 1.0 is equivalent to `"cost"` strategy
 * **autoOptimize** (false by default): if `true`, the state machine will apply the optimal configuration at the end of its execution
 * **autoOptimizeAlias** (string): if provided - and only if `autoOptimize` if `true`, the state machine will create or update this alias with the new optimal power value
 * **dryRun** (false by default): if true, the state machine will execute the input function only once and it will disable every functionality related to logs analysis, auto-tuning, and visualization; the dry-run mode is intended for testing purposes, for example to verify that IAM permissions are set up correctly
+* **preProcessorARN** (string): it must be the ARN of a Lambda function; if provided, the function will be invoked before every invocation of `lambdaARN`; more details below in the [Pre/Post-processing functions section](#user-content-prepost-processing)
+* **postProcessorARN** (string): it must be the ARN of a Lambda function; if provided, the function will be invoked after every invocation of `lambdaARN`; more details below in the [Pre/Post-processing functions section](#user-content-prepost-processing)
 
 
 Additionally, you can specify a list of power values at deploy-time in the `PowerValues` CloudFormation parameter. These power values will be used as the default in case no `powerValues` input parameter is provided.
@@ -205,6 +207,32 @@ To simplify these calculations, you could use weights that sum up to 100.
 Note: the number of weighted payloads must always be smaller or equal than `num` (or `num >= count(payloads)`). For example, if you have 50 weighted payloads, you'll need to set at least `num: 50` so that each payload will be used at least once.
 
 
+### Pre/Post-processing functions
+
+Sometimes you need to power-tune Lambda functions that have side effects such as creating or deleting records in a database. In these cases, you may need to execute some pre-processing or post-processing logic before and/or after each function invocation.
+
+For example, imagine that you are power-tuning a function that deletes one record from a downstream database. Since you want to execute this function `num` times you'd need to insert some records in advance and then find a way to delete all of them with a dynamic payload. Or you could simply configure a pre-processing function (using the `preProcessorARN` input parameter) that will create a brand new record before the actual function is executed.
+
+Here's the flow in pseudo-code:
+
+```
+function Executor:
+  iterate from 0 to num:
+    [payload = execute Pre-processor (payload)]
+    results = execute Main Function (payload)
+    [execute Post-processor (results)]
+```
+
+A few things to note:
+
+* You can configure a pre-processor and/or a post-processor independently
+* The pre-processor will receive the original payload
+* If the pre-processor returns a non-empty output, it will overwrite the original payload
+* The post-processor will receive the main function's output as payload
+* If a pre-processor or post-processor fails, the whole power-tuning state machine will fail
+* Pre/post-processors don't have to be in the same region of the main function
+* Pre/post-processors don't alter the statistics related to cost and performance
+
 ## State Machine Output
 
 The state machine will return the following output:
diff --git a/lambda/executor.js b/lambda/executor.js
index 4d067b6..7211c39 100644
--- a/lambda/executor.js
+++ b/lambda/executor.js
@@ -10,7 +10,16 @@ const minRAM = parseInt(process.env.minRAM, 10);
  */
 module.exports.handler = async(event, context) => {
     // read input from event
-    let {lambdaARN, value, num, enableParallel, payload, dryRun} = extractDataFromInput(event);
+    let {
+        lambdaARN,
+        value,
+        num,
+        enableParallel,
+        payload,
+        dryRun,
+        preProcessorARN,
+        postProcessorARN,
+    } = extractDataFromInput(event);
 
     validateInput(lambdaARN, value, num); // may throw
 
@@ -27,9 +36,9 @@ module.exports.handler = async(event, context) => {
     const payloads = generatePayloads(num, payload);
 
     if (enableParallel) {
-        results = await runInParallel(num, lambdaARN, lambdaAlias, payloads);
+        results = await runInParallel(num, lambdaARN, lambdaAlias, payloads, preProcessorARN, postProcessorARN);
     } else {
-        results = await runInSeries(num, lambdaARN, lambdaAlias, payloads);
+        results = await runInSeries(num, lambdaARN, lambdaAlias, payloads, preProcessorARN, postProcessorARN);
     }
 
     // get base cost
@@ -59,6 +68,8 @@ const extractDataFromInput = (event) => {
         enableParallel: !!event.parallelInvocation,
         payload: event.payload,
         dryRun: event.dryRun === true,
+        preProcessorARN: event.preProcessorARN,
+        postProcessorARN: event.postProcessorARN,
     };
 };
 
@@ -111,14 +122,14 @@ const convertPayload = (payload) => {
     return payload;
 };
 
-const runInParallel = async(num, lambdaARN, lambdaAlias, payloads) => {
+const runInParallel = async(num, lambdaARN, lambdaAlias, payloads, preARN, postARN) => {
     const results = [];
     // run all invocations in parallel ...
     const invocations = utils.range(num).map(async(_, i) => {
-        const data = await utils.invokeLambda(lambdaARN, lambdaAlias, payloads[i]);
+        const data = 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 ${payloads[i]}`);
+            throw new Error(`Invocation error (running in parallel): ${data.Payload} with payload ${JSON.stringify(payloads[i])}`);
         }
         results.push(data);
     });
@@ -127,14 +138,14 @@ const runInParallel = async(num, lambdaARN, lambdaAlias, payloads) => {
     return results;
 };
 
-const runInSeries = async(num, lambdaARN, lambdaAlias, payloads) => {
+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.invokeLambda(lambdaARN, lambdaAlias, payloads[i]);
+        const data = 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 ${payloads[i]}`);
+            throw new Error(`Invocation error (running in series): ${data.Payload} with payload ${JSON.stringify(payloads[i])}`);
         }
         results.push(data);
     }
diff --git a/lambda/utils.js b/lambda/utils.js
index fdb9850..a036f0b 100644
--- a/lambda/utils.js
+++ b/lambda/utils.js
@@ -172,6 +172,44 @@ module.exports.deleteLambdaAlias = (lambdaARN, alias) => {
     return lambda.deleteAlias(params).promise();
 };
 
+/**
+ * Invoke a (pre/post-)processor Lambda function and return its output (data.Payload).
+ */
+module.exports.invokeLambdaProcessor = async(processorARN, payload) => {
+    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)}`);
+    }
+    return processorData.Payload;
+};
+
+/**
+ * Wrapper around Lambda function invocation with pre/post-processor functions.
+ */
+module.exports.invokeLambdaWithProcessors = async(lambdaARN, alias, payload, preARN, postARN) => {
+    // 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);
+        if (preProcessorOutput) {
+            payload = preProcessorOutput;
+        }
+    }
+
+    // invoke function to be power-tuned
+    const data = await utils.invokeLambda(lambdaARN, alias, payload);
+
+    // 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);
+    }
+
+    return data;
+};
+
 /**
  * Invoke a given Lambda Function:Alias with payload and return its logs.
  */
diff --git a/package-lock.json b/package-lock.json
index 0146aa8..45a8b21 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,6 +1,6 @@
 {
   "name": "aws-lambda-power-tuning",
-  "version": "3.2.4",
+  "version": "3.2.5",
   "lockfileVersion": 1,
   "requires": true,
   "dependencies": {
@@ -125,9 +125,9 @@
       }
     },
     "@sinonjs/commons": {
-      "version": "1.7.2",
-      "resolved": "https://registry.npmjs.org/@sinonjs/commons/-/commons-1.7.2.tgz",
-      "integrity": "sha512-+DUO6pnp3udV/v2VfUWgaY5BIE1IfT7lLfeDzPVeMT1XKkaAp9LgSI9x5RtrFQoZ9Oi0PgXQQHPaoKu7dCjVxw==",
+      "version": "1.8.0",
+      "resolved": "https://registry.npmjs.org/@sinonjs/commons/-/commons-1.8.0.tgz",
+      "integrity": "sha512-wEj54PfsZ5jGSwMX68G8ZXFawcSglQSXqCftWX3ec8MDUzQdHgcKvw97awHbY0efQEL5iKUOAmmVtoYgmrSG4Q==",
       "dev": true,
       "requires": {
         "type-detect": "4.0.8"
@@ -2526,9 +2526,9 @@
       "dev": true
     },
     "spdx-expression-parse": {
-      "version": "3.0.0",
-      "resolved": "https://registry.npmjs.org/spdx-expression-parse/-/spdx-expression-parse-3.0.0.tgz",
-      "integrity": "sha512-Yg6D3XpRD4kkOmTpdgbUiEJFKghJH03fiC1OPll5h/0sO6neh2jqRDVHOQ4o/LMea0tgCkbMgea5ip/e+MkWyg==",
+      "version": "3.0.1",
+      "resolved": "https://registry.npmjs.org/spdx-expression-parse/-/spdx-expression-parse-3.0.1.tgz",
+      "integrity": "sha512-cbqHunsQWnJNE6KhVSMsMeH5H/L9EpymbzqTQ3uLwNCLZ1Q481oWaofqH7nO6V07xlXwY6PhQdQ2IedWx/ZK4Q==",
       "dev": true,
       "requires": {
         "spdx-exceptions": "^2.1.0",
diff --git a/scripts/deploy.sh b/scripts/deploy.sh
index 1def2e3..6007e13 100644
--- a/scripts/deploy.sh
+++ b/scripts/deploy.sh
@@ -1,6 +1,6 @@
 # config
-BUCKET_NAME=your-sam-templates-bucket
-STACK_NAME=lambda-power-tuning
+BUCKET_NAME=sam-templates-demos-cpt
+STACK_NAME=lambda-power-tuning-new
 PowerValues='128,256,512,1024,1536,3008'
 LambdaResource='*'
 


Test patch:
diff --git a/test/unit/test-lambda.js b/test/unit/test-lambda.js
index f0a1193..d39a4cc 100644
--- a/test/unit/test-lambda.js
+++ b/test/unit/test-lambda.js
@@ -67,6 +67,7 @@ var getLambdaAliasStub,
     updateLambdaAliasStub,
     deleteLambdaVersionStub,
     invokeLambdaStub,
+    invokeLambdaProcessorStub,
     deleteLambdaAliasStub;
 
 /** unit tests below **/
@@ -339,11 +340,14 @@ describe('Lambda Functions', async() => {
         const handler = require('../../lambda/executor').handler;
 
         var invokeLambdaCounter,
-            invokeLambdaPayloads;
+            invokeLambdaPayloads,
+            invokeProcessorCounter;
 
         beforeEach('mock utilities', () => {
             invokeLambdaCounter = 0;
             invokeLambdaPayloads = [];
+            invokeProcessorCounter = 0;
+
             invokeLambdaStub && invokeLambdaStub.restore();
             invokeLambdaStub = sandBox.stub(utils, 'invokeLambda')
                 .callsFake(async(_arn, _alias, payload) => {
@@ -354,7 +358,16 @@ describe('Lambda Functions', async() => {
                         StatusCode: 200,
                         LogResult: 'U1RBUlQgUmVxdWVzdElkOiA0NzlmYjUxYy0xZTM4LTExZTctOTljYS02N2JmMTYzNjA4ZWQgVmVyc2lvbjogOTkKMjAxNy0wNC0xMFQyMTo1NDozMi42ODNaCTQ3OWZiNTFjLTFlMzgtMTFlNy05OWNhLTY3YmYxNjM2MDhlZAl2YWx1ZTEgPSB1bmRlZmluZWQKMjAxNy0wNC0xMFQyMTo1NDozMi42ODNaCTQ3OWZiNTFjLTFlMzgtMTFlNy05OWNhLTY3YmYxNjM2MDhlZAl2YWx1ZTIgPSB1bmRlZmluZWQKMjAxNy0wNC0xMFQyMTo1NDozMi42ODNaCTQ3OWZiNTFjLTFlMzgtMTFlNy05OWNhLTY3YmYxNjM2MDhlZAl2YWx1ZTMgPSB1bmRlZmluZWQKRU5EIFJlcXVlc3RJZDogNDc5ZmI1MWMtMWUzOC0xMWU3LTk5Y2EtNjdiZjE2MzYwOGVkClJFUE9SVCBSZXF1ZXN0SWQ6IDQ3OWZiNTFjLTFlMzgtMTFlNy05OWNhLTY3YmYxNjM2MDhlZAlEdXJhdGlvbjogMS4wIG1zCUJpbGxlZCBEdXJhdGlvbjogMTAwIG1zIAlNZW1vcnkgU2l6ZTogMTI4IE1CCU1heCBNZW1vcnkgVXNlZDogMTUgTUIJCg==',
                         ExecutedVersion: '$LATEST',
-                        Payload: '{}' };
+                        Payload: '{}',
+                    };
+                });
+
+            invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
+            invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
+                .callsFake(async(_arn, _alias, payload) => {
+                    invokeProcessorCounter++;
+                    invokeLambdaCounter++;
+                    return '{"Processed": true}';
                 });
         });
 
@@ -580,12 +593,14 @@ describe('Lambda Functions', async() => {
         });
 
         it('should report an error if invocation fails', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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]]}',
+                    };
+                });
             await invokeForFailure(handler, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -594,12 +609,14 @@ describe('Lambda Functions', async() => {
         });
 
         it('should include payload in exception message if invocation fails (series)', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -612,12 +629,14 @@ describe('Lambda Functions', async() => {
         });
 
         it('should include payload in exception message if invocation fails (parallel)', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -632,12 +651,14 @@ describe('Lambda Functions', async() => {
 
 
         it('should include weighted payload in exception message if invocation fails (series)', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -653,12 +674,14 @@ describe('Lambda Functions', async() => {
         });
 
         it('should include weighted payload in exception message if invocation fails (parallel)', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -675,12 +698,14 @@ describe('Lambda Functions', async() => {
         });
 
         it('should report an error if invocation fails (parallel)', async() => {
-            utils.invokeLambda = async() => {
-                return {
-                    FunctionError: 'Unhandled',
-                    Payload: '{"errorType": "MemoryError", "stackTrace": [["/var/task/lambda_function.py", 11, "lambda_handler", "blabla"], ["/var/task/lambda_function.py", 7, "blabla]]}',
-                };
-            };
+            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]]}',
+                    };
+                });
             await invokeForFailure(handler, {
                 lambdaARN: 'arnOK',
                 value: '1024',
@@ -710,6 +735,111 @@ describe('Lambda Functions', async() => {
             expect(invokeLambdaCounter).to.be(1);
         });
 
+        it('should invoke pre-processor', async() => {
+            const num = 10;
+            await invokeForSuccess(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: num,
+                preProcessorARN: 'preArnOK',
+            });
+            expect(invokeLambdaCounter).to.be(num * 2);
+            expect(invokeProcessorCounter).to.be(num);
+        });
+
+        it('should invoke post-processor', async() => {
+            const num = 10;
+            await invokeForSuccess(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: num,
+                postProcessorARN: 'postArnOK',
+            });
+            expect(invokeLambdaCounter).to.be(num * 2);
+            expect(invokeProcessorCounter).to.be(num);
+        });
+
+        it('should invoke pre-processor and post-processor', async() => {
+            const num = 10;
+            await invokeForSuccess(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: num,
+                preProcessorARN: 'preArnOK',
+                postProcessorARN: 'postArnOK',
+            });
+            expect(invokeLambdaCounter).to.be(num * 3);
+            expect(invokeProcessorCounter).to.be(num * 2);
+        });
+
+        it('should invoke function with pre-processor output', async() => {
+            const num = 10;
+            await invokeForSuccess(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: num,
+                preProcessorARN: 'preArnOK',
+            });
+            expect(invokeLambdaPayloads[0].includes('Processed')).to.be(true);
+        });
+
+        it('should invoke function with original payload if pre-precessor does not return a new payload', async() => {
+
+            invokeLambdaProcessorStub && invokeLambdaProcessorStub.restore();
+            invokeLambdaProcessorStub = sandBox.stub(utils, 'invokeLambdaProcessor')
+                .callsFake(async(_arn, _alias, payload) => {
+                    invokeProcessorCounter++;
+                    invokeLambdaCounter++;
+                    return null; // empty output from pre-processor
+                });
+
+            const num = 10;
+            await invokeForSuccess(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: num,
+                payload: {Original: true},
+                preProcessorARN: 'preArnOK',
+            });
+            expect(invokeLambdaPayloads[0].includes('Original')).to.be(true);
+        });
+
+        it('should explode if pre-processor fails', 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');
+                });
+
+            await invokeForFailure(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: 10,
+                payload: {Original: true},
+                preProcessorARN: 'preArnOK',
+            });
+
+        });
+
+        it('should explode if post-processor fails', 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');
+                });
+
+            await invokeForFailure(handler, {
+                lambdaARN: 'arnOK',
+                value: '128',
+                num: 10,
+                payload: {Original: true},
+                postProcessorARN: 'postArnOK',
+            });
+
+        });
+
     });
 
     describe('analyzer', () => {
diff --git a/test/unit/test-utils.js b/test/unit/test-utils.js
index b5a4c2f..b900121 100644
--- a/test/unit/test-utils.js
+++ b/test/unit/test-utils.js
@@ -40,10 +40,12 @@ describe('Lambda Utils', () => {
         utils.updateLambdaAlias,
         utils.deleteLambdaAlias,
         utils.invokeLambda,
+        utils.invokeLambdaWithProcessors,
     ];
 
     // TODO fix me (use proper mocking in test-lambda.js)
     const getLambdaPower = utils.getLambdaPower;
+    const invokeLambdaProcessor = utils.invokeLambdaProcessor;
 
     // just returns the utility name for convenience
     function _fname(func) {
@@ -315,4 +317,44 @@ describe('Lambda Utils', () => {
             expect(utils.baseCostForRegion(prices, 'eu-west-1')).to.be(0.0000002083);
         });
     });
+
+    describe('invokeLambdaProcessor', () => {
+        // TODO proper mocking
+
+        var invokeLambdaCounter;
+        beforeEach('mock API call', () => {
+            invokeLambdaCounter = 0;
+        });
+
+        it('should invoke the processing function', async() => {
+            utils.invokeLambda = async(_arn, _alias, payload) => {
+                invokeLambdaCounter++;
+                return {
+                    Payload: '{"OK": "OK"}',
+                };
+            };
+            const data = await invokeLambdaProcessor('arnOK', {});
+            expect(invokeLambdaCounter).to.be(1);
+            expect(data).to.be('{"OK": "OK"}');
+        });
+
+        it('should explode if processor fails', async() => {
+            utils.invokeLambda = async(_arn, _alias, _payload) => {
+                invokeLambdaCounter++;
+                return {
+                    Payload: '{"KO": "KO"}',
+                    FunctionError: 'Unhandled',
+                };
+            };
+            try {
+                const data = await invokeLambdaProcessor('arnOK', {});
+                expect(data).to.be(null);
+            } catch (ex) {
+                expect(ex.message.includes('failed with error')).to.be(true);
+            }
+
+            expect(invokeLambdaCounter).to.be(1);
+        });
+    });
+
 });