DEV Community: Ammar Yasser

Passed AWS SAP, Here's my 2 cents about it

Ammar Yasser — Fri, 16 Sep 2022 13:05:45 +0000

I'd like to start with saying that my opinions are my own. feel free to disagree with me. that being said.. wish you a good read.

The usual debate about certifications 😖😒

You might look at the study time and resources it will take you to finish a certification like this and think wow, the industry is truly in a bad state for allowing such a phenomenon to appear. And i will tell you that sorry, this is just how the world works. We sometimes have a very difficult decision to make while selecting a leader, hiring an employee, choosing a major... etc. and we simply don't have enough data to make a choice. and so what we do is look for proxies. Same as what happens with coding interviews. the interviewer is looking for a proxy for you being a good software engineer and you solving the coding problem means there's less chance that you are a bad hire -although obviously it's not a zero chance-.

Certification will always matter in IT especially because there are new tools appearing all the time and it's hard to gauge how experienced a candidate is in this tool. people may lie on their resume, or make up stories about problems they solved.. but they can't fake passing an exam.

Why the SA pro is different 🤔

The SA pro is like all other certifications, a proxy. but it's a very good proxy and its a proxy for so many great skills that you will benefit from on your job. The exam is really difficult and this is a good thing. you won't be able to pass without a firm grasp of networking, automation, security, encryption and so many other topics.
It's also a proxy for your ability to learn and study, which is one of the most important skill you need as someone in tech. And so it means that it's successful in its job as a proxy and does indeed prove good things about its holders.

Play for the love of the sport. not to win 💡😎

The certificate itself doesn't matter. but its the knowledge you gained while studying for it. Taking the exam and passing just helps you feel the reward of the journey and being able to tell the world how far you stand in your journey, But it isn't the true gain.
With that being said, don't prepare for the cert with the cert as your end goal, You'll meet products that you may have not used before while learning. Use them, implement an architecture with them. open the product page at the console and see what different options you can choose while provisioning it.
As a professional level solutions architect it's very normal that you get asked by your employer to implement a highly available hybrid directory solution, to build a disaster recovery plan with very few minutes of RPO for a few hundred terabytes database, or to preform zero downtime migrations of very complex legacy systems. will your cert matter when find yourself unable to preform these tasks in real life ?

How to actually pass ? 💯

For practice exams i used the ones by tutorialsdojo and whizlabs. A special mention has to go to Adrian Cantrill's exquisite course and how well put it is. But mind you.. the course is a whopping 75 hours. and so the secret lies in doing it everyday. Find a way to embed your study into your routine and do it EVERYDAY!!
You don't have to do alot on a single day but being consistent with it is in my opinion the only way to beat this daunting challenge.
Also, you must be able to pinpoint your knowledge gaps, after a failed practice exam go through your incorrect answers and find out what wrong assumptions you had while answering the question and then take note of the correct fact about the product. and iterate.

Architectures to build as your prepare

You need hands on experience as you prepare, these are some architectures that i believe are essential for you to have built before the exam. Let me know your ideas in the comments:

Build a three tier architecture and create failover DNS records to another region
Serverless application using API gateway, Lambda, DynamoDB. Bonus points if you integrate a queuing service ⏩
Launch an RDS database with some dummy data, then migrate it to another region using DMS 🎒
Implement hybrid DNS between two VPCs.
Connect VPCs in different ways, try a Transit gateway, try Site-to-Site VPN, then try a peering connection.🚧
Create an organization and configure a monitoring account with an organizational trail 👀
Launch random EC2 instances with different types and AMI's then evaluate compliance using a config rule 🛑
Deploy a dummy app with Elastic Beanstalk using different strategies 🌱

Bonus points when you only use cloudformation. feel free to reach out to me if you're trying to build these 💌

Elastic, Scalable, Secure and highly available three tier app using AWS CDK and Python

Ammar Yasser — Fri, 08 Jul 2022 11:19:21 +0000

Introduction

The AWS CDK opens the possibilities for adding wonderful features to your IaC and help bring it to life. Today we explore how we can use it to quickly bootstrap a high security application that can be reconfigured and customized for many use cases.

For this project we use the AWS CDK, a framework by AWS for defining your infrastructure as actual code in your programming language of choice. If you're new to the CDK then the workshop is the best introduction you can get for it.

Outcome

Creating an infrastructure that is redundant across multiple availability zones, end users can only access the app through a domain name, the database should only be accessible from the application servers. a secure way to access the servers for troubleshooting must be provided. Networking resources should be managed separately

Networking Infrastructure

For this project we have two files, A file where the network is defined and another where the app is defined on the network created. Let's look at the network file and we'll explain it bit by bit

from aws_cdk import (
    Stack,
    aws_ec2 as ec2
)
from constructs import Construct


class vpcStack(Construct):

    def __init__(self, scope: Construct, construct_id: str, vpc_cidr='10.0.0.0/16', three_tier=True,**kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        ''' AWS CDK Construct that creates a configurable VPC for hosting a three tier architecture'''

First we ask the question of what is this Construct class that we use to define our VpcStack Class ?
A construct is a resource, when you define an EC2 Instance, a Lambda function, a DynamoDB table.. etc in a CDK template we're defining what is known as a construct. We can create our custom resources by inheriting from this class and defining the actual resources we want in this new object as attributes of our class! Quite revolutionary right ? We will see how we implement this soon.

For now, Our class is instantiated using two extra custom parameters we provide for flexibility. vpc_cidr and three_tier.
The VPC is created by default with a 10.0.0.0/16 CIDR. but parametrizing this will allow us to later pass the CIDR as a command line argument. three_tier is a boolean value for conditional infrastructure creation, in case you have an existing database you may not want to create the needed resources for it.

if three_tier:
    self.vpc = ec2.Vpc(
            self, 'vpc',
            cidr=vpc_cidr,
            max_azs=2,
            nat_gateways=2,
            enable_dns_hostnames=True,
            enable_dns_support=True,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name= 'load_balancer',
                    subnet_type= ec2.SubnetType.PUBLIC,
                    cidr_mask= 24
                ),
                ec2.SubnetConfiguration(
                    name= 'application',
                    subnet_type= ec2.SubnetType.PRIVATE_WITH_NAT,
                    cidr_mask= 24
                ), 
                ec2.SubnetConfiguration(
                    name= 'database',
                    subnet_type= ec2.SubnetType.PRIVATE_ISOLATED,
                    cidr_mask= 24
                )
            ]
        )
else: 
    self.vpc = ec2.Vpc(
            self, 'vpc',
            cidr=vpc_cidr,
            max_azs=2,
            nat_gateways=2,
            enable_dns_hostnames=True,
            enable_dns_support=True,
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name= 'load_balancer',
                    subnet_type= ec2.SubnetType.PUBLIC,
                    cidr_mask= 24
                ),
                ec2.SubnetConfiguration(
                    name= 'application',
                    subnet_type= ec2.SubnetType.PRIVATE_WITH_NAT,
                    cidr_mask= 24
                )
            ]
        )

In case we don't need a three tier app. our construct will create 2 subnets per AZ in the VPC. one will host the load balancer and is a public subnet. the other is private subnet with access to a NAT gateway. it's important to note here that if we were using any other IaC tool we would need to define an internet gateway, a NAT gateway, route tables, routes and route table associations. all of this is automated away with a few simple lines of code

Next, let's define our security groups

# Load balancer security group
self.lb_sg = ec2.SecurityGroup(
    self, 'lb-SG',
    vpc=self.vpc,
    allow_all_outbound=True
)

self.lb_sg.add_ingress_rule(
    ec2.Peer.any_ipv4(),
    ec2.Port.tcp(443)
)

self.lb_sg.add_ingress_rule(
    ec2.Peer.any_ipv4(),
    ec2.Port.tcp(80)
)


# Bastion security group
self.bastion_sg = ec2.SecurityGroup(
    self, 'bastion-SG',
    vpc=self.vpc,
    allow_all_outbound=True
)

self.bastion_sg.add_ingress_rule(
    ec2.Peer.any_ipv4(),
    ec2.Port.tcp(22)
)



# Application layer security group
self.asg_sg = ec2.SecurityGroup(
    self, 'asg-SG',
    vpc=self.vpc,
    allow_all_outbound=True
)

self.asg_sg.add_ingress_rule(
    self.lb_sg,
    ec2.Port.tcp(443)
)

self.asg_sg.add_ingress_rule(
    self.lb_sg,
    ec2.Port.tcp(80)
)

self.asg_sg.add_ingress_rule(
    self.bastion_sg,
    ec2.Port.tcp(22)
)



# Database security group if three tier is specified
if three_tier:
    self.db_SG = ec2.SecurityGroup(
        self, 'db-SG',
        vpc=self.vpc,
        allow_all_outbound=True
    )

    self.db_SG.add_ingress_rule(
        self.asg_sg,
        ec2.Port.tcp(5432)
    )

Note how all resources are created as attributes of the class. this is so we can access all resources individually later in the stack. for now we want our load balancer to be internet facing, and the application to receive web traffic only from the load balancer and SSH traffic from a bastion server. Bastion servers are a central point of access for SSH traffic, this limits the attack surface for your network and heavily restricts traffic to your servers.
Source targets con be written as other security groups as we see, we don't need to specify an ID, simply passing the construct of other security groups is good enough to write a valid ingress rule.

You can read more about the numerous security benefits from this setup here. and also more on bastion hosts here.

Web Layer

from aws_cdk import (
    Stack,
    aws_ec2 as ec2,
    aws_autoscaling as autoscaling,
    aws_elasticloadbalancingv2 as elbv2,
    aws_iam as iam,
    aws_rds as rds,
    Duration
)
from constructs import Construct
from .vpc_stack import vpcStack


class AppStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, three_tier=True, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Attempt to get custom cidr from context if provided
        context_cidr = self.node.try_get_context("cidr")

        # Creating the networking resources
        network = vpcStack(self, 'vpc', three_tier=three_tier, vpc_cidr=context_cidr if context_cidr else '10.0.0.0/16')

As we see, the app it self is inheriting from the Stack class, unlike the networking resources. This Stack overflow answer clearly explains the difference between the two.
Here we can see the benefit from having added vpc_cidr as a property. we can use the context variable to pass it dynamically during deployment like this:
cdk deploy -c cidr=172.0.0.0/16. if it's not passed the default of 10.0.0.0/16 will be used. learn more about the CDK context in this link.

Now, let's define our servers..

# Define our server execution role
        asg_role = iam.Role(self, "MyRole",
                assumed_by=iam.ServicePrincipal("ec2.amazonaws.com")
            )         
        asg_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonEC2RoleforSSM"))

        # Define our servers
        asg = autoscaling.AutoScalingGroup(self, "ASG",
                vpc=network.vpc,
                instance_type=ec2.InstanceType.of(ec2.InstanceClass.BURSTABLE3, ec2.InstanceSize.MICRO),
                machine_image=ec2.GenericLinuxImage({'us-east-1':'ami-0cff7528ff583bf9a'}),
                security_group=network.asg_sg,
                cooldown=Duration.minutes(1),
                min_capacity=2,
                max_capacity=8,
                role=asg_role,
                vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_NAT),
                user_data=ec2.UserData.custom(
                '''
                #!/bin/bash
                sudo yum update -y
                sudo yum install -y httpd.x86_64
                sudo systemctl start httpd
                sudo systemctl enable httpd
                sudo chmod 755 -R /var/www/html
                sudo echo "hello world from $(hostname -f)" >> /var/www/html/index.html
                '''
                )
            )

We want our servers in private subnets but not fully isolated, which is why we specify subnets with a route to the NAT gateway for outbound only connection to the internet.
We want them to be granted permissions to communicate with AWS SSM for secure management in case we don't want to use the bastion.

Also, we define our base OS using the GenericLinuxImage class despite that it's amazon linux. and amazon linux can be defined using a special class called AmazonlinuxImage. the reason for this is that using the amazon linux image class returned a very primitive Linux image that doesn't have systemctl, grep and other important linux modules so we use the default AMI ID for Amazon Linux in the us-east-1 region.

We also configure our servers to show a simple page for users, you can change the user data script to be anything else you want to deploy your own apps, such as cloning a repo and starting a nodeJS server. you can also create a machine image and deploy it instead of using Amazon Linux with user data

# Creating the bastion server
bastion = ec2.BastionHostLinux(self, "BastionHost",
        vpc=network.vpc,
        security_group=network.bastion_sg,
        subnet_selection=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC)
    )

# Configure load balancer
lb = elbv2.ApplicationLoadBalancer(self, "LB",
        vpc=network.vpc,
        internet_facing=True,
        security_group=network.lb_sg
    )

listener = lb.add_listener("Listener",
            port=80
    )

listener.add_targets("ApplicationFleet",
    port=80,
    targets=[asg]
    )

First, we define a bastion server using the handy BastionHostLinux class. this host has the security group that is the only allowed source for port 22 communication with the servers.
We define a load balancer with the application servers as targets when receiving HTTP traffic. many other properties can be defined depending on your application use case, such as session stickiness, connection draining period.. etc. this Page from the docs is a good summary of what you can do.

Data Layer

Again, only created if three_tier option is specified..

if three_tier:
# in case a three tier architecture is specified, create a database and a
# secrets manager secret, and grants get secret value role to the 
# application fleet
db_secrets = rds.DatabaseSecret(self, 'postgres-secret',
            username='postgres',
            secret_name='postgres-credentials'
            )

db = rds.DatabaseInstance(self, "db",
            engine=rds.DatabaseInstanceEngine.postgres(version=rds.PostgresEngineVersion.VER_13_4),
            instance_type=ec2.InstanceType.of(ec2.InstanceClass.BURSTABLE3, ec2.InstanceSize.MICRO),
            credentials=rds.Credentials.from_secret(db_secrets),
            vpc=network.vpc,
            multi_az=True,
            security_groups=[network.db_SG],
            vpc_subnets=ec2.SubnetSelection(
                subnet_type=ec2.SubnetType.PRIVATE_ISOLATED
            )
            )

asg.add_to_role_policy(iam.PolicyStatement(
    resources=[db_secrets.secret_arn],
    actions=[
            "secretsmanager:GetSecretValue"]
    ))

The database lives in a private isolated subnet, the only open connection is on the default engine port which is PostgreSQL in our case, and it's only open to the application layer's security group. we also define the database password as a secrets manager secret. and giving the application layer the ability to only retrieve the value of this secret only. In accordance with the least privilege principle. read more about AWS security best practices here

Conclusion

Application availability, security, quick and ease of deployment are no longer a hassle and are attainable for all organizations. and this is only the start...
this infrastructure can be easily turned to one which can stand regional failure. we can replace postgreSQL with Aurora global and configure DNS for failover routing. we can also enhance observability by exporting bastion SSH logs to Cloudwatch, and many other things.. I'm actually thinking of turning this into a mini series where i take this infrastructure and take it to its limits according to SRE principals. Let me know what you people think in the comments.

All the code for this article is available in this repo

Serverless ETL using AWS Lambda, Pandas, PostgreSQL, AWS CDK

Ammar Yasser — Sat, 25 Jun 2022 16:55:12 +0000

Introduction
Compute
Infrastructure as code
Continuous delivery
Conclusion

Introduction

Created in response to the #ACloudGuruChallenge posted by Forrest Brazeal.

Modern data infrastructure has become so exciting and with limitless possibilities, Today we explore a use case where we pull, transform and visualize covid-19 data from several sources in a serverless and extremely quick to deploy fashion using python as our tool of choice

Outcome

Having a job that downloads data about covid-19 for case counts, deaths and recoveries for every given day, and stores them in a database for later access and reporting

Tech stack

Compute: Python application on AWS Lambda
Storage: S3
Database: PostgreSQL on RDS
Reporting: Amazon Quicksight
IaaC and CI/CD: Github actions + AWS CDK

Data Sources

We pull data from This link and this John Hopkins dataset

Compute

AWS Lambda is a serverless platform for running small snippets of code in the cloud, in this section we will go over the code used to achieve the end goal

import boto3  
import requests  
import pandas as pd  # Creating a dataframe to store our data
import io  
import json  
import os  # accessing environment variables 
from db import get_latest_date, instantiate_db, load_to_db  # custom modules, will be discussed below
from data_transformation import join_dfs, change_to_datetime, clean_dataframe, rename_cols

And now we discuss the functions defined in our handler


# Instantiate our clients
s3_client = boto3.client('s3')
s3_resource = boto3.resource('s3')
sns_client = boto3.client('sns')

def download_file(url, file_name):
    ''' Downloads data from a file and saves it to s3 as an object'''
    body = requests.get(url).content.strip(b'\n')
    bucket_name = os.environ['S3_BUCKET']
    s3_path = "data/" + file_name
    s3_resource.Bucket(bucket_name).put_object(Key=s3_path, Body=body)


def load_data(key):
    ''' Loads data from s3 into a pandas Dataframe'''
    obj = s3_client.get_object(Bucket=os.environ['S3_BUCKET'], Key=key)
    csv_file = obj['Body'].read().strip(b'\n')
    df = pd.read_csv(io.BytesIO(csv_file), low_memory=False)
    return df

def send_db(df):
    ''' Sends the data to the database that only has a date higher than
    current latest in the db'''
    latest = pd.to_datetime(get_latest_date())
    data_to_send = df[df['Date'] > latest]
    data_to_send.apply(lambda x: load_to_db(x), axis=1)
    return len(data_to_send)

def post_to_sns(message):
    ''' Sends the output of the job to an SNS topic which fans out
    the result to any number of interested message consumers'''
    response = sns_client.publish(
        TargetArn=os.environ['SNS_TOPIC'],
        Message=json.dumps({'default': json.dumps(message)}),
        MessageStructure='json'
    )
    return response

Our workflow starts with instantiating boto3 clients for communicating with other AWS services, its best practice to start your clients outside of the handler to speed up the execution of your function, refer to Lambda best practices Here

Then we define functions to handle downloading the data and saving it to a storage platform, it's also a best practice to not hard code your s3 bucket name into you functions. especially in our case because we use infrastructure as code that creates the bucket on runtime, we don't know the bucket name in advance

We then have a function for loading our final dataframe into the database

def send_db(df):
    ''' Sends the data to the database that only has a date higher than
    current latest in the db'''
    latest = pd.to_datetime(get_latest_date())
    data_to_send = df[df['Date'] > latest]
    data_to_send.apply(lambda x: load_to_db(x), axis=1)
    return `len(data_to_send)`

Let's discuss this function in detail, this job runs daily and so we only want to post the new days into the database instead of loading the entire data everytime, to handle this we query the database for its current latest date. and we apply load_to_db(record) only if this record has a date higher than the latest.

this way we also have unified logic for initial loading or daily runs. and in the end we want to know how many records we processed by returning len(data_to_send)

Lastly, we define a function that posts the results of our run to SNS. we will see how we use all these functions in the handler after we define the functions in the other modules

Lets inspect out data_transformation module first

def join_dfs(df, jh, col='Date'):
    ''' Takes in two dataframes and preforms a left join
    that happens on the Date column, change the col argument
    for a different column'''
    final = df.merge(jh, on=col, how='left')
    return final


def change_to_datetime(df, col='Date'):
    ''' Changes a column in a dataframe to type datetime'''
    df[col] = df[col].astype('datetime64[ns]')
    return

def rename_cols(df, names_dict):
    ''' Renames a group of columns in a df by passing a dict of old names and new names  '''
    df.rename(columns=names_dict, inplace=True)
    return


def clean_dataframe(jh, country, columns, col='Country/Region'):
    ''' Filters a dataframe column from entries that don't
    match a specific condition and removes needless columns,
    used for cleaning the John Hopkins data'''
    jh.drop(jh[jh[col] != country].index, inplace=True)
    jh.drop(columns, axis=1, inplace=True)
    return

Our end dataframe should combine the case counts and deaths in the first data source to the recoveries column in the second data source. the second data source also contains data from many countries, for the sake of our case we only want USA data. and so we define functions for dropping the columns and data we don't need, unifying the naming structure and data types in both dataframes, and finally join them together.

And now, let's look at our database module

import psycopg2 #postgres client
import boto3
import json

sm = boto3.client('secretsmanager')

secret = sm.get_secret_value(SecretId='postgres-credentials')
secrets_dict = json.loads(secret['SecretString'])

def connect():
    ''' Instantiates a connection to the db'''
    conn = psycopg2.connect(
        host=secrets_dict['host'],
        database='postgres',
        user=secrets_dict['username'],
        password=secrets_dict['password'])
    return conn

def instantiate_db():
    conn = connect()
    cursor = conn.cursor()
    try:
        cursor.execute('SELECT * FROM covid19 LIMIT 1;')
        return
    except UndefinedTable:
        print('creating table')
        cursor.execute(
            '''CREATE TABLE covid19 (
                date TIMESTAMP PRIMARY KEY,
                cases NUMERIC,
                deaths NUMERIC,
                recovered NUMERIC
            );

            INSERT INTO covid19 (date, cases, deaths, recovered) VALUES ('2019-01-01', 0, 0, 0);'''
        )
        conn.commit()
        cursor.close()
        conn.close()


def get_latest_date():
    ''' Returns the current ltest record in the db'''
    conn = connect()
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM covid19 ORDER BY date DESC LIMIT 1;')
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results[0][0]


def load_to_db(row):
    ''' Insert new rows to the db'''
    conn = connect()
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO covid19 (date, cases, deaths, recovered) VALUES (%s, %s, %s, %s)', (row['Date'], row['cases'], row['deaths'], row['Recovered'])
        )
    conn.commit()
    cursor.close()
    conn.close()

Firstly, we don't have the database credentials hardcoded or even known ahead of deploying. we use AWS's secrets manager to create a secure password for us and retrieve it from the API at runtime.
This approach is extremely better for security as the credentials are never visible to anyone and only available during processing

After having retrieved the credentials we instantiate a connection using psycopg2, the python API for PostgreSQL

def connect():
    ''' Instantiates a connection to the db'''
    conn = psycopg2.connect(
        host=secrets_dict['host'],
        database='postgres',
        user=secrets_dict['username'],
        password=secrets_dict['password'])
    return conn

We return the connection object as its essential for any database operation, this function is a dependency for all other functions in the module.

Second, we have a function for creating our table. this function will first attempt to query for the existence of the table and return a random record from it. if it fails to do so due to UndefinedTable it will create the table and insert a dummy record that dates back to before the data in our sources. this is important for the get_latest_date() function to work properly, because if it retrieves this dummy record, initial loading is performed.

lastly, we have the actual function that loads a record into the database. this function expects a dataframe row as input.

The justification behind this approach is that dataframe operations that involve iteration are slow according to this Stackoverflow question.
One of the alternatives suggested is to use df.apply(your_function), which is how we formulate our approach.

And now, we look at the actual handler code that executes at invocation of the function

def lambda_handler(event, context):
    try:
        download_file('https://raw.githubusercontent.com/nytimes/covid-19-data/master/us.csv', 'data.csv')
        download_file('https://raw.githubusercontent.com/datasets/covid-19/master/data/time-series-19-covid-combined.csv', 'john_hopkins.csv')
        df = load_data('data/data.csv')
        jh = load_data('data/john_hopkins.csv')
        rename_cols(df, {'date':'Date'}) #unify naming method
        clean_dataframe(jh, 'US', ['Deaths', 'Country/Region' ,'Province/State',  'Confirmed']) # remove needless columns and non US data
        change_to_datetime(df)
        change_to_datetime(jh)
        final = join_dfs(df, jh)
        instantiate_db() # create the table if it doesnt exist
        data_sent = send_db(final) 
        res = post_to_sns('Job ran successfully, Updated {} records in the database'.format(data_sent)) # report if successful
    except Exception as e:
        res = post_to_sns('Job failed with error: {}'.format(e)) # report errors

Infrastructure as Code

For this project we use the AWS CDK, a framework by AWS for defining your infrastructure as actual code in your programming language of choice, which is worth saying that it is a true wonder of engineering and i had so much fun using it.
However it is worth saying that navigating the docs was extremely tedious and some of the processes are not straight forward. for the sake of keeping the project entirely in Python we deploy CDK in python flavor.

Resources in a CDK stack are defined with the following parameters:

the construct that will use the resource
an identifier
any properties specific to the resource

Learn more about the CDK in this workshop

Let's take a look at our CDK stack

class CdkStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Define Lambda layers
        pandas = lambda_.LayerVersion.from_layer_version_attributes(self, 'Pandas',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-pandas:4")

        requests = lambda_.LayerVersion.from_layer_version_attributes(self, 'Requests',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-requests-html:4")

        psycopg = lambda_.LayerVersion.from_layer_version_attributes(self, 'Psycopg',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-psycopg2-binary:1")

Before anything, Huge shoutout to Keith Rozario for providing us with this awesome Repo.
where he defined layers for lambda functions to add libraries that doesn't exist by default in the lambda runtime without you having to create them yourself. I used his repo to add the requests, pandas and psycopg2 libraries to the project.

Next up, let's define our database..

# instantiate DB credentials using secrets manager
        db_secrets = rds.DatabaseSecret(self, 'postgres-secret',
                    username='postgres',
                    secret_name='postgres-credentials'
                    )

        # Create the database
        db = rds.DatabaseInstance(self, "db",
                    engine=rds.DatabaseInstanceEngine.postgres(version=rds.PostgresEngineVersion.VER_13_4),
                    instance_type=ec2.InstanceType.of(ec2.InstanceClass.BURSTABLE3, ec2.InstanceSize.MICRO),
                    credentials=rds.Credentials.from_secret(db_secrets),
                    vpc=ec2.Vpc(self, 'vpc'),
                    vpc_subnets=ec2.SubnetSelection(
                        subnet_type=ec2.SubnetType.PUBLIC
                    )
                    )

db.connections.allow_default_port_from_any_ipv4()

We instantiate the database secret using Secrets Manager and then create the database using this secret using the rds.Credentials.from_secret(db_secrets) method, we also launch this DB in a public subnet for easier reach from AWS Lambda.
However in a production setting and in case the database is going to hold sensitive data its best to launch it in a private subnet and configure Lambda to work in your private network, check this article for more details.

And now, let's create our S3 bucket and SNS topic. pretty straight forward

# Create the bucket used to store the data
        s3_bucket = s3.Bucket(self, 'dataBucket')

        # Create the delivery topic
        topic = sns.Topic(self, 'deliveryTopic')

And defining our actual function with appropriate permissions..

lambda_role = iam.Role(self, "lambda_role",
                    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
                    )
        lambda_role.add_to_policy(iam.PolicyStatement(
                    resources=["*"],
                    actions=[
                            "s3:GetObject",
                            "s3:PutObject",
                            "sns:Publish",
                            "secretsmanager:GetSecretValue"]
                    ))
function = lambda_.Function(self, "Serverless-ETL",
                    runtime=lambda_.Runtime.PYTHON_3_9,
                    code=lambda_.Code.from_asset("./code"),
                    handler="lambda_function.lambda_handler",
                    layers=[pandas, requests, psycopg],
                    timeout=Duration.minutes(5),
                    role=lambda_role,
                    memory_size=512,
                    environment={
                        'S3_BUCKET':s3_bucket.bucket_name,
                        'SNS_TOPIC': topic.topic_arn
                    }
                    )
# allow our function to post to sns
topic.grant_publish(function)

As we see here, the bucket name and SNS topic ARN are passed to the function on deployment through its environment variables. and we load the lambda code saved in a folder in the directory called "code" using the method lambda_.Code.from_asset("./code"). we also configure the function with 512MB of memory, this is because this function instantiates several clients and they take up alot of memory to work with.

Lastly, we define the Eventbridge rule to trigger our function

        # Create the event rule and schedule
        rule = events.Rule(self, "Rule",
                    schedule=events.Schedule.expression('cron(0 0 * * ? *)'),
                    )        
        rule.add_target(targets.LambdaFunction(function))

Continuous Delivery

The beauty of using the AWS CDK as your tool of choice for deploying is its simplicity in use and maintaining of infrastructure state using Cloudformation changesets. You can read more about them here

The only commands we need to launch our infrastructure are cdk diff to detect changes and cdk deploy to commit changes. we can wrap these commands in a nice and easy Github actions workflow as after having configured your AWS access keys in Github. here's a link explaining how its done.


# run only on commits to main
on:
  push:
    branches: [ "main" ]
  pull_request:
    branches: [ "main" ]

  workflow_dispatch:


jobs:
  aws-cdk:
    runs-on: ubuntu-latest
    steps:

      - name: Check out repository code
        uses: actions/checkout@v3

      - name: cdk diff
        uses: youyo/aws-cdk-github-actions@v2
        with:
          cdk_version: 2.28.1
          cdk_subcommand: 'diff'
          cdk_stack: CdkStack
          actions_comment: false
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: 'us-east-1'


      - name: cdk deploy
        uses: youyo/aws-cdk-github-actions@v2.1.1
        with:
          cdk_version: 2.28.1
          cdk_subcommand: 'deploy'
          cdk_stack: 'CdkStack'
          cdk_args: '--require-approval never'
          actions_comment: false
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: 'us-east-1

This workflow uses actions from the actions marketplace, find out more here. For every new commit to change our code for the function or the infrastructure, the CDK CLI will run and show us what changed and will deploy these changes to your account.

In the end i hooked up Amazon Quicksight to my RDS database to extract insights on the data. RDS integrates with Quicksight very seamlessly

Conclusion

Modern cloud infrastructure is being pushed to entirely new limits and possibilities and its only up to your imagination how you will use it. I have enjoyed a lot and learned so much while creating this. all the code for this article can be found in this Repo

DEV Community: Ammar Yasser

Passed AWS SAP, Here's my 2 cents about it

The usual debate about certifications 😖😒

Why the SA pro is different 🤔

Play for the love of the sport. not to win 💡😎

How to actually pass ? 💯

Architectures to build as your prepare

Elastic, Scalable, Secure and highly available three tier app using AWS CDK and Python

Table of Contents

Introduction

Outcome

Networking Infrastructure

Web Layer

Data Layer

Conclusion

Serverless ETL using AWS Lambda, Pandas, PostgreSQL, AWS CDK

Table of contents

Introduction

Outcome

Tech stack

Data Sources

Compute

Infrastructure as Code

Continuous Delivery

Conclusion