Added pipelines based on s3 source and functions

anandshah123 · anandshah123 · commit 7f37543851a3 · 2022-09-08T17:50:08.000+02:00
diff --git a/infrastructure-cdk/data-source-function/app.py b/infrastructure-cdk/data-source-function/app.py
@@ -0,0 +1,7 @@
+import boto3
+import os
+
+kinesis = boto3.client("kinesis")
+
+def lambda_handler(event, context):
+    kinesis.put_record(StreamName=os.environ["STREAM_NAME"], Data=b'This is test message from the Lambda function', PartitionKey="default")
diff --git a/infrastructure-cdk/flink-app-redeploy-hook/app.py b/infrastructure-cdk/flink-app-redeploy-hook/app.py
@@ -0,0 +1,76 @@
+import json
+import os
+import traceback
+
+import boto3
+
+asset_bucket_arn = os.environ["ASSET_BUCKET_ARN"]
+kinesis_apps = boto3.client("kinesisanalyticsv2")
+code_pipeline = boto3.client('codepipeline')
+
+
+def put_job_success(job, message):
+    """Notify CodePipeline of a successful job
+
+    Args:
+        job: The CodePipeline job ID
+        message: A message to be logged relating to the job status
+
+    Raises:
+        Exception: Any exception thrown by .put_job_success_result()
+
+    """
+    print('Putting job success')
+    print(message)
+    code_pipeline.put_job_success_result(jobId=job)
+
+
+def put_job_failure(job, message):
+    """Notify CodePipeline of a failed job
+
+    Args:
+        job: The CodePipeline job ID
+        message: A message to be logged relating to the job status
+
+    Raises:
+        Exception: Any exception thrown by .put_job_failure_result()
+
+    """
+    print('Putting job failure')
+    print(message)
+    code_pipeline.put_job_failure_result(jobId=job, failureDetails={'message': message, 'type': 'JobFailed'})
+
+
+def lambda_handler(event, context):
+    print(json.dumps(event))
+    # Extract the Job ID
+    job_id = event['CodePipeline.job']['id']
+
+    if "APP_NAME" in event:
+        app_name = event["APP_NAME"]
+    else:
+        app_name = os.environ["APP_NAME"]
+
+    if "FILE_KEY" in event:
+        file_key = event["FILE_KEY"]
+    else:
+        file_key = os.environ["FILE_KEY"]
+
+    try:
+        token = kinesis_apps.describe_application(ApplicationName=app_name)["ApplicationDetail"]["ConditionalToken"]
+        kinesis_apps.update_application(ApplicationName=app_name,
+                                        ApplicationConfigurationUpdate={
+                                            "ApplicationCodeConfigurationUpdate": {
+                                                "CodeContentUpdate": {
+                                                    "S3ContentLocationUpdate": {"BucketARNUpdate": asset_bucket_arn,
+                                                                                "FileKeyUpdate": file_key}}}
+
+                                        }, ConditionalToken=token)
+        put_job_success(job_id, "Success")
+    except Exception as e:
+        traceback.print_exc()
+        put_job_failure(job_id, 'Function exception: ' + str(e))
+
+
+if __name__ == '__main__':
+    lambda_handler({}, {})
diff --git a/infrastructure-cdk/lib/application-pipeline-stack.ts b/infrastructure-cdk/lib/application-pipeline-stack.ts
@@ -1,32 +1,37 @@
-import {Aws, CfnOutput, Stack, StackProps} from 'aws-cdk-lib';
+import {Aws, CfnOutput, RemovalPolicy, Stack, StackProps} from 'aws-cdk-lib';
 import {Construct} from 'constructs';
 import {JavaBuildPipeline} from "./constructs/java-build-pipeline";
 import {BlockPublicAccess, Bucket, BucketEncryption} from "aws-cdk-lib/aws-s3";
-import {ASSET_BUCKET_EXPORT_NAME, REPOSITORY_FILE_PATH} from "./shared-vars";
+import {APPLICATION_NAME, ASSET_BUCKET_EXPORT_NAME} from "./shared-vars";
 
 export class ApplicationPipelineStack extends Stack {
     constructor(scope: Construct, id: string, props?: StackProps) {
         super(scope, id, props);
 
-        const artefactBucket = new Bucket(this, 'ArtefactBucket', {
+        const artifactBucket = new Bucket(this, 'ArtefactBucket', {
             blockPublicAccess: BlockPublicAccess.BLOCK_ALL,
             enforceSSL: true,
-            encryption: BucketEncryption.S3_MANAGED
+            encryption: BucketEncryption.S3_MANAGED,
+            versioned: true,
+            autoDeleteObjects: true,
+            removalPolicy: RemovalPolicy.DESTROY
         });
 
         new JavaBuildPipeline(this, 'java-app', {
-            deployBucket: artefactBucket, repositoryName: REPOSITORY_FILE_PATH
+            appName: APPLICATION_NAME,
+            deployBucket: artifactBucket,
+            repositoryName: APPLICATION_NAME
         });
 
-        new CfnOutput(this, 'AssetBucketName', {
-            value: artefactBucket.bucketName,
-            description: "Artefact Bucket name storing application binaries",
+        new CfnOutput(this, 'ArtifactBucketName', {
+            value: artifactBucket.bucketName,
+            description: "Artifact Bucket name storing application binaries",
             exportName: ASSET_BUCKET_EXPORT_NAME
         });
 
-        new CfnOutput(this, 'AssetBucketLink', {
-            value: "https://s3.console.aws.amazon.com/s3/buckets/" + artefactBucket.bucketName + "?region=" + Aws.REGION + "&tab=objects",
-            description: "Artefact Bucket Link"
+        new CfnOutput(this, 'ArtifactBucketLink', {
+            value: "https://s3.console.aws.amazon.com/s3/buckets/" + artifactBucket.bucketName + "?region=" + Aws.REGION + "&tab=objects",
+            description: "Artifact Bucket Link"
         });
 
     }
diff --git a/infrastructure-cdk/lib/application-stack.ts b/infrastructure-cdk/lib/application-stack.ts
@@ -0,0 +1,49 @@
+import {Aws, Fn, Stack, StackProps} from 'aws-cdk-lib';
+import {Construct} from 'constructs';
+import {Bucket} from "aws-cdk-lib/aws-s3";
+import {APPLICATION_NAME, ASSET_BUCKET_EXPORT_NAME} from "./shared-vars";
+import {Stream, StreamEncryption, StreamMode} from "aws-cdk-lib/aws-kinesis";
+import {Code, Function, Runtime} from "aws-cdk-lib/aws-lambda";
+import * as kda from "@aws-cdk/aws-kinesisanalytics-flink-alpha";
+
+export class ApplicationStack extends Stack {
+    constructor(scope: Construct, id: string, props?: StackProps) {
+        super(scope, id, props);
+        const assetBucket = Bucket.fromBucketName(this, 'imported-asset-bucket', Fn.importValue(ASSET_BUCKET_EXPORT_NAME));
+
+        const stream = new Stream(this, 'raw', {
+            streamMode: StreamMode.PROVISIONED,
+            shardCount: 1,
+            encryption: StreamEncryption.MANAGED
+        });
+
+        const dataSourceFn = new Function(this, 'data-source', {
+            code: Code.fromAsset("data-source-function"),
+            handler: "app.lambda_handler",
+            runtime: Runtime.PYTHON_3_9,
+            environment: {
+                STREAM_NAME: stream.streamName
+            }
+        });
+
+        stream.grantWrite(dataSourceFn);
+
+        const application = new kda.Application(this, 'app', {
+                code: kda.ApplicationCode.fromBucket(assetBucket, "jars/" + APPLICATION_NAME + "-latest.jar"),
+                runtime: kda.Runtime.FLINK_1_13,
+                propertyGroups: {
+                    "KinesisReader": {
+                        "input.stream.name": stream.streamName,
+                        "aws.region": Aws.REGION,
+                        "flink.stream.initpos": "LATEST"
+                    },
+                },
+                snapshotsEnabled: false,
+                parallelismPerKpu: 1
+            }
+        );
+
+        stream.grantRead(application);
+
+    }
+}
diff --git a/infrastructure-cdk/lib/constructs/java-build-pipeline.ts b/infrastructure-cdk/lib/constructs/java-build-pipeline.ts
@@ -6,17 +6,22 @@ import {
     CodeBuildAction,
     LambdaInvokeAction,
     ManualApprovalAction,
-    S3DeployAction
+    S3DeployAction,
+    S3SourceAction
 } from "aws-cdk-lib/aws-codepipeline-actions";
-import {IFunction} from "aws-cdk-lib/aws-lambda/lib/function-base";
+import * as path from "path";
+import {Code, Function, Runtime} from "aws-cdk-lib/aws-lambda";
+import {Aws, Duration} from "aws-cdk-lib";
+import {PolicyStatement} from "aws-cdk-lib/aws-iam";
+import {SOURCE_CODE_ZIP} from "../shared-vars";
 
 
 interface JavaBuildPipelineProps {
+    appName: string
     repositoryName: string
     deployBucket: IBucket
     projectRoot?: string
     deployBucketBasePath?: string
-    postActionLambda?: IFunction
 }
 
 export class JavaBuildPipeline extends Construct {
@@ -43,8 +48,6 @@ export class JavaBuildPipeline extends Construct {
                 },
                 build: {
                     commands: [
-                        `curl ${props.repositoryName} --output app.zip`, // Download zip directly from Github
-                        'unzip app.zip',
                         `cd ${directory}`,
                         'mvn clean package -B',
                         `mkdir -p ${s3BasePath}`,
@@ -83,9 +86,17 @@ export class JavaBuildPipeline extends Construct {
                 //         repo: props.repositoryName
                 //     })]
                 // },
+                {
+                    stageName: "source", actions: [new S3SourceAction({
+                        output: sourceAsset,
+                        actionName: "Checkout",
+                        bucket: props.deployBucket,
+                        bucketKey: SOURCE_CODE_ZIP
+                    })]
+                },
                 {
                     stageName: "build", actions: [new CodeBuildAction({
-                        actionName: "CodeBuild", input: sourceAsset, project: project, outputs: [buildOutput]
+                        input: sourceAsset, actionName: "CodeBuild", project: project, outputs: [buildOutput]
                     })]
                 }, {
                     stageName: "saveArtefact", actions: [new S3DeployAction({
@@ -102,13 +113,33 @@ export class JavaBuildPipeline extends Construct {
                 }]
         });
 
-        if (props.postActionLambda) {
-            pipeline.addStage({
-                stageName: "deploy", actions: [new LambdaInvokeAction({
-                    actionName: "Deploy",
-                    lambda: props.postActionLambda
-                })]
-            });
-        }
+        const versionUpdateFn = new Function(this, 'version-update-fn', {
+            code: Code.fromAsset(path.join(__dirname, '../../flink-app-redeploy-hook')),
+            handler: "app.lambda_handler",
+            runtime: Runtime.PYTHON_3_9,
+            environment: {
+                ASSET_BUCKET_ARN: props.deployBucket.bucketArn,
+                FILE_KEY: s3BasePath + "/" + props.appName + "-latest.jar",
+                APP_NAME: props.appName
+            },
+            timeout: Duration.minutes(1)
+        });
+
+        versionUpdateFn.addToRolePolicy(new PolicyStatement({
+            actions: ["kinesisanalytics:DescribeApplication", "kinesisanalytics:UpdateApplication"],
+            resources: ["arn:aws:kinesisanalytics:" + Aws.REGION + ":" + Aws.ACCOUNT_ID + ":application/*"]
+        }));
+        versionUpdateFn.addToRolePolicy(new PolicyStatement({
+            resources: ["*"],
+            actions: ["codepipeline:PutJobSuccessResult", "codepipeline:PutJobFailureResult"]
+        }));
+
+        pipeline.addStage({
+            stageName: "deploy", actions: [new LambdaInvokeAction({
+                actionName: "Deploy",
+                lambda: versionUpdateFn
+            })]
+        });
+
     }
 }
diff --git a/infrastructure-cdk/lib/infra-pipeline-stack.ts b/infrastructure-cdk/lib/infra-pipeline-stack.ts
@@ -1,20 +1,22 @@
-import {Stack, StackProps} from 'aws-cdk-lib';
+import {Fn, Stack, StackProps} from 'aws-cdk-lib';
 import {Construct} from 'constructs';
-import {CodePipeline, ShellStep} from "aws-cdk-lib/pipelines";
+import {CodePipeline, CodePipelineSource, ShellStep} from "aws-cdk-lib/pipelines";
 import {RealtimeApplication} from "./real-time-application";
-import {REPOSITORY_FILE_PATH} from "./shared-vars";
+import {Bucket} from "aws-cdk-lib/aws-s3";
+import {BUCKET_NAME_OUTPUT} from "aws-cdk/lib";
+import {SOURCE_CODE_ZIP} from "./shared-vars";
 
 export class InfraPipelineStack extends Stack {
     constructor(scope: Construct, id: string, props?: StackProps) {
         super(scope, id, props);
 
+        const artifactBucket = Bucket.fromBucketName(this, 'artifactBucket-import', Fn.importValue(BUCKET_NAME_OUTPUT))
 
         const pipeline = new CodePipeline(this, 'Pipeline', {
             selfMutation: false,
             synth: new ShellStep('Synth', {
+                input: CodePipelineSource.s3(artifactBucket, SOURCE_CODE_ZIP),
                 commands: [
-                    "curl " + REPOSITORY_FILE_PATH + " --output app.zip",
-                    "unzip app.zip",
                     "cd infrastructure-cdk",
                     "npm ci",
                     "npm run build",
diff --git a/infrastructure-cdk/lib/real-time-application.ts b/infrastructure-cdk/lib/real-time-application.ts
@@ -1,56 +1,12 @@
-import {Aws, Fn, Stage, StageProps} from "aws-cdk-lib";
+import {Stage, StageProps} from "aws-cdk-lib";
 import {Construct} from "constructs";
-import {Code, Function, Runtime} from "aws-cdk-lib/aws-lambda";
-import {Stream, StreamEncryption, StreamMode} from "aws-cdk-lib/aws-kinesis";
-import * as kda from "@aws-cdk/aws-kinesisanalytics-flink-alpha";
-import {APPLICATION_NAME, ASSET_BUCKET_EXPORT_NAME} from "./shared-vars";
-import {Bucket} from "aws-cdk-lib/aws-s3";
+import {ApplicationStack} from "./application-stack";
 
 
 export class RealtimeApplication extends Stage {
     constructor(scope: Construct, id: string, props?: StageProps) {
         super(scope, id, props);
 
-        const assetBucket = Bucket.fromBucketName(this, 'imported-asset-bucket', Fn.importValue(ASSET_BUCKET_EXPORT_NAME));
-
-        const stream = new Stream(this, 'raw', {
-            streamMode: StreamMode.PROVISIONED,
-            shardCount: 1,
-            encryption: StreamEncryption.MANAGED
-        });
-
-        const dataSourceFn = new Function(this, 'data-source', {
-            code: Code.fromInline(
-                "import boto3\n" +
-                "import os\n\n" +
-                "kinesis = boto3.client('kinesis')\n\n" +
-                "def lambda_handler(event, context):\n" +
-                "   kinesis.put_record(StreamName=os.environ['STREAM_NAME']), Data=b'{\"message\":\"This is the test message from the lambda\"}', PartitionKey='default')\n"
-            ),
-            handler: "index.lambda_handler",
-            runtime: Runtime.PYTHON_3_9,
-            environment: {
-                STREAM_NAME: stream.streamName
-            }
-        });
-
-        stream.grantWrite(dataSourceFn);
-
-        const application = new kda.Application(this, 'app', {
-                code: kda.ApplicationCode.fromBucket(assetBucket, "jars/" + APPLICATION_NAME + "-1.0.0.jar"),
-                runtime: kda.Runtime.FLINK_1_13,
-                propertyGroups: {
-                    "KinesisReader": {
-                        "input.stream.name": stream.streamName,
-                        "aws.region": Aws.REGION,
-                        "flink.stream.initpos": "LATEST"
-                    },
-                },
-                snapshotsEnabled: false,
-                parallelismPerKpu: 1
-            }
-        );
-
-        stream.grantRead(application);
+        new ApplicationStack(this, 'ApplicationStack');
     }
 }
diff --git a/infrastructure-cdk/lib/shared-vars.ts b/infrastructure-cdk/lib/shared-vars.ts
@@ -1,3 +1,3 @@
-export const REPOSITORY_FILE_PATH = "aws-samples/automate-deployment-and-version-update-of-kda-application"
 export const APPLICATION_NAME = "kinesis-analytics-application"
-export const ASSET_BUCKET_EXPORT_NAME = "Blog::Artefact::BucketName"
+export const SOURCE_CODE_ZIP = "automate-deployment-and-version-update-of-kda-application.zip"
+export const ASSET_BUCKET_EXPORT_NAME = "Blog::Artifact::BucketName"
