Serverless: Better support EventBridge Pipes and Lambda Event Source Mapping

[gregfurman]

This Lambda Event Source Mapping (ESM) issue raised a few months ago highlighted that we can much better support the Bento serverless offering. Namely, there are some common patterns and functionality that I think can be made much easier to implement and more performant (which when working with AWS Lambda is a huge consideration!).

It's probably best to make multiple individual issues out of each of these but having all of this in a central place is a good first step.

Improve Support for Lambda Trigger Types

• AWS provides the https://github.com/aws/aws-lambda-go library which has Go structs for (presumably) all possible event types a Go Lambda can ingest from AWS.
• This could be leveraged to parse incoming batches into Bento messages, allowing for a more common representation that could (1) be better suited to handling multiple message types and (2) abstract away the differences in formatting between events (i.e are ID fields called messageID or eventID)
• In addition, much of the data being passed around in these events could be set as metadata (i.e AWS region, event source ARNs, etc).

Reporting partial batch failures

• ESM and EventBridge Pipes have the concept of partial batch failures, where one can specify exactly which failed items of a batch processed by Lambda should be re-processed. The default case being that a single failure will cause the whole batch to repeat.
• While this is currently possible to implement in Bento, I think we can do better. For example, the solution below (credit to @ethanae) could be simplified with better internal handling of AWS triggers/events:

  pipeline:
  processors:
  - mapping: root = this.Records
  - unarchive:
      format: json_array
  - resource: foo_process_record
  - catch:
      - mapping: |
          root.error = "true"
          root.messageId = this.messageId
          root.error_msg = error()
  - archive:
      format: json_array
  - mapping: |
      root.batchItemFailures = this.filter(record -> record.error == "true").map_each(r -> {"itemIdentifier":r.messageId})

• Ideally, I think we could be relying on Bento message errors to signify failed items of a batch:
  • We can possibly add an option block to the serverless config where options.reportbatchitemfailures: true could ensure partial batches are handled.
  • Alternatively, a report_batch_item_failures processor could take in batches of messages and return the correct partial batch payload as a response for all errored messges:

    { 
    "batchItemFailures": [ 
    {
        "itemIdentifier": "<SequenceNumber/MessageID>"
    }
    ]
    }