Forem: Danny Kay

Getting up and running on AWS EKS Fargate

Danny Kay — Wed, 11 May 2022 16:09:08 +0000

Introduction

EKS Fargate has been on my "investigate" list in Notion for quite some time now. With some spare time the other weekend I thought I'd do some digging and see if I could get something mega basic but repeatable up and running for if I need to use EKS Fargate for a proof of concept or something similar.

This post walks through how to get a basic application running on EKS Fargate complete with logging in CloudWatch.

Want to follow along?

Alrighty, so before we do anything constructive the repository for all the bits and bobs I'm showing you below can be found here. Some of the code in the snippets has been left out for brevity.

Be wary, AWS is fun but that fun comes at a cost. Following this will cost you some dough so keep an eye on the Billing & Cost Management Dashboard people!

This post assumes you understand the basics of AWS, Kubernetes and Terraform as I won't be going in to deep on these subjects.

I'll admit, my Terraform could be better, this isn't production worthy Terraform code in anyway or form, just me hacking some things together :)

Some of my examples are based off the official Terraform EKS Examples Github repository which can be viewed here.

Let's Get Started

So before we can even think about doing anything we need somewhere for the Fargate nodes to run i.e. Subnets within a VPC....

module "vpc" {
  source  = "terraform-aws-modules/vpc/aws"
  version = "~> 3.0"

  name = local.name
  cidr = "10.0.0.0/16"

  azs             = ["${local.region}a", "${local.region}b", "${local.region}c"]
  private_subnets = ["10.0.1.0/24", "10.0.2.0/24", "10.0.3.0/24"]
  public_subnets  = ["10.0.4.0/24", "10.0.5.0/24", "10.0.6.0/24"]

  enable_nat_gateway   = true
  single_nat_gateway   = true

  enable_dns_hostnames = true

  public_subnet_tags = {
    "kubernetes.io/cluster/${local.name}" = "shared"
    "kubernetes.io/role/elb"              = 1
  }

  private_subnet_tags = {
    "kubernetes.io/cluster/${local.name}" = "shared"
    "kubernetes.io/role/internal-elb"     = 1
  }

  tags = local.tags
}

So here we are utilising the VPC Terraform module which allows us to spin up a basic network structure for our EKS Cluster.

Next we can use the EKS Terraform module for the EKS Cluster...

module "eks" {
  source                          = "terraform-aws-modules/eks/aws"
  version                         = "~> 18.0"

  cluster_name                    = local.name
  cluster_version                 = local.cluster_version
  cluster_endpoint_private_access = true
  cluster_endpoint_public_access  = true

  vpc_id     = module.vpc.vpc_id
  subnet_ids = module.vpc.private_subnets
}

resource "null_resource" "kubectl" {
  depends_on = [module.eks]
  provisioner "local-exec" {
    command = "aws eks --region ${local.region} update-kubeconfig --name ${local.name}"
  }
}

Utilising the EKS Terraform Module means we can get a EKS Cluster up and running super quick without having to get bogged down in Terraform code.

In the above code snippet we are also using the null_resource to be able to run a command which updates the local kubeconfig so we can use kubectl commands against the EKS Cluster as soon as the Terraform apply command has finished.

The outputs and providers file haven't been shown in this post but are available in Github Repository.

Right, now we can go ahead and deploy our VPC and EKS Cluster by running the below in a Terminal...

terraform init
terraform apply

Deploying the EKS Cluster can take some time so feel free to grab a Coffee or alternative beverage whilst it does it's thing.

Once Terraform has finished deploying everything you should see a success message followed by the ARN of the Cluster, this specific output has been specified in the outputs file...

Apply complete! Resources: 0 added, 0 changed, 0 destroyed.

Outputs:

cluster_arn = "arn:aws:eks:eu-west-2:012345678901:cluster/ex-eks-fargate"

To verify everything has gone according to plan we can use kubectl to interact with out newly created Cluster...

➜ kubectl get ns
NAME              STATUS   AGE
default           Active   3m
kube-node-lease   Active   3m
kube-public       Active   3m
kube-system       Active   3m

➜ kubectl get pods -n kube-system
NAME                       READY   STATUS    RESTARTS   AGE
coredns-679c9bd45b-f7x8b   0/1     Pending   0          3m
coredns-679c9bd45b-zw52k   0/1     Pending   0          3m

Hmm, that's interesting, the CoreDNS Pods are stuck in Pending...let's dig into that.

Look Ma, No Nodes

So for those of you have worked with the EKS Terraform Module before might have realised that the Worker Groups are missing from the EKS Terraform code.

Instead of using Worker Groups we utilise Fargate Profiles instead.

A Fargate Profile is used to determine which pods use Fargate when launched. The beauty of this is that you could have a EKS Cluster which comprises of Self Managed/AWS Managed nodes and Fargate Profiles, using the Fargate Profiles to run a specific workload.

So how does the Fargate Profile work?

The Profile contains selectors, if a Pod Deployment contains a selector that matches those in the Profile, it's deployed to Fargate.

A basic Fargate Profile and IAM configuration is shown here...

resource "aws_eks_fargate_profile" "default" {
  depends_on             = [module.eks]
  cluster_name           = local.name
  fargate_profile_name   = "default"
  pod_execution_role_arn = aws_iam_role.eks_fargate_profile_role.arn
  subnet_ids             = flatten([module.vpc.private_subnets])

  selector {
    namespace = "kube-system"
  }

  selector {
    namespace = "default"
  }
}

resource "aws_iam_role" "eks_fargate_profile_role" {
  name = "eks-fargate-profile-role"

  assume_role_policy = jsonencode({
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        Service = "eks-fargate-pods.amazonaws.com"
      }
    }]
    Version = "2012-10-17"
  })
}

resource "aws_iam_role_policy_attachment" "AmazonEKSFargatePodExecutionRolePolicy" {
  policy_arn = "arn:aws:iam::aws:policy/AmazonEKSFargatePodExecutionRolePolicy"
  role       = aws_iam_role.eks_fargate_profile_role.name
}

We give the Profile a name, specify an Execution Role and specify the Subnets we want our Pods deployed into amongst other attributes.

The Selector is responsible for selecting Pods which are deployed to the kube-system and default namespace and running them on Fargate Nodes.

On the IAM side of things, we create a IAM Role and attach the AmazonEKSFargatePodExecutionRolePolicy to it so it can download container images from ECR and attach the IAM Role to the Fargate Profile as shown in the above code. More information regarding the IAM side of things can be viewed here.

Right, now we can go ahead and deploy our Fargate Profile and associated IAM configuration by running terraform apply as we did in the last section and verifying that the operation was successful.

But, as you can see below, the CorsDNS Pods are still Pending...

➜ kubectl get pods -n kube-system
NAME                       READY   STATUS    RESTARTS   AGE
coredns-679c9bd45b-m948h   0/1     Pending   0          2m
coredns-679c9bd45b-wwlfn   0/1     Pending   0          2m

Lets dig into why that's happening.

So Close, but we have something to "Patch" up

So we have our EKS Cluster, a Fargate Profile which is setup to run Pods in the kube-system namespace...what else do we need to do?

By default, the CoreDNS deployment is configured to run on Amazon EC2 instances, we need to change that.

If we run the describe kubectl command against one of the CoreDNS Pods we'll see something interesting...

➜  eks-fargate kubectl describe pod/coredns-6f6f94db9c-fjhjn -n kube-system
Name:                 coredns-6f6f94db9c-fjhjn
Namespace:            kube-system
Priority:             2000000000
Priority Class Name:  system-cluster-critical
Node:                 <none>
Labels:               eks.amazonaws.com/component=coredns
                      k8s-app=kube-dns
                      pod-template-hash=6f6f94db9c
Annotations:          eks.amazonaws.com/compute-type: ec2
                      kubectl.kubernetes.io/restartedAt: 2022-05-12T15:11:14+01:00
                      kubernetes.io/psp: eks.privileged
Status:               Pending

We can see that there is an eks.amazonaws.com/compute-type annotation with the value of ec2, we need to remove that annotation, but as we have no control over the Deployment files we'll have to run a patch command against the deployment followed by a rollout restart...

kubectl patch deployment coredns \
      --namespace kube-system \
      --type=json -p='[{"op": "remove", "path": "/spec/template/metadata/annotations", "value": "eks.amazonaws.com/compute-type"}]' \

kubectl rollout restart -n kube-system deployment coredns

The rollout restart command is needed to delete and re-create the existing CoreDNS pods so that they are scheduled on Fargate.

To make life slightly easier, we can place the above code in a script that is ran once the Fargate Profile has been created...

resource "null_resource" "fargate_patch_coredns" {
  depends_on = [aws_eks_fargate_profile.default]
  provisioner "local-exec" {
    command = "/bin/bash ./scripts/patch_coredns_deployment.sh"
  }
}

As in previous steps we can run terraform apply and wait for Terraform to do it's thing.

Fargate Finale

It will take around a minute for the CoreDNS deployment to be deployed onto Fargate after Terraform has finished applying...

➜  kubectl get pods -n kube-system
NAME                       READY   STATUS    RESTARTS   AGE
coredns-86dd9db845-m968g   1/1     Running   0          83s
coredns-86dd9db845-pk27d   1/1     Running   0          83s

Happy days! Let's dig a little deeper...

➜  kubectl get nodes
NAME                                               STATUS   ROLES    AGE   VERSION
fargate-ip-10-0-2-184.eu-west-2.compute.internal   Ready    <none>   89s   v1.22.6-eks-7d68063
fargate-ip-10-0-2-36.eu-west-2.compute.internal    Ready    <none>   84s   v1.22.6-eks-7d68063

So we can see that two Fargate nodes have been fired up, one for each CoreDNS Pod. How exciting!

If we dig into the logs of one of the Pods we can see the events that took place...

Events:
  Type     Reason           Age    From               Message
  ----     ------           ----   ----               -------
  Warning  LoggingDisabled  3m19s  fargate-scheduler  Disabled logging because aws-logging configmap was not found. configmap "aws-logging" not found
  Normal   Scheduled        2m28s  fargate-scheduler  Successfully assigned kube-system/coredns-86dd9db845-m968g to fargate-ip-10-0-3-190.eu-west-2.compute.internal
  Normal   Pulling          2m27s  kubelet            Pulling image "602401143452.dkr.ecr.eu-west-2.amazonaws.com/eks/coredns:v1.8.7-eksbuild.1"
  Normal   Pulled           2m25s  kubelet            Successfully pulled image "602401143452.dkr.ecr.eu-west-2.amazonaws.com/eks/coredns:v1.8.7-eksbuild.1" in 1.992500087s
  Normal   Created          2m25s  kubelet            Created container coredns
  Normal   Started          2m24s  kubelet            Started container coredns

Alright, great job!

Let's move onto to the next step which is getting the logs from the Pods into CloudWatch, so we can get rid of the Disabled logging because aws-logging configmap was not found error event.

What's going on in there?

In order for the logs of our Pods to be transported to CloudWatch, there are two things that need setting up on the Kubernetes side of things:

A Namespace
A ConfigMap

I won't go into what the two artifacts above look like as they can be viewed here, but rest assured these are also in the Github repository which can be viewed here.

Utilising a null_resource and a shell script the above mentioned files are deployed once the Fargate Profile has been configured...

resource "null_resource" "deploy_logging_artifacts" {
  depends_on = [null_resource.fargate_patch_coredns]
  provisioner "local-exec" {
    command = "/bin/bash ./logging/deploy.sh"
  }
}

One other thing that we need to do is assign a new policy to our Fargate Profile Role to grant permission to communicate with CloudWatch...

resource "aws_iam_role_policy_attachment" "cloudwatch_transport_policy_attachment" {
  policy_arn = aws_iam_policy.cloudwatch_transport_iam_policy.arn
  role       = aws_iam_role.eks_fargate_profile_role.name
}

resource "aws_iam_policy" "cloudwatch_transport_iam_policy" {
  name = "cloudwatch-transport-iam-policy"
  path = "/"
  policy = jsonencode({
    "Version" : "2012-10-17",
    "Statement" : [{
      "Effect" : "Allow",
      "Action" : [
        "logs:CreateLogStream",
        "logs:CreateLogGroup",
        "logs:DescribeLogStreams",
        "logs:PutLogEvents",
      ],
      "Resource" : "*"
    }]
  })
}

As in previous steps, we can run terraform apply and wait for Terraform to do it's thing.

As the CoreDNS Pods aren't the most chattiest of things, to verify the logging is working we can just delete a Pod...

kubectl delete pod/coredns-86dd9db845-pk27d -n kube-system

We can then jump into CloudWatch or use the amazing awslogs to view the output...

➜ awslogs get fluent-bit-cloudwatch --watch --aws-region eu-west-2

fluent-bit-cloudwatch from-fluent-bit-kube.var.log.containers.coredns-5679cff8b6-cmmw8_kube-system_coredns-6e7399229b358343d0469c28a9ca3beaf5646d066250e8c2b76c3b1208f54459.log .:53
fluent-bit-cloudwatch from-fluent-bit-kube.var.log.containers.coredns-5679cff8b6-cmmw8_kube-system_coredns-6e7399229b358343d0469c28a9ca3beaf5646d066250e8c2b76c3b1208f54459.log [INFO] plugin/reload: Running configuration MD5 = 47d57903c0f0ba4ee0626a17181e5d94
fluent-bit-cloudwatch from-fluent-bit-kube.var.log.containers.coredns-5679cff8b6-cmmw8_kube-system_coredns-6e7399229b358343d0469c28a9ca3beaf5646d066250e8c2b76c3b1208f54459.log CoreDNS-1.8.7
fluent-bit-cloudwatch from-fluent-bit-kube.var.log.containers.coredns-5679cff8b6-cmmw8_kube-system_coredns-6e7399229b358343d0469c28a9ca3beaf5646d066250e8c2b76c3b1208f54459.log linux/amd64, go1.17.7, a9adfd56

The log group name is set to fluent-bit-cloudwatch, this is defined in the ConfigMap we deployed at the start of this section.

Basic Application Up & Running

The official Kubernetes site has a basic application that we can try out on EKS Fargate, it can be viewed here.

For the purpose of this post I've created a small script which deploys the application once the EKS Fargate Terraform code and scripts have finished running. So when the application is deployed the Fargate Nodes will have been patched and the logging configuration will have been deployed...

#!/bin/bash

cd guestbook-app

kubectl apply -f https://k8s.io/examples/application/guestbook/redis-leader-deployment.yaml
kubectl apply -f https://k8s.io/examples/application/guestbook/redis-leader-service.yaml
kubectl apply -f https://k8s.io/examples/application/guestbook/redis-follower-deployment.yaml
kubectl apply -f https://k8s.io/examples/application/guestbook/redis-follower-service.yaml
kubectl apply -f https://k8s.io/examples/application/guestbook/frontend-deployment.yaml
kubectl apply -f https://k8s.io/examples/application/guestbook/frontend-service.yaml

As the Guestbook deployment has no namespace specified it will be deploy into the default one, as we specified the default namespace as a selector earlier on, we don't have to make any changes to accommodate the Guestbook.

Right, to verify everything works correctly I've destroyed the existing setup by running terraform destroy then running terraform apply as we did earlier.

When Terraform has finished applying we should be able to verify that the Guestbook has been deployed successfully...

➜ kubectl get pods                         
NAME                             READY   STATUS    RESTARTS   AGE
frontend-85595f5bf9-9nc27        1/1     Running   0          5m19s
frontend-85595f5bf9-hkj5s        1/1     Running   0          5m23s
frontend-85595f5bf9-jnnpt        1/1     Running   0          5m21s
redis-follower-dddfbdcc9-88cr9   1/1     Running   0          5m25s
redis-follower-dddfbdcc9-j95b9   1/1     Running   0          5m24s
redis-leader-fb76b4755-b4zpx     1/1     Running   0          5m27s

Lets connect to the frontend service we created using port-forward...

➜  kubectl port-forward svc/frontend 8080:80
Forwarding from 127.0.0.1:8080 -> 80
Forwarding from [::1]:8080 -> 80

If everything is working correctly, you should be able to navigate to http://localhost:8080 and interact with the Guestbook application.

Wrapping Up

So there we are, we've run through how to get a basic EKS Fargate Cluster up and running, how to schedule Pods on to Fargate and how to get basic logging functioning.

I hope everyone who has read this post has enjoyed it and if anyone has any questions drop me a comment or a tweet!

Cover Photo by Chuttersnap on Unsplash.

I've dropped a few links in the post but there are a few more below which you may find beneficial.

https://docs.aws.amazon.com/eks/latest/userguide/fargate-logging.html
https://docs.aws.amazon.com/eks/latest/userguide/fargate-profile.html
https://docs.aws.amazon.com/eks/latest/userguide/fargate-getting-started.html
https://github.com/hashicorp/terraform-provider-kubernetes/issues/723
https://github.com/aws-ia/terraform-aws-eks-blueprints/issues/394
https://github.com/terraform-aws-modules/terraform-aws-eks/issues/1286
https://github.com/vmware-tanzu/octant

Stay safe!

Cheers
DK

Triggering Lambda Functions From Amazon MSK

Danny Kay — Mon, 21 Sep 2020 15:52:27 +0000

Introduction

I was on Twitter last month and came across the below tweet.

// Detect dark theme var iframe = document.getElementById('tweet-1294178133164339200-758'); if (document.body.className.includes('dark-theme')) { iframe.src = "https://platform.twitter.com/embed/Tweet.html?id=1294178133164339200&theme=dark" }

This announcement was pretty big in my opinion.

Having the ability to trigger AWS Lambda functions from Amazon MSK records allows us to utilize many benefits of Serverless computing.

I had a bit of spare time on my hands so thought I'd have a little dig into this new feature.

Want to follow along?

Alrighty, so before we do anything constructive the repository for all the bits and bobs I'm showing you below can be found here.

Be wary, AWS is fun but that fun comes at a cost. Following this will cost you some dough so keep an eye on the Billing & Cost Management Dashboard people!

This post assumes you know your way around AWS and understand how Kafka works.

Setting up the Infrastructure

Before writing our Lambda functions we need some infrastructure, I'm using CloudFormation in this example. AWS has kindly provided a template that provides a basic architecture of a 3 node Kafka Cluster, VPC, Subnet setup, and an EC2 bastion.

https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-msk-cluster.html#aws-resource-msk-cluster--examples

I've tweaked this ever so slightly so that we can communicate with the broker using TLS_PLAINTEXT instead of TLS, I've modified the MSK Security group to allow MSK to communicate with itself and I'm running 2 Brokers instead of 3.

You will also need an EC2 Key Pair so go ahead into the EC2 Console and create one, making sure you download the Key and keep it somewhere safe and out of harm's way.

With the two pieces of pivotal information, we can go ahead and run the script.

➜ deploy-msk-infrastructure.sh IP_ADDRESS KEY_PAIR

➜ deploy-msk-infrastructure.sh 1.2.3.4/32 demo-keypair.pem

It takes around 20–30 minutes for everything to be up, running, and functional.

Once we are cooking with gas we should be able to SSH into our Bastion, create a Topic, and produce a record to it.

➜ chmod 400 demo-keypair.pem

➜ ssh -i "demo-keypair.pem" ec2-user@ec2-1-2-3-4.eu-west-2.compute.amazonaws.com


__|  __|_  )
       _|  (     /   Amazon Linux 2 AMI
      ___|\___|___|
https://aws.amazon.com/amazon-linux-2/


[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-topics.sh --create --topic example \
--bootstrap-server b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092 --partitions 1 --replication-factor 1

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-console-producer.sh --topic example --broker-list b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092

>Hello!

And then consume it.

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-console-consumer.sh --topic example --bootstrap-server b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092 --from-beginning

Hello!

Alrighty, we have verified that our Cluster is alive and kicking. Now we can go ahead and create a Lambda function to verify the connectivity.

Simple Lambda Function

Before we get into the code lets just go over how a Lambda function is actually invoked by Amazon MSK.

A Lambda function is responsible for handling our records. But there is another process that happens before this which is an Event Source Mapping.

An Event Source Mapping is a Lambda process that reads data from a source system and invokes a function with the data it has received from the source system.

The Event Source Mapping that we create in the two examples below is responsible for reading the records from the Topics and invoking the functions.

The Event Source Mapping can be created from the AWS Lambda Console but it can also be created from a CloudFormation Template which we'll be doing in this example.

Right let's create this basic Lambda function, and when I say basic, I mean basic!

Think of this Lambda as verifying that it can communicate with Kafka and all the permissions are set up correctly.

package com.danieljameskay.main

import com.amazonaws.services.lambda.runtime.Context
import com.amazonaws.services.lambda.runtime.RequestHandler
import com.danieljameskay.models.MSKEvent
import com.google.gson.Gson
import com.google.gson.GsonBuilder

class App(var gson: Gson = GsonBuilder().setPrettyPrinting().create()) : RequestHandler<MSKEvent, String> {
    override fun handleRequest(event: MSKEvent, context: Context): String {
        val logger = context.logger
        logger.log(gson.toJson(event))
        return "OK"
    }
}

Quickly running through the code. The JSON Event is Mapped into an MSKEvent object and the contents are logged out.

Following the link here, we have to create a Role with some extra permissions and configure the Event Source for the Lambda function as shown below in the below CloudFormation Template.

AWSTemplateFormatVersion: '2010-09-09'
Description: "Basic Lambda Function to verify MSK connectivity."
Parameters:
  EventSourceArn:
    Description: ARN of the Event Source, Kafka in this case
    Type: String
  TopicName:
    Description: Topic for the Lambda to consume from
    Type: String
  S3Bucket:
    Description: S3 Bucket for the JAR
    Type: String
  S3Key:
    Description: JAR Name
    Type: String
  Handler:
    Description: Application Handler
    Type: String
Resources:
  BasicKafkaConsumerLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName:
        Fn::Sub: BasicKafkaConsumerLambdaRole
      AssumeRolePolicyDocument:
        Statement:
          - Action:
              - sts:AssumeRole
            Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
        Version: 2012-10-17
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AWSLambdaExecute
        - arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole
      Path: /

  BasicKafkaConsumerLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Name: BasicKafkaConsumerLambdaFunction
      Description: BasicKafkaConsumerLambdaFunction
      Runtime: java11
      Code:
        S3Bucket: !Ref S3Bucket
        S3Key: !Ref S3Key
      Handler: !Ref Handler
      MemorySize: 128
      Timeout: 10
      Role:
        Fn::GetAtt:
          - BasicKafkaConsumerLambdaRole
          - Arn

  EventSourceMapping:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      EventSourceArn: !Ref EventSourceArn
      FunctionName: !Ref BasicKafkaConsumerLambdaFunction
      StartingPosition : LATEST
      Topics:
        - !Ref TopicName

With the Lambda, Role, and Event Source Mapping configured in CloudFormation, we can go ahead and run the script.

➜ ./deploy-lambda.sh UBER_JAR_LOCATION \
S3_BUCKET_TO_UPLOAD_TOO \
STACK_NAME \
EVENT_SOURCE_ARN \
TOPIC_NAME \
S3_BUCKET_WHICH_CONTAINS_JAR \
JAR_NAME \
LAMBDA_HANDLER 

➜ ./deploy-lambda.sh ./basic-function-all.jar \
s3://my-bucket/basic-function-all.jar \
Basic-Lambda-Stack \     
arn:aws:kafka:eu-west-2:111111111111:cluster/MSKCluster/f0705c96-e239-4f74-a0f7-f82031a2fc65-4 \
example \
my-bucket \
basic-function-all.jar \
com.danieljameskay.main.App::handleRequest

This script is extremely basic, it uses Gradle to build the app, creates a JAR, uploads it to S3, and finally creates the CloudFormation stack. It should take a couple of minutes for everything to be deployed.

As we did before, connect back to the Bastion and produce some basic test records.

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-console-producer.sh --topic example --broker-list b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092 --property "parse.key=true" --property "key.separator=:"
> 1:Hello!
> 2:Hey!

Then we can head to the Terminal and query the CloudWatch Logs.

➜ aws logs get-log-events --log-group-name /aws/lambda/BasicKafkaConsumerLambdaFunction --log-stream-name '2020/09/20/[$LATEST]8dd5e9881b7f456897d62d39b654e36d'

We can see in the logs the records printed out. Below is a prettier version that has been escaped, values restored and is easier to read.

{
  "records": {
    "example-0": [
      {
        "topic": "example",
        "partition": 0,
        "offset": 0,
        "timestamp": 1600598451721,
        "timestampType": "CREATE_TIME",
        "key": "MQ==",
        "value": "SGVsbG8h"
      }
    ]
  }
}

and

{
  "records": {
    "example-1": [
      {
        "topic": "example",
        "partition": 1,
        "offset": 0,
        "timestamp": 1600598468528,
        "timestampType": "CREATE_TIME",
        "key": "Mg==",
        "value": "SGV5IQ=="
      }
    ]
  }
}

So we can see the records have been grouped into an array based on the partition they came from. We also get some metadata such as the timestamp and the value of the record has been encoded in base64.

The custom class I created which is used by the function handler to map the JSON into an MSKEvent object in this example drops the EventSource and EventSourceARN fields and if a Key isn't provided a zero will be visible in the payload.

Below is an example of what the payload would look like if our function consumed multiple records from different partitions.

{
  "records": {
    "example-1": [
      {
        "topic": "example",
        "partition": 1,
        "offset": 0,
        "timestamp": 1599337456523E12,
        "timestampType": "CREATE_TIME",
        "key": "0",
        "value": "cWVxd2Vxd2U="
      }
    ],
    "example-2": [
      {
        "topic": "example",
        "partition": 2,
        "offset": 0,
        "timestamp": 1599337455636E12,
        "timestampType": "CREATE_TIME",
        "key": "0",
        "value": "cXdlcXdl"
      }
    ]
  }
}

Consumer Groups

A quick note on Consumer Groups. Multiple Consumers can work as part of a Consumer Group to increase parallelism. This allows the Partitions in the Topic to be divided between the available Consumers for consumption.

When the Consumer Group is registered with MSK, the group has the same name as the event source mapping UUID. If you aren't sure what the UUID is of your Event Source Mapping follow the instructions in the first link at the bottom of this post.

➜ aws lambda get-event-source-mapping --uuid 02d6c44a-46a7-41fe-8dd0-0f8ab903fee2
{
    "UUID": "02d6c44a-46a7-41fe-8dd0-0f8ab903fee2",
    "BatchSize": 100,
    "EventSourceArn": "arn:aws:kafka:eu-west-2:111111111111:cluster/MSKCluster/f0705c96-e239-4f74-a0f7-f82031a2fc65-4",
    "FunctionArn": "arn:aws:lambda:eu-west-2:111111111111:function:BasicKafkaConsumerLambdaFunction",
    "LastModified": "2020-09-20T11:18:49.525000+01:00",
    "LastProcessingResult": "OK",
    "State": "Enabled",
    "StateTransitionReason": "USER_INITIATED",
    "Topics": [
        "example"
    ]
}

Then cross-reference this using the Kafka Consumer Group CLI.

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-consumer-groups.sh  --list  --bootstrap-server b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092

02d6c44a-46a7-41fe-8dd0-0f8ab903fee2

So we can see that the UUID of our Event Source Mapping does match the name of the Consumer Group. The last command will display the current offset and lag for each Partition which is useful to understand how our Consumer Group is performing.

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-consumer-groups.sh --bootstrap-server b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092 --describe --group 02d6c44a-46a7-41fe-8dd0-0f8ab903fee2

Integrating SQS

Now we're going to step it up a notch.

We're going to insert some real-world data into Kafka then trigger a Lambda function to send that data into an SQS queue.

Why SQS you might ask?

SQS was the very first AWS service that launched back in November 2004 and to this very day, it is still an extremely popular service used to build queue-based solutions.

One of my favorite Microservice patterns is the Anti-Corruption Layer Pattern. It allows us to provide a layer between two systems that need to be integrated where the semantics are different.

We're going to build a tiny ACL in this example. This means mapping the event generated by MSK into a smaller data structure, serializing the data to JSON, and sending it into an SQS queue.

I have a small application running which is receiving updates every 20 seconds from the Poloniex 24 Hour Exchange Volume API. Its sending the updates it receives onto MSK. The data received has the below structure.

{
  "USDC": "2811608.201",
  "PAX": "100.623",
  "DAI": "202.840",
  "USDJ": "14553.678",
  "USDT": "36356839.873",
  "TRX": "73155426.619",
  "BUSD": "7581.134",
  "BTC": "559.254",
  "BNB": "1717.156",
  "ETH": "70.466",
  "XMR": "0.000"
}

This application is running on the Bastion EC2 instance just for the purpose of this demo. Before we do anything, we have to create a Topic for records to be sent to.

[ec2-user@ip-1-2-3-4 ~]➜ ./kafka/kafka_2.12-2.2.1/bin/kafka-topics.sh --create --topic example.poloniex.24hr.exchange.volume --bootstrap-server b-1.mskcluster.vxbd2z.c4.kafka.eu-west-2.amazonaws.com:9092 --partitions 3 --replication-factor 2

As we did earlier in this article, we'll need to deploy our new Lambda function and some Cloudformation Templates for our SQS resource.

First, we'll deploy our SQS Queue.

➜ ./SQS-Integration/deploy-infrastructure STACK_NAME NAME_OF_QUEUE

The CloudFormation Template for SQS is really simple, it just requires the name of the Queue we want to create.

AWSTemplateFormatVersion: "2010-09-09"
Description: "Creates an SQS queue for the Poloniex Exchange Volume Lambda to send records to."
Parameters:
  QueueName:
    Description: Name of the Queue
    Type: String
Resources: 
  SQSQueue: 
    Type: AWS::SQS::Queue
    Properties: 
      QueueName: !Ref QueueName

The Application is pretty straight forward. A record is produced to Kafka, Lambda is triggered, the JSON event is mapped into an MSKEvent object, the records inside the event are looped over and mapped into a basic ExchangeVolumeUpdate object and sent to SQS.

The CloudFormation template is pretty similar to the earlier one we used for the basic Lambda function.

AWSTemplateFormatVersion: '2010-09-09'
Description: "Poloniex Exchange Volume Lambda Function"
Parameters:
  EventSourceArn:
    Description: ARN of the Event Source, Kafka in this case
    Type: String
  TopicName:
    Description: Topic for the Lambda to consume from
    Type: String
  S3Bucket:
    Description: S3 Bucket for the JAR
    Type: String
  S3Key:
    Description: JAR Name
    Type: String
  Handler:
    Description: Application Handler
    Type: String

Resources:
  PoloniexExchangeVolumeLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName:
        Fn::Sub: PoloniexExchangeVolumeLambdaRole
      AssumeRolePolicyDocument:
        Statement:
          - Action:
              - sts:AssumeRole
            Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
        Version: 2012-10-17
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AWSLambdaExecute
        - arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole
        - arn:aws:iam::aws:policy/AmazonSQSFullAccess
      Path: /

  PoloniexExchangeVolumeLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: PoloniexExchangeVolumeLambdaFunction
      Description: PoloniexExchangeVolumeLambdaFunction
      Runtime: java11
      Code:
        S3Bucket: !Ref S3Bucket
        S3Key: !Ref S3Key
      Handler: !Ref Handler
      MemorySize: 256
      Timeout: 10
      Role:
        Fn::GetAtt:
          - PoloniexExchangeVolumeLambdaRole
          - Arn

  EventSourceMapping:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      EventSourceArn: !Ref EventSourceArn
      FunctionName: !Ref PoloniexExchangeVolumeLambdaFunction
      StartingPosition : LATEST
      Topics:
        - !Ref TopicName

Then we'll deploy our Lambda.

➜ ./deploy-lambda.sh UBER_JAR_LOCATION \
S3_BUCKET_TO_UPLOAD_TOO \
STACK_NAME \
EVENT_SOURCE_ARN \
TOPIC_NAME \
S3_BUCKET_WHICH_CONTAINS_JAR \
JAR_NAME \
LAMBDA_HANDLER

➜ ./deploy-lambda.sh ./sqs-integration-all.jar \
s3://my-bucket/sqs-integration-all.jar \
MSK-SQS-Lambda-Stack \     
arn:aws:kafka:eu-west-2:111111111111:cluster/MSKCluster/f0705c96-e239-4f74-a0f7-f82031a2fc65-4 \
example.poloniex.24hr.exchange.volume \
my-bucket \
sqs-integration-all.jar \
com.danieljameskay.sqs.integration.main.App::handleRequest

This script behaves in the same way as the one used in the basic example above, it uses Gradle to build the app, creates a JAR, uploads it to S3, and finally creates the CloudFormation stack. It should take a couple of minutes for everything to be deployed.

The application is running and sending records to MSK so as soon as our Lambda is deployed we should be able to use the SQS web interface to verify records are being sent to SQS with no issues.

We can also view the CloudWatch graphs to gain insight into how our function is behaving.

Happy days! Everything is working as it should!

Wrap Up

So there we are, we've run through how to invoke some basic Lambda Functions from MSK and utilized CloudFormation to create the infrastructure.

I hope everyone who has read this post has enjoyed it and if anyone has any questions drop me a comment or a tweet!

Cover photo by Pieter van de Sande on Unsplash

Useful links:

https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html
https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html
https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/

Stay safe!

Cheers
DK

Building a serverless COVID-19 notification pipeline

Danny Kay — Sat, 18 Apr 2020 11:05:59 +0000

Intro

Due to the lockdown in place because of COVID-19, I found myself with more spare time than usual so I decided to put it to better use. I'd built plenty of Lambda's over the years but I'd never built a fully serverless solution...so whatever I was building was going to be serverless.

I was having a browse around Github at COVID-19 datasets and came across a repository that contained data for the UK and was updated daily...then I had a lightbulb moment!

I'd seen a ton of amazing dashboards tracking cases all over the globe but my angle was different. I'd ingest the UK data from Github and somehow send out an email showing the increase in confirmed cases, tests and deaths.

I put my AWS cap on and came up with a basic solution...

A Lambda that ingests a CSV document from Github and stores it in S3, triggered by a CloudWatch event
A Lambda that reads the CSV doc from S3 and stores the latest data in DynamoDB, utilising DynamoDB as a materialised view
Another Lambda which is triggered by a DynamoDB Stream and calculates the difference between the old data and new data then sends out an email notification advising of the changes

Then I put the wheels in motion...

First steps - Installing AWS SAM and Creating our Project

We are going to be using the AWS SAM framework to build the pipeline. It's pretty straightforward to get installed and the instructions can be found here.

With SAM installed, we can create a new project by running sam init and following the CLI prompts.


▶ sam init

Which template source would you like to use?

1 - AWS Quick Start Templates
2 - Custom Template Location

Choice: 1

Which runtime would you like to use?
1 - nodejs12.x
2 - python3.8
3 - ruby2.7
4 - go1.x
5 - java11
6 - dotnetcore3.1
7 - nodejs10.x
8 - python3.7
9 - python3.6
10 - python2.7
11 - ruby2.5
12 - java8
13 - dotnetcore2.1
14 - dotnetcore2.0
15 - dotnetcore1.0

Runtime: 8

Project name [sam-app]: Serverless-Covid-19-Pipeline
Cloning app templates from https://github.com/awslabs/aws-sam-cli-app-templates.git

AWS quick start application templates:

1 - Hello World Example
2 - EventBridge Hello World
3 - EventBridge App from scratch (100+ Event Schemas)

Template selection: 1

-----------------------
Generating application:
-----------------------
Name: Serverless-Covid-19-Pipeline
Runtime: python3.7
Application Template: hello-world
Output Directory: .

Next steps can be found in the README file at ./Covid-19-Serverless-Pipeline/README.md

I chose to write the Lambda's in Python because I'm a big fan of the language and its something I want to use more in my projects even though I'm still learning the language.

With the project created we should have a folder structure like the following...

Serverless-Covid-19-Pipeline/
├── README.md
├── events/
│ └── event.json
├── HelloWorld/
│ ├── __init__.py
│ └── app.py
│ └── requirements.txt
├── template.yaml
├── .gitignore
└── tests/

We can trigger a build by running the sam build command.

▶ sam build
Building resource 'HelloWorldFunction'
Running PythonPipBuilder:ResolveDependencies
Running PythonPipBuilder:CopySource

Build Succeeded

Built Artifacts  : .aws-sam/build
Built Template   : .aws-sam/build/template.yaml

Commands you can use next
=========================
[*] Invoke Function: sam local invoke
[*] Deploy: sam deploy --guided

As you can see our first build succeeded. In order to test the function we need to have Docker installed on our machine, we'll do this later when we have our first Lambda function written.

As this project is going to contain multiple Lambdas, we have to create the directories for the two other Lambdas, add the relevant files and rename the existing Lambda folder.

Serverless-Covid-19-Pipeline/
├── README.md
├── events/
│ └── event.json
├── data_ingestion_function/
│ ├── __init__.py
│ └── app.py
│ └── requirements.txt
├── s3_processing_function/
│ ├── __init__.py
│ └── app.py
│ └── requirements.txt
├── ddb_stream_notification/
│ ├── __init__.py
│ └── app.py
│ └── requirements.txt
├── template.yaml
├── .gitignore
└── tests/

I'm not the best person when it comes to naming things, the data_ingestion_function will contain the Lamba which downloads the data from Github and save it to S3, the s3_processing_function will contain the Lambda which downloads the data from S3 and saves it to DynamoDB and lastly, the ddb_stream_notification will contain the Lambda which receives the DynamoDB stream and sends the message to SNS.

First Function - Data Ingestion

The first Lambda function we have to build is responsible for downloading the data from the Github repository and saving it to S3. This Lambda is going to be triggered by a CloudWatch Scheduled Event, so it can be invoked at the same time every day.

We're using the requests and boto3 libraries in this example so make sure they are included in the requirements.txt file and install them.

The first thing we want to do open the app.py file and replace the boilerplate code with our function code.

When the function is invoked it downloads the raw data from Github, writes the data to a CSV file in the tmp folder and then uploads this file to the specified S3 Bucket.

One of the amazing features of the AWS SAM Framework is the ability to test Lambda functions locally before deploying them to AWS.

In order to test our function, we need a mock scheduled event. We can create one by running sam local generate-event cloudwatch scheduled-event. This will generate a JSON snippet which we can modify as necessary and paste into a new JSON file named scheduled-event.json within the events folder of the parent directory.

But, we have a little problem.

Creating an S3 Bucket with CloudFormation

When we test this Lambda it will try and upload the file to our S3 Bucket...which we don't have. We need to modify the template.yaml and deploy our Cloudformation Stack to create the S3 Bucket.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  Serverless-Covid-19-Pipeline

  SAM Template for Serverless-Covid-19-Pipeline

Parameters:

  S3Bucket:
    Type: String
    Default: danieljameskay-data-ingestion-bucket

Resources:
  DataIngestionBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref S3Bucket

To deploy our CloudFormation Stack, we need to run sam deploy --guided, this will allow us to save values for deployment which will be used for future deployments.

$ sam build && sam deploy --guided

Build Succeeded

Built Artifacts  : .aws-sam/build
Built Template   : .aws-sam/build/template.yaml

Commands you can use next
=========================
[*] Invoke Function: sam local invoke
[*] Deploy: sam deploy --guided


Configuring SAM deploy
======================

        Looking for samconfig.toml :  Not found

        Setting default arguments for 'sam deploy'
        =========================================
        Stack Name [sam-app]: serverless-covid-19-pipeline
        AWS Region [us-east-1]: eu-west-2
        #Shows you resources changes to be deployed and require a 'Y' to initiate deploy
        Confirm changes before deploy [y/N]: y
        #SAM needs permission to be able to create roles to connect to the resources in your template
        Allow SAM CLI IAM role creation [Y/n]: y
        Save arguments to samconfig.toml [Y/n]: y

        Looking for resources needed for deployment: Found!

                Managed S3 bucket: aws-sam-cli-managed-default-samclisourcebucket-1nw0fx8sanitw
                A different default S3 bucket can be set in samconfig.toml

        Saved arguments to config file
        Running 'sam deploy' for future deployments will use the parameters saved above.
        The above parameters can be changed by modifying samconfig.toml
        Learn more about samconfig.toml syntax at 
        https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-config.html

        Deploying with following values
        ===============================
        Stack name                 : serverless-covid-19-pipeline
        Region                     : eu-west-2
        Confirm changeset          : True
        Deployment s3 bucket       : aws-sam-cli-managed-default-samclisourcebucket-1nw0fx8sanitw
        Capabilities               : ["CAPABILITY_IAM"]
        Parameter overrides        : {}

Initiating deployment
=====================

Waiting for changeset to be created..

CloudFormation stack changeset
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Operation                                                    LogicalResourceId                                            ResourceType                                               
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Add                                                        DataIngestionBucket                                          AWS::S3::Bucket                                            
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Changeset created successfully. arn:aws:cloudformation:eu-west-2:983527076849:changeSet/samcli-deploy1586858125/37a6647c-ef63-4175-8b5c-dd28a8bd201c


Previewing CloudFormation changeset before deployment
======================================================
Deploy this changeset? [y/N]: y

2020-04-14 10:56:15 - Waiting for stack create/update to complete

CloudFormation events from changeset
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ResourceStatus                               ResourceType                                 LogicalResourceId                            ResourceStatusReason                       
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
CREATE_IN_PROGRESS                           AWS::S3::Bucket                              DataIngestionBucket                          -                                          
CREATE_IN_PROGRESS                           AWS::S3::Bucket                              DataIngestionBucket                          Resource creation Initiated                
CREATE_COMPLETE                              AWS::S3::Bucket                              DataIngestionBucket                          -                                          
CREATE_COMPLETE                              AWS::CloudFormation::Stack                   serverless-covid-19-pipeline                 -                                          
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Successfully created/updated stack - serverless-covid-19-pipeline in eu-west-2

So now if we head over to the AWS Console and open the S3 Explorer we should see the newly created Bucket. Good job!

Updating the Lambda configuration in the Cloudformation template

We need to update the Cloudformation template with our Data Ingestion function configuration.

Resources:
  DataIngestionFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: DataIngestionFunction
      CodeUri: data_ingestion_function/
      Handler: app.lambda_handler
      Runtime: python3.7
      Policies:
        - AmazonS3FullAccess
        - AWSLambdaBasicExecutionRole
        - AWSXrayWriteOnlyAccess
      Tracing: Active
      Environment:
        Variables:
          AWS_DESTINATION_BUCKET: !Ref S3Bucket
          DATA_URL: https://raw.githubusercontent.com/tomwhite/covid-19-uk-data/master/data/covid-19-totals-uk.csv

We have assigned some basic policies to the Lambda function and specified the Bucket name and URL of the Github data as environment variables.

Now we should be able to build and test the function...

$ sam build && sam local invoke "DataIngestionFunction" -e events/scheduled-event.json
Building resource 'DataIngestionFunction'
Running PythonPipBuilder:ResolveDependencies
Running PythonPipBuilder:CopySource

Build Succeeded

Built Artifacts  : .aws-sam/build
Built Template   : .aws-sam/build/template.yaml

Commands you can use next
=========================
[*] Invoke Function: sam local invoke
[*] Deploy: sam deploy --guided

Invoking app.lambda_handler (python3.7)

Fetching lambci/lambda:python3.7 Docker container image......
Mounting /Users/daniel.kay/dev/Serverless-Covid-19-Pipeline/Serverless-Covid-19-Pipeline/.aws-sam/build/DataIngestionFunction as /var/task:ro,delegated inside runtime container
[INFO]  2020-04-14T13:28:13.880Z                Found credentials in environment variables.

START RequestId: 46f1780b-7e04-124f-f233-6ce13fb27a11 Version: $LATEST
[INFO]  2020-04-14T13:28:14.452Z        46f1780b-7e04-124f-f233-6ce13fb27a11    Requesting data from Github

[INFO]  2020-04-14T13:28:14.946Z        46f1780b-7e04-124f-f233-6ce13fb27a11    Data downloaded successfully

[INFO]  2020-04-14T13:28:15.758Z        46f1780b-7e04-124f-f233-6ce13fb27a11    Data uploaded to S3

END RequestId: 46f1780b-7e04-124f-f233-6ce13fb27a11
REPORT RequestId: 46f1780b-7e04-124f-f233-6ce13fb27a11  Init Duration: 1513.60 ms       Duration: 1310.72 ms    Billed Duration: 1400 ms        Memory Size: 128 MB     Max Memory Used: 43 MB

{"message":"Lambda completed"}

Looking good, we can see from the log lines that the data was downloaded and uploaded to S3. If we navigate to the S3 Explorer and open up the Bucket we should see the CSV file.

We can now download the file and verify the contents. The file should contain data up until the previous day.

Configuring a Scheduled Events

So our Data Ingestion Lambda works perfectly. It downloads the data and it uploads it to S3. But we don't want to invoke the function manually, we want it invoked at a particular time of the day. We can utilize a Cloudwatch Scheduled Event to invoke the function for us.

We have to update our Cloudformation Template to include the event rule and permission for EventBridge to invoke the Lambda function.

DataIngestionScheduledRule:
  Type: AWS::Events::Rule
    Properties:
      Name: DataIngestionScheduledRule
      Description: Triggers the Data Ingestion lambda once per day
      ScheduleExpression: cron(45 14 * * ? *)
      State: ENABLED
      Targets:
        -
          Arn: !GetAtt DataIngestionFunction.Arn
          Id: DataIngestionScheduledRule

DataIngestionInvokePermission:
  Type: AWS::Lambda::Permission
  Properties:
    FunctionName: !Ref DataIngestionFunction
    Action: lambda:InvokeFunction
    Principal: events.amazonaws.com
    SourceArn: !GetAtt DataIngestionScheduledRule.Arn

We are setting the scheduled expression to be a cron expression which will invoke the function daily at 1545 GMT due to daylight saving in the UK 😄. The target is the Data Ingestion function as this is the function we want to be invoked.

Right, it's ready to be built and deployed...

$ sam build && sam deploy
Building resource 'DataIngestionFunction'
Running PythonPipBuilder:ResolveDependencies
Running PythonPipBuilder:CopySource

Build Succeeded

Built Artifacts  : .aws-sam/build
Built Template   : .aws-sam/build/template.yaml

Commands you can use next
=========================
[*] Invoke Function: sam local invoke
[*] Deploy: sam deploy --guided


        Deploying with following values
        ===============================
        Stack name                 : serverless-covid-19-pipeline
        Region                     : eu-west-2
        Confirm changeset          : True
        Deployment s3 bucket       : aws-sam-cli-managed-default-samclisourcebucket-1nw0fx8sanitw
        Capabilities               : ["CAPABILITY_IAM"]
        Parameter overrides        : {}

Initiating deployment
=====================
Uploading to serverless-covid-19-pipeline/175b90da9d229e544ad442eeb0c9e280.template  1711 / 1711.0  (100.00%)

Waiting for changeset to be created..

CloudFormation stack changeset
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Operation                                                    LogicalResourceId                                            ResourceType                                               
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Add                                                        DataIngestionInvokePermission                                AWS::Lambda::Permission                                    
+ Add                                                        DataIngestionScheduledRule                                   AWS::Events::Rule                                          
* Modify                                                     DataIngestionFunctionRole                                    AWS::IAM::Role                                             
* Modify                                                     DataIngestionFunction                                        AWS::Lambda::Function                                      
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Changeset created successfully. arn:aws:cloudformation:eu-west-2:983527076849:changeSet/samcli-deploy1586873775/cd0112a5-5534-4d7e-95b7-a64503823b64


Previewing CloudFormation changeset before deployment
======================================================
Deploy this changeset? [y/N]: y

2020-04-14 15:16:25 - Waiting for stack create/update to complete

CloudFormation events from changeset
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ResourceStatus                               ResourceType                                 LogicalResourceId                            ResourceStatusReason                       
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
UPDATE_IN_PROGRESS                           AWS::Lambda::Function                        DataIngestionFunction                        Requested update requires the creation of  
                                                                                                                                       a new physical resource; hence creating    
                                                                                                                                       one.                                       
UPDATE_COMPLETE                              AWS::Lambda::Function                        DataIngestionFunction                        -                                          
UPDATE_IN_PROGRESS                           AWS::Lambda::Function                        DataIngestionFunction                        Resource creation Initiated                
CREATE_IN_PROGRESS                           AWS::Events::Rule                            DataIngestionScheduledRule                   -                                          
CREATE_IN_PROGRESS                           AWS::Events::Rule                            DataIngestionScheduledRule                   Resource creation Initiated                
CREATE_COMPLETE                              AWS::Events::Rule                            DataIngestionScheduledRule                   -                                          
CREATE_IN_PROGRESS                           AWS::Lambda::Permission                      DataIngestionInvokePermission                Resource creation Initiated                
CREATE_IN_PROGRESS                           AWS::Lambda::Permission                      DataIngestionInvokePermission                -                                          
CREATE_COMPLETE                              AWS::Lambda::Permission                      DataIngestionInvokePermission                -                                          
UPDATE_COMPLETE_CLEANUP_IN_PROGRESS          AWS::CloudFormation::Stack                   serverless-covid-19-pipeline                 -                                          
UPDATE_COMPLETE                              AWS::CloudFormation::Stack                   serverless-covid-19-pipeline                 -                                          
DELETE_COMPLETE                              AWS::Lambda::Function                        DataIngestionFunction                        -                                          
DELETE_IN_PROGRESS                           AWS::Lambda::Function                        DataIngestionFunction                        -                                          
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Successfully created/updated stack - serverless-covid-19-pipeline in eu-west-2

We can see from the above that the invoke permissions and the scheduled rule have been created successfully. To validate this, we can head into CloudWatch Console and check out the Rules within the Events section.

If we navigate to the Lambda Console in AWS and open our Lambda from within the Functions section, we should see that the CloudWatch Event is now set up as a trigger for this Lambda.

Exciting times!

So how do we know if this function was in fact triggered at 1545?

We can do this in a few different ways:

We can check the timestamp on the CSV file that was created when we were testing as this will have a new modified date and time
If the file wasn't there to begin with there should be a CSV file in the bucket
We can check the CloudWatch logs
CloudWatch ServiceLens (my favorite)

If we check the CloudWatch logs it will look very similar to the log lines we saw when we were testing earlier.

We can see the data was downloaded and uploaded to S3 without any issues.

CloudWatch ServiceLens provides us with complete observability of our applications and services by providing logs, metrics, tracing and alarms in one friendly dashboard.

I'm planning on writing another post about ServiceLens at a later date.

Second Function - Data Processing

We are now going to start working on the second Lambda function. This function will be responsible for listening to S3 events from the Bucket we created earlier, using the event to download the CSV file and insert the data for the most recent date into DynamoDB.

As we've covered how we interact with the AWS SAM CLI, these steps won't be as heavily detailed as they were in previous sections

First things first, we need a DynamoDB table. We can update our CloudFormation template to include a DynamoDB resource type with some basic attributes.

Whilst we are doing this we can also add our Lambda configuration.

DynamoDBTable:
  Type:  AWS::DynamoDB::Table
  Properties:
    TableName:  danieljameskay-covid19-data
    KeySchema:
      -
        AttributeName:  country
        KeyType:  HASH
    AttributeDefinitions:
      -
        AttributeName:  country
        AttributeType:  S
    ProvisionedThroughput:
      ReadCapacityUnits:  5
      WriteCapacityUnits:  5
    StreamSpecification:
      StreamViewType:  NEW_AND_OLD_IMAGES

S3ProcessingFunction:
  Type: AWS::Serverless::Function
  Properties:
    FunctionName: S3ProcessingFunction
    CodeUri: s3_processing_function/
    Handler: app.lambda_handler
    Runtime: python3.7
    Tracing: Active
    Environment:
      Variables:
        AWS_DOWNLOAD_BUCKET: !Ref S3Bucket
        DDB_TABLE: danieljameskay-covid19-data
    Policies:
      - AWSLambdaBasicExecutionRole
      - AWSXrayWriteOnlyAccess
      - AmazonDynamoDBFullAccess
      - AmazonS3FullAccess
    Events:
      s3Notification:
        Type: S3
        Properties:
          Bucket: !Ref DataIngestionBucket
          Events: s3:ObjectCreated:*

We are specifying the key of our table is the country. The type is hash which is also known as the partition key. We're also specifying a StreamSpecification of type NEW_AND_OLD_IMAGES which means that when the values for the country are updated, DynamoDB will stream the new value and old value to any listeners...in our case, our Notification Lambda.

We should now be able to run sam build && sam deploy to deploy our DynamoDB Table.

Now the code...

When the function is invoked it downloads the CSV file from Github and stores it in the tmp directory, processes it and inserts the most recent record into DynamoDB along with the date that the data corresponds to.

There's probably a nicer way to write some of this logic but as I'm still learning Python...it does the job 👨‍💻

As we did earlier we can test this Lambda by running sam build "S3ProcessingFunction" && sam local invoke "S3ProcessingFunction" -e events/s3-put-event.json

We can verify the Lambda has worked correctly by verifying the log lines...

[INFO]  2020-04-16T10:25:10.372Z        bfcc35ee-2908-1a44-afe0-e02816837dbc    Downloading CSV file from S3
[INFO]  2020-04-16T10:25:10.663Z        bfcc35ee-2908-1a44-afe0-e02816837dbc    File downloaded to tmp directory
[INFO]  2020-04-16T10:25:10.669Z        bfcc35ee-2908-1a44-afe0-e02816837dbc    Inserting record into DynamoDB...
[INFO]  2020-04-16T10:25:10.937Z        bfcc35ee-2908-1a44-afe0-e02816837dbc    Record added to DynamoDB

and checking DynamoDB via the AWS CLI...

$ aws dynamodb scan --table-name danieljameskay-covid19-data                                             
{
    "Count": 1, 
    "Items": [
        {
            "date": {
                "S": "04-13-2020"
            }, 
            "country": {
                "S": "UK"
            }, 
            "confirmed_cases": {
                "S": "88621"
            }, 
            "tests": {
                "S": "290720"
            }, 
            "deaths": {
                "S": "11329"
            }
        }
    ], 
    "ScannedCount": 1, 
    "ConsumedCapacity": null
}

We can see that our test has successfully run and the data has been inserted into DynamoDB.

We should now be able to run sam build && sam deploy to deploy our new Lambda.

Lets see if what we've done so far works 🤔

So, we've tested our 2 Lambdas locally and 1 of them in AWS. We can wait until the time we set earlier to see if our Lambda's work or we can trigger the Data Ingestion Lambda from the AWS CLI...

$ aws lambda invoke --function-name DataIngestionFunction out --log-type None
{
    "ExecutedVersion": "$LATEST", 
    "StatusCode": 200
}

and then we can check DynamoDB.

$ aws dynamodb scan --table-name danieljameskay-covid19-data                 
{
    "Count": 1, 
    "Items": [
        {
            "date": {
                "S": "04-15-2020"
            }, 
            "country": {
                "S": "UK"
            }, 
            "confirmed_cases": {
                "S": "98476"
            }, 
            "tests": {
                "S": "313769"
            }, 
            "deaths": {
                "S": "12868"
            }
        }
    ], 
    "ScannedCount": 1, 
    "ConsumedCapacity": null
}

Looking good! So we manually triggered the Data Ingestion Lambda which in turn triggered the S3 Processing Lambda using the S3 Event which added the new data to the Table.

Third Function - Notification

We have a Lambda which downloads the data and another which processes this data and saves it to DynamoDB.

The last Lambda in this solution listens to a stream from our DynamoDB Table which contains the old data and the new data. Using this data it works out the percent difference between the two datasets and builds a message which is then used to be sent out using AWS SNS to an email address.

Right, CloudFormation additions...

DDBNotificationFunction:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: ddb_notification_function
    FunctionName: 'DDBNotificationFunction'
    Handler: app.lambda_handler
    Runtime: python3.7
    Tracing: Active
    Environment:
      Variables:
        SNS_TOPIC: !Ref SNSTopic
    Policies:
      - AmazonSNSFullAccess
      - AWSLambdaBasicExecutionRole
      - AWSXrayWriteOnlyAccess
    Events:
      Stream:
        Type: DynamoDB
        Properties:
          Stream: !GetAtt  DynamoDBTable.StreamArn
          StartingPosition: TRIM_HORIZON

SNSTopic:
  Type: AWS::SNS::Topic
  Properties:
  TopicName: "ddb-stream-covid-19-daily-data"

SNSSubscription:
  Type: AWS::SNS::Subscription
  Properties:
    Endpoint: 4i4nbyt@2go-mail.com
    Protocol: email
    TopicArn: !Ref  'SNSTopic'

Here we have our last Lambda configuration. The only thing to really point out here is we are specifying DynamoDB as the type of event source for this Lambda and the Stream ARN of our DynamoDB Table.

Lastly, we are creating an SNS Topic and Subscription. Our Lambda will send the message to the SNS Topic and the Subscription will be responsible for sending it to our temporary email address.

Code time people...

We are retrieving the old and new records from the event which triggered the Lambda function. Then we establish the percentage difference between the old attributes and new attributes and finally publishing a message with the percentage change to SNS.

We should now be able to run sam build && sam deploy to deploy our new Lambda.

Testing from the Lambda Console

In the previous sections, we tested our functions from the AWS CLI and AWS SAM CLI but we can invoke our function from Lambda Console.

With the Notification Lambda open in the Lambda Console, we can click "Test" followed by "Configure Test Events". Using the "Hello World" template, we can just overwrite the JSON payload with a JSON payload generated by running sam local generate-event dynamodb update and changing the values.

Below is an example of the event that the DynamoDB Notification Lambda receives when the data is changed.

{
    "Records": [
        {
            "eventID": "24df338c43fa112256a16d7d13c6eb14",
            "eventName": "MODIFY",
            "eventVersion": "1.1",
            "eventSource": "aws:dynamodb",
            "awsRegion": "eu-west-2",
            "dynamodb": {
                "ApproximateCreationDateTime": 1587043428,
                "Keys": {
                    "country": {
                        "S": "UK"
                    }
                },
                "NewImage": {
                    "date": {
                        "S": "04-13-2020"
                    },
                    "country": {
                        "S": "UK"
                    },
                    "confirmed_cases": {
                        "S": "98621"
                    },
                    "tests": {
                        "S": "291720"
                    },
                    "deaths": {
                        "S": "12329"
                    }
                },
                "OldImage": {
                    "date": {
                        "S": "04-12-2020"
                    },
                    "country": {
                        "S": "UK"
                    },
                    "confirmed_cases": {
                        "S": "88621"
                    },
                    "tests": {
                        "S": "290720"
                    },
                    "deaths": {
                        "S": "11329"
                    }
                },
                "SequenceNumber": "5571000000000004914106821",
                "SizeBytes": 125,
                "StreamViewType": "NEW_AND_OLD_IMAGES"
            },
            "eventSourceARN": "arn:aws:dynamodb:eu-west-2:983527076849:table/danieljameskay-covid19-data/stream/2020-04-15T09:00:54.674"
        }
    ]
}

We can then use this saved event and some fake data to trigger our Lambda.

As our Lambda has successfully run we should have received an email. Let's check our inbox...

Oh yeah 😎😎😎

And there we have it, the email with the percentage changes. It looks amazing, doesn't it?

End to end

We haven't seen the solution work end to end as of yet, let's change that!

I modified the time for the DataIngestionScheduledRule to 10 minutes in the future and altered the DynamoDB record so it had the values for the 15th April so when the Data Ingestion Lambda ran it would pull down the data for the 16th April.

So I deployed the changes, made a coffee and came back to see the email in my inbox with the difference between the two days.

I also checked SerivceLens and everything had run smoothly with no errors :)

Wrapping Up

Working on this has been extremely rewarding. I know my way around AWS pretty well but, I've never used CloudFormation before and I'm still learning Python which added an extra challenge.

We could have used LocalStack and local DynamoDB to aid with the testing locally so we wouldn't have to interact with AWS until we deployed.

Things I would have liked to do and may do in the future:

Utilize AWS Amplify and build a React/GraphQL application that interacts with the full UK dataset as opposed to the most recent dates
Use some more datasets for USA and Europe
Use the X-Ray SDK to track AWS SDK calls
Learn more about CloudFormation and tidy up the template.yaml file

With everything that's been going on with COVID-19 building this solution and writing it up has kept me preoccupied in these uncertain times. Hopefully, there will be another post or two coming in the future.

Big thanks to Tom White for the data used in this post. For anyone interested in his Github profile, it can be found here.

Cover Photo by Dhaya Eddine Bentaleb on Unsplash.

I hope everyone who has read this post has enjoyed it and if anyone has any questions drop me a comment or drop a tweet!

Stay safe!

Cheers