Support natively processing s3 events with lambda deployments

[mallikvala]

Hey @Jeffail , this can probably be an enhancement but any suggestions on how to trigger an s3 input in lambda in a scenario where a lambda is triggered during a putObject operation on an s3 bucket.

currently, when invoked a single part is created with the event and processor chain is invoked without any input. the event as such wont make much sense until the file is downloaded. looking for suggestions.

[Jeffail]

Hey @mallikvala, if I understand your goals correctly you can trigger a download from the event by using a cache processor with an S3 cache resource. It would look something like this:

pipeline:
  processors:
  - cache:
      cache: bucket
      operator: get
      key: ${! json("path.to.s3.item") }
      
resources:
  caches:
    bucket:
      s3:
        bucket: foo

My mind's a little fuzzy on what the event looks like that gets sent to the lambda function but if you can log it and post it here I can update this example.

[Jeffail]

Labelling this as documentation as it would be cool to have some examples like this in the lambda section of the docs.

[mallikvala]

@Jeffail wont using cache.get make the whole file a single message part rather than each line like how it happens when we use an input.file processor?

[Jeffail]

Hey @mallikvala, if you want to process the S3 files as line delimited messages then you can use the unarchive processor in order to cut the file into a batch, and then optionally follow it with a split processor if you want to dispatch the batch as individual messages:

pipeline:
  processors:
  - cache:
      cache: bucket
      operator: get
      key: ${! json("path.to.s3.item") }
  - unarchive:
      format: lines
  - split: {}
      
resources:
  caches:
    bucket:
      s3:
        bucket: foo

Another thing to keep in mind is that the cache processor might fail (if the key is not found, etc) and the message will continue through the pipeline, you can choose from a range of options for handling the errors: https://www.benthos.dev/docs/configuration/error_handling

[Jeffail]

I've repurposed this issue as an enhancement because using caches is a pretty weak user experience here and since this is a somewhat common pattern with AWS there ought to be a better way to do this, ideally with minimal effort on the configuration side. However, this still needs to be explicit in order to preserve backwards compatibility.