Merge branch 'develop' into feat/idempotency_helper

Author: Tom McCarthy

CHANGELOG.md

@@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.10.2] - 2021-02-04
+
+### Fixed
+
+* **Utilities**: Correctly handle and list multiple exceptions in SQS batch processing utility.
+* **Docs*:: Fix typos on AppConfig docstring import, and `SnsModel` typo in parser.
+* **Utilities**: `typing_extensions` package is now only installed in Python < 3.8
+
 ## [1.10.1] - 2021-01-19
 
 ### Fixed

aws_lambda_powertools/tracing/tracer.py

@@ -14,6 +14,10 @@
 
 is_cold_start = True
 logger = logging.getLogger(__name__)
+# Set the streaming threshold to 0 on the default recorder to force sending
+# subsegments individually, rather than batching them.
+# See https://github.com/awslabs/aws-lambda-powertools-python/issues/283
+aws_xray_sdk.core.xray_recorder.configure(streaming_threshold=0)
 
 
 class Tracer:

aws_lambda_powertools/utilities/batch/base.py

@@ -85,7 +85,7 @@ def success_handler(self, record: Any, result: Any):
         self.success_messages.append(record)
         return entry
 
-    def failure_handler(self, record: Any, exception: Exception):
+    def failure_handler(self, record: Any, exception: Tuple):
         """
         Failure callback
 
@@ -94,8 +94,9 @@ def failure_handler(self, record: Any, exception: Exception):
         tuple
             "fail", exceptions args, original record
         """
-        entry = ("fail", exception.args, record)
-        logger.debug(f"Record processing exception: {exception}")
+        exception_string = f"{exception[0]}:{exception[1]}"
+        entry = ("fail", exception_string, record)
+        logger.debug(f"Record processing exception: {exception_string}")
         self.exceptions.append(exception)
         self.fail_messages.append(record)
         return entry

aws_lambda_powertools/utilities/batch/exceptions.py

@@ -1,7 +1,25 @@
 """
 Batch processing exceptions
 """
+import traceback
 
 
 class SQSBatchProcessingError(Exception):
     """When at least one message within a batch could not be processed"""
+
+    def __init__(self, msg="", child_exceptions=()):
+        super().__init__(msg)
+        self.msg = msg
+        self.child_exceptions = child_exceptions
+
+    # Overriding this method so we can output all child exception tracebacks when we raise this exception to prevent
+    # errors being lost. See https://github.com/awslabs/aws-lambda-powertools-python/issues/275
+    def __str__(self):
+        parent_exception_str = super(SQSBatchProcessingError, self).__str__()
+        exception_list = [f"{parent_exception_str}\n"]
+        for exception in self.child_exceptions:
+            extype, ex, tb = exception
+            formatted = "".join(traceback.format_exception(extype, ex, tb))
+            exception_list.append(formatted)
+
+        return "\n".join(exception_list)

aws_lambda_powertools/utilities/batch/sqs.py

@@ -4,6 +4,7 @@
 Batch SQS utilities
 """
 import logging
+import sys
 from typing import Callable, Dict, List, Optional, Tuple
 
 import boto3
@@ -90,10 +91,10 @@ def _process_record(self, record) -> Tuple:
             An object to be processed.
         """
         try:
-            result = self.handler(record)
-            return self.success_handler(record, result)
-        except Exception as exc:
-            return self.failure_handler(record, exc)
+            result = self.handler(record=record)
+            return self.success_handler(record=record, result=result)
+        except Exception:
+            return self.failure_handler(record=record, exception=sys.exc_info())
 
     def _prepare(self):
         """
@@ -123,7 +124,11 @@ def _clean(self):
             logger.debug(f"{len(self.fail_messages)} records failed processing, but exceptions are suppressed")
         else:
             logger.debug(f"{len(self.fail_messages)} records failed processing, raising exception")
-            raise SQSBatchProcessingError(list(self.exceptions))
+            raise SQSBatchProcessingError(
+                msg=f"Not all records processed succesfully. {len(self.exceptions)} individual errors logged "
+                f"separately below.",
+                child_exceptions=self.exceptions,
+            )
 
         return delete_message_response
 

aws_lambda_powertools/utilities/parser/models/dynamodb.py

@@ -2,7 +2,8 @@
 from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel
-from typing_extensions import Literal
+
+from ..types import Literal
 
 
 class DynamoDBStreamChangedRecordModel(BaseModel):

aws_lambda_powertools/utilities/parser/models/kinesis.py

@@ -5,7 +5,8 @@
 
 from pydantic import BaseModel, validator
 from pydantic.types import PositiveInt
-from typing_extensions import Literal
+
+from ..types import Literal
 
 logger = logging.getLogger(__name__)
 

aws_lambda_powertools/utilities/parser/models/s3.py

@@ -5,7 +5,8 @@
 from pydantic.fields import Field
 from pydantic.networks import IPvAnyNetwork
 from pydantic.types import PositiveInt
-from typing_extensions import Literal
+
+from ..types import Literal
 
 
 class S3EventRecordGlacierRestoreEventData(BaseModel):

aws_lambda_powertools/utilities/parser/models/ses.py

@@ -4,7 +4,8 @@
 from pydantic import BaseModel, Field
 from pydantic.networks import EmailStr
 from pydantic.types import PositiveInt
-from typing_extensions import Literal
+
+from ..types import Literal
 
 
 class SesReceiptVerdict(BaseModel):

aws_lambda_powertools/utilities/parser/models/sns.py

@@ -3,7 +3,8 @@
 
 from pydantic import BaseModel, root_validator
 from pydantic.networks import HttpUrl
-from typing_extensions import Literal
+
+from ..types import Literal
 
 
 class SnsMsgAttributeModel(BaseModel):

aws_lambda_powertools/utilities/parser/models/sqs.py

@@ -2,7 +2,8 @@
 from typing import Dict, List, Optional
 
 from pydantic import BaseModel
-from typing_extensions import Literal
+
+from ..types import Literal
 
 
 class SqsAttributesModel(BaseModel):

aws_lambda_powertools/utilities/parser/types.py

@@ -1,6 +1,13 @@
 """Generics and other shared types used across parser"""
+
 from typing import TypeVar
 
 from pydantic import BaseModel
 
+# We only need typing_extensions for python versions <3.8
+try:
+    from typing import Literal  # noqa: F401
+except ImportError:
+    from typing_extensions import Literal  # noqa: F401
+
 Model = TypeVar("Model", bound=BaseModel)

benchmark/.gitignore

@@ -0,0 +1 @@
+.aws-sam

benchmark/README.md

@@ -0,0 +1,26 @@
+# Cold Start Benchmark
+
+The [benchmark.sh script](./benchmark.sh) is a bash script to compare the cold-start time of using the AWS Lambda Powertools in a semi-automated way. It does so by deploying two Lambda functions which both have the aws-lambda-powertools module installed. One Lambda function will import and initialize the three core utilities (`Metrics`, `Logger`, `Tracer`), while the other one will not.
+
+Please note that this requires the [SAM CLI](https://github.com/aws/aws-sam-cli) version 1.2.0 or later.
+
+## Usage
+
+> **NOTE**: This script is expected to run in Unix-based systems only, and can incur charges on your AWS account.
+
+To use the script, you should move into the benchmark folder and run the benchmark script:
+
+```
+export S3_BUCKET=code-artifact-s3-bucket
+
+cd benchmark
+./benchmark.sh
+```
+
+This will:
+
+* Deploy a CloudFormation stack using guided SAM deployment (*you will need to answer a few questions*).
+* Run loops to update the memory setting of the functions to force a cold start, then invoke them. This process is repeated a number of time to get more consistent results.
+* Wait 2.5 minutes to ensure data propagates from CloudWatch Logs to CloudWatch Logs Insights.
+* Run a query on CloudWatch Logs insights, looking at the **REPORT** line from the logs.
+* Delete the CloudFormation stack.

benchmark/benchmark.sh

@@ -0,0 +1,83 @@
+#!/bin/bash
+
+set -e
+trap cleanup EXIT
+
+if [ -z "S3_BUCKET" ]; then
+    echo "Missing S3_BUCKET environment variabe"
+    exit 1
+fi
+
+export BENCHMARK_STACK_NAME=${BENCHMARK_STACK_NAME:-"powertools-benchmark"}
+
+function cleanup {
+    echo "Cleaning up stack..."
+    aws cloudformation delete-stack --stack-name $BENCHMARK_STACK_NAME
+}
+
+function run_function {
+    # Update function to force a cold start
+    aws lambda update-function-configuration --function-name $1 --memory-size 256 >/dev/null
+    aws lambda update-function-configuration --function-name $1 --memory-size 128 >/dev/null
+    # Cold-start invoke
+    aws lambda invoke --function-name $1 --payload '{}' /dev/null >/dev/null && echo -n . || echo -n e
+}
+
+# Retrieve statistics
+function get_stats {
+    # Gather results from CloudWatch Logs Insights
+    query_id=$(aws logs start-query --log-group-name $1 --query-string 'filter @type = "REPORT" | stats pct(@initDuration, 50) as init_duration, pct(@duration, 50) as duration' --start-time $(expr $(date +%s) - 86400) --end-time $(expr $(date +%s) + 0) --query 'queryId' --output text)
+    while true; do
+        result=$(aws logs get-query-results --query-id $query_id --query 'status' --output text)
+        if [ $result == "Complete" ]; then
+            break
+        fi
+        sleep 1
+    done
+
+    # Check if greater than threshold and print result
+    init_duration=$(aws logs get-query-results --query-id $query_id --query 'results[0][?field==`init_duration`].value' --output text)
+    duration=$(aws logs get-query-results --query-id $query_id --query 'results[0][?field==`duration`].value' --output text)
+    echo "$init_duration,$duration"
+}
+
+# Build and deploy the benchmark stack
+echo "Building and deploying..."
+sam build
+sam deploy --stack-name $BENCHMARK_STACK_NAME --s3-bucket $S3_BUCKET --capabilities CAPABILITY_IAM
+
+# Retrieve output values
+echo "Retrieve values..."
+export INSTRUMENTED_FUNCTION=$(aws cloudformation describe-stacks --stack-name $BENCHMARK_STACK_NAME --query 'Stacks[0].Outputs[?OutputKey==`InstrumentedFunction`].OutputValue' --output text)
+export REFERENCE_FUNCTION=$(aws cloudformation describe-stacks --stack-name $BENCHMARK_STACK_NAME --query 'Stacks[0].Outputs[?OutputKey==`ReferenceFunction`].OutputValue' --output text)
+export INSTRUMENTED_LOG_GROUP=$(aws cloudformation describe-stacks --stack-name $BENCHMARK_STACK_NAME --query 'Stacks[0].Outputs[?OutputKey==`InstrumentedLogGroup`].OutputValue' --output text)
+export REFERENCE_LOG_GROUP=$(aws cloudformation describe-stacks --stack-name $BENCHMARK_STACK_NAME --query 'Stacks[0].Outputs[?OutputKey==`ReferenceLogGroup`].OutputValue' --output text)
+
+echo INSTRUMENTED_FUNCTION=$INSTRUMENTED_FUNCTION
+echo REFERENCE_FUNCTION=$REFERENCE_FUNCTION
+echo INSTRUMENTED_LOG_GROUP=$INSTRUMENTED_LOG_GROUP
+echo REFERENCE_LOG_GROUP=$REFERENCE_LOG_GROUP
+
+# Running cold starts
+echo "Running functions..."
+for i in {0..20}; do
+    run_function $INSTRUMENTED_FUNCTION
+done &
+process_id=$!
+for i in {0..20}; do
+    run_function $REFERENCE_FUNCTION
+done &
+wait $process_id
+wait $!
+echo
+
+# Gather statistics
+# Waiting 2.5 minutes to make sure the data propagates from CloudWatch Logs
+# into CloudWatch Logs Insights.
+echo "Waiting for data to propagate in CloudWatch Logs Insights..."
+sleep 150
+return_code=0
+echo "INSTRUMENTED=$(get_stats $INSTRUMENTED_LOG_GROUP)"
+echo "REFERENCE=$(get_stats $REFERENCE_LOG_GROUP)"
+
+exit $return_code

benchmark/src/instrumented/main.py

@@ -0,0 +1,17 @@
+from aws_lambda_powertools import (Logger, Metrics, Tracer)
+
+
+# Initialize core utilities
+logger = Logger()
+metrics = Metrics()
+tracer = Tracer()
+
+
+# Instrument Lambda function
+@logger.inject_lambda_context
+@metrics.log_metrics
+@tracer.capture_lambda_handler
+def handler(event, context):
+    return {
+        "message": "success"
+    }

benchmark/src/instrumented/requirements.txt

@@ -0,0 +1 @@
+aws-lambda-powertools

benchmark/src/reference/main.py

@@ -0,0 +1,4 @@
+def handler(event, context):
+    return {
+        "message": "success"
+    }

benchmark/src/reference/requirements.txt

@@ -0,0 +1 @@
+aws-lambda-powertools

benchmark/template.yaml

@@ -0,0 +1,48 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+
+Globals:
+  Function:
+    Handler: main.handler
+    Runtime: python3.8
+    MemorySize: 128
+    Tracing: Active
+    Environment:
+      Variables:
+        POWERTOOLS_SERVICE_NAME: benchmark
+        POWERTOOLS_METRICS_NAMESPACE: LambdaPowertools
+        POWERTOOLS_LOGGER_LOG_EVENT: "true"
+        LOG_LEVEL: INFO
+
+Resources:
+  InstrumentedFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ./src/instrumented/
+
+  ReferenceFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ./src/reference/
+
+  InstrumentedLogGroup:
+    Type: AWS::Logs::LogGroup
+    Properties:
+      LogGroupName: !Sub "/aws/lambda/${InstrumentedFunction}"
+      RetentionInDays: 7
+
+  ReferenceLogGroup:
+    Type: AWS::Logs::LogGroup
+    Properties:
+      LogGroupName: !Sub "/aws/lambda/${ReferenceFunction}"
+      RetentionInDays: 7
+
+Outputs:
+  InstrumentedFunction:
+    Value: !Ref InstrumentedFunction
+  ReferenceFunction:
+    Value: !Ref ReferenceFunction
+  InstrumentedLogGroup:
+    Value: !Ref InstrumentedLogGroup
+  ReferenceLogGroup:
+    Value: !Ref ReferenceLogGroup