feat(celery): add celery services

src/constructs/ad-hoc/app/index.ts

@@ -1,7 +1,7 @@
 import { CfnOutput, Stack } from 'aws-cdk-lib';
 import { IVpc, ISecurityGroup } from 'aws-cdk-lib/aws-ec2';
 import { Repository } from 'aws-cdk-lib/aws-ecr';
-import { Cluster, EcrImage } from 'aws-cdk-lib/aws-ecs';
+import { Cluster, ContainerImage, EcrImage } from 'aws-cdk-lib/aws-ecs';
 import { IApplicationLoadBalancer, ApplicationListener } from 'aws-cdk-lib/aws-elasticloadbalancingv2';
 import { DatabaseInstance } from 'aws-cdk-lib/aws-rds';
 import { CnameRecord, HostedZone } from 'aws-cdk-lib/aws-route53';
@@ -11,7 +11,9 @@ import { Construct } from 'constructs';
 // import { HighestPriorityRule } from '../../internal/customResources/highestPriorityRule';
 import { EcsRoles } from '../../internal/ecs/iam';
 import { ManagementCommandTask } from '../../internal/ecs/management-command';
+import { RedisService } from '../../internal/ecs/redis';
 import { WebService } from '../../internal/ecs/web';
+import { WorkerService } from '../../internal/ecs/worker';
 
 export interface AdHocAppProps {
   readonly baseStackName: string;
@@ -58,6 +60,7 @@ export class AdHocApp extends Construct {
     });
 
     const serviceDiscoveryNamespace = props.serviceDiscoveryNamespace.namespaceName;
+
     // shared environment variables
     const environmentVariables: { [key: string]: string }= {
       S3_BUCKET_NAME: props.assetsBucket.bucketName,
@@ -80,6 +83,19 @@ export class AdHocApp extends Construct {
       zone: hostedZone,
     });
 
+    new RedisService(this, 'RedisService', {
+      cluster,
+      environmentVariables: {},
+      vpc: props.vpc,
+      appSecurityGroup: props.appSecurityGroup,
+      taskRole: ecsRoles.ecsTaskRole,
+      executionRole: ecsRoles.taskExecutionRole,
+      image: ContainerImage.fromRegistry('redis:5.0.3-alpine'),
+      containerName: 'redis',
+      family: 'redis',
+      serviceDiscoveryNamespace: props.serviceDiscoveryNamespace,
+    });
+
     // api service
     // const backendService =
     new WebService(this, 'ApiService', {
@@ -125,6 +141,18 @@ export class AdHocApp extends Construct {
     });
 
     // worker service
+    new WorkerService(this, 'DefaultCeleryWorker', {
+      cluster,
+      environmentVariables,
+      vpc: props.vpc,
+      appSecurityGroup: props.appSecurityGroup,
+      taskRole: ecsRoles.ecsTaskRole,
+      executionRole: ecsRoles.taskExecutionRole,
+      image: backendImage,
+      command: ['celery', '--app=backend.celery_app:app', 'worker', '--loglevel=INFO', '-Q', 'default'],
+      containerName: 'default-worker',
+      family: 'default-worker',
+    });
 
     // scheduler service
 

src/constructs/internal/ecs/redis/index.ts

@@ -0,0 +1,115 @@
+import { Duration, RemovalPolicy, Stack } from 'aws-cdk-lib';
+import { ISecurityGroup, IVpc } from 'aws-cdk-lib/aws-ec2';
+import {
+  LogDriver,
+  Cluster,
+  ContainerImage,
+  FargateService,
+  FargateTaskDefinition,
+} from 'aws-cdk-lib/aws-ecs';
+import { Role } from 'aws-cdk-lib/aws-iam';
+import {
+  LogGroup,
+  LogStream,
+  RetentionDays,
+} from 'aws-cdk-lib/aws-logs';
+import { DnsRecordType, PrivateDnsNamespace } from 'aws-cdk-lib/aws-servicediscovery';
+import { Construct } from 'constructs';
+
+
+export interface RedisProps {
+  readonly cluster: Cluster;
+  readonly vpc: IVpc;
+  readonly cpu?: number;
+  readonly memorySize?: number;
+  readonly appSecurityGroup: ISecurityGroup;
+  readonly image: ContainerImage;
+  readonly useSpot?: boolean;
+  readonly containerName: string;
+  readonly taskRole: Role;
+  readonly executionRole: Role;
+  readonly family: string;
+  readonly environmentVariables: { [key: string]: string };
+  readonly serviceDiscoveryNamespace: PrivateDnsNamespace;
+};
+
+export class RedisService extends Construct {
+  constructor(scope: Construct, id: string, props: RedisProps) {
+    super(scope, id);
+
+    const stackName = Stack.of(this).stackName;
+
+    // define log group and logstream
+    const logGroupName = `/ecs/${stackName}/${props.containerName}/`;
+    const streamPrefix = props.containerName;
+
+    // define log group and logstream
+    const logGroup = new LogGroup(this, 'LogGroup', {
+      logGroupName,
+      retention: RetentionDays.ONE_DAY,
+      removalPolicy: RemovalPolicy.DESTROY,
+    });
+
+    new LogStream(this, 'LogStream', {
+      logGroup,
+      logStreamName: props.containerName,
+    });
+
+    // task definition
+    const taskDefinition = new FargateTaskDefinition(this, 'TaskDefinition', {
+      cpu: props.cpu ?? 256,
+      executionRole: props.executionRole,
+      taskRole: props.taskRole,
+      family: props.family,
+    });
+
+    taskDefinition.addContainer(props.containerName, {
+      image: props.image,
+      containerName: props.containerName,
+      environment: props.environmentVariables,
+      essential: true,
+      logging: LogDriver.awsLogs({
+        streamPrefix,
+        logGroup,
+      }),
+    });
+
+    const useSpot = props.useSpot ?? false;
+
+
+    // TODO: add service discovery service
+    const cloudMapService = props.serviceDiscoveryNamespace.createService('Service', {
+      name: `${stackName}-redis`,
+      dnsRecordType: DnsRecordType.A,
+      dnsTtl: Duration.seconds(30),
+      healthCheck: {
+        failureThreshold: 1,
+      },
+    });
+
+    const fargateService = new FargateService(this, 'Service', {
+      cluster: props.cluster,
+      taskDefinition,
+      assignPublicIp: true,
+      capacityProviderStrategies: [
+        {
+          capacityProvider: 'FARGATE_SPOT',
+          weight: useSpot ? 100 : 0,
+        },
+        {
+          capacityProvider: 'FARGATE',
+          weight: useSpot ? 0 : 100,
+        },
+      ],
+      desiredCount: 1,
+      enableExecuteCommand: true,
+      securityGroups: [props.appSecurityGroup],
+      serviceName: `${stackName}-${props.containerName}`,
+      vpcSubnets: {
+        subnets: props.vpc.privateSubnets,
+      },
+    });
+
+    fargateService.associateCloudMapService({ service: cloudMapService });
+  }
+}

src/constructs/internal/ecs/worker/index.ts

@@ -0,0 +1,105 @@
+import { RemovalPolicy, Stack } from 'aws-cdk-lib';
+import { ISecurityGroup, IVpc } from 'aws-cdk-lib/aws-ec2';
+import {
+  LogDriver,
+  Cluster,
+  ContainerImage,
+  FargateService,
+  FargateTaskDefinition,
+} from 'aws-cdk-lib/aws-ecs';
+import { Role } from 'aws-cdk-lib/aws-iam';
+import {
+  LogGroup,
+  LogStream,
+  RetentionDays,
+} from 'aws-cdk-lib/aws-logs';
+import { Construct } from 'constructs';
+
+
+export interface WorkerProps {
+  readonly cluster: Cluster;
+  readonly vpc: IVpc;
+  readonly cpu?: number;
+  readonly memorySize?: number;
+  readonly appSecurityGroup: ISecurityGroup;
+  readonly image: ContainerImage;
+  readonly command: string[];
+  readonly useSpot?: boolean;
+  readonly containerName: string;
+  readonly taskRole: Role;
+  readonly executionRole: Role;
+  readonly family: string;
+  readonly environmentVariables: { [key: string]: string };
+};
+
+export class WorkerService extends Construct {
+  constructor(scope: Construct, id: string, props: WorkerProps) {
+    super(scope, id);
+
+    const stackName = Stack.of(this).stackName;
+
+    // define log group and logstream
+    const logGroupName = `/ecs/${stackName}/${props.containerName}/`;
+    const streamPrefix = props.containerName;
+
+    // define log group and logstream
+    const logGroup = new LogGroup(this, 'LogGroup', {
+      logGroupName,
+      retention: RetentionDays.ONE_DAY,
+      removalPolicy: RemovalPolicy.DESTROY,
+    });
+
+    new LogStream(this, 'LogStream', {
+      logGroup,
+      logStreamName: props.containerName,
+    });
+
+    // task definition
+    const taskDefinition = new FargateTaskDefinition(this, 'TaskDefinition', {
+      cpu: props.cpu ?? 256,
+      executionRole: props.executionRole,
+      taskRole: props.taskRole,
+      family: props.family,
+    });
+
+    taskDefinition.addContainer(props.containerName, {
+      image: props.image,
+      command: props.command,
+      containerName: props.containerName,
+      environment: props.environmentVariables,
+      essential: true,
+      logging: LogDriver.awsLogs({
+        streamPrefix,
+        logGroup,
+      }),
+    });
+
+    const useSpot = props.useSpot ?? false;
+
+    // we don't need autoscaling in ad hoc environments
+    // TODO: add prop to enable autoscaling for the worker
+    // const service =
+    new FargateService(this, 'Service', {
+      cluster: props.cluster,
+      taskDefinition,
+      assignPublicIp: true,
+      capacityProviderStrategies: [
+        {
+          capacityProvider: 'FARGATE_SPOT',
+          weight: useSpot ? 100 : 0,
+        },
+        {
+          capacityProvider: 'FARGATE',
+          weight: useSpot ? 0 : 100,
+        },
+      ],
+      desiredCount: 1,
+      enableExecuteCommand: true,
+      securityGroups: [props.appSecurityGroup],
+      serviceName: `${stackName}-${props.containerName}`,
+      vpcSubnets: {
+        subnets: props.vpc.privateSubnets,
+      },
+    });
+  }
+}