Better error handling (#3)

CloudWatch2S3.template

@@ -307,6 +307,8 @@ Resources:
                     - ParameterName: BufferIntervalInSeconds
                       ParameterValue:
                         Ref: ProcessorBufferIntervalHint
+                    - ParameterName: NumberOfRetries
+                      ParameterValue: '3'
                   Type: Lambda
             - Ref: AWS::NoValue
         RoleARN:
@@ -512,15 +514,27 @@ Resources:
                 result = []
                 size = 0
 
-                for record in handle_records(event["records"]):
-                    size += len(str(record))
-                    if size < 6000000:
-                        # lambda limits output to 6mb
-                        # kinesis will treat records not here as failed and retry
-                        # TODO or do we need to reingest?
-                        result.append(record)
+                for i, record in enumerate(handle_records(event["records"])):
+                    record_size = len(str(record))
+                    if record_size >= 6000000:
+                        print("Dropping single record that's over 6MB")
+                        result.append({
+                            "result": "Dropped",
+                            "recordId": record["recordId"]
+                        })
                     else:
-                        break
+                        size += record_size
+                        if size < 6000000:
+                            # lambda limits output to 6mb
+                            # kinesis will treat records not here as failed and retry
+                            # TODO or do we need to reingest?
+                            result.append(record)
+                        else:
+                            print("Failing record as output is over 6MB")
+                            result.append({
+                                "result": "ProcessingFailed",
+                                "recordId": record["recordId"]
+                            })
 
                 return {"records": result}
       Handler: index.handler

README.md

@@ -42,6 +42,10 @@ supply the value you copied as the `LogDestination` parameter.
 There are a lot of parameters to play with, but the defaults should be good enough for most. If you have a lot of log
 records coming in (more than 1000/s or 1MB/s), you might want to increase Kinesis shard count.
 
+### Known Limitations
+
+Single CloudWatch records can't be over 6MB when using anything else but raw log format. Kinesis uses Lambda to convert data and Lambda output is limited to 6MB. Note that data comes in compressed from CloudWatch but has to come out decompressed from Lambda. So the decompressed record can't be over 6MB. You will see record failures in CloudWatch metrics for the Kinesis stream for this and errors in the log for the processor Lambda function.
+
 ### Troubleshooting
 
 * Make sure the right CloudWatch log groups are subscribed