Merge pull request #1 from kazfuku/master

add documentdb and lambda triggers samples

Author: kazfuku

README.md

@@ -18,7 +18,8 @@ It contains:
 2. `mvn install` package lambda source files into jar and install to maven local repository
 3. `cd ../cdk`
 4. `mvn compile`     compile cdk source files
-5. `cdk deploy`      deploy resources via cdk
+5. `cdk bootstrap`   deploy bootstrap stack (If you deploy first at region)
+6. `cdk deploy`      deploy resources via cdk
 
 ## License
 

cdk/pom.xml

@@ -70,6 +70,11 @@
             <artifactId>lambda</artifactId>
             <version>${aws.cdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awscdk</groupId>
+            <artifactId>lambda-event-sources</artifactId>
+            <version>${aws.cdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>software.amazon.awscdk</groupId>
             <artifactId>s3</artifactId>
@@ -95,6 +100,26 @@
             <artifactId>secretsmanager</artifactId>
             <version>${aws.cdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awscdk</groupId>
+            <artifactId>docdb</artifactId>
+            <version>${aws.cdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awscdk</groupId>
+            <artifactId>kinesis</artifactId>
+            <version>${aws.cdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awscdk</groupId>
+            <artifactId>sqs</artifactId>
+            <version>${aws.cdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awscdk</groupId>
+            <artifactId>sns</artifactId>
+            <version>${aws.cdk.version}</version>
+        </dependency>
 
         <!-- https://mvnrepository.com/artifact/junit/junit -->
         <dependency>

cdk/src/main/java/com/amazon/aws/prototyping/CdkStack.java

@@ -1,7 +1,28 @@
 package com.amazon.aws.prototyping;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import com.amazon.aws.prototyping.apigateway.DocumentDbFunction;
+import com.amazon.aws.prototyping.apigateway.DynamoDBFunction;
+import com.amazon.aws.prototyping.apigateway.EC2Function;
+import com.amazon.aws.prototyping.apigateway.HttpFunction;
+import com.amazon.aws.prototyping.apigateway.JdbcFunction;
+import com.amazon.aws.prototyping.apigateway.JsonFunction;
+import com.amazon.aws.prototyping.apigateway.KinesisProduceFunction;
+import com.amazon.aws.prototyping.apigateway.S3Function;
+import com.amazon.aws.prototyping.apigateway.SnsFunction;
+import com.amazon.aws.prototyping.apigateway.SqsFunction;
+import com.amazon.aws.prototyping.eventsource.DynamoDBStreamFunction;
+import com.amazon.aws.prototyping.eventsource.KinesisConsumeFunction;
+import com.amazon.aws.prototyping.eventsource.SnsSubscribedFunction;
+import com.amazon.aws.prototyping.eventsource.SqsReceiveFunction;
 
 import software.amazon.awscdk.core.Construct;
 import software.amazon.awscdk.core.Duration;
@@ -12,9 +33,14 @@
 import software.amazon.awscdk.services.apigateway.Resource;
 import software.amazon.awscdk.services.apigateway.RestApi;
 import software.amazon.awscdk.services.apigateway.StageOptions;
+import software.amazon.awscdk.services.docdb.CfnDBCluster;
+import software.amazon.awscdk.services.docdb.CfnDBClusterParameterGroup;
+import software.amazon.awscdk.services.docdb.CfnDBInstance;
+import software.amazon.awscdk.services.docdb.CfnDBSubnetGroup;
 import software.amazon.awscdk.services.dynamodb.Attribute;
 import software.amazon.awscdk.services.dynamodb.AttributeType;
 import software.amazon.awscdk.services.dynamodb.GlobalSecondaryIndexProps;
+import software.amazon.awscdk.services.dynamodb.StreamViewType;
 import software.amazon.awscdk.services.dynamodb.Table;
 import software.amazon.awscdk.services.ec2.AmazonLinuxGeneration;
 import software.amazon.awscdk.services.ec2.AmazonLinuxImage;
@@ -31,14 +57,22 @@
 import software.amazon.awscdk.services.ec2.Vpc;
 import software.amazon.awscdk.services.iam.Effect;
 import software.amazon.awscdk.services.iam.PolicyStatement;
+import software.amazon.awscdk.services.kinesis.Stream;
 import software.amazon.awscdk.services.lambda.Code;
 import software.amazon.awscdk.services.lambda.Function;
 import software.amazon.awscdk.services.lambda.Runtime;
+import software.amazon.awscdk.services.lambda.StartingPosition;
+import software.amazon.awscdk.services.lambda.eventsources.DynamoEventSource;
+import software.amazon.awscdk.services.lambda.eventsources.KinesisEventSource;
+import software.amazon.awscdk.services.lambda.eventsources.SnsEventSource;
+import software.amazon.awscdk.services.lambda.eventsources.SqsEventSource;
 import software.amazon.awscdk.services.rds.DatabaseInstance;
 import software.amazon.awscdk.services.rds.DatabaseInstanceEngine;
 import software.amazon.awscdk.services.s3.Bucket;
 import software.amazon.awscdk.services.secretsmanager.Secret;
 import software.amazon.awscdk.services.secretsmanager.SecretStringGenerator;
+import software.amazon.awscdk.services.sns.Topic;
+import software.amazon.awscdk.services.sqs.Queue;
 
 public class CdkStack extends Stack {
     public CdkStack(final Construct scope, final String id) {
@@ -48,8 +82,8 @@ public CdkStack(final Construct scope, final String id) {
     public CdkStack(final Construct scope, final String id, final StackProps props) {
         super(scope, id, props);
 
-        StageOptions deployOptions = StageOptions.builder().loggingLevel(MethodLoggingLevel.INFO).dataTraceEnabled(true)
-                .build();
+        StageOptions deployOptions =
+                StageOptions.builder().loggingLevel(MethodLoggingLevel.INFO).dataTraceEnabled(true).build();
         RestApi api = RestApi.Builder.create(this, "JavaSamplesRestApi").deployOptions(deployOptions).build();
 
         List<SubnetConfiguration> subnetConfigurations = Arrays.asList(
@@ -63,6 +97,10 @@ public CdkStack(final Construct scope, final String id, final StackProps props)
         createEC2Sample(api, vpc);
         createJsonSample(api);
         createRdbSample(api, vpc);
+        createDocumentDbSample(api, vpc);
+        createKinesisSample(api);
+        createSqlSample(api);
+        createSnsSample(api);
     }
 
     private void createHttpSample(RestApi api) {
@@ -95,7 +133,7 @@ private void createDynamoDbSample(RestApi api) {
         Attribute attributeYear = Attribute.builder().name("year").type(AttributeType.NUMBER).build();
         Attribute attributeTitle = Attribute.builder().name("title").type(AttributeType.STRING).build();
         Table table = Table.Builder.create(this, "SampleTable").partitionKey(attributeYear).sortKey(attributeTitle)
-                .build();
+                .stream(StreamViewType.NEW_AND_OLD_IMAGES).build();
         GlobalSecondaryIndexProps index = GlobalSecondaryIndexProps.builder().indexName("title-year-index")
                 .partitionKey(attributeTitle).sortKey(attributeYear).build();
         table.addGlobalSecondaryIndex(index);
@@ -112,6 +150,11 @@ private void createDynamoDbSample(RestApi api) {
         Resource dynamodbResource = api.getRoot().addResource("dynamodb");
         dynamodbResource.addMethod("GET", LambdaIntegration.Builder.create(getFunction).proxy(true).build());
         dynamodbResource.addMethod("PUT", LambdaIntegration.Builder.create(putFunction).proxy(true).build());
+
+        Function streamFunction = createFunction(DynamoDBStreamFunction.class, "handle");
+        DynamoEventSource eventSource =
+                DynamoEventSource.Builder.create(table).startingPosition(StartingPosition.LATEST).build();
+        streamFunction.addEventSource(eventSource);
     }
 
     private void createEC2Sample(RestApi api, Vpc vpc) {
@@ -121,20 +164,20 @@ private void createEC2Sample(RestApi api, Vpc vpc) {
                 .instanceType(InstanceType.of(InstanceClass.BURSTABLE2, InstanceSize.MICRO)).vpc(vpc).build();
 
         PolicyStatement describeInstancesPolicy = PolicyStatement.Builder.create().effect(Effect.ALLOW)
-                .actions(Arrays.asList("ec2:describeInstances")).resources(Arrays.asList("*")).build();
+                .actions(Collections.singletonList("ec2:describeInstances")).resources(Arrays.asList("*")).build();
 
         Function startFunction = createFunction(EC2Function.class, "startAndWait", Duration.minutes(3), null);
         startFunction.addEnvironment("INSTANCE_ID", instance.getInstanceId());
         startFunction.addToRolePolicy(PolicyStatement.Builder.create().effect(Effect.ALLOW)
-                .actions(Arrays.asList("ec2:startInstances")).resources(Arrays.asList(String
+                .actions(Collections.singletonList("ec2:startInstances")).resources(Arrays.asList(String
                         .format("arn:aws:ec2:%s:%s:instance/%s", getRegion(), getAccount(), instance.getInstanceId())))
                 .build());
         startFunction.addToRolePolicy(describeInstancesPolicy);
 
         Function stopFunction = createFunction(EC2Function.class, "stopAndWait", Duration.minutes(3), null);
         stopFunction.addEnvironment("INSTANCE_ID", instance.getInstanceId());
         stopFunction.addToRolePolicy(PolicyStatement.Builder.create().effect(Effect.ALLOW)
-                .actions(Arrays.asList("ec2:stopInstances")).resources(Arrays.asList(String
+                .actions(Collections.singletonList("ec2:stopInstances")).resources(Arrays.asList(String
                         .format("arn:aws:ec2:%s:%s:instance/%s", getRegion(), getAccount(), instance.getInstanceId())))
                 .build());
         stopFunction.addToRolePolicy(describeInstancesPolicy);
@@ -158,11 +201,12 @@ private void createJsonSample(RestApi api) {
     }
 
     private void createRdbSample(RestApi api, Vpc vpc) {
-        Secret secret = Secret.Builder.create(this, "RdsSampleUserPassword")
-                .generateSecretString(
-                        SecretStringGenerator.builder().secretStringTemplate(String.format("{\"username\": \"test\"}"))
+        Secret secret =
+                Secret.Builder.create(this, "RdsSampleUserPassword")
+                        .generateSecretString(SecretStringGenerator.builder()
+                                .secretStringTemplate(String.format("{\"username\": \"test\"}"))
                                 .generateStringKey("password").excludePunctuation(true).build())
-                .build();
+                        .build();
 
         String databaseName = "sample";
         SecurityGroup rdsSecurityGroup = SecurityGroup.Builder.create(this, "SampleRdsSecurityGroup").vpc(vpc).build();
@@ -193,6 +237,108 @@ private void createRdbSample(RestApi api, Vpc vpc) {
                 LambdaIntegration.Builder.create(selectFunction).proxy(true).build());
     }
 
+    private void createDocumentDbSample(RestApi api, Vpc vpc) {
+        Secret secret = Secret.Builder.create(this, "DocumentDbSampleUserPassword")
+                .generateSecretString(SecretStringGenerator.builder().secretStringTemplate("{\"username\": \"test\"}")
+                        .generateStringKey("password").excludePunctuation(true).build())
+                .build();
+
+        SecurityGroup securityGroup =
+                SecurityGroup.Builder.create(this, "SampleDocumentDBSecurityGroup").vpc(vpc).build();
+        securityGroup.addIngressRule(Peer.ipv4(vpc.getVpcCidrBlock()), Port.tcp(27017));
+
+        Map<String, String> parameterMap = new HashMap<>();
+        parameterMap.put("tls", "disabled");
+        CfnDBClusterParameterGroup parameterGroup = CfnDBClusterParameterGroup.Builder
+                .create(this, "SampleDocumentDBParameterGroup").name("sample-document-db-parameter-group")
+                .description("DBParameterGroup for SampleDocumentDBCluster").family("docdb3.6").parameters(parameterMap)
+                .build();
+
+        List<String> subnetIds =
+                new ArrayList<>(CollectionUtils.collect(vpc.getPrivateSubnets(), (subnet) -> subnet.getSubnetId()));
+        // both db subnet group name and cluster name need to be lower case, because it
+        // is changed to lower case by cdk for some reason.
+        CfnDBSubnetGroup subnetGroup = CfnDBSubnetGroup.Builder.create(this, "SampleDocumentDBSubnetGroup")
+                .subnetIds(subnetIds).dbSubnetGroupName("sample-document-db-subnet-group")
+                .dbSubnetGroupDescription("DBSubnetGroup for SampleDocumentDBCluster").build();
+
+        CfnDBCluster cluster = CfnDBCluster.Builder.create(this, "SampleDocumentDBCluster")
+                .dbClusterIdentifier("sample-document-db-cluster").dbClusterParameterGroupName(parameterGroup.getName())
+                .dbSubnetGroupName(subnetGroup.getDbSubnetGroupName())
+                .vpcSecurityGroupIds(Collections.singletonList(securityGroup.getSecurityGroupId()))
+                .masterUsername(secret.secretValueFromJson("username").toString())
+                .masterUserPassword(secret.secretValueFromJson("password").toString()).build();
+        cluster.addDependsOn(subnetGroup);
+
+        CfnDBInstance instance = CfnDBInstance.Builder.create(this, "SampleDocumentDB")
+                .dbClusterIdentifier(cluster.getDbClusterIdentifier()).dbInstanceClass("db.r5.large").build();
+        instance.addDependsOn(cluster);
+
+        Function insertFunction = createFunction(DocumentDbFunction.class, "insert", Duration.seconds(30), vpc);
+        Function deleteFunction = createFunction(DocumentDbFunction.class, "delete", Duration.seconds(30), vpc);
+        Function findFunction = createFunction(DocumentDbFunction.class, "find", Duration.seconds(30), vpc);
+
+        for (Function function : Arrays.asList(insertFunction, deleteFunction, findFunction)) {
+            function.addEnvironment("CLUSTER_ENDPOINT", cluster.getAttrEndpoint());
+            function.addEnvironment("SECRET_ARN", secret.getSecretArn());
+            secret.grantRead(function);
+        }
+
+        Resource documentDbResource = api.getRoot().addResource("documentdb");
+        documentDbResource.addResource("insert").addMethod("POST",
+                LambdaIntegration.Builder.create(insertFunction).proxy(true).build());
+        documentDbResource.addResource("delete").addMethod("POST",
+                LambdaIntegration.Builder.create(deleteFunction).proxy(true).build());
+        documentDbResource.addResource("find").addMethod("GET",
+                LambdaIntegration.Builder.create(findFunction).proxy(true).build());
+    }
+
+    private void createKinesisSample(RestApi api) {
+        Stream stream = Stream.Builder.create(this, "SampleKinesisStream").build();
+
+        Function putRecordFunction = createFunction(KinesisProduceFunction.class, "putRecord");
+        putRecordFunction.addEnvironment("KINESIS_STREAM_NAME", stream.getStreamName());
+        stream.grantWrite(putRecordFunction);
+
+        api.getRoot().addResource("kinesis").addMethod("PUT",
+                LambdaIntegration.Builder.create(putRecordFunction).proxy(true).build());
+
+        Function function = createFunction(KinesisConsumeFunction.class, "handle");
+
+        KinesisEventSource eventSource =
+                KinesisEventSource.Builder.create(stream).startingPosition(StartingPosition.TRIM_HORIZON).build();
+        function.addEventSource(eventSource);
+    }
+
+    private void createSqlSample(RestApi api) {
+        Queue queue = Queue.Builder.create(this, "SampleSqsQueue").build();
+
+        Function sendFunction = createFunction(SqsFunction.class, "send");
+        queue.grantSendMessages(sendFunction);
+        sendFunction.addEnvironment("QUEUE_URL", queue.getQueueUrl());
+
+        api.getRoot().addResource("sqs").addMethod("PUT",
+                LambdaIntegration.Builder.create(sendFunction).proxy(true).build());
+
+        Function receiveFunction = createFunction(SqsReceiveFunction.class, "handle");
+        SqsEventSource eventSource = SqsEventSource.Builder.create(queue).build();
+        receiveFunction.addEventSource(eventSource);
+    }
+
+    private void createSnsSample(RestApi api) {
+        Topic topic = Topic.Builder.create(this, "SampleTopic").build();
+
+        Function publishFunction = createFunction(SnsFunction.class, "publish");
+        topic.grantPublish(publishFunction);
+        publishFunction.addEnvironment("TOPIC_ARN", topic.getTopicArn());
+
+        api.getRoot().addResource("sns").addMethod("POST",
+                LambdaIntegration.Builder.create(publishFunction).proxy(true).build());
+
+        Function subscribedFunction = createFunction(SnsSubscribedFunction.class, "handle");
+        subscribedFunction.addEventSource(new SnsEventSource(topic));
+    }
+
     private Function createFunction(Class<?> functionClass, String handler) {
         return createFunction(functionClass, handler, Duration.seconds(10), null);
     }

lambda/.gitignore

@@ -0,0 +1,4 @@
+target
+.settings
+.classpath
+.project

lambda/pom.xml

@@ -35,6 +35,7 @@
     </build>
     <properties>
         <aws.sdk.version>1.11.695</aws.sdk.version>
+        <jackson.version>2.10.1</jackson.version>
     </properties>
     <dependencies>
         <dependency>
@@ -72,21 +73,47 @@
             <artifactId>aws-java-sdk-dynamodb</artifactId>
             <version>${aws.sdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-kinesis</artifactId>
+            <version>${aws.sdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sqs</artifactId>
+            <version>${aws.sdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sns</artifactId>
+            <version>${aws.sdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
             <version>8.0.18</version>
         </dependency>
+        <!-- use same version in jackson core and dataformat-cbor for using kinesis sdk -->
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.10.1</version>
+            <version>${jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-cbor</artifactId>
+            <version>${jackson.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpclient</artifactId>
             <version>4.5.10</version>
         </dependency>
+        <dependency>
+            <groupId>org.mongodb</groupId>
+            <artifactId>mongodb-driver</artifactId>
+            <version>3.12.3</version>
+        </dependency>
         <dependency>
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>

lambda/src/main/java/com/amazon/aws/prototyping/apigateway/AbstractFunction.java

@@ -0,0 +1,12 @@
+package com.amazon.aws.prototyping.apigateway;
+
+import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent;
+
+abstract class AbstractFunction {
+    protected APIGatewayProxyResponseEvent ok() {
+        APIGatewayProxyResponseEvent response = new APIGatewayProxyResponseEvent();
+        response.setStatusCode(Integer.valueOf(200));
+        response.setBody("ok");
+        return response;
+    }
+}