-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
airgradient secret fix and node 16 upgrade
- Loading branch information
Showing
7 changed files
with
429 additions
and
1,582 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| LCS_API=https://api.openaq.org | ||
| STACK=my-dev-stack | ||
| SECRET_STACK=my-prod-stack | ||
| STACK=lcs-etl-pipeline | ||
| SECRET_STACK=lcs-etl-pipeline | ||
| BUCKET=openaq-fetches | ||
| VERBOSE=1 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,10 +1,42 @@ | ||
| { | ||
| "app": "npx ts-node --prefer-ts-exts cdk/app.ts", | ||
| "watch": { | ||
| "include": [ | ||
| "**" | ||
| ], | ||
| "exclude": [ | ||
| "README.md", | ||
| "cdk*.json", | ||
| "**/*.d.ts", | ||
| "**/*.js", | ||
| "tsconfig.json", | ||
| "package*.json", | ||
| "yarn.lock", | ||
| "node_modules", | ||
| "test" | ||
| ] | ||
| }, | ||
| "context": { | ||
| "@aws-cdk/core:enableStackNameDuplicates": "true", | ||
| "aws-cdk:enableDiffNoFail": "true", | ||
| "@aws-cdk/core:stackRelativeExports": "true", | ||
| "@aws-cdk/aws-ecr-assets:dockerIgnoreSupport": true, | ||
| "@aws-cdk/aws-secretsmanager:parseOwnedSecretName": true | ||
| "@aws-cdk/aws-apigateway:usagePlanKeyOrderInsensitiveId": true, | ||
| "@aws-cdk/core:stackRelativeExports": true, | ||
| "@aws-cdk/aws-rds:lowercaseDbIdentifier": true, | ||
| "@aws-cdk/aws-lambda:recognizeVersionProps": true, | ||
| "@aws-cdk/aws-lambda:recognizeLayerVersion": true, | ||
| "@aws-cdk/aws-cloudfront:defaultSecurityPolicyTLSv1.2_2021": true, | ||
| "@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true, | ||
| "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true, | ||
| "@aws-cdk/core:checkSecretUsage": true, | ||
| "@aws-cdk/aws-iam:minimizePolicies": true, | ||
| "@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true, | ||
| "@aws-cdk/core:validateSnapshotRemovalPolicy": true, | ||
| "@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true, | ||
| "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true, | ||
| "@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true, | ||
| "@aws-cdk/aws-apigateway:disableCloudWatchRole": true, | ||
| "@aws-cdk/core:enablePartitionLiterals": true, | ||
| "@aws-cdk/core:target-partitions": [ | ||
| "aws", | ||
| "aws-cn" | ||
| ] | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,133 +1,152 @@ | ||
| import * as events from "@aws-cdk/aws-events"; | ||
| import * as eventTargets from "@aws-cdk/aws-events-targets"; | ||
| import * as iam from "@aws-cdk/aws-iam"; | ||
| import * as lambda from "@aws-cdk/aws-lambda"; | ||
| import { SqsEventSource } from "@aws-cdk/aws-lambda-event-sources"; | ||
| import * as s3 from "@aws-cdk/aws-s3"; | ||
| import * as sqs from "@aws-cdk/aws-sqs"; | ||
| import * as cdk from "@aws-cdk/core"; | ||
| import { execSync } from "child_process"; | ||
| import { Interval, Source } from "./types"; | ||
| import * as events from 'aws-cdk-lib/aws-events'; | ||
| import * as eventTargets from 'aws-cdk-lib/aws-events-targets'; | ||
| import * as iam from 'aws-cdk-lib/aws-iam'; | ||
| import * as lambda from 'aws-cdk-lib/aws-lambda'; | ||
| import { SqsEventSource } from 'aws-cdk-lib/aws-lambda-event-sources'; | ||
| import * as s3 from 'aws-cdk-lib/aws-s3'; | ||
| import * as sqs from 'aws-cdk-lib/aws-sqs'; | ||
| import * as cdk from 'aws-cdk-lib/core'; | ||
| import { execSync } from 'child_process'; | ||
| import { Interval, Source } from './types'; | ||
| import { Construct } from 'constructs'; | ||
|
|
||
| export class EtlPipeline extends cdk.Stack { | ||
| constructor( | ||
| scope: cdk.Construct, | ||
| id: string, | ||
| { fetcherModuleDir, schedulerModuleDir, sources, lcsApi, bucketName, ...props }: StackProps | ||
| ) { | ||
| super(scope, id, props); | ||
| constructor( | ||
| scope: Construct, | ||
| id: string, | ||
| { | ||
| fetcherModuleDir, | ||
| schedulerModuleDir, | ||
| sources, | ||
| lcsApi, | ||
| bucketName, | ||
| ...props | ||
| }: StackProps | ||
| ) { | ||
| super(scope, id, props); | ||
|
|
||
| const queue = new sqs.Queue(this, "FetcherQueue", { | ||
| queueName: `${cdk.Stack.of(this).stackName}-fetch-queue`, | ||
| visibilityTimeout: cdk.Duration.seconds(2880), | ||
| }); | ||
| const bucket = s3.Bucket.fromBucketName(this, "Data", bucketName); | ||
| const queue = new sqs.Queue(this, 'FetcherQueue', { | ||
| queueName: `${cdk.Stack.of(this).stackName}-fetch-queue`, | ||
| visibilityTimeout: cdk.Duration.seconds(2880), | ||
| }); | ||
| const bucket = s3.Bucket.fromBucketName(this, 'Data', bucketName); | ||
|
|
||
| this.buildFetcherLambda({ moduleDir: fetcherModuleDir, queue, bucket, lcsApi }); | ||
| this.buildSchedulerLambdas({ moduleDir: schedulerModuleDir, queue, sources }); | ||
| } | ||
| this.buildFetcherLambda({ | ||
| moduleDir: fetcherModuleDir, | ||
| queue, | ||
| bucket, | ||
| lcsApi, | ||
| }); | ||
| this.buildSchedulerLambdas({ | ||
| moduleDir: schedulerModuleDir, | ||
| queue, | ||
| sources, | ||
| }); | ||
| } | ||
|
|
||
| private buildFetcherLambda(props: { | ||
| moduleDir: string; | ||
| queue: sqs.Queue; | ||
| bucket: s3.IBucket; | ||
| lcsApi: string; | ||
| }): lambda.Function { | ||
| this.prepareNodeModules(props.moduleDir); | ||
| const handler = new lambda.Function(this, "Fetcher", { | ||
| description: "Fetch a single source for a given time period", | ||
| runtime: lambda.Runtime.NODEJS_12_X, | ||
| handler: "index.handler", | ||
| code: lambda.Code.fromAsset(props.moduleDir), | ||
| timeout: cdk.Duration.seconds(900), | ||
| memorySize: 512, | ||
| environment: { | ||
| BUCKET: props.bucket.bucketName, | ||
| STACK: cdk.Stack.of(this).stackName, | ||
| VERBOSE: '1', | ||
| LCS_API: props.lcsApi | ||
| }, | ||
| }); | ||
| handler.addEventSource( | ||
| new SqsEventSource(props.queue, { | ||
| batchSize: 1, | ||
| }) | ||
| ); | ||
| props.queue.grantConsumeMessages(handler); | ||
| props.bucket.grantReadWrite(handler); | ||
| handler.addToRolePolicy( | ||
| new iam.PolicyStatement({ | ||
| effect: iam.Effect.ALLOW, | ||
| actions: [ | ||
| "secretsmanager:DescribeSecret", | ||
| "secretsmanager:GetSecretValue", | ||
| ], | ||
| resources: [ | ||
| `arn:aws:secretsmanager:*:*:secret:${cdk.Stack.of(this).stackName}/*`, | ||
| ], | ||
| }) | ||
| ); | ||
| return handler; | ||
| } | ||
| private buildFetcherLambda(props: { | ||
| moduleDir: string; | ||
| queue: sqs.Queue; | ||
| bucket: s3.IBucket; | ||
| lcsApi: string; | ||
| }): lambda.Function { | ||
| this.prepareNodeModules(props.moduleDir); | ||
| const handler = new lambda.Function(this, 'Fetcher', { | ||
| description: 'Fetch a single source for a given time period', | ||
| runtime: lambda.Runtime.NODEJS_16_X, | ||
| handler: 'index.handler', | ||
| code: lambda.Code.fromAsset(props.moduleDir), | ||
| timeout: cdk.Duration.seconds(900), | ||
| memorySize: 512, | ||
| environment: { | ||
| BUCKET: props.bucket.bucketName, | ||
| STACK: cdk.Stack.of(this).stackName, | ||
| VERBOSE: '1', | ||
| LCS_API: props.lcsApi, | ||
| }, | ||
| }); | ||
| handler.addEventSource( | ||
| new SqsEventSource(props.queue, { | ||
| batchSize: 1, | ||
| }) | ||
| ); | ||
| props.queue.grantConsumeMessages(handler); | ||
| props.bucket.grantReadWrite(handler); | ||
| handler.addToRolePolicy( | ||
| new iam.PolicyStatement({ | ||
| effect: iam.Effect.ALLOW, | ||
| actions: [ | ||
| 'secretsmanager:DescribeSecret', | ||
| 'secretsmanager:GetSecretValue', | ||
| ], | ||
| resources: [ | ||
| `arn:aws:secretsmanager:*:*:secret:${ | ||
| cdk.Stack.of(this).stackName | ||
| }/*`, | ||
| ], | ||
| }) | ||
| ); | ||
| return handler; | ||
| } | ||
|
|
||
| private buildSchedulerLambdas(props: { | ||
| moduleDir: string; | ||
| queue: sqs.Queue; | ||
| sources: Source[]; | ||
| }): lambda.Function[] { | ||
| const durations: Record<Interval, cdk.Duration> = { | ||
| minute: cdk.Duration.minutes(1), | ||
| hour: cdk.Duration.hours(1), | ||
| day: cdk.Duration.days(1), | ||
| }; | ||
| return Object.entries(durations).map(([interval, duration]) => { | ||
| const scheduler = new lambda.Function( | ||
| this, | ||
| `${interval}Scheduler`, | ||
| { | ||
| description: `${interval}Scheduler`, | ||
| runtime: lambda.Runtime.NODEJS_12_X, | ||
| handler: "index.handler", | ||
| code: lambda.Code.fromAsset(props.moduleDir), | ||
| timeout: cdk.Duration.seconds(25), | ||
| memorySize: 128, | ||
| environment: { | ||
| QUEUE_URL: props.queue.queueUrl, | ||
| SOURCES: props.sources | ||
| .filter((source) => source.frequency === interval) | ||
| .map((source) => source.provider) | ||
| .join(","), | ||
| }, | ||
| } | ||
| ); | ||
| props.queue.grantSendMessages(scheduler); | ||
| new events.Rule(this, `${interval}Rule`, { | ||
| schedule: events.Schedule.rate(duration), | ||
| targets: [new eventTargets.LambdaFunction(scheduler)], | ||
| }); | ||
| return scheduler; | ||
| }); | ||
| } | ||
| private buildSchedulerLambdas(props: { | ||
| moduleDir: string; | ||
| queue: sqs.Queue; | ||
| sources: Source[]; | ||
| }): lambda.Function[] { | ||
| const durations: Record<Interval, cdk.Duration> = { | ||
| minute: cdk.Duration.minutes(1), | ||
| hour: cdk.Duration.hours(1), | ||
| day: cdk.Duration.days(1), | ||
| }; | ||
| return Object.entries(durations).map(([interval, duration]) => { | ||
| const scheduler = new lambda.Function( | ||
| this, | ||
| `${interval}Scheduler`, | ||
| { | ||
| description: `${interval}Scheduler`, | ||
| runtime: lambda.Runtime.NODEJS_16_X, | ||
| handler: 'index.handler', | ||
| code: lambda.Code.fromAsset(props.moduleDir), | ||
| timeout: cdk.Duration.seconds(25), | ||
| memorySize: 128, | ||
| environment: { | ||
| QUEUE_URL: props.queue.queueUrl, | ||
| SOURCES: props.sources | ||
| .filter((source) => source.frequency === interval) | ||
| .map((source) => source.provider) | ||
| .join(','), | ||
| }, | ||
| } | ||
| ); | ||
| props.queue.grantSendMessages(scheduler); | ||
| new events.Rule(this, `${interval}Rule`, { | ||
| schedule: events.Schedule.rate(duration), | ||
| targets: [new eventTargets.LambdaFunction(scheduler)], | ||
| }); | ||
| return scheduler; | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Install node_modules in module directory for the sake of easy packaging. | ||
| * @param moduleDir string | ||
| */ | ||
| private prepareNodeModules(moduleDir: string): void { | ||
| const cmd = [ | ||
| "yarn", | ||
| "--prod", | ||
| "--frozen-lockfile", | ||
| `--modules-folder ${moduleDir}/node_modules`, | ||
| ].join(" "); | ||
| execSync(cmd); | ||
| } | ||
| /** | ||
| * Install node_modules in module directory for the sake of easy packaging. | ||
| * @param moduleDir string | ||
| */ | ||
| private prepareNodeModules(moduleDir: string): void { | ||
| const cmd = [ | ||
| 'yarn', | ||
| '--prod', | ||
| '--frozen-lockfile', | ||
| `--modules-folder ${moduleDir}/node_modules`, | ||
| ].join(' '); | ||
| execSync(cmd); | ||
| } | ||
| } | ||
|
|
||
| interface StackProps extends cdk.StackProps { | ||
| fetcherModuleDir: string; | ||
| schedulerModuleDir: string; | ||
| lcsApi: string; | ||
| bucketName: string; | ||
| sources: Source[]; | ||
| fetcherModuleDir: string; | ||
| schedulerModuleDir: string; | ||
| lcsApi: string; | ||
| bucketName: string; | ||
| sources: Source[]; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.