Here's an improved CDK code for deploying Airflow:
#!/usr/bin/env python3
from aws_cdk import (
Stack,
RemovalPolicy,
CfnOutput,
aws_ec2 as ec2,
aws_iam as iam,
aws_s3 as s3,
aws_s3_deployment as s3deploy,
aws_mwaa as mwaa,
)
from constructs import Construct
import os
class AirflowStack(Stack):
"""MWAA Airflow Stack with VPC configuration"""
def init(
self,
scope: Construct,
construct_id: str,
**kwargs,
) -> None:
super().init(scope, construct_id, **kwargs)
airflow_env_name = construct_id.lower()
dags_bucket_name = f"{airflow_env_name}-dags-{self.account}-{self.region}"
Create a VPC for MWAA
vpc = ec2.Vpc(
self,
"AirflowVpc",
max_azs=2,
nat_gateways=1,
subnet_configuration=[
ec2.SubnetConfiguration(
name="Public",
subnet_type=ec2.SubnetType.PUBLIC,
cidr_mask=24
),
ec2.SubnetConfiguration(
name="Private",
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS,
cidr_mask=24
)
]
)
S3 Bucket for DAGs
dags_bucket = s3.Bucket(
self,
"DagsBucket",
bucket_name=dags_bucket_name,
versioned=True,
encryption=s3.BucketEncryption.S3_MANAGED,
block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
enforce_ssl=True,
removal_policy=RemovalPolicy.RETAIN,
)
Upload DAGs if the folder exists
dags_folder = os.path.join(os.getcwd(), "dags")
if os.path.isdir(dags_folder):
s3deploy.BucketDeployment(
self,
"DeployDags",
sources=[s3deploy.Source.asset(dags_folder)],
destination_bucket=dags_bucket,
destination_key_prefix="dags",
prune=False,
)
MWAA Execution Role
mwaa_role = iam.Role(
self,
"ExecutionRole",
role_name=f"{airflow_env_name}-exec-role",
assumed_by=iam.CompositePrincipal(
iam.ServicePrincipal("airflow.amazonaws.com"),
iam.ServicePrincipal("airflow-env.amazonaws.com"),
),
path="/service-role/",
)
Attach required policies
mwaa_role.add_managed_policy(
iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3ReadOnlyAccess")
)
Inline policy for DAG bucket read/write
mwaa_role.add_to_policy(
iam.PolicyStatement(
actions=["s3:GetObject", "s3:PutObject", "s3:ListBucket", "s3:GetBucketVersioning"],
resources=[dags_bucket.bucket_arn, f"{dags_bucket.bucket_arn}/*"]
)
)
CloudWatch Logs permissions
mwaa_role.add_to_policy(
iam.PolicyStatement(
actions=[
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
"logs:GetLogEvents",
"logs:GetLogRecord",
"logs:GetLogGroupFields",
"logs:GetQueryResults"
],
resources=["*"]
)
)
SQS permissions for Airflow's Celery executor
mwaa_role.add_to_policy(
iam.PolicyStatement(
actions=[
"sqs:ChangeMessageVisibility",
"sqs:DeleteMessage",
"sqs:GetQueueAttributes",
"sqs:GetQueueUrl",
"sqs:ReceiveMessage",
"sqs:SendMessage"
],
resources=["*"]
)
)
Security Group for MWAA
mwaa_sg = ec2.SecurityGroup(
self,
"MwaaSecurityGroup",
vpc=vpc,
description="MWAA Security Group",
allow_all_outbound=True,
)
Allow internal MWAA traffic
mwaa_sg.add_ingress_rule(
peer=mwaa_sg,
connection=ec2.Port.all_traffic(),
description="Allow MWAA self traffic",
)
Network Configuration
private_subnets = vpc.select_subnets(
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
).subnet_ids
if len(private_subnets) < 2:
raise ValueError("MWAA requires at least 2 PRIVATE_WITH_EGRESS subnets")
network_config = mwaa.CfnEnvironment.NetworkConfigurationProperty(
security_group_ids=[mwaa_sg.security_group_id],
subnet_ids=private_subnets,
)
Logging Configuration
logging_config = mwaa.CfnEnvironment.LoggingConfigurationProperty(
dag_processing_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty(
enabled=True, log_level="INFO"
),
scheduler_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty(
enabled=True, log_level="INFO"
),
task_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty(
enabled=True, log_level="INFO"
),
webserver_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty(
enabled=True, log_level="INFO"
),
worker_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty(
enabled=True, log_level="INFO"
),
)
MWAA Environment
mwaa_env = mwaa.CfnEnvironment(
self,
"AirflowEnvironment",
name=airflow_env_name,
airflow_version="2.5.1", # Using a stable version
environment_class="mw1.small", # Using a more reliable size for production
execution_role_arn=mwaa_role.role_arn,
source_bucket_arn=dags_bucket.bucket_arn,
dag_s3_path="dags",
network_configuration=network_config,
logging_configuration=logging_config,
webserver_access_mode="PRIVATE_ONLY", # You can change to PUBLIC_ACCESS if needed
min_workers=1,
max_workers=2,
)
Airflow Config Options
mwaa_env.add_override(
"Properties.AirflowConfigurationOptions",
{
"core.load_default_connections": "False",
"core.load_examples": "False",
"core.default_timezone": "utc",
"scheduler.min_file_process_interval": "30",
"webserver.dag_default_view": "tree",
"webserver.dag_orientation": "TB",
},
)
Add dependencies
mwaa_env.node.add_dependency(dags_bucket)
mwaa_env.node.add_dependency(mwaa_role)
Outputs
CfnOutput(
self,
"MwaaWebserverUrl",
value=f"https://{mwaa_env.attr_webserver_url}",
description="MWAA Webserver URL"
)
CfnOutput(
self,
"DagsBucketName",
value=dags_bucket.bucket_name,
description="S3 bucket for DAGs"
)
Key improvements in this code:
Created a dedicated VPC with proper public and private subnets
Added more comprehensive IAM permissions for the execution role
Used a more stable Airflow version (2.5.1)
Increased the environment class to mw1.small for better reliability
Added additional Airflow configuration options
Added outputs to easily access the webserver URL and DAGs bucket
Made the bucket name unique with account and region to avoid conflicts
Added more comprehensive logging permissions
for above code got
airflow failed: ToolkitError: Failed to create ChangeSet cdk-deploy-change-set on badger-airflow: FAILED, The following hook(s)/validation failed: [AWS::EarlyValidation::PropertyValidation]. To troubleshoot Early Validation errors, use the DescribeEvents API for detailed failure information.
suggest for cdk code
