Register multiple DynamoDB streams

[jtrollia]

Hello,

I did manage to add a DynamoDB stream (one single table) as a resource, following serverless and aws docs, however, what would be the correct syntax to implement two or more tables?

Thanks

[pmuens]

Hey @Malivuk we've just started to work on the stream event support. So Serverless will support DynamoDB and Kinesis streams out of the box pretty soon --> #2250.

It also supports multiple streams for one function (e.g. for two different tables).

Would be great if you could try it out and report back if it works for you.

[jtrollia]

Hi @pmuens, thanks for your support. I updated my local environment with your pull request but I have an issue while scanning tables. AWS returns {lambda} is not authorized to perform: dynamodb:Scan on resource {db stream}. I manually added Scan to the iamRoleStatements (even wildcard), without success.

Any idea what I could be doing wrong?

[pmuens]

Hmm. That's interesting. I used the permission definitions from this table:

http://docs.aws.amazon.com/lambda/latest/dg/lambda-api-permissions-ref.html

Not sure about the scan operation as it's not listed there...

[jtrollia]

Hi,

I guess the problem is coming from my settings anyways. I ended up redeploying the service using the following .yml:

provider:
  name: aws
  runtime: nodejs4.3
  stage: production
  region: ap-northeast-1
  iamRoleStatements:
  - Effect: "Allow"
    Resource: "*"
    Action:
      - "dynamodb:*"

functions:
  all:
    handler: handler.all
    events:
      - http: GET all/{class}
      - stream: arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*
      - stream: arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*
      - stream: arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*

It seems to override the default policy, which finally looks like:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:ap-northeast-1:*:*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:*"
            ],
            "Resource": "*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator",
                "dynamodb:DescribeStream",
                "dynamodb:ListStreams",
                "dynamodb:Scan"
            ],
            "Resource": "arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator",
                "dynamodb:DescribeStream",
                "dynamodb:ListStreams",
                "dynamodb:Scan"
            ],
            "Resource": "arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*",
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator",
                "dynamodb:DescribeStream",
                "dynamodb:ListStreams",
                "dynamodb:Scan"
            ],
            "Resource": "arn:aws:dynamodb:ap-northeast-1:*:table/*/stream/*",
            "Effect": "Allow"
        }
    ]
}

It's kind of ugly and I know I'm probably missing something out.

But well, it's working (demo). Each view is calling all function with a {class} parameter, which determines what table to scan.

[pmuens]

Awesome @Malivuk 💃

Great to hear that it's working now. Yes, you can always overwrite specific configuration that way. So it's completely fine how you've done it 👍

Let me know if you need anything else.

[pmuens]

This is resolved 🎉

[mdgbayly]

@Malivuk We're trying to setup Lambda to listen on a dynamodb stream using the wildcard syntax similar to what you are using above. Trying it this way because the stream is actually setup by an independent DynamoDB deployment and I'm trying to avoid hardcoding the stream label.

However, when I include a wildcard in the ARN it doesn't seem to work. Just wondering if you ran into similar issues. (x'd out our acct id)

Serverless: Deployment failed!
(node:26044) UnhandledPromiseRejectionWarning: Unhandled promise rejection (rejection id: 1): ServerlessError: An error occurred while provisioning your stack: ScheduleEventSourceMappingDynamodbDEVStreamObject - Stream not found: arn:aws:dynamodb:us-west-2:xxxxxxxxxx:table/DEV_StreamObject/stream/*.

from serverless.yml

functions:
  schedule:
    handler: handler.schedule
    events:
      - stream: arn:aws:dynamodb:us-west-2:xxxxxxxxxxxxx:table/DEV_StreamObject/stream/*

from cloudformation-template-update-stack.json

    "ScheduleEventSourceMappingDynamodbDEVStreamObject": {
      "Type": "AWS::Lambda::EventSourceMapping",
      "DependsOn": "IamRoleLambdaExecution",
      "Properties": {
        "BatchSize": 10,
        "EventSourceArn": "arn:aws:dynamodb:us-west-2:xxxxxxxxxxxxxx:table/DEV_StreamObject/stream/*",
        "FunctionName": {
          "Fn::GetAtt": [
            "ScheduleLambdaFunction",
            "Arn"
          ]
        },
        "StartingPosition": "TRIM_HORIZON",
        "Enabled": "True"
      }
    }

Thanks for any input!

[jtrollia]

Hi @mdgbayly if you refer to the (...)stream/* part, I just wrote that as a generic value, where * is actually my stream ID (e.g., 2017-03-06T08:33:11.683 ). I don't think you can replace your table stream ID by an asterisk, sorry for the confusion :(

[marckaraujo]

@Malivuk That's no way to get streamId dynamically? I have multiple stages, then I need to hardcode all streamIds...

[jtrollia]

@marckaraujo you might try another approach then, making full use of Serverless Variables.

You can declare some variables inside your .yml configuration file, via the CLI tool or even from another file. For instance:

service: your-service
provider: aws
custom:
  tableStageOne: arn:aws:dynamodb:your-region:000:table/tableOne/stream/000
  tableStageTwo: arn:aws:dynamodb:your-region:000:table/tableTwo/stream/000

Then, later in your file:

functions:
  foo:
    handler: handler.foo
    events:
      - stream: ${self:custom.tableStageTwo}

I don't know if that helps but it sure provides more flexibility.

[pmuens]

Thanks for jumping in @Malivuk 👍

Yes, Serverless Variables would be the go-to solution here! A very feature rich real-world application which uses them heavily is hello-retail by Nordstrom.

Here you could e.g. see their serverless.yml file which is focused on streams: https://github.com/Nordstrom/hello-retail/blob/94bcfca3add213a0f407bbc6acb8d33b4cde046f/retail-stream/serverless.yml

This file is then used in other places to reference the streams:
https://github.com/Nordstrom/hello-retail/blob/94bcfca3add213a0f407bbc6acb8d33b4cde046f/product-catalog/api/serverless.yml#L10

[marckaraujo]

@Malivuk and @pmuens , I just found an easier solution:

iamRoleStatements:
     - Effect: Allow
       Action:
          - dynamodb:DescribeTable
          - dynamodb:Query
          - dynamodb:Scan
          - dynamodb:GetItem
          - dynamodb:PutItem
          - dynamodb:UpdateItem
          - dynamodb:DeleteItem
          - dynamodb:DescribeStream
          - dynamodb:GetRecords
          - dynamodb:GetShardIterator
          - dynamodb:ListStreams

function
  streamRoom:
                  description: "Stream Room"
                  handler: api/stream/room/handler.default
                  memorySize: 256
                  timeout: 36
                  events:
                    - stream:
                        type: dynamodb
                        arn:
                          Fn::GetAtt:
                            - RoomDynamoDbTable
                            - StreamArn
                        batchSize: 1

resources:
  Resources:
    RoomDynamoDbTable:
      Type: 'AWS::DynamoDB::Table'
      DeletionPolicy: Retain
      Properties:
        AttributeDefinitions:
          -
            AttributeName: id
            AttributeType: S
          -
            AttributeName: status
            AttributeType: S
          -
            AttributeName: game
            AttributeType: S
        KeySchema:
          -
            AttributeName: id
            KeyType: HASH
        ProvisionedThroughput:
          ReadCapacityUnits: 5
          WriteCapacityUnits: 5
        StreamSpecification:
          StreamViewType: NEW_AND_OLD_IMAGES
        TableName: 'Room'

With this you dont need streamId 👍

[pmuens]

@Malivuk and @pmuens , I just found an easier solution:

Great. Even better @marckaraujo 👍 Thanks for sharing!