init

Author: Michael Bahr

.gitignore

@@ -0,0 +1,23 @@
+# Distribution / packaging
+.Python
+*.pyc
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Serverless directories
+.serverless
+.idea
+venv

README.md

@@ -0,0 +1,35 @@
+# aws-scheduler-testing
+
+This is a project to help with testing and performance measurements of the [aws-scheduler](https://github.com/bahrmichael/aws-scheduler).
+
+## Setup
+
+### Prerequisites
+You must have the following tools installed:
+- serverless framework 1.48.3 or later
+- node
+- npm
+- python3
+- pip
+
+Run `setup/init_table.py <stage>` to setup the database. Replace `<stage>` with the stage of your application (e.g. `dev`).
+
+Run `setup/init_output_topic.py <stage>`. Replace `<stage>` with the stage of your application (e.g. `dev`).
+
+### Deploy
+1. Navigate into the project folder
+2. With a tooling of your choice create and activate a venv
+3. `pip install -r requirements.txt`
+4. `npm i serverless-python-requirements`
+5. `sls deploy`
+6. Optional: `pip install matplotlib` if you want to plot the delays later 
+
+Wait for the deployment to finish. Test the service by first attaching a function to the output topic and then send a few events to the input topic.
+
+## Run it
+
+Once you created the database and topic and deployed the stack, run the producer with `python producer.py <your-output-topic> [amount] [input-topic]`. Replace `<your-output-topic>` with the arn of the topic you created during the setup. You may specify a second argument to set the number of events, the default is 100. You may specify a third argument to use a different input topic. You may also trigger the function from the aws lambda console.
+
+Run `sls logs -f consumer -t` to tail the logs. You may receive an error that the log stream does not exist until the first event has arrived.
+
+After 5 minutes all events should have arrived at the output topic and the measured delays are stored in the database. You can now run `python evaluate_measurement.py` to plot the delays.

consumer.py

@@ -0,0 +1,45 @@
+import time
+from datetime import datetime
+from uuid import uuid4
+
+from pynamodb.exceptions import PutError
+
+from model import MeasuredDuration
+
+
+def handle(event, context):
+    delays = []
+    for record in event['Records']:
+        print(record)
+        payload = record['Sns']['Message']
+        execution_time = datetime.fromisoformat(payload)
+        delta = int((datetime.utcnow() - execution_time).total_seconds() * 1000)
+        print(f'delay: {delta}')
+        delays.append(delta)
+
+    with MeasuredDuration.batch_write() as batch:
+        retries = []
+        for delay in delays:
+            item = MeasuredDuration()
+            item.id = str(uuid4())
+            item.delay = delay
+            try:
+                batch.save(item)
+            except PutError as e:
+                print(e)
+                time.sleep(.200)
+                retries.append(delay)
+
+        while len(retries) > 0:
+            delay = retries.pop(0)
+            item = MeasuredDuration()
+            item.id = str(uuid4())
+            item.delay = delay
+            try:
+                batch.save(item)
+            except PutError as e:
+                print(e)
+                time.sleep(.200)
+                retries.append(delay)
+
+    print('Processed %d records' % len(event['Records']))

evaluate_measurements.py

@@ -0,0 +1,34 @@
+import matplotlib.pyplot as plt
+from model import MeasuredDuration
+
+
+def download():
+    delays = []
+    counter = 0
+    for item in MeasuredDuration.scan():
+        delays.append(item.delay)
+        counter += 1
+        if counter % 1000 == 0:
+            print(f'{counter}')
+    return delays
+
+
+if __name__ == '__main__':
+    data = download()
+
+    print(f'total: {len(data)}')
+    plt.hist(data, bins=100)
+
+    plt.title(f'Regular Scaled ({len(data)} events)')
+    plt.xlabel("Delay after scheduled time (milliseconds)")
+    plt.ylabel("Number of events")
+    plt.savefig('regular.png')
+
+    plt.yscale('log')
+    plt.xscale('log')
+    plt.title(f'Log Scaled ({len(data)} events)')
+    plt.xlabel("Delay after scheduled time (milliseconds)")
+    plt.ylabel("Number of events")
+    plt.savefig('log.png')
+
+    print('Diagrams saved to regular.png and log.png')

init_output_topic.py

@@ -0,0 +1,38 @@
+import sys
+
+import boto3
+
+client = boto3.client('sns')
+
+
+if __name__ == '__main__':
+
+    if len(sys.argv) < 2:
+        print('Missing argument for stage.')
+        exit()
+    stage = sys.argv[1]
+
+    name = f'scheduler-output-{stage}'
+
+    print(f'Creating topic {name}')
+    create_response = client.create_topic(
+        Name=name,
+    )
+
+    arn = create_response['TopicArn']
+    print(f'Created topic {name} with arn {arn}')
+
+    if len(sys.argv) == 3:
+        role = sys.argv[2]
+    else:
+        role = 'arn:aws:sts::256608350746:assumed-role/aws-scheduler-prod-us-east-1-lambdaRole/aws-scheduler-prod-emitter'
+    print(f'Granting publish rights to {name} topic for role {role}')
+
+    permission_response = client.add_permission(
+        TopicArn=arn,
+        Label=f'{name}-publish-access',
+        AWSAccountId=[str(role)],
+        ActionName=['Publish']
+    )
+
+    print('Done')

init_table.py

@@ -0,0 +1,32 @@
+import boto3
+
+client = boto3.client('dynamodb')
+
+
+def create():
+    name = 'aws-scheduler-testing'
+    response = client.list_tables()
+    if name in response['TableNames']:
+        print('Table %s already exists. Please delete it first.' % name)
+        return
+    client.create_table(
+        TableName=name,
+        AttributeDefinitions=[
+            {
+                'AttributeName': 'id',
+                'AttributeType': 'S'
+            }
+        ],
+        KeySchema=[
+            {
+                'AttributeName': 'id',
+                'KeyType': 'HASH'
+            }
+        ],
+        BillingMode='PAY_PER_REQUEST',
+    )
+    print('%s created' % name)
+
+
+if __name__ == '__main__':
+    create()

model.py

@@ -0,0 +1,13 @@
+from pynamodb.attributes import (
+    UnicodeAttribute,
+    NumberAttribute)
+from pynamodb.models import Model
+
+
+class MeasuredDuration(Model):
+
+    class Meta:
+        table_name = 'aws-scheduler-testing'
+
+    id = UnicodeAttribute(hash_key=True)
+    delay = NumberAttribute()

producer.py

@@ -0,0 +1,57 @@
+import json
+import os
+import sys
+from datetime import datetime, timedelta
+from random import randrange
+
+import boto3
+
+client = boto3.client('sns')
+
+
+def handle(event, context):
+    input_topic = os.environ.get('INPUT_TOPIC_ARN')
+    target = os.environ.get('OUTPUT_TOPIC_ARN')
+    run(100, input_topic, target)
+
+
+def run(target, input_topic, amount):
+    start = datetime.utcnow()
+    for i in range(0, amount):
+        delay = randrange(60 * 5)
+        scheduled_time = (datetime.utcnow() + timedelta(seconds=delay)).isoformat()
+        payload = {
+            'payload': scheduled_time,
+            'date': scheduled_time,
+            'target': target
+        }
+
+        client.publish(
+            TopicArn=input_topic,
+            Message=json.dumps(payload)
+        )
+
+        if i % 10 == 0:
+            total_millis = (datetime.utcnow() - start).total_seconds() * 1000
+            print(f'total: {i}, {int(total_millis / i)} per event')
+
+
+if __name__ == '__main__':
+    arguments = sys.argv[1:]
+    if len(arguments) >= 1:
+        target = arguments[0]
+    else:
+        print('You must specify the output topic as the first parameter')
+        exit()
+
+    if len(arguments) >= 3:
+        input_topic = arguments[2]
+    else:
+        input_topic = 'arn:aws:sns:us-east-1:256608350746:scheduler-input-prod'
+
+    if len(arguments) >= 2:
+        amount = int(arguments[1])
+    else:
+        amount = 100
+
+    run(target, input_topic, amount)

requirements.txt

@@ -0,0 +1,9 @@
+boto3==1.9.197
+botocore==1.12.197
+docutils==0.14
+jmespath==0.9.4
+pynamodb==3.4.1
+python-dateutil==2.8.0
+s3transfer==0.2.1
+six==1.12.0
+urllib3==1.25.3

serverless.yml

@@ -0,0 +1,44 @@
+service: aws-scheduler-testing
+
+provider:
+  name: aws
+  runtime: python3.7
+  stage: ${opt:stage, 'dev'}
+  region: ${opt:region, 'us-east-1'}
+  iamRoleStatements:
+    - Effect: Allow
+      Action:
+        - dynamodb:PutItem
+      Resource: { "Fn::Join": [":", ["arn:aws:dynamodb:${self:provider.region}", { "Ref": "AWS::AccountId" }, "table/aws-scheduler-testing" ] ]  }
+
+custom:
+  output_topic:
+    name:
+    arn: { "Fn::Join": [":", ["arn:aws:sns:${self:provider.region}", { "Ref": "AWS::AccountId" }, "table/scheduler-output-${self:provider.stage}" ] ]  }
+
+functions:
+  consumer:
+    handler: consumer.handle
+    events:
+      - sns:
+          arn:
+            Fn::Join:
+              - ':'
+              - - 'arn:aws:sns'
+                - Ref: 'AWS::Region'
+                - Ref: 'AWS::AccountId'
+                - "${self:custom.inbound.name}"
+          topicName: "${self:custom.inbound.name}"
+  producer:
+    handler: producer.handle
+    environment:
+      INPUT_TOPIC_ARN: 'arn:aws:sns:us-east-1:256608350746:scheduler-input-prod'
+      OUTPUT_TOPIC_ARN: { "Fn::Join": [":", ["arn:aws:sns:${self:provider.region}", { "Ref": "AWS::AccountId" }, "table/scheduler-output-${self:provider.stage}" ] ]  }
+
+plugins:
+  - serverless-python-requirements
+
+package:
+  exclude:
+    - venv/**
+    - node_modules/**