speed improvements

Author: Michael Bahr

consumer.py

@@ -12,6 +12,7 @@
 
 
 def publish_to_failure_topic(event, reason):
+    # todo: prepare against failure of publish sns
     print('Event failed: %s' % event)
     if 'failure_topic' in event:
         payload = {
@@ -22,19 +23,23 @@ def publish_to_failure_topic(event, reason):
 
 
 def handle(events):
+    received = datetime.utcnow()
     to_be_scheduled = []
     event_wrappers = []
     for event in events:
         print(event)
 
         if 'date' not in event:
             publish_to_failure_topic(event, 'date is required')
+            print('error.date_required %s' % (json.dumps({'event': event})))
             continue
         if 'payload' not in event:
             publish_to_failure_topic(event, 'payload is required')
+            print('error.payload_required %s' % (json.dumps({'event': event})))
             continue
         if 'target' not in event:
             publish_to_failure_topic(event, 'target is required')
+            print('error.target_required %s' % (json.dumps({'event': event})))
             continue
 
         event_wrapper = EventWrapper()
@@ -43,14 +48,16 @@ def handle(events):
 
         if not isinstance(event['payload'], str):
             publish_to_failure_topic(event, 'payload must be a string')
+            print('error.payload_is_not_string %s' % (json.dumps({'event': event})))
             continue
 
         event_wrapper.payload = event['payload']
         event_wrapper.target = event['target']
 
         if 'user' not in event:
-            if 'true' == os.environ.get('ENFORCE_USER'):
+            if os.environ.get('ENFORCE_USER'):
                 publish_to_failure_topic(event, 'user is required')
+                print('error.event_has_no_user %s' % (json.dumps({'event': event})))
                 continue
         else:
             event_wrapper.user = event['user']
@@ -59,6 +66,7 @@ def handle(events):
         if has_less_then_ten_minutes(event_wrapper.date):
             to_be_scheduled.append(event_wrapper.id)
 
+        print('event.consumed %s' % (json.dumps({'id': event_wrapper.id, 'timestamp': str(received), 'user': event_wrapper.user})))
         event_wrappers.append(event_wrapper)
 
     # we must save before delegating, because the downstream function will access the DB entity

emitter.py

@@ -1,47 +1,41 @@
 import json
+import time
+from datetime import datetime
 
-from db_helper import delete_with_retry, save_with_retry
+from db_helper import save_with_retry
 from model import EventWrapper
-
 from sns_client import publish_sns
 
 
 def handle(items):
-    # todo: remove the message from the queue?
-    processed_ids = []
     failed_ids = []
     print(f'Processing {len(items)} records')
+
+    # sort the items so that we process the earliest first
+    items.sort(key=lambda x: x['date'])
+
     for item in items:
         event_id = item['id']
-        try:
-            if EventWrapper.count(hash_key=event_id) == 0:
-                # if the event was already deleted from the database, then don't send it again
-                continue
-        except Exception as e:
-            print(e)
-            # if we can't determine if the event was already processed, then we'll send it
-            # to make sure we have at least one delivery
+
+        # the event we received may have been scheduled early
+        scheduled_execution = datetime.fromisoformat(item['date'])
+
+        delay = (scheduled_execution - datetime.utcnow()).total_seconds()
+        # remove another 10ms as there will be a short delay between the emitter, the target sns and its consumer
+        delay -= 0.01
+        # if there is a positive delay then wait until it's time
+        if delay > 0:
+            time.sleep(delay)
 
         try:
             publish_sns(item['target'], item['payload'])
-            processed_ids.append(event_id)
+            print('event.emitted %s' % (json.dumps({'id': event_id, 'timestamp': str(datetime.utcnow()), 'scheduled': str(scheduled_execution)})))
         except Exception as e:
-            print(e)
+            print(str(e))
             failed_ids.append(event_id)
 
-    to_delete = []
-    for event_id in processed_ids:
-        try:
-            to_delete.append(EventWrapper.get(hash_key=event_id))
-        except Exception as e:
-            print(f'Skipped {event_id} because it doesn\'t exist anymore')
-            print(e)
-
-    delete_with_retry(to_delete)
-
     failed_items = []
     for event_id in failed_ids:
-        # todo: maybe also emit this to an error topic?
         try:
             event = EventWrapper.get(hash_key=event_id)
             event.status = 'FAILED'
@@ -53,15 +47,9 @@ def handle(items):
                     'error': 'ERROR',
                     'event': event.payload
                 }
-                # todo: let the publish_sns/sqs methods do the json dumping themselves if they encounter a non-string
-                # i believe json.dumps("test") results in "test", if that's correct then we can always apply json.dumps()
                 publish_sns(event.failure_topic, json.dumps(payload))
         except Exception as e:
-            print(f'Skipped {event_id} because it doesn\'t exist anymore')
-            print(e)
+            print(f'Failure update: Skipped {event_id} because it doesn\'t exist anymore')
+            print(str(e))
 
     save_with_retry(failed_items)
-
-
-if __name__ == '__main__':
-    print(json.dumps("123"))

event_loader.py

@@ -47,13 +47,13 @@ def run():
     print('Batched %d entries' % count)
 
 
-def load_data(until, last_evaluated_key, limit=5000):
+def load_data(until, last_evaluated_key, limit=5000, status='NEW'):
     if last_evaluated_key is None:
         response = table.query(
             IndexName=os.environ.get('INDEX_NAME'),
             KeyConditionExpression='#status = :st and #date < :until',
             ExpressionAttributeNames={"#status": "status", "#date": "date"},
-            ExpressionAttributeValues={":until": until, ':st': 'NEW'},
+            ExpressionAttributeValues={":until": until, ':st': status},
             Limit=limit,
             ProjectionExpression='id'
         )
@@ -62,7 +62,7 @@ def load_data(until, last_evaluated_key, limit=5000):
             IndexName=os.environ.get('INDEX_NAME'),
             KeyConditionExpression='#status = :st and #date < :until',
             ExpressionAttributeNames={"#status": "status", "#date": "date"},
-            ExpressionAttributeValues={":until": until, ':st': 'NEW'},
+            ExpressionAttributeValues={":until": until, ':st': status},
             Limit=limit,
             ProjectionExpression='id',
             ExclusiveStartKey=last_evaluated_key

scheduler.py

@@ -12,7 +12,6 @@
 
 
 def handle(events):
-    now = datetime.utcnow()
     successful_ids = []
     failed_ids = []
 
@@ -30,18 +29,25 @@ def handle(events):
 
         events_by_id[event_id] = item
 
-        delta = datetime.fromisoformat(item.date) - now
+        delta = datetime.fromisoformat(item.date) - datetime.utcnow()
         delay = delta.total_seconds()
         rounded_delay = math.ceil(delay)
         if rounded_delay < 0:
             rounded_delay = 0
 
+        # schedule the event a second earlier to help with delays in sqs/lambda cold start
+        # the emitter will wait accordingly
+        rounded_delay -= 1
+
+        print(f'ID {event_id} is supposed to emit in {rounded_delay}s which is {delay - rounded_delay}s before target.')
+
         to_be_scheduled.append({
             'Id': event_id,
             'MessageBody': json.dumps({
                 'payload': item.payload,
                 'target': item.target,
-                'id': item.id
+                'id': item.id,
+                'date': item.date
             }),
             'DelaySeconds': rounded_delay
         })
@@ -92,11 +98,11 @@ def send_to_sqs(to_be_scheduled):
             for element in response['Successful']:
                 successful_ids.append(element['Id'])
         if 'Failed' in response:
-            print(response['Failed'])
+            print(f'ERROR: Failed to process entry: {response["Failed"]}')
             for element in response['Failed']:
                 failed_ids.append(element['Id'])
 
     except Exception as e:
-        print(e)
+        print(str(e))
         failed_ids = to_be_scheduled
     return successful_ids, failed_ids

serverless.yml

@@ -5,30 +5,35 @@ provider:
   runtime: python3.7
   stage: ${opt:stage, 'dev'}
   region: ${opt:region, 'us-east-1'}
+  environment:
+    STAGE: "${self:provider.stage}"
+  tags:
+    name: 'aws-scheduler'
   iamRoleStatements:
     - Effect: Allow
       Action:
+        - dynamodb:DescribeTable
+        - dynamodb:DeleteItem
+        - dynamodb:PutItem
         - dynamodb:DescribeTable
         - dynamodb:Query
         - dynamodb:GetItem
         - dynamodb:GetShardIterator
-        - dynamodb:PutItem
-        - dynamodb:DeleteItem
       Resource:
-        - "${self:custom.table.arn}"
         - "${self:custom.table.index.arn}"
-    - Effect: Allow
-      Action:
-        - SQS:SendMessage
-      Resource: "${self:custom.queue.arn}"
+        - "${self:custom.table.arn}"
     - Effect: Allow
       Action:
         - lambda:InvokeFunction
       Resource: { "Fn::Join": [":", ["arn:aws:lambda:${self:provider.region}", { "Ref": "AWS::AccountId" }, "function", "${self:custom.scheduleFunction}" ] ]  }
-  environment:
-    STAGE: "${self:provider.stage}"
+    - Effect: Allow
+      Action:
+        - SQS:SendMessage
+      Resource: "${self:custom.queue.arn}"
 
 custom:
+  lumigo:
+    token: t_e74a5589b5524099bdc04
   inbound:
     name: "scheduler-input-${self:provider.stage}"
     arn: { "Fn::Join": [":", ["arn:aws:sns:${self:provider.region}", { "Ref": "AWS::AccountId" }, "${self:custom.inbound.name}" ] ]  }
@@ -60,20 +65,29 @@ functions:
           topicName: "${self:custom.inbound.name}"
     environment:
       SCHEDULE_FUNCTION: "${self:custom.scheduleFunction}"
-      ENFORCE_USER: false
+      ENFORCE_USER: true
+    tags:
+      name: 'aws-scheduler-consumer'
+
   eventLoader:
     handler: handler.event_loader
     events:
       - schedule: rate(1 minute)
     environment:
       SCHEDULE_FUNCTION: "${self:custom.scheduleFunction}"
       INDEX_NAME: "${self:custom.table.index.name}"
+    tags:
+      name: 'aws-scheduler-event-loader'
+
   scheduler:
     handler: handler.scheduler
     environment:
       QUEUE_URL: "${self:custom.queue.url}"
     # when we have to wait for DynamoDB autoscaling we may exceed the default of 6s
     timeout: 30
+    tags:
+      name: 'aws-scheduler-scheduler'
+
   emitter:
     handler: handler.emitter
     events:
@@ -87,6 +101,16 @@ functions:
                 - Ref: AWS::Region
                 - Ref: AWS::AccountId
                 - "${self:custom.queue.name}"
+    timeout: 20
+    tags:
+      name: 'aws-scheduler-emitter'
+  cleanupFailed:
+    handler: cleanup_failed.handle
+    timeout: 300
+    tags:
+      name: 'aws-scheduler-cleanup'
+    environment:
+      INDEX_NAME: "${self:custom.table.index.name}"
 
 plugins:
   - serverless-python-requirements

setup/init_table.py

@@ -1,4 +1,5 @@
 import sys
+import time
 
 import boto3
 
@@ -11,10 +12,13 @@ def events():
     stage = sys.argv[1]
 
     name = f'aws-scheduler-events-{stage}'
-    response = client.list_tables()
-    if name in response['TableNames']:
-        print('Table %s already exists. Please delete it first.' % name)
-        return
+    while True:
+        response = client.list_tables()
+        if name in response['TableNames']:
+            print('Table %s already exists. Please delete it first. Waiting 5 seconds until trying again...' % name)
+            time.sleep(5)
+        else:
+            break
     client.create_table(
         TableName=name,
         AttributeDefinitions=[