AWS examples in C# – AWS CLI commands

Last Updated on by

Post summary: Important AWS CLI commands used in AWS examples in C#.

This post is part of AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS series. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository.

Introduction

In AWS examples in C# – run the solution post I have described how to install/uninstall current examples. In the current post, I am going to show in detail individual commands used. The configuration parameters in the command below will be given with capital letters and starting with a dollar sign, e.g. $CONFIGURATION_PARAMETER. Each AWS command has its code representation in the SDK for the desired programming language.

AWS Command Line Interface

The AWS Command Line Interface (CLI) is a unified tool to manage AWS services. Control of multiple AWS services from the command line and automate them through scripts can be done with just one tool to download and configure. The full list of services that can be controlled is listed in the AWS Command Line Interface reference page. Each service has a subpage with a list of all available commands. All commands return JSON as a response. In a subsequent post, I will describe how to manage the JSON in the command line. All operations in the current post are done after AWS credentials are set as environment variables:

export AWS_ACCESS_KEY_ID=KIA57FV4.....
export AWS_SECRET_ACCESS_KEY=mSgsxOWVh...
export AWS_DEFAULT_REGION=us-east-1

SQS operations

The full list can be found in aws sqs CLI reference page. More information about SQS can be found in AWS examples in C# – create a service working with SQS post.

Create

Initially, all queues are listed with list-queues, in order to check if the queue already exists.

aws sqs list-queues

The queue is created with create-queue command, the result of the command returns the queue URL.

aws sqs create-queue --queue-name $QUEUE_NAME

After queues are created, the re-drive policy has to be set up. The ARN of the dead-letter queue can be obtained with get-queue-attributes command by providing the queue URL.

aws sqs get-queue-attributes \
	--queue-url $DEAD_LETTER_QUEUE_URL \
	--attribute-names QueueArn

The re-drive policy is set with set-queue-attributes command.

aws sqs set-queue-attributes \
	--queue-url $QUEUE_URL \
	--attributes "{\"RedrivePolicy\":\"{\\\"maxReceiveCount\\\":\\\"3\\\",\\\"deadLetterTargetArn\\\":\\\"$DEAD_LETTER_QUEUE_ARN\\\"}\",\"ReceiveMessageWaitTimeSeconds\":\"$LONG_POLLING_TIMEOUT\"}"

Delete

In order to delete the queue, its URL is needed. The URL is obtained with get-queue-url command.

aws sqs get-queue-url --queue-name $QUEUE_NAME

Deletion happens with delete-queue command.

aws sqs delete-queue --queue-url $QUEUE_URL

DynamoDB operations

The full list can be found in aws dynamodb CLI reference page. More information about DynamoDB can be found in AWS examples in C# – create a service working with DynamoDB post.

Create

The table data is obtained with describe-table command.

aws dynamodb describe-table --table-name $TABLE_NAME

If the table does not exist, it is created with create-table command. The table command has all the data needed. See more about table attributes in AWS examples in C# – create a service working with DynamoDB post.

aws dynamodb create-table \
	--table-name $TABLE_NAME \
	--attribute-definitions 'AttributeName=FirstName,AttributeType=S' 'AttributeName=LastName,AttributeType=S' \
	--key-schema 'AttributeName=FirstName,KeyType=HASH' 'AttributeName=LastName,KeyType=RANGE' \
	--provisioned-throughput 'ReadCapacityUnits=5,WriteCapacityUnits=5' \
	--stream-specification 'StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES'

Delete

The table is deleted by name with delete-table command.

aws dynamodb delete-table --table-name $TABLE_NAME

IAM roles operations

The full list can be found in aws iam CLI reference page.

Create

Roles are listed with list-roles command to check if the role exists.

aws iam list-roles

The role is created with create-role command.

aws iam create-role \
	--role-name $ROLE_NAME \
	--assume-role-policy-document file://assume-role-policy-document.json

This is the only case in the current examples where an additional JSON document is needed alongside a command. It is not possible to pass this JSON inline as it is with aws sqs set-queue-attributes command. This JSON allows certain services to be accessed by this role.

{
	"Version": "2012-10-17",
	"Statement": [
		{
			"Effect": "Allow",
			"Principal": {
				"Service": [
					"lambda.amazonaws.com",
					"ec2.amazonaws.com",
					"ecs.amazonaws.com",
					"ecs-tasks.amazonaws.com",
					"batch.amazonaws.com"
				]
			},
			"Action": "sts:AssumeRole"
		}
	]
}

List policies to get the policy ARN with list-policies command. Basically, to make things easier, AdministratorAccess existing policy is used with its ARN.

aws iam list-policies

Attach the policy to the role with attach-role-policy command.

aws iam attach-role-policy \
	--role-name $ROLE_NAME \
	--policy-arn $POLICY_ARN

Delete

List policies with list-policies command to get the ARN, then detach the policy from the role.

aws iam detach-role-policy \
	--role-name $ROLE_NAME \
	--policy-arn $POLICY_ARN

After the policy is detached, role is deleted with delete-role command.

aws iam delete-role --role-name $ROLE_NAME

AWS Lambda operations

The full list can be found in aws lambda CLI reference page.

Create

List functions with list-functions command to check if the function exists.

aws lambda list-functions

Creating a function is done with create-function command and takes many arguments. Most of the parameters are self-explanatory. Timeout is important, the lambda function execution is suspended after the timeout passes, in current examples, it is 30 seconds, I found that cold start could take up to 15 seconds some times. The lambda configurations are described in AWS examples in C# – create basic Lambda function post.

aws lambda create-function \
	--function-name $FUNCTION_NAME \
	--runtime dotnetcore2.1 \
	--role $ROLE_ARN \
	--handler $HANDLER_STRING_WITH_NAMESPACE_CLASS_METHOD \
	--environment "Variables={AWS_SQS_QUEUE_NAME=$QUEUE_NAME, AWS_SQS_IS_FIFO=$IS_QUEUE_FIFO}" \
	--timeout $FUNCTION_TIMEOUT \
	--zip-file fileb://$PATH_TO_ZIP_FILE)

Once the function is created, it can be linked to an event source, such as DynamoDB. This happens by DynamoDB stream ARN. Once a record is inserted, updated or deleted in DynamoDB, the lambda function is called with this event.

aws lambda create-event-source-mapping \
	--function-name $FUNCTION_NAME \
	--event-source-arn $DYNAMODB_STREAM_ARN \
	--starting-position LATEST)

In case of function already exists, but its code has to be updated, this is done with update-function-code command.

aws lambda update-function-code \
	--function-name $FUNCTION_NAME \
	--zip-file fileb://$PATH_TO_ZIP_FILE)

Along with the code, function configuration can be updated as well with update-function-configuration command.

aws lambda update-function-configuration \
	--function-name $FUNCTION_NAME  \
	--role $ROLE_ARN\
	--handler $HANDLER_STRING_WITH_NAMESPACE_CLASS_METHOD  \
	--environment "Variables={AWS_SQS_QUEUE_NAME=$QUEUE_NAME, AWS_SQS_IS_FIFO=$IS_QUEUE_FIFO}" \
	--timeout $FUNCTION_TIMEOUT

Delete

In order to delete, then the event source UUID has to be obtained, this is done with list-event-source-mappings command.

aws lambda list-event-source-mappings --function-name $FUNCTION_NAME

Then event source mapping is deleted with delete-event-source-mapping command.

aws lambda delete-event-source-mapping --uuid $EVNET_SOURCE_UUID

And finally, the function itself is deleted with delete-function command.

aws lambda delete-function --function-name $FUNCTION_NAME

ECS (Elastic Container Service) operations

The full list can be found in aws ecs CLI reference page.

Create

Before doing anything with ECR, docker login command should be created with get-login, so docker is authenticated with AWS ECR. With eval function, the docker login command is directly executed.

eval $(aws ecr get-login --no-include-email)

Clusters are first listed, in order to evaluate if the application is already deployed.

aws ecs list-clusters

Cluster is created with create-cluster command. A cluster consists of services.

aws ecs create-cluster --cluster-name $CLUSTER_NAME

Existing task definitions are listed, to evaluate whether they are published or not. Task definitions are Docker configurations.

aws ecs describe-task-definition --task-definition $TASK_DEFINITION_NAME

Task definition is created with register-task-definition command.

aws ecs register-task-definition \
	--family $TASK_DEFINITION_NAME \
	--execution-role-arn $ROLE_ARN\
	--network-mode awsvpc \
	--container-definitions $CONTAINER_DEFINITIONS \
	--requires-compatibilities "FARGATE" \
	--cpu "256" \
	--memory "512"

$CONTAINER_DEFINITIONS is a Docker configuration which defines the task definition:

name=$TASK_DEFINITION_NAME,\
image=$IMAGE_TAG,\
environment=[\
	{name=AwsQueueIsFifo,value=$_IS_QUEUE_FIFO},\
	{name=AwsRegion,value=$REGION},\
	{name=AwsQueueName,value=$QUEUE_NAME},\
	{name=AwsAccessKey,value=$AWS_ACCESS_KEY},\
	{name=AwsSecretKey,value=$AWS_SECRET_KEY},\
	{name=AwsQueueAutomaticallyCreate,value=$AWS_QUEUE_AUTO_CREATE},\
	{name=AwsQueueLongPollTimeSeconds,value=$AWS_POLL_TIME_SECONDS}\
],\
logConfiguration={\
	logDriver=awslogs,\
	options={\
		awslogs-group=ecs/$SERVICE_NAME,\
		awslogs-region=$REGION,\
		awslogs-stream-prefix=ecs\
	}\
}

Before creating a service, existing ones are listed with describe-services command. Service has one or more running instances of a task definition. This is how service can scale.

aws ecs describe-services \
	--cluster $CLUSTER_NAME\
	--services $SERVICE_NAME

Creating a service is done with create-service command. $TASK_REVISION is the result of the register-task-definition command. $SUBNET_ID is returned by aws ec2 describe-subnets command.

aws ecs create-service --cluster $CLUSTER_NAME \
	--service-name $SERVICE_NAME \
	--task-definition "$TASK_DEFINITION_NAME:$TASK_REVISION" \
	--desired-count 1 \
	--launch-type "FARGATE" \
	--network-configuration "awsvpcConfiguration={subnets=[$SUBNET_ID],securityGroups=[$SECURITY_GROUP_ID],assignPublicIp=ENABLED}")

Updating of the service is done with a very update-service similar command.

aws ecs update-service --cluster $CLUSTER_NAME \
	--service $SERVICE_NAME \
	--task-definition "$TASK_DEFINITION_NAME:$TASK_REVISION" \
	--desired-count 1 \
	--network-configuration "awsvpcConfiguration={subnets=[$SUBNET_ID],securityGroups=[$SECURITY_GROUP_ID],assignPublicIp=ENABLED}")

Delete

In order to delete task definitions, they should be first listed with list-task-definitions command, so the task definition version is available.

aws ecs list-task-definitions

Removing of the task definition is done with deregister-task-definition command. Note that the command does what it says, deregister, it does not delete. The task definition is kept in history in status INACTIVE.

aws ecs deregister-task-definition --task-definition "$TASK_DEFINITION_VERSION"

Deleting the service is done with the delete-service command, the –force parameter also stops the running tasks.

aws ecs delete-service \
	--cluster $CLUSTER_NAME \
	--service $SERVICE_NAME \
	--force

In the end, the whole cluster is deleted with delete-cluster command.

aws ecs delete-cluster --cluster $CLUSTER_NAME

ECR (Elastic Container Registry) operations

The full list can be found in aws ecr CLI reference page.

Delete

The repository is created by Docker when the image is pushed to it. Repository and images inside are deleted with delete-repository command.

aws ecr delete-repository \
	--repository-name $REPOSITORY_NAME \
	--force

EC2 (Elastic Compute Cloud) operations

The full list can be found in aws ec2 CLI reference page.

Create

EC2 is responsible for security groups, which expose the service to the world by applying firewall rules. Before creating the group, it is first searched for presence with describe-security-groups command.

aws ec2 describe-security-groups

The security group is created with create-security-group command.

aws ec2 create-security-group \
	--description $SECIRITY_GROUP_DESCRIPTION\
	--group-name $SECIRITY_GROUP_NAME

Inbound rules are defined with authorize-security-group-ingress command, where ip_permission is a bash function generation the JSON for better reuse.

aws ec2 authorize-security-group-ingress \
	--group-id $SECURITY_GROUP_ID \
	--ip-permissions "[$(ip_permission $SERVICE_PORT)]"

Function generation firewall rule JSON, $1 is an argument given to the function.

function ip_permission() {
	echo "{\"IpProtocol\": \"tcp\", \"FromPort\": $1, \"ToPort\": $1, \"IpRanges\": [{\"CidrIp\": \"0.0.0.0/0\", \"Description\": \"Port $1\"}]}"
}

Subnets are listed with describe-subnets command. Each subnet has 3 availability zones.

aws ec2 describe-subnets

Finally, in order to report the IP of the deployed service, describe-network-interfaces command is used.

aws ec2 describe-network-interfaces --filters "Name=network-interface-id,Values=$networkInterfaceId"

Delete

A security group is deleted by name with delete-security-group command.

aws ec2 delete-security-group --group-name $SECURITY_GROUP

CloudWatch operations

The full list can be found in aws logs CLI reference page.

Delete

CloudWatch logs are created by default from the services. Deleting the logs is done with delete-log-group command. Note that I am using Git Bash on Windows and MSYS_NO_PATHCONV=1 is mandatory because the log group name starts with /.

MSYS_NO_PATHCONV=1 aws logs delete-log-group --log-group-name ecs/$SERVICE_NAME

Conclusion

AWS command-line interface provides tooling to handle all needed operations of the AWS services. It is the preferred way to manage services over the Web user interface.

Related Posts

Read more...

AWS examples in C# – create basic Lambda function

Last Updated on by

Post summary: Code examples on how to create AWS Lambda function in .NET Core.

This post is part of AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS series. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository.

AWS Lambda

AWS Lambda allows easy ramp-up of service without all the hassle to manage servers and environments. The ready code is uploaded to Lambda and automatically run. More detailed information and design considerations about AWS Lambda can be found in AWS examples in C# – working with Lambda functions post.

Create Lambda

Amazon provides Amazon.Lambda.Templates NuGet package which contains a lot of templates for Lambda functions. The NuGet package can be installed with dotnet new -i Amazon.Lambda.Templates. Once templates are installed, a new empty function is created with:

dotnet new lambda.EmptyFunction --name MyFunction

Two projects are created: src and test. The source project is with netcoreapp2.1 runtime and has reference to Amazon.Lambda.Core and Amazon.Lambda.Serialization.Json NuGet packages. It has only one simple function that takes a string input and converts it to uppercase. The test project has a unit test that tests this function.

Once the function is ready, it can be deployed to AWS Lambda and tested. In order to deploy, the Amazon lambda tools should be installed with: dotnet tool install -g Amazon.Lambda.Tools. Once the tool is installed, there should be an IAM role created with name MyRole for e.g. Before deploying the lambda, environment variable with AWS access information should be set:

export AwsAccessKey=KIA57FV4.....
export AwsSecretKey=mSgsxOWVh...
export AwsRegion=us-east-1

Then lambda is deployed with:

dotnet lambda deploy-function MyFunction \
	--function-role MyRole \
	--project-location src/MyFunction \
	--region $AwsRegion \
	--aws-access-key-id $AwsAccessKey \
	--aws-secret-key $AwsSecretKey

The function can be tested with:

dotnet lambda invoke-function MyFunction \
	--payload "Just Testing the Payload" \
	--project-location src/MyFunction \
	--region $AwsRegion \
	--aws-access-key-id $AwsAccessKey \
	--aws-secret-key $AwsSecretKey

The result is shown:

Amazon Lambda Tools for .NET Core applications (3.3.1)
Project Home: https://github.com/aws/aws-extensions-for-dotnet-cli, https://github.com/aws/aws-lambda-dotnet

Payload:
"JUST TESTING THE PAYLOAD"

Log Tail:
START RequestId: 3f1844e0-7437-4219-a391-621a55dec0e9 Version: $LATEST
END RequestId: 3f1844e0-7437-4219-a391-621a55dec0e9
REPORT RequestId: 3f1844e0-7437-4219-a391-621a55dec0e9  Duration: 925.38 ms    Billed Duration: 1000 ms Memory Size: 256 MB     Max Memory Used: 62 MB  Init Duration: 196.85 ms

The duration in the example above is 925.38ms, billed duration is 1000ms. This is the first run, the cold start as described in the previous section, which took almost a second for a very simple function. Next run the duration was 0.28ms.

The lambda can also manually be deployed and tested, see how in Run a “Hello, World!” with AWS Lambda article.

Listen to DynamoDB events

In AWS examples in C# – create a service working with DynamoDB post, I have described more about DynamoDB and its streams are very well integrated with AWS Lambda. In the current examples, the lambda functions are designed to process DynamoDB stream events. DynamoDB stream ARN (Amazon Resource Name) is defined as an event source for the lambda. Then the lambda receives DynamoDBEvent object, which is defined in Amazon.Lambda.DynamoDBEvents NuGet package. There is not much business logic in the lambda function, once the event object is received, it is read, logged to AWS CloudWatch with ILambdaContext, and its data is written to another DynamoDB table and to an SQS queue. For this purpose, references to AWSSDK.DynamoDBv2 and AWSSDK.SQS NuGet packages are made. In the example code, I have created SQS and DynamoDB proxies, so functionalities are isolated and only what is needed is exposed, these are DynamoDbWriter and SqsWriter. In both proxies, the configuration is done with Region, which is read by AWS_REGION environment variable. This variable is always present in the AWS Lambda environment. In this case, there is no need for authentication with AWSCredentials class, as everything happens inside AWS. The lambda function has two constructors, one is receiving instances of both proxies, if none are passed it instantiates them. An automatic dependency injection is not used with lambda. This constructor is used by the unit tests to pass mocked objects. The parameterless constructor is needed by AWS Lambda to instantiate the function class.

MoviesFunction

public class MoviesFunction
{
	private readonly ISqsWriter _sqsWriter;
	private readonly IDynamoDbWriter _dynamoDbWriter;

	public MoviesFunction() : this(null, null) { }

	public MoviesFunction(ISqsWriter sqsWriter, IDynamoDbWriter dynamoDbWriter)
	{
		_sqsWriter = sqsWriter ?? new SqsWriter();
		_dynamoDbWriter = dynamoDbWriter ?? new DynamoDbWriter();
	}

	public async Task FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
	{
		context.Logger.LogLine($"Beginning to process {dynamoEvent.Records.Count} records...");

		foreach (var record in dynamoEvent.Records)
		{
			context.Logger.LogLine($"Event ID: {record.EventID}");
			context.Logger.LogLine($"Event Name: {record.EventName}");

			var streamRecordJson = _dynamoDbWriter.SerializeStreamRecord(record.Dynamodb);
			context.Logger.LogLine($"DynamoDB Record:{streamRecordJson}");
			context.Logger.LogLine(streamRecordJson);

			var logEntry = new LogEntry
			{
				Message = $"Movie '{record.Dynamodb.NewImage["Title"].S}' processed by lambda",
				DateTime = DateTime.Now
			};
			await _sqsWriter.WriteLogEntryAsync(logEntry);
			await _dynamoDbWriter.PutLogEntryAsync(logEntry);
		}

		context.Logger.LogLine("Stream processing complete.");
	}
}

DynamoDbWriter

public interface IDynamoDbWriter
{
	Task PutLogEntryAsync(LogEntry logEntry);
	string SerializeStreamRecord(StreamRecord streamRecord);
}

public class DynamoDbWriter : IDynamoDbWriter
{
	private static readonly string Region = Environment.GetEnvironmentVariable("AWS_REGION") ?? "us-east-1";

	private readonly IAmazonDynamoDB _dynamoDbClient;
	private readonly JsonSerializer _jsonSerializer;

	public DynamoDbWriter()
	{
		var dynamoDbConfig = new AmazonDynamoDBConfig
		{
			RegionEndpoint = RegionEndpoint.GetBySystemName(Region)
		};
		_dynamoDbClient = new AmazonDynamoDBClient(dynamoDbConfig);
		_jsonSerializer = new JsonSerializer();
	}

	public async Task PutLogEntryAsync(LogEntry logEntry)
	{
		var request = new PutItemRequest
		{
			TableName = "LogEntries",
			Item = new Dictionary<string, AttributeValue>
			{
				{"Message", new AttributeValue {S = logEntry.Message}},
				{"DateTime", new AttributeValue {S = logEntry.ToString()}}
			}
		};

		await _dynamoDbClient.PutItemAsync(request);
	}

	public string SerializeStreamRecord(StreamRecord streamRecord)
	{
		using (var writer = new StringWriter())
		{
			_jsonSerializer.Serialize(writer, streamRecord);
			return writer.ToString();
		}
	}
}

SqsWriter

public interface ISqsWriter
{
	Task WriteLogEntryAsync(LogEntry logEntry);
}

public class SqsWriter : ISqsWriter
{
	private static readonly string QueueName = Environment.GetEnvironmentVariable("AWS_SQS_QUEUE_NAME");
	private static readonly bool IsQueueFifo = bool.Parse(Environment.GetEnvironmentVariable("AWS_SQS_IS_FIFO") ?? "false");
	private static readonly string Region = Environment.GetEnvironmentVariable("AWS_REGION") ?? "us-east-1";

	private readonly IAmazonSQS _sqsClient;

	public SqsWriter()
	{
		var sqsConfig = new AmazonSQSConfig
		{
			RegionEndpoint = RegionEndpoint.GetBySystemName(Region)
		};
		_sqsClient = new AmazonSQSClient(sqsConfig);
	}

	public async Task WriteLogEntryAsync(LogEntry logEntry)
	{
		var queueUrl = await _sqsClient.GetQueueUrlAsync(QueueName);
		var sendMessageRequest = new SendMessageRequest
		{
			QueueUrl = queueUrl.QueueUrl,
			MessageBody = JsonConvert.SerializeObject(logEntry),
			MessageAttributes = SqsMessageTypeAttribute.CreateAttributes(typeof(LogEntry).Name)
		};
		if (IsQueueFifo)
		{
			sendMessageRequest.MessageGroupId = typeof(LogEntry).Name;
			sendMessageRequest.MessageDeduplicationId = Guid.NewGuid().ToString();
		}

		await _sqsClient.SendMessageAsync(sendMessageRequest);
	}
}

Conclusion

As shown in the current post, it is very easy to create lambda function, deploy and run it. The current post shows a lambda function that listens to DynamoDB events and processes those events by writing into another DynamoDB table and SQS queue.

Related Posts

Read more...

AWS examples in C# – create a service working with DynamoDB

Last Updated on by

Post summary: Introduction to NoSQL, introduction to DynamoDB and what are its basic features and capabilities.

This post is part of AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS series. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository. In the current post, I give an overview of DyanmoDB and what it can be used for.

NoSQL database

NoSQL database provides a mechanism for storage and retrieval of data that is modeled in means other than the tabular relations used in relational databases (RDBMS). There are several types of NoSQL databases:

  • Key-value stores – every single item in the database is stored as an attribute name (or ‘key’), together with its value.
  • Document databases – pair each key with a complex data structure known as a document, usually, it is a JSON document. Documents can contain many different key-value pairs, or key-array pairs, or even nested documents.
  • Graph stores – used to store information about networks of data. Data is organized in the form of nodes and connections between the nodes.
  • Wide-column stores – store columns of data together, instead of rows. It can query large data volumes faster than conventional relational databases.

A very good article on the NoSQL topic is NoSQL Databases Explained.

AWS DynamoDB

Amazon DynamoDB is a key-value and document database that delivers single-digit millisecond performance at any scale. It’s a fully managed, multi-region, multi-master, durable database with built-in security, backup and restore, and in-memory caching for internet-scale applications.

DynamoDB tables

DynamoDB stores data in tables. The data is represented as items, which have attributes. When a table is created, along with its name, a primary key should be provided. The primary key can consist only of a partition key (HASH), it is mandatory. DynamoDB uses an internal hash function to evenly distribute data items across partitions, based on their partition key values. The primary key can also consist of the partition key and sort key (RANGE), which is complementary to the partition. DynamoDB stores items with the same partition key physically close together, in sorted order by the sort key value.

Secondary indexes

DynamoDB offers the possibility to define so-called secondary indexes. There are two types – global and local. A global secondary index is a one that has a partition, a HASH, key different than the HASH key or the table, each table has a limit of 20 global indexes. A local index is one that has the same partition key but different sorting key. Up to 5 local secondary indexes per table are allowed. Properly managing those indexes is the key to using efficiently DynamoDB as a storage unit.

Streams

DynamoDB Streams is an optional feature that captures data modification events in DynamoDB tables. The data about different DynamoDB events appear in the stream in near-real-time, and in the order that the events occurred. Each event is represented by a stream record in case of add, update or delete an item. Stream records can be configured what data to hold, they can have the old and the new item, or only one of them if needed, or even only the keys. Stream records have a lifetime of 24 hours, after that, they are automatically removed from the stream. Streams are used together with AWS Lambda to create a trigger code that executes automatically whenever an event appears in a stream.

Read/Write Capacity Mode

Amazon DynamoDB has two read/write capacity modes for processing reads and writes on your tables: on-demand and provisioned, which is the default, free-tier eligible mode. The read/write capacity mode controls how charges are applied to read and write throughput and how to manage capacity. The capacity mode is set when the table is created and it can be changed later. The provisioned mode is the default one, it is recommended to be used in case of known workloads. The on-demand mode is recommended to be used in case of unpredictable and unknown workloads. DynamoDB provides auto-scaling capabilities so the table’s provisioned capacity is adjusted automatically in response to traffic changes.

Understanding the concept around read and write capacity units is tricky. One write capacity unit is up to 1KB of data per second. If write is done in a transaction though, then the capacity unit count doubles. An example is if there is 2KB of data to be written per second, then the table definition needs 2 write capacity units. If the write is done in a transaction though, then 4 capacity units have to be defined. Read capacity unit is similar, with the difference that there are two flavors of reading – strongly consistent read and eventually consistent read. An eventually consistent read means, that data returned by DynamiDB might not be up to date and some write operation might not have been refracted to it. If data should be guaranteed to be propagated on all DynamoDB nodes and it is up-to-date data, then strongly consistent read is needed. One read capacity unit gives one strongly consistent read or two eventually consistent reads for data up to 4KB. Transactions double the count if read units needed, hence two units are required to read data up to 4KB. For example, if the data to be read is 8 KB, then 2 read capacity units are required to sustain one strongly consistent read per second, 1 read capacity unit if in case of eventually consistent reads, or 4 read capacity units for a transactional read request.

In case the application exceeds the provisioned throughput capacity on a table or index, then it is subject to request throttling. Throttling prevents the application from consuming too many capacity units. When a request is throttled, it fails with an HTTP 400 code (Bad Request) and a ProvisionedThroughputExceededException. The AWS SDKs have built-in support for retrying throttled requests, so no custom logic is needed.

Different programmatic interfaces

Every AWS SDK provides one or more programmatic interfaces for working with Amazon DynamoDB. These interfaces range from simple low-level DynamoDB wrappers to object-oriented persistence layers. The available interfaces vary depending on the AWS SDK and programming language that you use. For C# available interfaces are low-level interface, document interface and object persistence interface. In AWS examples in C# – basic DynamoDB operations post I have given detailed code examples of all of them.

Low-level interface

The low-level interface lets the consumer manage all the details and do the data mapping. Data is mapped manually to its proper data type. Supported data types are:

  • B – binary value, a MemoryStream
  • BS – list of MemoryStream objects
  • S – string
  • SS – list of string objects
  • N – number converted into a string
  • NS – list of number strings
  • BOOL – boolean
  • L – list of AttributeValue objects
  • M – map, dictionary of AttributeValue objects
  • NULL – if set to true, then this is a null value

If the low-level interface is used for querying then a KeyConditionExpression is used to query the data. It is called a query, but it not actually a query in terms of RDBMS way of thinking, as the HASH key should be only used with an equality operator. For the RANGE key, there is a variety of operators to be used, such as:

  • sortKeyName = :sortkeyval – true if the sort key value is equal to :sortkeyval
  • sortKeyName < :sortkeyval – true if the sort key value is less than :sortkeyval
  • sortKeyName <= :sortkeyval – true if the sort key value is less than or equal to :sortkeyval
  • sortKeyName > :sortkeyval – true if the sort key value is greater than :sortkeyval
  • sortKeyName >= :sortkeyval – true if the sort key value is greater than or equal to :sortkeyval
  • sortKeyName BETWEEN :sortkeyval1 AND :sortkeyval2 – true if the sort key value is greater than or equal to :sortkeyval1, and less than or equal to :sortkeyval2
  • begins_with ( sortKeyName, :sortkeyval ) – true if the sort key value begins with a particular operand. (You cannot use this function with a sort key that is of type Number.) Note that the function name begins_with is case-sensitive.

Document interface

The document programming interface returns the full document by its unique HASH key. The document is actually a JSON.

{
	"Title": {
		"Value": "Die Hard",
		"Type": 0
	},
	"Genre": {
		"Value": "0",
		"Type": 1
	}
}

Object persistence interface

WIth object persistency client classes are mapped to DynamoDB tables. There are several attributes that can be applied to database model classes, such as  DynamoDBTable, DynamoDBHashKey, DynamoDBRangeKey, DynamoDBProperty, DynamoDBIgnore, etc. To save the client-side objects to the tables, the object persistence model provides the DynamoDBContext class, an entry point to DynamoDB. This class provides a connection to DynamoDB and enables you to access tables, perform various CRUD operations.

Architectural constraints

Understanding DynamoDB nature is important in order to design a service that works with it. It is important to cost-efficiently define the table capacity. If less capacity is defined, then consumers can get 400 responses, the other extreme is to generate way too much cost. Another aspect is reading the data. DynamoDB does not provide a way to search for data. In any case, the application that used DynamoDB has to have a proper way to access the data by key.

Using DynamoDB in a service

DynamoDB can be straight forward used in a service, such as SqsReader or ActorsServerlessLambda and MoviesServerlessLambda functions, see the bigger picture in AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS post. An AmazonDynamoDBClient is instantiated and used with one of the programming interfaces described above.

Another important usage is to subscribe to and process stream events. This is done in both ActorsLambdaFunction and MoviessLambdaFunction. See more details about Lambda usage in AWS examples in C# – working with Lambda functions post.

More information on how to run the solution can be found in AWS examples in C# – run the solution post.

Conclusion

In the current post, I have given a basic overview of DynamoDB. It is important to understand its specifics in order to use it efficiently.

Related Posts

Read more...

AWS examples in C# – basic DynamoDB operations

Last Updated on by

Post summary: Code examples with DynamoDB write and read operations.

This post is part of AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS series. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository. In the current post, I give practical code examples of how to work with DynamoDB.

Instantiate Amazon DynamoDB client

In the current examples, in SqsReader project, a configuration class called AppConfig is used. Its values are injected from the environment variables by .NET Core framework in Startup class. In order to work with DynamoDB, a client is needed. The DynamoDB client interface is called IAmazonDynamoDB and comes from AWS C# SDK. The NuGet package is called AWSSDK.DynamoDBv2. The concrete AWS client implementation is AmazonDynamoDBClient and an object is instantiated in DynamoDbClientFactory class and used as a singleton. RegionEndpoint is used to instantiate AmazonDynamoDBConfigAwsCredentials class extends the AWS’ abstract AWSCredentials and is used in order to manage the credentials.

DynamoDbClientFactory.cs

public static AmazonDynamoDBClient CreateClient(AppConfig appConfig)
{
	var dynamoDbConfig = new AmazonDynamoDBConfig
	{
		RegionEndpoint = RegionEndpoint.GetBySystemName(appConfig.AwsRegion)
	};
	var awsCredentials = new AwsCredentials(appConfig);
	return new AmazonDynamoDBClient(awsCredentials, dynamoDbConfig);
}

AwsCredentials.cs

public class AwsCredentials : AWSCredentials
{
	private readonly AppConfig _appConfig;

	public AwsCredentials(AppConfig appConfig)
	{
		_appConfig = appConfig;
	}

	public override ImmutableCredentials GetCredentials()
	{
		return new ImmutableCredentials(_appConfig.AwsAccessKey,
						_appConfig.AwsSecretKey, null);
	}
}

AppConfig.cs

public class AppConfig
{
	public string AwsRegion { get; set; }
	public string AwsAccessKey { get; set; }
	public string AwsSecretKey { get; set; }
}

Creating tables

DatabaseClient class that uses and exposes just a few methods of IAmazonDynamoDB is custom created. It checks if the table is created and if it is not, then it creates the table. Afterward, it waits for the table to become in status ACTIVE. Movies and Actors tables creation is done in separate classes with CreateTableRequest, which needs the table name. KeySchema specifies the attributes that build the primary key for a table or an index. The attributes must also be defined in the AttributeDefinitions list. KeyType has two possible values – HASH and RANGE. In the case of the Movies table, there is only a HASH key, which is always mandatory and unique, this means no two items can have the same partition key value, the second insert overwrites the first one. In the case of the Actors table, along with the partition key, there is also a sort key with KeyType of RANGE which is complimentary to the HASH. I have not used secondary indexes in the current example, but DynamoDB provides this functionality. They can be defined with GlobalSecondaryIndexes and LocalSecondaryIndexes elements of the CreateTableRequest. A stream is defined with StreamSpecification element in the CreateTableRequest, its StreamViewType is NEW_AND_OLD_IMAGES. This means that in case of add, update or delete, the DynamoDBEvent, which is later used in a lambda, holds both the new values and the old values of the item. ProvisionedThroughput is used to set the read and write capacity mode. In the current example, it is 5 capacity units for reading and the same for writing. The ProvisionedThroughput is needed because the default BillingMode is PROVISIONED. It can be changed to PAY_PER_REQUEST and then ProvisionedThroughput should not be specified. A more detailed explanation of each parameter can be found in AWS examples in C# – create a service working with DynamoDB post.

MoviesRepository.cs

public async Task CreateTableAsync()
{
	var request = new CreateTableRequest
	{
		TableName = TableName,
		KeySchema = new List<KeySchemaElement>
		{
			new KeySchemaElement
			{
				AttributeName = "Title",
				KeyType = "HASH"
			}
		},
		AttributeDefinitions = new List<AttributeDefinition>
		{
			new AttributeDefinition
			{
				AttributeName = "Title",
				AttributeType = "S"
			}
		},
		ProvisionedThroughput = new ProvisionedThroughput
		{
			ReadCapacityUnits = 5,
			WriteCapacityUnits = 5
		},
		StreamSpecification = new StreamSpecification
		{
			StreamEnabled = true,
			StreamViewType = StreamViewType.NEW_AND_OLD_IMAGES
		}
	};

	await _client.CreateTableAsync(request);
}

ActorsRepository.cs

public async Task CreateTableAsync()
{
	var request = new CreateTableRequest
	{
		TableName = TableName,
		KeySchema = new List<KeySchemaElement>
		{
			new KeySchemaElement
			{
				AttributeName = "FirstName",
				KeyType = "HASH"
			},
			new KeySchemaElement
			{
				AttributeName = "LastName",
				KeyType = "RANGE"
			}
		},
		AttributeDefinitions = new List<AttributeDefinition>
		{
		   new AttributeDefinition
			{
				AttributeName = "FirstName",
				AttributeType = "S"
			},
			new AttributeDefinition
			{
				AttributeName = "LastName",
				AttributeType = "S"
			}
		},
		ProvisionedThroughput = new ProvisionedThroughput
		{
			ReadCapacityUnits = 5,
			WriteCapacityUnits = 5
		},
		StreamSpecification = new StreamSpecification
		{
			StreamEnabled = true,
			StreamViewType = StreamViewType.NEW_AND_OLD_IMAGES
		}
	};

	await _client.CreateTableAsync(request);
}

DatabaseClient.cs

private const string StatusUnknown = "UNKNOWN";
private const string StatusActive = "ACTIVE";

private readonly IAmazonDynamoDB _client;

public DatabaseClient(IAmazonDynamoDB client)
{
	_client = client;
}

public async Task CreateTableAsync(CreateTableRequest createTableRequest)
{
	var status = await GetTableStatusAsync(createTableRequest.TableName);
	if (status != StatusUnknown)
	{
		return;
	}

	await _client.CreateTableAsync(createTableRequest);

	await WaitUntilTableReady(createTableRequest.TableName);
}

public async Task PutItemAsync(PutItemRequest putItemRequest)
{
	await _client.PutItemAsync(putItemRequest);
}

private async Task<string> GetTableStatusAsync(string tableName)
{
	try
	{
		var response = await _client.DescribeTableAsync(new DescribeTableRequest
		{
			TableName = tableName
		});
		return response?.Table.TableStatus;
	}
	catch (ResourceNotFoundException)
	{
		return StatusUnknown;
	}
}

private async Task WaitUntilTableReady(string tableName)
{
	var status = await GetTableStatusAsync(tableName);
	for (var i = 0; i < 10 && status != StatusActive; ++i)
	{
		await Task.Delay(500);
		status = await GetTableStatusAsync(tableName);
	}
}

Different programmatic interfaces

In AWS examples in C# – create a service working with DynamoDB post, I have explained more details about the three different programmatic interfaces, that DynamoDB offers, a low-level interface, document interface, and object persistence interface.

Writing using the low-level interface

The low-level interface lets the consumer manage all the details and do the data mapping. Here is an example of how to create an Actor using the low-level interface. Data is mapped manually to its proper data type. In this case, the actor.FirstName and actor.LastName is assigned to the S property of the AttributeValue, which is a string type.

private readonly IDatabaseClient _client;
public async Task SaveActorAsync(Actor actor)
{
	var request = new PutItemRequest
	{
		TableName = TableName,
		Item = new Dictionary<string, AttributeValue>
		{
			{"FirstName", new AttributeValue {S = actor.FirstName}},
			{"LastName", new AttributeValue {S = actor.LastName}}
		}
	};
	await _client.PutItemAsync(request);
}

The full code is in ActorsRepository.cs.

Writing using the object persistence interface

With the object persistency interface, client classes are mapped to DynamoDB tables. The example given below comes from the original AWS documentation and shows explicit mapping. With DynamoDBTable the mapping to the table is created, then DynamoDBHashKey and DynamoDBRangeKey annotate the keys. With DynamoDBProperty a specific name can be given, so it is different from the table field name. Title is directly mapped to Title field in the database table. DynamoDBIgnore attribute ignores writing and reading this particular property to and from the table.


[DynamoDBTable("ProductCatalog")]
public class Book
{
	[DynamoDBHashKey]
	public int Id { get; set; }

	public string Title { get; set; }

	[DynamoDBRangeKey]
	public int ISBN { get; set; }

	[DynamoDBProperty("Authors")]
	public List<string> BookAuthors { get; set; }

	[DynamoDBIgnore]
	public string CoverPage { get; set; }
}

To save the client-side objects to the tables, the object persistence model provides the DynamoDBContext class, an entry point to DynamoDB. This class provides a connection to DynamoDB and enables you to access tables and perform various CRUD operations. The current examples are slightly different since Movie model is very simple, there are no DynamoDB attributes on it, so DynamoDBContext uses its default mapping features to map them. The movie has two properties, Title, which is a string and is the HASH key in the table and Genre, which is an enum, practically an integer.


public enum MovieGenre
{
	[EnumMember(Value = "Action Movie")]
	Action,
	[EnumMember(Value = "Drama Movie")]
	Drama
}

public class Movie
{
	public string Title { get; set; }

	[JsonConverter(typeof(StringEnumConverter))]
	public MovieGenre Genre { get; set; }
}

Since there is no DynamoDBTable attribute of the model, then DynamoDBContext is trying to map it by default to a table with the name Movie, but such a table does not exist. This is why DynamoDBOperationConfig is needed to map to the correct table name.


private readonly IDynamoDBContext _context;

public async Task SaveMovieAsync(Movie movie)
{
	var operationConfig = new DynamoDBOperationConfig
	{
		OverrideTableName = "Movies"
	};
	await _context.SaveAsync(movie, operationConfig);
}

The full code is in MoviesRepository.cs. Object persistence interface is a wide topic, full details can be found in .NET: Object Persistence Model page.

Querying using the low-level interface

An example is given for query request for Actors table that has FirstName as a HASH key and LastName as RANGE key. Important here is KeyConditionExpression, it holds the actual query. It is called a query, but it not actually a query in terms of RDBMS way of thinking, as the HASH key should be only used with an equality operator. For the RANGE key, there is a variety of operators to be used, in the example given equality operator is used as well. To add value to the value placeholder, :FirstName in the example, ExpressionAttributeValues is used. The dictionary key is the placeholder value, and AttributeValue is the value mapped to a specific value type, in the example, it is S, for a string. It is also possible to give placeholder value for the table field name as well, which is then replaced with the actual value in ExpressionAttributeNames dictionary, such as #LastName.

private static QueryRequest BuildQueryRequest(string firstName, string lastName)
{
	var request = new QueryRequest("Actors")
	{
		KeyConditionExpression = "FirstName = :FirstName"
	};
	request.ExpressionAttributeValues.Add(":FirstName", new AttributeValue
	{
		S = firstName
	});

	if (!string.IsNullOrEmpty(lastName))
	{
		request.KeyConditionExpression += " AND #LastName = :LastName";
		request.ExpressionAttributeNames.Add("#LastName", "LastName");
		request.ExpressionAttributeValues.Add(":LastName", new AttributeValue
		{
			S = lastName
		});
	}

	return request;
}

The full code is in ActorsHandler.cs. More about DynamoDB queries can be found in DynamoDB API_Query page.

Get item using document interface

The document programming interface returns the full document by its unique HASH key. The table is accessed with public static Table LoadTable(IAmazonDynamoDB ddbClient, TableConfig config) and then the document is loaded with public Task<Document> GetItemAsync(Primitive hashKey). In current examples, a proxy class is defined, which isolates the IAmazonDynamoDB operations:

GetDocumentAsync


public async Task<Document> GetDocumentAsync(string tableName, string documentKey)
{
	var table = Table.LoadTable(_dynamoDbClient, new TableConfig(tableName));
	return await table.GetItemAsync(new Primitive(documentKey));
}

GetDocumentAsync

var document = await _dynamoDbReader.GetDocumentAsync(TableName, title);

var movie = new Movie
{
	Title = document["Title"],
	Genre = (MovieGenre)int.Parse(document["Genre"])
};

The document is actually a JSON.

{
	"Title": {
		"Value": "Die Hard",
		"Type": 0
	},
	"Genre": {
		"Value": "0",
		"Type": 1
	}
}

The full code is in MoviesHandler.cs.

Conclusion

In the current post, I have given practical code examples of how to do the basic DynamoDB operations in C#. This post is complimentary to AWS examples in C# – create a service working with DynamoDB post.

Related Posts

Read more...

AWS examples in C# – create a service working with SQS

Last Updated on by

Post summary: To give a basic overview of AWS SQS, how to write a message to it and how to make a consumer that constantly polls the queue for new messages.

This post is part of AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS series. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository.

Event-driven architecture

I would like to briefly touch the topic of event-driven architecture since message service providers, such as SQS or RabbitMQ are the basis of its implementation. This is a software architecture paradigm promoting the production, detection, consumption of, and reaction to events. An event is a significant change in the state of an object, to which someone might be interested in. All communication happens asynchronously and systems are loosely coupled. An event-driven system typically consists of event emitters, event consumers, and event channels. Emitters have the responsibility to detect, gather, and transfer events. Emitters do not know the consumers of the events, they do not even know if a consumer exists. Consumers have the responsibility of applying a reaction as soon as an event is presented in a dedicated channel. This leads to the pattern commonly known as eventual consistency, which pushes the complexity of consistency to the application tier, which is the biggest challenge to solve in an event-driven architecture.

Apart from SQS, there is even more sophisticated service from AWS called EventBridge, which makes it easy to build event-driven applications because it takes care of event ingestion and delivery, security, authorization, and error handling. It is basically a serverless event bus that makes it easy to connect applications together using data from its own applications, integrated Software-as-a-Service (SaaS) applications, and AWS services.

AWS SQS

SQS stands for Simple Queue Service, it is a fully managed message queuing service that enables to decouple and scale microservices, distributed systems, and serverless applications. SQS eliminates the complexity and overhead associated with managing and operating message-oriented middleware and empowers developers to focus on differentiating work.

Types of queues

SQS offers two types of message queues:

  • Standard queues – they offer maximum throughput, best-effort ordering, and at-least-once delivery. This means there is no guaranteed order and messages can be duplicated.
  • FIFO queues – they are designed to guarantee that messages are processed exactly once, in the exact order that they are sent.

Dead-letter queue

In addition to those, there is a special type of queues, called dead-letter queues. They are used mainly for debugging and failure proofing applications. If a message cannot be successfully processed after several retries from one of the source queues above, it ends in the dead-letter queue, from which it can be analyzed and returned back to source queue for reprocessing.

Message processing

It is important to know how SQS operates, in order to make good architectural decisions. When a message is published to the queue it becomes visible. When some consumer reads the message, then the message becomes not visible, but still present in the queue, its status now is in-flight. There is visibility timeout which by default is 30 seconds, the maximum value is 12 hours. After the visibility timeout passes then the message is visible again to be read by consumers. In case there is no dead-letter queue, this process happens over and over until the message retention period is reached, afterward message gets automatically deleted. The retention period default value is 4 days, the maximum value is 14 days. In case of a dead-letter queue, after the message cannot be processed for more than maximum receive count times, then it goes to dead-letter queue and stays in the dead-letter for its message retention period. See more info on SQS on How Amazon SQS Works page.

Architectural approaches

One queue or many queues

Since many event emitters can write messages to the queue it gets tricky to process the messages properly. One option is to have a separate queue for separate types of messages, another option is to put some metadata into the messages. I have decided to go for the solution with one queue because I have just one consumer which knows which message processor to call and thus simplify the code. In the case of many SQS queues, there should be many consumers defined in the code, which is better to split into many micro-services, for each SQS queue.

Dead-letter

I would say a dead-letter queue with the maximum retention period of 14 days is a good idea. In this case, messages can be quarantined which will not slow down the normal queue operations. In the case of no dead-letter queue and default timeouts, if a message cannot be processed, then it will appear every 30 seconds for a period of 4 days, this makes 2880 times a day, 11520 times in total. Now imagine there are thousands of messages like this one. I have decided to go for a dead-letter queue with the default retention period.

Long polling

Long polling is another aspect that has to be considered. It can be enabled in two ways. One is on a queue level, by setting the ReceiveMessageWaitTimeSeconds when creating the queue, it can be from 1 to 20 seconds. Other way to enable it is when messages are read from the queue, there is WaitTimeSeconds setting in the request, which can be from 1 to 20 seconds. In case both options are combined, then WaitTimeSeconds takes precedence.

Unknown messages

Another architectural decision in case there is only one queue is what to be done with unknown messages. In the case of no dead-letter queue, messages are good to be deleted, otherwise, they will keep showing for the queue’s retention period. I throw an error in the logs and after 3 unsuccessful attempts, which is the receive count times I have configured, the message goes to the dead-letter queue.

Standard vs. FIFO queues

SQS is able to handle a high amount of messages, theoretically an unlimited amount of messages per second. Standard SQS queue does not maintain any order of messages and also it is possible that there is a duplication of messages delivery. For this reason, AWS offers a FIFO (First-In-First-Out) queue, they provide message order and ensure exactly-once processing. The limitation of the FIFO queue is its number of transactions per second, which are 300 messages per second or 3000 if they are in batched mode.

SQS queue operations at a glance

In AWS examples in C# – basic SQS queue operations post following the operations briefed below were described in more details:

  • Create queue with dead-letter queue
  • Read messages from the queue
  • Write a message to the queue (comes in two flavors)
  • Delete messages to the queue
  • Move messages from dead-letter to source queue

Creating SQS message consumer

In order to read the messages, there should be a consumer that constantly polls the queue and processes the messages. ProcessMessageAsync uses the strategy design pattern to get the proper message processor based on MessageType attribute. Processors are stored in _messageProcessors which is IEnumerable<IMessageProcessor> and is injected by .NET Core dependency injection. If a processor is found, then the processor is invoked, if not an error is shown in the logs. This logic can be subject to change if unknown messages are tolerated in the queue. In ProcessAsync method there is a while loop, which constantly reads for messages by _sqsClient which SqsClient class described in previous sections. SQS returns the response if there are some messages or if WaitTimeSeconds time expired when reading the message or ReceiveMessageWaitTimeSeconds configured by AwsQueueLongPollTimeSeconds environment variable has expired. This while loop is a little tricky to unit test though as it consumes the main thread, and the mocked object should be instructed to wait. Everything is controlled by a CancellationTokenSource, when this is canceled, then consumption is stopped.

ProcessMessageAsync

private async Task ProcessMessageAsync(Message message)
{
	try
	{
		var messageType = message.MessageAttributes.GetMessageTypeAttributeValue();
		if (messageType == null)
		{
			throw new Exception($"No 'MessageType' attribute present in message {JsonConvert.SerializeObject(message)}");
		}

		var processor = _messageProcessors.SingleOrDefault(x => x.CanProcess(messageType));
		if (processor == null)
		{
			throw new Exception($"No processor found for message type '{messageType}'");
		}

		await processor.ProcessAsync(message);
		await _sqsClient.DeleteMessageAsync(message.ReceiptHandle);
	}
	catch (Exception ex)
	{
		_logger.LogError(ex, $"Cannot process message [id: {message.MessageId}, receiptHandle: {message.ReceiptHandle}, body: {message.Body}] from queue {_sqsClient.GetQueueName()}");
	}
}

ProcessAsync

private async void ProcessAsync()
{
	try
	{
		while (!_tokenSource.Token.IsCancellationRequested)
		{
			var messages = await _sqsClient.GetMessagesAsync(_tokenSource.Token);
			messages.ForEach(async x => await ProcessMessageAsync(x));
		}
	}
	catch (OperationCanceledException)
	{
		//operation has been canceled but it shouldn't be propagated
	}
}

StartConsuming

public void StartConsuming()
{
	if (!IsConsuming())
	{
		_tokenSource = new CancellationTokenSource();
		ProcessAsync();
	}
}

private bool IsConsuming()
{
	return _tokenSource != null && !_tokenSource.Token.IsCancellationRequested;
}

Message processors

In the current example, I have taken the architectural design decision to have one queue and different messages into it. For each different type of message, there is a relevant processor. With the strategy design pattern, the appropriate message processor is picked based on MessageType attribute. Processors implement a very simple interface IMessageProcessor. In the current example, they take the message as a string, serialize it to an object and save this object to DynamoDB. A sample implementation is shown below:

IMessageProcessor

public interface IMessageProcessor
{
	bool CanProcess(string messageType);
	Task ProcessAsync(Message message);
}

ActorMessageProcessor

public bool CanProcess(string messageType)
{
	return messageType == typeof(Actor).Name;
}

public async Task ProcessAsync(Message message)
{
	var actor = JsonConvert.DeserializeObject<Actor>(message.Body);
	await _actorsRepository.SaveActorAsync(actor);
	_logger.LogInformation($"ActorMessageProcessor invoked with: {message.Body}");
}

AWS ECS and AWS ECR

ECS stands for Elastic Container Service is a fully managed container orchestration service. Containers can be run in clusters using AWS Fargate, which is a serverless compute for containers. Fargate removes the need to provision and manage servers, lets you specify and pay for resources per application, and improves security through application isolation by design.

ECR stands for Elastic Container Registry is a fully-managed Docker container registry that makes it easy for developers to store, manage, and deploy Docker container images. ECR is integrated with ECS, eliminating the need to operate own container repositories or worry about scaling the underlying infrastructure.

SqsWriter and SqsReader

SqsWriter is a .NET Core 3.0 application, that is dockerized and run in AWS ECS with Fargate, and its container images are stored in ECR. It exposes an API that can be used to publish Actor or Movie objects as messages with separate MessageType attributes in the SQS queue.

SqsReader is a .NET Core 3.0 application, that is dockerized and run in AWS ECS with Fargate, and its container images are stored in ECR. It has a consumer that listens to the SQS queue and processes the messages by writing them into appropriate AQS DynamoDB tables. It also exposes API to stop or start processing, as long as reprocess the dead-letter queue or simply get the queue status.

More information on how to run the solution can be found in AWS examples in C# – run the solution post.

Conclusion

In the current post, I have given some concepts of event-driven architecture and how SQS fits in it. Also, I have described some architectural considerations when using SQS queues, such as dead-letter queues, one queue with different message type or several queues, etc. In the end, I have given practical code on how to make a consumer for the SQS queue.

Related Posts

Read more...

AWS examples in C# – working with SQS, DynamoDB, Lambda, ECS

Last Updated on by

Post summary: Overview of the AWS examples in C# series.

In several blog posts, I give some practical examples of how to use AWS SQS, DynamoDB, Lambda with C# code. The code used for this series of blog posts is located in aws.examples.csharp GitHub repository.

Introduction

AWS stands for Amazon Web Services, it is a subsidiary of Amazon that provides on-demand cloud computing platforms and APIs to individuals, companies, and governments, on a metered pay-as-you-go basis. AWS is one of the big cloud service providers. The others are Microsoft Azure and Google Cloud. All three cloud service providers have functions that are semantically common but differ in practical implementation. Also, every one of them has its own flavors. I have chosen to use AWS for these examples as it is something I have used before and I am most comfortable with it.

Architectural overview

In order to get a full understanding of the architecture, I have prepared this very basic diagram. It illustrates what services are there and how they communicate.

SqsReader and SqsWriter

Both are .NET Core 3.0 microservices running in docker containers. The images are uploaded in AWS ECR (Elastic Container Registry) and containers are run in AWS ECS (Elastic Container Service). SqsReader has a REST endpoint, by which an Actor or Movie can be posted. Both are pushed as a message to AWS SQS (Simple Queue Service). SqsWriter is listening to the SQS and in case of a message, it processes it. If the message is from type Actor or Movie then SqsReader saves it to the respective AWS DynamoDB tables. If the message is LogEntry, then the message is only output into SqsReader logs.

ActorsLambdaFunction and MoviessLambdaFunction

Both are .NET Core 2.1 lambda functions run in AWS Lambda. They listen to Actors and Movies DynamoDB tables changes and in case of new entries, they write to LogEntries DynamoDB table. Also, they write SQS messages from type LogEntry, which are then read by SqsReader.

ActorsServerlessLambda and MoviesServerlessLambda

Those are again lambda functions by are fully managed by the Serverless framework. They have a lambda application defined as well as Cloud Formation templates. They expose a REST API trough AWS API Gateway, by which the Actors table can be queried or a movie can be got from the Movies table.

Post in the series

This is a long series of posts describing into detail all the pieces of the architectural diagram above. Also, every aspect of the code in the repository is explained in detail in subsequent blog posts. It was a very interesting learning opportunity for me, which I would like to share. Here are the posts in the series:

Future plans

There are several topics I would like to go into as well, but there is no code yet for them into the GitHub repository. Those are:

  • AWS examples in C# – manage with Terraform
  • AWS examples in C# – use AWS Cognito for API Gateway authorizer

Conclusion

These series of posts are intended to give some basic overview of important AWS services and how to use them in C# code.

Related Posts

Read more...