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

Issue:
Add wait/delay between invocations for heavily rate limited downstream dependencies
When downstream dependencies are heavily rate limited, there's no easy way to natively 'throttle' the power tuner to accommodate.

One existing potential solution would be to have a Lambda function that essentially does a `Thread.Sleep(time)` to add a delay (and of course not run the tuner in parallel).

Given that involves paying for Lambda and a State Transition, could it be possible to add an additional parameter of `sleepBetweenRunsMs` or something similar to indicate a sleep period between power tuner runs?

There are some downsides to this approach though. It would involve a state transition for every run, even if the wait/delay time was set to 0. A possible approach to this is to have a Choice state at the start that either does the current branch work flow, or does a 'new' branch/workflow that is all essentially a clone of the current flow but now has a Wait task as part of the map/loop. This could look something like:
![image](https://user-images.githubusercontent.com/1962883/196672338-81f84f0b-d8d6-477a-bb9d-36ead5ec2091.png)
This approach would only add a single additional state transition, which whilst not ideal as every user would incur that cost, it may be an acceptable addition for the benefit of an easy to configure/use Wait/Delay between each invocation.

Maybe this has been discussed before and was rejected because of the ability to add a function that does a naïve sleep?

Pull request:
Add configurable sleep parameter between invocations
- Add a configurable time in milliseconds that the executor function will wait between invoking the users function.
- Updated readme docs with details about the usage of the parameter

Fixes #179 

Patch:
diff --git a/README-INPUT-OUTPUT.md b/README-INPUT-OUTPUT.md
index 57eede5..f83859b 100644
--- a/README-INPUT-OUTPUT.md
+++ b/README-INPUT-OUTPUT.md
@@ -21,7 +21,7 @@ The state machine accepts the following input parameters:
 * **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-functions)
 * **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-functions)
 * **discardTopBottom** (number between 0.0 and 0.4, by default is 0.2): By default, the state machine will discard the top/bottom 20% of "outliers" (the fastest and slowest), to filter out the effects of cold starts that would bias the overall averages. You can customize this parameter by providing a value between 0 and 0.4, with 0 meaning no results are discarded and 0.4 meaning that 40% of the top/bottom results are discarded (i.e. only 20% of the results are considered).
-
+* **sleepBetweenRunsMs** (integer) If provided, the time in milliseconds that the tuner function will sleep/wait after invoking your function, but before carrying out the Post-Processing step, should that be provided. This could be used if you have agressive downstream rate limits you need to respect. By default this will be set to 0 and the function won't sleep between invocations. Setting this value will have no effect if running the invocations in parallel.
 
 ## State machine configuration (at deployment time)
 
@@ -35,7 +35,7 @@ The CloudFormation template accepts the following parameters:
 * **payloadS3Bucket** (string): the S3 bucket name used for large payloads (>256KB); if provided, it's added to a custom managed IAM policy that grants read-only permission to the S3 bucket; more details below in the [S3 payloads section](#user-content-s3-payloads)
 * **payloadS3Key** (string, default=`*`): they S3 object key used for large payloads (>256KB); the default value grants access to all S3 objects in the bucket specified with `payloadS3Bucket`; more details below in the [S3 payloads section](#user-content-s3-payloads)
 
-Please note that the total execution time should stay below 300 seconds (5 min), which is the default timeout. You can easily estimate the total execution timeout based on the average duration of your functions. For example, if your function's average execution time is 5 seconds and you haven't enabled `parallelInvocation`, you should set `totalExecutionTimeout` to at least `num * 5`: 50 seconds if `num=10`, 500 seconds if `num=100`, and so on. If you have enabled `parallelInvocation`, usually you don't need to tune the value of `totalExecutionTimeout` unless your average execution time is above 5 min.
+Please note that the total execution time should stay below 300 seconds (5 min), which is the default timeout. You can easily estimate the total execution timeout based on the average duration of your functions. For example, if your function's average execution time is 5 seconds and you haven't enabled `parallelInvocation`, you should set `totalExecutionTimeout` to at least `num * 5`: 50 seconds if `num=10`, 500 seconds if `num=100`, and so on. If you have enabled `parallelInvocation`, usually you don't need to tune the value of `totalExecutionTimeout` unless your average execution time is above 5 min. If you have a sleep between invocations set, you should include that in your timeout calculations.
 
 ### Usage in CI/CD pipelines
 
diff --git a/lambda/executor.js b/lambda/executor.js
index 0fff1eb..c1d37af 100644
--- a/lambda/executor.js
+++ b/lambda/executor.js
@@ -26,6 +26,7 @@ module.exports.handler = async(event, context) => {
         preProcessorARN,
         postProcessorARN,
         discardTopBottom,
+        sleepBetweenRunsMs,
     } = await extractDataFromInput(event);
 
     validateInput(lambdaARN, value, num); // may throw
@@ -46,10 +47,20 @@ module.exports.handler = async(event, context) => {
     // pre-generate an array of N payloads
     const payloads = utils.generatePayloads(num, payload);
 
+    const runInput = {
+        num: num,
+        lambdaARN: lambdaARN,
+        lambdaAlias: lambdaAlias,
+        payloads: payloads,
+        preARN: preProcessorARN,
+        postARN: postProcessorARN,
+        sleepBetweenRunsMs: sleepBetweenRunsMs,
+    };
+
     if (enableParallel) {
-        results = await runInParallel(num, lambdaARN, lambdaAlias, payloads, preProcessorARN, postProcessorARN);
+        results = await runInParallel(runInput);
     } else {
-        results = await runInSeries(num, lambdaARN, lambdaAlias, payloads, preProcessorARN, postProcessorARN);
+        results = await runInSeries(runInput);
     }
 
     // get base cost for Lambda
@@ -90,10 +101,21 @@ const extractDiscardTopBottomValue = (event) => {
     return Math.min(Math.max(discardTopBottom, 0.0), 0.4);
 };
 
+const extractSleepTime = (event) => {
+    let sleepBetweenRunsMs = event.sleepBetweenRunsMs;
+    if (isNaN(sleepBetweenRunsMs)) {
+        sleepBetweenRunsMs = 0;
+    } else {
+        sleepBetweenRunsMs = parseInt(sleepBetweenRunsMs, 10);
+    }
+    return sleepBetweenRunsMs;
+};
+
 const extractDataFromInput = async(event) => {
     const input = event.input; // original state machine input
     const payload = await extractPayloadValue(input);
     const discardTopBottom = extractDiscardTopBottomValue(input);
+    const sleepBetweenRunsMs = extractSleepTime(input);
     return {
         value: parseInt(event.value, 10),
         lambdaARN: input.lambdaARN,
@@ -104,10 +126,11 @@ const extractDataFromInput = async(event) => {
         preProcessorARN: input.preProcessorARN,
         postProcessorARN: input.postProcessorARN,
         discardTopBottom: discardTopBottom,
+        sleepBetweenRunsMs: sleepBetweenRunsMs,
     };
 };
 
-const runInParallel = async(num, lambdaARN, lambdaAlias, payloads, preARN, postARN) => {
+const runInParallel = async({num, lambdaARN, lambdaAlias, payloads, preARN, postARN}) => {
     const results = [];
     // run all invocations in parallel ...
     const invocations = utils.range(num).map(async(_, i) => {
@@ -123,7 +146,7 @@ const runInParallel = async(num, lambdaARN, lambdaAlias, payloads, preARN, postA
     return results;
 };
 
-const runInSeries = async(num, lambdaARN, lambdaAlias, payloads, preARN, postARN) => {
+const runInSeries = async({num, lambdaARN, lambdaAlias, payloads, preARN, postARN, sleepBetweenRunsMs}) => {
     const results = [];
     for (let i = 0; i < num; i++) {
         // run invocations in series
@@ -132,6 +155,9 @@ const runInSeries = async(num, lambdaARN, lambdaAlias, payloads, preARN, postARN
         if (invocationResults.FunctionError) {
             throw new Error(`Invocation error (running in series): ${invocationResults.Payload} with payload ${JSON.stringify(actualPayload)}`);
         }
+        if (sleepBetweenRunsMs > 0) {
+            await utils.sleep(sleepBetweenRunsMs);
+        }
         results.push(invocationResults);
     }
     return results;
diff --git a/lambda/utils.js b/lambda/utils.js
index a03fcfa..5e6e327 100644
--- a/lambda/utils.js
+++ b/lambda/utils.js
@@ -462,7 +462,7 @@ module.exports.computeAverageDuration = (durations, discardTopBottom) => {
         // not an error, but worth logging
         // this happens when you have less than 5 invocations
         // (only happens if dryrun or in tests)
-        console.log("not enough results to discard");
+        console.log('not enough results to discard');
     }
 
     const newN = durations.length - 2 * toBeDiscarded;
@@ -562,3 +562,8 @@ module.exports.baseCostForRegion = (priceMap, region) => {
     console.log(region + ' not found in base price map, using default: ' + priceMap['default']);
     return priceMap['default'];
 };
+
+
+module.exports.sleep = async(sleepBetweenRunsMs) => {
+    await new Promise(resolve => setTimeout(resolve, sleepBetweenRunsMs));
+};


Test patch:
diff --git a/test/unit/test-lambda.js b/test/unit/test-lambda.js
index 01499b5..181132f 100644
--- a/test/unit/test-lambda.js
+++ b/test/unit/test-lambda.js
@@ -29,7 +29,8 @@ var setLambdaPowerCounter,
     publishLambdaVersionCounter,
     createLambdaAliasCounter,
     updateLambdaAliasCounter,
-    waitForFunctionUpdateCounter;
+    waitForFunctionUpdateCounter,
+    sleepCounter;
 
 // utility to invoke handler (success case)
 const invokeForSuccess = async(handler, event) => {
@@ -84,6 +85,7 @@ describe('Lambda Functions', async() => {
         createLambdaAliasCounter = 0;
         updateLambdaAliasCounter = 0;
         waitForFunctionUpdateCounter = 0;
+        sleepCounter = 0;
 
         sandBox.stub(utils, 'regionFromARN')
             .callsFake((arn) => {
@@ -129,6 +131,11 @@ describe('Lambda Functions', async() => {
                 waitForFunctionUpdateCounter++;
                 return 'OK';
             });
+        sandBox.stub(utils, 'sleep')
+            .callsFake(async(_) => {
+                sleepCounter++;
+                return 'OK, no need to wait =)';
+            });
     });
 
     afterEach('Global mock utilities afterEach', () => {
@@ -442,6 +449,49 @@ describe('Lambda Functions', async() => {
             expect(getLambdaArchitectureCounter).to.be(1);
         });
 
+        it('should invoke the given cb, when done with between function sleep', async() => {
+
+            await invokeForSuccess(handler, {
+                value: '128',
+                input: {
+                    lambdaARN: 'arnOK',
+                    num: 10,
+                    sleepBetweenRunsMs: 50,
+                },
+            });
+
+            expect(getLambdaArchitectureCounter).to.be(1);
+            expect(sleepCounter).to.be(10);
+
+        });
+
+        it('should invoke the given cb, when done (parallelInvocation) and ignore function sleep', async() => {
+            await invokeForSuccess(handler, {
+                value: '128',
+                input: {
+                    lambdaARN: 'arnOK',
+                    num: 20,
+                    sleepBetweenRunsMs: 50,
+                    parallelInvocation: true,
+                },
+            });
+            expect(getLambdaArchitectureCounter).to.be(1);
+            expect(sleepCounter).to.be(0);
+        });
+
+        it('should invoke the given cb, when done with invalid function sleep', async() => {
+            await invokeForSuccess(handler, {
+                value: '128',
+                input: {
+                    lambdaARN: 'arnOK',
+                    num: 100,
+                    sleepBetweenRunsMs: 'Not a number',
+                },
+            });
+            expect(getLambdaArchitectureCounter).to.be(1);
+            expect(sleepCounter).to.be(0);
+        });
+
         it('should return statistics (default)', async() => {
             const response = await invokeForSuccess(handler, {
                 value: '128',
diff --git a/test/unit/test-utils.js b/test/unit/test-utils.js
index 98a7e92..b5f6f87 100644
--- a/test/unit/test-utils.js
+++ b/test/unit/test-utils.js
@@ -835,4 +835,23 @@ describe('Lambda Utils', () => {
 
     });
 
+    describe('sleep', () => {
+
+        it('should wait X milliseconds', async() => {
+
+            const clock = sinon.useFakeTimers();
+
+            // sleep 10 seconds
+            // without await (time is "locked" with fake timers)
+            utils.sleep(10000);
+
+            // release all timers so we don't have to actually wait
+            await clock.runAllAsync();
+            clock.restore();
+
+        });
+
+
+    });
+
 });