modernization

airflow/Dockerfile

@@ -1,4 +1,5 @@
-FROM apache/airflow:1.10.11
+FROM apache/airflow:2.2.2-python3.8
+
 
 ENV AIRFLOW_HOME=/usr/local/airflow
 
@@ -8,9 +9,24 @@ RUN apt-get update && apt-get install -y python3-pip \
     libcurl4-gnutls-dev \
     librtmp-dev \
     python3-dev \
-    libpq-dev
-
-RUN python3 -m pip install PyGreSQL argcomplete pycurl
+    python3-setuptools \
+    libpq-dev \
+    gcc \
+    build-essential \
+    g++ \
+    git-all \
+    unixodbc-dev \
+    apt-utils \
+    apt-transport-https \
+    debconf-utils \
+    vim \
+    telnet
+
+RUN python3 -m pip install --upgrade pip
+
+RUN python3 -m pip install awscli
+RUN export PATH=~/.local/bin:$PATH; \
+    aws --version
 
 COPY ./config/* /
 COPY ./dags ${AIRFLOW_HOME}/dags
@@ -21,6 +37,8 @@ EXPOSE 8080
 
 USER airflow
 
+RUN python3 -m pip install --user argcomplete pycurl concurrent-log-handler
+
 WORKDIR ${AIRFLOW_HOME}
 
 # ENTRYPOINT ["/entrypoint.sh"]

airflow/config/webserver_entry.sh

@@ -1,11 +1,10 @@
 #!/usr/bin/env bash
 
 set -Eeuxo pipefail
-
-airflow initdb
+airflow db init
 sleep 5
 
-airflow create_user -r Admin -u admin -f FirstName -l LastName -p ${ADMIN_PASS} -e [email protected]
+airflow users create -r Admin -u admin -f FirstName -l LastName -p ${ADMIN_PASS} -e [email protected]
 sleep 5
 
 airflow webserver

airflow/config/worker_entry.sh

@@ -1,5 +1,6 @@
 #!/usr/bin/env bash
 
 set -Eeuxo pipefail
+
 sleep 30
-airflow worker
+airflow celery worker

app/config.ts

@@ -1,6 +1,6 @@
 import { DBConfig } from "./constructs/rds";
-import {InstanceClass, InstanceSize, InstanceType} from "@aws-cdk/aws-ec2";
-import { RetentionDays } from "@aws-cdk/aws-logs";
+import {InstanceClass, InstanceSize, InstanceType} from "aws-cdk-lib/aws-ec2";
+import { RetentionDays } from "aws-cdk-lib/aws-logs";
 
 export interface AirflowTaskConfig {
   readonly cpu: number;
@@ -66,7 +66,7 @@ export const defaultDBConfig: DBConfig = {
   dbName: "farflow",
   port: 5432,
   masterUsername: "airflow",
-  instanceType: InstanceType.of(InstanceClass.T2, InstanceSize.SMALL),
+  instanceType: InstanceType.of(InstanceClass.T3, InstanceSize.SMALL),
   allocatedStorageInGB: 25,
   backupRetentionInDays: 30
 };

app/constructs/airflow-construct.ts

@@ -1,10 +1,11 @@
-import {CfnOutput, Construct} from "@aws-cdk/core";
-import { IVpc } from "@aws-cdk/aws-ec2";
-
-import ecs = require('@aws-cdk/aws-ecs');
-import ec2 = require("@aws-cdk/aws-ec2");
-import { DockerImageAsset } from '@aws-cdk/aws-ecr-assets';
-import { FargateTaskDefinition } from '@aws-cdk/aws-ecs';
+import { CfnOutput } from "aws-cdk-lib/core";
+import { Construct } from 'constructs';
+import { IVpc } from "aws-cdk-lib/aws-ec2";
+
+import ecs = require('aws-cdk-lib/aws-ecs');
+import ec2 = require("aws-cdk-lib/aws-ec2");
+import { DockerImageAsset } from 'aws-cdk-lib/aws-ecr-assets';
+import { FargateTaskDefinition } from 'aws-cdk-lib/aws-ecs';
 
 import {airflowTaskConfig, ContainerConfig} from "../config";
 import { ServiceConstruct } from "./service-construct";

app/constructs/dag-tasks.ts

@@ -1,9 +1,10 @@
-import { Construct } from "@aws-cdk/core";
-import {AwsLogDriver, } from "@aws-cdk/aws-ecs";
-import { RetentionDays } from "@aws-cdk/aws-logs";
-import {IVpc, ISecurityGroup, Port} from "@aws-cdk/aws-ec2";
-import efs = require('@aws-cdk/aws-efs');
-import { LogGroup } from '@aws-cdk/aws-logs';
+
+import { Construct } from 'constructs';
+import {AwsLogDriver, } from "aws-cdk-lib/aws-ecs";
+import { RetentionDays } from "aws-cdk-lib/aws-logs";
+import {IVpc, ISecurityGroup, Port} from "aws-cdk-lib/aws-ec2";
+import efs = require('aws-cdk-lib/aws-efs');
+import { LogGroup } from 'aws-cdk-lib/aws-logs';
 
 import { AirflowDagTaskDefinition, EfsVolumeInfo } from "./task-construct"
 

app/constructs/rds.ts

@@ -1,16 +1,17 @@
-import { Duration, Construct } from "@aws-cdk/core";
+import { Duration} from "aws-cdk-lib/core";
+import { Construct } from 'constructs';
 import {
   DatabaseInstance,
   DatabaseInstanceEngine, PostgresEngineVersion,
   StorageType
-} from "@aws-cdk/aws-rds";
-import { ISecret, Secret } from "@aws-cdk/aws-secretsmanager";
+} from "aws-cdk-lib/aws-rds";
+import { ISecret, Secret } from "aws-cdk-lib/aws-secretsmanager";
 import {
   InstanceType,
   ISecurityGroup,
   IVpc,
   SubnetType
-} from "@aws-cdk/aws-ec2";
+} from "aws-cdk-lib/aws-ec2";
 
 import { defaultDBConfig } from "../config";
 
@@ -60,13 +61,13 @@ export class RDSConstruct extends Construct {
 
     this.rdsInstance = new DatabaseInstance(this, "RDSInstance", {
       engine: DatabaseInstanceEngine.postgres({
-        version: PostgresEngineVersion.VER_12_4
+        version: PostgresEngineVersion.VER_13_4
       }),
       instanceType: defaultDBConfig.instanceType,
       instanceIdentifier: defaultDBConfig.dbName,
       vpc: props.vpc,
       securityGroups: [props.defaultVpcSecurityGroup],
-      vpcPlacement: { subnetType: SubnetType.PRIVATE },
+      vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_NAT },
       storageEncrypted: true,
       multiAz: false,
       autoMinorVersionUpgrade: false,
@@ -94,6 +95,6 @@ export class RDSConstruct extends Construct {
     endpoint: string,
     password: string
   ): string {
-    return `postgresql+pygresql://${dbConfig.masterUsername}:${password}@${endpoint}:${dbConfig.port}/${dbConfig.dbName}`;
+    return `postgresql+psycopg2://${dbConfig.masterUsername}:${password}@${endpoint}:${dbConfig.port}/${dbConfig.dbName}`;
   }
 }

app/constructs/service-construct.ts

@@ -1,12 +1,13 @@
-import {CfnOutput, Construct, Duration} from "@aws-cdk/core";
-import {IVpc} from "@aws-cdk/aws-ec2";
-import {FargatePlatformVersion, FargateTaskDefinition} from '@aws-cdk/aws-ecs';
+import { CfnOutput, Duration } from "aws-cdk-lib/core";
+import { Construct } from 'constructs';
+import {IVpc} from "aws-cdk-lib/aws-ec2";
+import { FargatePlatformVersion, FargateTaskDefinition } from 'aws-cdk-lib/aws-ecs';
 
 import {PolicyConstruct} from "../policies";
 import {workerAutoScalingConfig} from "../config";
-import ecs = require('@aws-cdk/aws-ecs');
-import ec2 = require("@aws-cdk/aws-ec2");
-import elbv2 = require("@aws-cdk/aws-elasticloadbalancingv2");
+import ecs = require('aws-cdk-lib/aws-ecs');
+import ec2 = require("aws-cdk-lib/aws-ec2");
+import elbv2 = require("aws-cdk-lib/aws-elasticloadbalancingv2");
 
 export interface ServiceConstructProps {
   readonly vpc: IVpc;
@@ -29,14 +30,15 @@ export class ServiceConstruct extends Construct {
       policies.managedPolicies.forEach(managedPolicy => props.taskDefinition.taskRole.addManagedPolicy(managedPolicy));
     }
     if (policies.policyStatements) {
-      policies.policyStatements.forEach(policyStatement => props.taskDefinition.taskRole.addToPolicy(policyStatement));
+      policies.policyStatements.forEach(policyStatement => props.taskDefinition.taskRole.addToPrincipalPolicy(policyStatement));
     }
 
     // Create Fargate Service for Airflow
     this.fargateService = new ecs.FargateService(this, name, {
       cluster: props.cluster,
       taskDefinition: props.taskDefinition,
-      securityGroup: props.defaultVpcSecurityGroup,
+      securityGroups: [props.defaultVpcSecurityGroup,],
+      enableExecuteCommand: true,
       platformVersion: FargatePlatformVersion.VERSION1_4
     });
     const allowedPorts = new ec2.Port({

app/constructs/task-construct.ts

@@ -1,9 +1,12 @@
-import { Construct } from "@aws-cdk/core";
+//import { Construct } from "aws-cdk-lib/core";
+import { Construct } from 'constructs';
+
+import ecs = require('aws-cdk-lib/aws-ecs');
+import { DockerImageAsset } from 'aws-cdk-lib/aws-ecr-assets';
+import { FargateTaskDefinition } from 'aws-cdk-lib/aws-ecs';
+import {ManagedPolicy} from "aws-cdk-lib/aws-iam";
+
 
-import ecs = require('@aws-cdk/aws-ecs');
-import { DockerImageAsset } from '@aws-cdk/aws-ecr-assets';
-import { FargateTaskDefinition } from '@aws-cdk/aws-ecs';
-import {ManagedPolicy} from "@aws-cdk/aws-iam";
 
 export interface AirflowDagTaskDefinitionProps {
   readonly taskFamilyName: string;

app/farflow.ts

@@ -1,6 +1,6 @@
-import ec2 = require('@aws-cdk/aws-ec2');
-import ecs = require('@aws-cdk/aws-ecs');
-import cdk = require('@aws-cdk/core');
+import ec2 = require('aws-cdk-lib/aws-ec2');
+import ecs = require('aws-cdk-lib/aws-ecs');
+import cdk = require('aws-cdk-lib/core');
 import {RDSConstruct} from "./constructs/rds";
 import {AirflowConstruct} from "./constructs/airflow-construct";
 import { DagTasks } from './constructs/dag-tasks';

app/policies.ts

@@ -1,5 +1,6 @@
-import { Construct } from "@aws-cdk/core";
-import { IManagedPolicy, ManagedPolicy, PolicyStatement } from "@aws-cdk/aws-iam";
+//import { Construct } from "aws-cdk-lib/core";
+import { Construct } from 'constructs';
+import { IManagedPolicy, ManagedPolicy, PolicyStatement } from "aws-cdk-lib/aws-iam";
 
 export class PolicyConstruct extends Construct {
     public readonly policyStatements?: PolicyStatement[];