PUB SUB in AWS Lambda via SNS using C#

March 21, 2018 by boro

AWS Lambda’s are a great replacement for things like Windows Services which need to run common tasks periodically. A few examples would be triggering scheduled backups or polling urls.

You can set many things as the trigger for a lambda, for scheduled operations this can be a CRON value triggered from a CloudWatch event. Alternatively lambda’s can be triggered via a subscription to an SNS topic.

Depending on the type of operation you want to perform on a schedule you might find it takes longer than the timeout restriction imposed by AWS. If that’s the case then a simple PUB SUB (publisher, subscriber) configuration should help.

Sample scenario

We want to move databases backups between 2 buckets in S3. There are several databases to copy, each of which being a large file.

In one lambda you can easily find all the files to copy but if you also try to copy them all at some point your function will timeout.

Pub sub to the rescue

Why not setup 2 lambda functions? One as the publisher and one as the subscriber, and then glue the two together with an SNS topic (Simple Notification Service)

The publisher

Typically this would be triggered from a schedule and would look to raise events for each operations. Lets assume we use a simple POCO for converying the information we need:

class UrlRequestMessage
{
    public string[] Urls {get;set;}
}

class UrlRequestMessage

{

public string[] Urls {get;set;}

}

public class Function
{
	public string FunctionHandler(object input, ILambdaContext context)
	{
                //This could gather urls to poll from a file, db or anywhere
		var urlsToScan = LoadUrls();

		var snsClient = new AmazonSimpleNotificationServiceClient();

		int batchSize = 4;

		context.Logger.LogLine($"Batch size: {batchSize}");

		foreach (var urlToScan in urlsToScan.Batch(batchSize))
		{
			snsClient.PublishAsync("topic arn e.g. arn:aws:sns:eu-west-1:98976####:UrlPoller_Topic",
				JsonConvert.SerializeObject(new UrlRequestMessage {Urls = urlToScan.ToArray()})).Wait();

			context.Logger.LogLine($"Raised event for urls: {String.Join(" | ", urlToScan)}");
		}

		return "ok";
	}
}

public class Function

{

public string FunctionHandler(object input, ILambdaContext context)

{

//This could gather urls to poll from a file, db or anywhere

var urlsToScan = LoadUrls();

var snsClient = new AmazonSimpleNotificationServiceClient();

int batchSize = 4;

context.Logger.LogLine($"Batch size: {batchSize}");

foreach (var urlToScan in urlsToScan.Batch(batchSize))

{

snsClient.PublishAsync("topic arn e.g. arn:aws:sns:eu-west-1:98976####:UrlPoller_Topic",

JsonConvert.SerializeObject(new UrlRequestMessage {Urls = urlToScan.ToArray()})).Wait();

context.Logger.LogLine($"Raised event for urls: {String.Join(" | ", urlToScan)}");

}

return "ok";

}

The batching can be ignored if needs be – in this scenario this allows multiple urls to be handled by one subscriber.

The subscriber

Next we need to listen for the messages – you want to configure the subscriber function to have an SNS trigger that uses the same topic you posted to before.

public class Function
{
	public string FunctionHandler(SNSEvent message, ILambdaContext context)
	{
		foreach (var record in message.Records)
		{
			var decodedMessage = JsonConvert.DeserializeObject<UrlRequestMessage>(record.Sns.Message);

			foreach (var url in decodedMessage.Urls)
			{
                                //here you just need to implement your logic for polling a url
                                // e.g. var result = new WebClient().DownloadStringTaskAsync(url).Result;
				var requestSummary = RequestUrl(url, context.Logger);				
			}
		}

		return "OK!";
	}
}

public class Function

{

public string FunctionHandler(SNSEvent message, ILambdaContext context)

{

foreach (var record in message.Records)

{

var decodedMessage = JsonConvert.DeserializeObject<UrlRequestMessage>(record.Sns.Message);

foreach (var url in decodedMessage.Urls)

{

//here you just need to implement your logic for polling a url

// e.g. var result = new WebClient().DownloadStringTaskAsync(url).Result;

var requestSummary = RequestUrl(url, context.Logger);

}

return "OK!";

}

Debugging things
You can either run each function on demand and see any output directly in the Lambda test window, or dig into your cloudwatch logs for each function.

AWS Lambda now supports Serverless applications including WebApi

February 16, 2018 by boro

One of the most exciting areas that I’ve seen emerging in the Cloud space recently is Serverless computing. Both AWS and Azure have their own flavour: AWS Lambda and Azure Functions.

An intro into Serverless

It really does what it says on the tin. You can run code but without dedicated infrastructure that you host. A good example is when building Alexa Skills.

You create AWS lambda function, in most of the languages of your choice, and then deploy into the cloud. Whenever someone uses your skill the lambda gets invoked and returns the content you need.

Behind the scenes AWS host your function in a container, if it receives traffic the container remains hot. If it doesn’t receive traffic its ‘frozen’. There is a very good description of this at https://medium.com/@tjholowaychuk/aws-lambda-lifecycle-and-in-memory-caching-c9cd0844e072

Your language of choice

AWS Lambda supports a raft of languages: Python, Node, Java, .net core and others. Recently this has been upgraded so that it supports .net core 2.

Doing the legwork

With a basic lambda function you can concoct different handlers (methods) which respond to requests. This allows one lambda to service several endpoints. However, you need to do quite a lot of wiring and it doesn’t feel quite like normal WebApi programming.

Enter the serverless applications

This came right out the blue, but was very cool – Amazon released some starter kits that allow you run both RazorPage and WebApi applications in Lambdas!!! https://aws.amazon.com/blogs/developer/serverless-asp-net-core-2-0-applications/

Woah, you can write normal WebApi and deploy into a lambda. That is big.

Quick, migrate all the things

So I tried this. And in the most part everything worked pretty seamlessly. All the code I’d already written easily mapped into WebApi controllers I could then run locally. Tick.

Deploying was simple, either via Visual Studio or the dotnet lambda tools. Tick.

Using the serverless.template that ships with the starter pack it even setup my Api Gateway. Tick.

Dependency injection thats inherently available in .net core all worked. Tick.

WebApi attribute routing all works. Tick.

So far so good right 🙂

What I haven’t quite cracked yet?

In my original deployment (pre WebApi) I was using API level caching over a couple specific endpoints. This was path based as it was for specific methods. The new API Gateway deployment directs all traffic to a {/proxy+} url in order to route any request to the routing in your WebApi. If you turn caching on here, its a bit of a race, whichever url is hit first will fill the cache for all requests. Untick!

Debugging errors locally don’t always bubble startup errors very well. I have a feeling this isn’t anything Amazon related but is something worth being aware of. E.g. if you mess up your DI, it takes some ctor null debugging to find the cause. Untick.

Summary

I was hugely impressed with WebApi integration. Once the chinks in the path based caching at the API Gateway can get ironed out I’d consider this a very good option for handling API requests.

Watch this space 🙂

Copying large files between S3 buckets

January 30, 2018 by boro

There are many different scenarios you might face where you need to migrate data between S3 buckets, and folders. Depending on the use case you have several options for the language to select.

Lambda’s – this could be Python, Java, JavaScript or C#
Bespoke code – again, this could be any language you select. To keep things different from above, lets add Powershell to the mix

Behind the scenes a lot of these SDK’s call into common endpoints Amazon host. As a user you don’t really need to delve too deeply into the specific endpoints unless you really need to.

Back to the issue at hand – copying large files
Amazon impose a limit of roughly 5GB on regular copy operations. If you use the standard copy operations you will probably hit exceptions when the file sizes grow.

The solution is to use the multipart copy. It sounds complex but all the code is provided for you:

Python
This is probably the easiest to do as the boto3 library already does this. Rather than using copy_object, the copy function already handles multipart uploads: http://boto3.readthedocs.io/en/latest/reference/services/s3.html#S3.Client.copy

C#
The C# implementation is slightly more complex however Amazon provide good worked examples: https://docs.aws.amazon.com/AmazonS3/latest/dev/CopyingObjctsUsingLLNetMPUapi.html

Powershell
A close mimic of the C# implementation – someone has ported the C# example into powershell: https://stackoverflow.com/a/32635525/1065332

Happy copying!

Serving personalized content as JSON from Sitecore

October 5, 2017 by boro

As with many tools and approaches to solving technical issues, you can often find many ways to achieve the same output. The challenge I was up against was how to serve personalized content as JSON when being served from Sitecore.

The solution below is one way you can achieve this, undoubtedly there are many more!

The over-arching setup

Think of your JSON feed like a Sitecore page. You will need to break a rule of REST as Sitecore personalization requires session, and therefore isn’t stateless. This will need to be reflected in your consuming app, if you don’t provide an identifier every request it will need to understand and persist cookies between requests.

First up you need to select a device, a layout and some renderings. None of this differs to normal Sitecore development. I’ve found for debugging purposes using a new device works better as you can view the content as a web page as well as JSON.

The layout

Assuming you’ve decided on a device, you will need to setup a layout:

@using Sitecore.Analytics
@using Sitecore.Analytics.Model.Entities
@using Sitecore.Mvc
@{
	Response.ContentType = "text/javascript";
	//note you could easily get the userId from other means e.g. headers / cookies. In fact Querystring isn't great as it shows in path analyzer as different urls.
	var userId = Request.QueryString["userId"];

	if (!String.IsNullOrWhiteSpace(userId))
	{
		if (Tracker.Current != null && Tracker.Current.Session != null)
		{
			Tracker.Current.Session.Identify(userId);			
		}
	}
}
{
	@Html.Sitecore().Placeholder("mainjson")
}

@using Sitecore.Analytics

@using Sitecore.Analytics.Model.Entities

@using Sitecore.Mvc

Response.ContentType = "text/javascript";

//note you could easily get the userId from other means e.g. headers / cookies. In fact Querystring isn't great as it shows in path analyzer as different urls.

var userId = Request.QueryString["userId"];

if (!String.IsNullOrWhiteSpace(userId))

{

if (Tracker.Current != null && Tracker.Current.Session != null)

{

Tracker.Current.Session.Identify(userId);

}

{

@Html.Sitecore().Placeholder("mainjson")

}

The renderings
Again, much like you would for a page, you can create e.g. Controller Renderings which output the content as you need. One thing to note, these will want to render as JSON e.g.:

@model Website.Models.Response

"name":"@Model.Name",
"content":"@Model.Text",
"variation":"@Model.Variation",
"formatSlots":"@Model.FormatSlots",
"key":"@Model.Key"

<span class="crayon-s">"content"</span>,</span></div><div class="crayon-line" id="crayon-66a06260d9210411961626-5"><span class="crayon-s">"variation"</span><span class="crayon-o">:</span><span class="crayon-s">"@Model.Variation"</span>

These components can have datasources setup as normal, and hence personalization is available into your JSON feed.

Via a browser you would then load the url as normal, remembering to specify the device you’ve selected and you should see content based off rules, user information and behaviours and more.

Taking it to the next level
The simpler approach assumes you have one component per page. Complexity comes in when you need to generate valid JSON based off multiple controls – this can be achieved but requires you to either configure things via rendering parameters or at the point the page is rendered, interrogate the counterpart presentation components and work out whether you have sibling controls.

If you find you have sibling components to render you’d need to add commas after your controls to ensure valid JSON.

Who makes the tea? Why not ask Alexa?

October 4, 2017 by boro

Following on from the previous post, AlexaCore, this post will explain some of the challenges you might encounter when launching Alexa Skills into the store. It will also cover some cool things you can do if you want to enrich the feedback users receive as they use their skill.

Time for a brew?

If you need to solve your debate of who makes the tea why not check out Tea Round, recently enabled on the Skill Store.

First up, how can you find skills?

They are all available on the Amazon site, or accessible via the alexa.amazon.com portal.

Running test versions of your skill

This is pretty straight forwards. You need to login to developer.amazon.com, run through the wizard to create a skill which includes pairing up with an AWS Lambda (or controller endpoint). You should then see the newly created skill in your Alexa app but marked with a ‘dev’ flag.

Testing your skill

You have a few options, either you can simply talk to your Alexa with voice commands or use the text test tool within the developer.amazon.com console.

Getting certified gotcha’s

The certification process looks to validate and check a few things:

Are there bugs in the skill?
Do the descriptions and prompts align with your skill’s intents?
Do you leave the users hanging?

Things that caught me out during this process were:

Testing the skill where you skip past the launch intent
- E.g. rather than asking ‘Alexa, open the tea round’ and then allowing the LaunchIntent to run, you can ask ‘Alexa, open the tea round and spin the wheel’. My logic around initializing the session originally ran in the LaunchIntent, some simple refactoring resolved this.
Leaving the users hanging – in my opinion this isn’t great UX but rules are rules
- If you respond to a user without a prompt e.g. without a request of the user, the rules define you should end the session. My AddIntent would respond to ‘Add Nick’ with ‘Ok, Nick is now in the team’. To get past the certification it needed updating to ‘Ok, Nick is now in the team. Why not spin the wheel?’
Make sure the suggested prompts you include match up to the text set in your intents. The best bet here is to look at other skills and see how they phrase the prompts.

Saving data beyond a session

Much like a session in a web request, for the context of the lifetime of a skill a session gets persisted. This can be used to store anything you want, some simple examples would be an array of the names of the people in the Tea Round. That’s fine, but next time someone loads the skill, the session will be empty and the user will need to re-add each user – with all the extra questioning needed for certification this could get painful.

AWS provide a document model DB, Dynamo, that’s very well suited to this kind of thing. The Tea Round stores the permanent team in Dynamo, updated from the Lambda function that sits behind the skill.

Understanding users names

This can be tricky as subtle variations behind names can lead to them being spelt, and pronounced very differently – especially when regional dialect comes into play. The best success I’ve found is to provide as broad a set of example names as possible when setting up your {slots} in the intent.

Enriching the responses

A typical request & response cycle will send information that Amazon decodes from speech into your Lambda function. From there you can then return text that gets read out by your Alexa. By using Sitecore as a headless backend, the response text can be driven from the CMS – some recent updates to AlexaCore provide helpers for making these requests.

Where things then get interesting are that you can personalize messaging based off behaviour and user interactions. Big brother is watching?!?!

Closing thoughts

Gasp, after all that I’m thirsty – time for a brew! (how English eh :))

If you fancy allowing Alexa into your tea making process, have a look at Tea Round

AlexaCore – a c# diversion into writing Alexa skills

September 11, 2017 by boro

Following the recent Amazon Prime day, I thought it was time to jump on the home assistant bandwagon – £80 seemed a pretty good deal for an Alexa Echo.

If you’ve not tried writing Alexa skills there are some really good blog posts to help get started at: http://timheuer.com/blog/archive/2016/12/12/amazon-alexa-skill-using-c-sharp-dotnet-core.aspx

Skills can be underpinned by an AWS lambda function. In experimenting with writing these I’ve started putting together some helpers which remove a lot of the boiler-plate code needed for C# Alexa Lambda functions, including some fluent tools for running unit tests.

The code and some examples are available at https://github.com/boro2g/AlexaCore. Hopefully you will find them help get your skills off the ground!

Make request with fiddler based off a timer

July 5, 2017 by boro

Fiddler is a great tool for debugging web requests. Things like the composer section allow you to concoct requests and then test out the responses you get. If you’re using composer and look at the raw view you will see something like:

GET http://www.bbc.co.uk/ HTTP/1.1
Host: www.bbc.co.uk

1 2	GET http://www.bbc.co.uk/ HTTP/1.1 Host: www.bbc.co.uk

If you want to run this request every N seconds you can setup a quick script to achieve this:

The full script will be shown below, you need to simply add into the class Handlers code, save the script and then use the new menu options that get added to the Tools menu (Request by Timer, Stop Timer and Request Once):

One thing to note, in your script ensure you keep the trailing \r\n\r\n in the request url otherwise you’ll receive an error.

class Handlers
{
    static var oTimer: System.Windows.Forms.Timer = new Timer();
    public static ToolsAction("Request by Timer")
    function doIt()
    {     
        oTimer = new Timer();
        oTimer.add_Tick(OnTimer);
        oTimer.Interval = 3000;
        oTimer.Start();
    }
      
    public static ToolsAction("Stop Timer")
    function stopIt()
    { 
        oTimer.Stop();
    }
            
    public static ToolsAction("Request Once")
    function doItOnce()
    {
        OnTimer(null, null);
    }

    public static function OnTimer(sender: Object, ea: EventArgs)
    {
        var s = "GET http://www.bbc.co.uk/ HTTP/1.1\r\nHost: www.bbc.co.uk\r\n\r\n";  
        try{ 
            FiddlerObject.utilIssueRequest(s); 
        } 
        catch(e){ 
            MessageBox.Show("send failed" + e.ToString()); 
        } 
    }

class Handlers

{

static var oTimer: System.Windows.Forms.Timer = new Timer();

public static ToolsAction("Request by Timer")

function doIt()

{

oTimer = new Timer();

oTimer.add_Tick(OnTimer);

oTimer.Interval = 3000;

oTimer.Start();

}

public static ToolsAction("Stop Timer")

function stopIt()

{

oTimer.Stop();

}

public static ToolsAction("Request Once")

function doItOnce()

{

OnTimer(null, null);

}

public static function OnTimer(sender: Object, ea: EventArgs)

{

var s = "GET http://www.bbc.co.uk/ HTTP/1.1\r\nHost: www.bbc.co.uk\r\n\r\n";

try{

FiddlerObject.utilIssueRequest(s);

}

catch(e){

MessageBox.Show("send failed" + e.ToString());

}

Enjoy

Log aggregation in AWS – part 3 – enriching the data sent into ElasticSearch

June 9, 2017 by boro

This is the third, and last part of the series that details how to aggregate all your log data within AWS. See Part 1 and Part 2 for getting started and keeping control of the size of your index.

By this point you should have a self sufficient ElasticSearch domain running that pools logs from all the CloudWatch log groups that have been configured with the correct subscriber.

The final step will be how we can enrich the data being sent into the index?

By default AWS will set you up a lambda function that extracts information from the CloudWatch event. It will contain things like: instanceId, event timestamp, the source log group and a few more. This is handled in the lambda via:

var source = buildSource(logEvent.message, logEvent.extractedFields);
source['@id'] = logEvent.id;
source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString();
source['@message'] = logEvent.message;
source['@owner'] = payload.owner;
source['@log_group'] = payload.logGroup;
source['@log_stream'] = payload.logStream;

var source = buildSource(logEvent.message, logEvent.extractedFields);

source['@id'] = logEvent.id;

source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString();

source['@message'] = logEvent.message;

source['@owner'] = payload.owner;

source['@log_group'] = payload.logGroup;

source['@log_stream'] = payload.logStream;

Note, a tip around handling numeric fields – in order for ElasticSearch to believe fields are numbers rather than strings you can multiply the value by 1 e.g.: source[‘@fieldName’] = 1*value;

What to enrich the data with?

This kind of depends on your use case. As we were aggregating logs from a wide range of boxes, applications and services, we wanted to enrich the data in the index with the tags applied to each instance. This sounds simple in practice but needed some planning around how to access the tags for each log entry – I’m not sure AWS would look kindly on making 500,000 API requests in 15 mins!

Lambda and caching

Lambda functions are a very interesting offering and I’m sure you will start to see a lot more use cases for them over the next few years. One challenge they bring is they are stateless – in our scenario we need to provide a way of querying an instance for its tags based of its InstanceId. Enter DynamoDb, another AWS service that provides scalable key-value pair storage.

Amazon define Dynamo as: Amazon DynamoDB is a fully managed non-relational database service that provides fast and predictable performance with seamless scalability.

Our solution

There were 2 key steps to the process:

Updating dynamo to gather tag information from our running instances
Updating the lambda script to pull the tags from dynamo as log entries are processed

1. Pushing instance data into Dynamo

Setup a lambda function that would periodically scan all running instances in our account and push the respective details into Dynamo.

Setup a new Dynamo db table
1. Named: kibana-instanceLookup
2. Region: eu-west-1 (note, adjust this as you need)
3. Primary partition key: instanceId (of type string)
  1. Note – we will tweak the read capacity units once things are up and running – for production we average about 50
Setup a new lambda function
1. Named: LogsToElasticSearch_InstanceQueries_regionName
2. Add environment variable: region=eu-west-1
  1. Note, if you want this to pool logs from several regions into one dynamo setup a lambda function per region and set different environment variables for each. You can use the same trigger and role for each region
3. Use the script shown below
4. Set the execution timeout to be: 1 minute (note, tweak this if the function takes longer to run)
5. Create a new role and give the following permissions:
  1. AmazonEC2ReadOnlyAccess (assign the OTB policy)
  2. Plus add the following policy:
  3. { "Effect": "Allow", "Action": "dynamoDb:*", "Resource": "arn:aws:dynamodb:eu-west-1:###:table/kibana-instanceLookup" }
    
    1
    2
    3
    4
    5
    
    {
    "Effect": "Allow",
    "Action": "dynamoDb:*",
    "Resource": "arn:aws:dynamodb:eu-west-1:###:table/kibana-instanceLookup"
    }
    1. Note, the ### in the role wants to be your account id
6. Setup a trigger within Cloudwatch -> Rules
  1. To run hourly, set the cron to be: 0 * * * ? *
  2. Select the target to be your new lambdas
    1. Note, you can always test your lambda by simply running on demand with any test event

And the respective script:

var AWS = require("aws-sdk");

var tableName = 'kibana-instanceLookup';

exports.handler = function (input, context)
{
	AWS.config.update({ region: process.env.region });

	this._ec2 = new AWS.EC2();

	var queryParams = {
		MaxResults: 200
	};

	var instances = [];

	this._ec2.describeInstances(queryParams, function (err, data)
	{
		if (err) console.log(err, err.stack);
		else
		{
			data.Reservations.forEach((r) =>
			{
				r.Instances.forEach((i) =>
				{
					instances.push({ "instanceId": i.InstanceId, "tags": i.Tags });
				});
			});

			console.log(JSON.stringify(instances));

			instances.forEach((instance) =>
			{
				pushInstanceToDynamo(instance);
			});
		}
	});

	function pushInstanceToDynamo(instance)
	{
		AWS.config.update({ region: 'eu-west-1' });

		var params = {
			Key: {
				"instanceId": {
					S: instance.instanceId
				}
			},
			TableName: tableName
		};

		new AWS.DynamoDB().getItem(params, function (err, data)
		{
			if (err)
			{
				console.log(err, err.stack);
			}
			else
			{
				if (data.Item)
				{
					console.log("Item found in dynamo - not updating " + instance.instanceId);
				}
				else
				{
					var tags = {};
					tags.L = [];
					instance.tags.forEach((tag) =>
					{
						if (tag.Key.indexOf("aws:") === -1)
						{
							tags.L.push(buildArrayEntry(tag));
						}
					});

					var insertParams = {
						Item: {
							"instanceId": {
								S: instance.instanceId
							},
							"tags": tags
						},
						ReturnConsumedCapacity: "TOTAL",
						TableName: tableName
					};

					new AWS.DynamoDB().putItem(insertParams, function (err, data)
					{
						if (err) console.log(err, err.stack);
						else console.log(data);
					});
				}
			}
		});
	}

	function buildArrayEntry(tag)
	{
		var m = {};

		m[tag.Key] = { "S": tag.Value };

		return { "M": m };
	}
}

100

101

102

103

104

105

var AWS = require("aws-sdk");

var tableName = 'kibana-instanceLookup';

exports.handler = function (input, context)

{

AWS.config.update({ region: process.env.region });

this._ec2 = new AWS.EC2();

var queryParams = {

MaxResults: 200

};

var instances = [];

this._ec2.describeInstances(queryParams, function (err, data)

{

if (err) console.log(err, err.stack);

else

{

data.Reservations.forEach((r) =>

{

r.Instances.forEach((i) =>

{

instances.push({ "instanceId": i.InstanceId, "tags": i.Tags });

});

console.log(JSON.stringify(instances));

instances.forEach((instance) =>

{

pushInstanceToDynamo(instance);

});

}

});

function pushInstanceToDynamo(instance)

{

AWS.config.update({ region: 'eu-west-1' });

var params = {

Key: {

"instanceId": {

S: instance.instanceId

}

TableName: tableName

};

new AWS.DynamoDB().getItem(params, function (err, data)

{

if (err)

{

console.log(err, err.stack);

}

else

{

if (data.Item)

{

console.log("Item found in dynamo - not updating " + instance.instanceId);

}

else

{

var tags = {};

tags.L = [];

instance.tags.forEach((tag) =>

{

if (tag.Key.indexOf("aws:") === -1)

{

tags.L.push(buildArrayEntry(tag));

}

});

var insertParams = {

Item: {

"instanceId": {

S: instance.instanceId

"tags": tags

ReturnConsumedCapacity: "TOTAL",

TableName: tableName

};

new AWS.DynamoDB().putItem(insertParams, function (err, data)

{

if (err) console.log(err, err.stack);

else console.log(data);

});

}

});

}

function buildArrayEntry(tag)

{

var m = {};

m[tag.Key] = { "S": tag.Value };

return { "M": m };

}

Note, if your dynamo runs in a different region to eu-west-1, update the first line of the pushInstanceToDynamo method and set the desired target region.

Running on demand should then fill your dynamo with data e.g.:

2. Querying dynamo when you process log entries

The final piece of the puzzle is to update the streaming function to query dynamo as required. This needs a few things:

Update the role used for the lambda that streams data from CloudWatch into ElasticSearch
{ "Effect": "Allow", "Action": "dynamodb:GetItem", "Resource": "arn:aws:dynamodb:eu-west-1:###:table/kibana-instanceLookup" }

1
2
3
4
5

{
"Effect": "Allow",
"Action": "dynamodb:GetItem",
"Resource": "arn:aws:dynamodb:eu-west-1:###:table/kibana-instanceLookup"
}

where ### is your account id
Update the lambda script setup in stage 1 and tweak as shown below

Add the AWS variable to the requires at the top of the file:

var AWS = require("aws-sdk");

1	var AWS = require("aws-sdk");

Update the exports.handler & transform methods and add loadFromDynamo to be:

exports.handler = function (input, context)
{
	this._dynamoDb = new AWS.DynamoDB();	

	// decode input from base64
	var zippedInput = new Buffer(input.awslogs.data, 'base64');

	// decompress the input
	zlib.gunzip(zippedInput, function (error, buffer)
	{
		if (error) { context.fail(error); return; }

		// parse the input from JSON
		var awslogsData = JSON.parse(buffer.toString('utf8'));

		// transform the input to Elasticsearch documents
		transform(awslogsData, (elasticsearchBulkData) =>
		{
			// skip control messages
			if (!elasticsearchBulkData)
			{
				console.log('Received a control message');
				context.succeed('Control message handled successfully');
				return;
			}

			// post documents to the Amazon Elasticsearch Service
			post(elasticsearchBulkData, function (error, success, statusCode, failedItems)
			{
				console.log('Response: ' + JSON.stringify({
					"statusCode": statusCode
				}));

				if (error)
				{
					console.log('Error: ' + JSON.stringify(error, null, 2));

					if (failedItems &amp;&amp; failedItems.length > 0)
					{
						console.log("Failed Items: " +
							JSON.stringify(failedItems, null, 2));
					}

					context.fail(JSON.stringify(error));
				} else
				{
					console.log('Success: ' + JSON.stringify(success));
					context.succeed('Success');
				}
			});
		});
	});
};

function transform(payload, callback)
{
	if (payload.messageType === 'CONTROL_MESSAGE')
	{
		return null;
	}

	var bulkRequestBody = '';

	var instanceId = payload.logStream;

	if (instanceId.indexOf(".") > -1)
	{
		instanceId = instanceId.substring(0, instanceId.indexOf("."));
	}

	loadFromDynamo(instanceId,
		(dynamoTags) =>
		{
			payload.logEvents.forEach(function (logEvent)
			{
				var timestamp = new Date(1 * logEvent.timestamp);

				// index name format: cwl-YYYY.MM.DD
				var indexName = [
					'cwl-' + timestamp.getUTCFullYear(),              // year
					('0' + (timestamp.getUTCMonth() + 1)).slice(-2),  // month
					('0' + timestamp.getUTCDate()).slice(-2)          // day
				].join('.');				

				var source = buildSource(logEvent.message, logEvent.extractedFields);
				source['@id'] = logEvent.id;
				source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString();
				source['@message'] = logEvent.message;
				source['@owner'] = payload.owner;
				source['@log_group'] = payload.logGroup;
				source['@log_stream'] = payload.logStream;				

				var action = { "index": {} };
				action.index._index = indexName;
				action.index._type = payload.logGroup;
				action.index._id = logEvent.id;

				bulkRequestBody += [
					JSON.stringify(action),
					JSON.stringify(Object.assign({}, source, dynamoTags))
				].join('\n') + '\n';
			});
			callback(bulkRequestBody);
		});
}

function loadFromDynamo(instanceId, callback)
{
	var tagsSource = {};

	try
	{
		var params = {
			Key: {
				"instanceId": {
					S: instanceId
				}
			},
			TableName: "kibana-instanceLookup"
		};
		this._dynamoDb.getItem(params, function (err, data)
		{
			if (err)
			{
				console.log(err, err.stack);
				callback(tagsSource);
			}
			else
			{
				if (data.Item) 
				{
					data.Item.tags.L.forEach((tag) =>
						{

							var key = Object.keys(tag.M)[0];
							tagsSource['@' + key] = tag.M[key].S;
						});
				}

				callback(tagsSource);
			}
		});
	}
	catch (exception)
	{
		console.log(exception);
		callback(tagsSource);
	}
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

exports.handler = function (input, context)

{

this._dynamoDb = new AWS.DynamoDB();

// decode input from base64

var zippedInput = new Buffer(input.awslogs.data, 'base64');

// decompress the input

zlib.gunzip(zippedInput, function (error, buffer)

{

if (error) { context.fail(error); return; }

// parse the input from JSON

var awslogsData = JSON.parse(buffer.toString('utf8'));

// transform the input to Elasticsearch documents

transform(awslogsData, (elasticsearchBulkData) =>

{

// skip control messages

if (!elasticsearchBulkData)

{

console.log('Received a control message');

context.succeed('Control message handled successfully');

return;

}

// post documents to the Amazon Elasticsearch Service

post(elasticsearchBulkData, function (error, success, statusCode, failedItems)

{

console.log('Response: ' + JSON.stringify({

"statusCode": statusCode

}));

if (error)

{

console.log('Error: ' + JSON.stringify(error, null, 2));

if (failedItems && failedItems.length > 0)

{

console.log("Failed Items: " +

JSON.stringify(failedItems, null, 2));

}

context.fail(JSON.stringify(error));

} else

{

console.log('Success: ' + JSON.stringify(success));

context.succeed('Success');

}

});

};

function transform(payload, callback)

{

if (payload.messageType === 'CONTROL_MESSAGE')

{

return null;

}

var bulkRequestBody = '';

var instanceId = payload.logStream;

if (instanceId.indexOf(".") > -1)

{

instanceId = instanceId.substring(0, instanceId.indexOf("."));

}

loadFromDynamo(instanceId,

(dynamoTags) =>

{

payload.logEvents.forEach(function (logEvent)

{

var timestamp = new Date(1 * logEvent.timestamp);

// index name format: cwl-YYYY.MM.DD

var indexName = [

'cwl-' + timestamp.getUTCFullYear(), // year

('0' + (timestamp.getUTCMonth() + 1)).slice(-2), // month

('0' + timestamp.getUTCDate()).slice(-2) // day

].join('.');

var source = buildSource(logEvent.message, logEvent.extractedFields);

source['@id'] = logEvent.id;

source['@timestamp'] = new Date(1 * logEvent.timestamp).toISOString();

source['@message'] = logEvent.message;

source['@owner'] = payload.owner;

source['@log_group'] = payload.logGroup;

source['@log_stream'] = payload.logStream;

var action = { "index": {} };

action.index._index = indexName;

action.index._type = payload.logGroup;

action.index._id = logEvent.id;

bulkRequestBody += [

JSON.stringify(action),

JSON.stringify(Object.assign({}, source, dynamoTags))

].join('\n') + '\n';

});

callback(bulkRequestBody);

});

}

function loadFromDynamo(instanceId, callback)

{

var tagsSource = {};

try

{

var params = {

Key: {

"instanceId": {

S: instanceId

}

TableName: "kibana-instanceLookup"

};

this._dynamoDb.getItem(params, function (err, data)

{

if (err)

{

console.log(err, err.stack);

callback(tagsSource);

}

else

{

if (data.Item)

{

data.Item.tags.L.forEach((tag) =>

{

var key = Object.keys(tag.M)[0];

tagsSource['@' + key] = tag.M[key].S;

});

}

callback(tagsSource);

}

});

}

catch (exception)

{

console.log(exception);

callback(tagsSource);

}

The final step is to refresh the index definition within Kibana: Management -> Index patterns -> Refresh field list.

Final thoughts
There are a few things to keep an eye on as you roll this out – bear in mind these may need tweaking over time:

The lambda function that scans EC2 times out, if so, up the timeout
The elastic search index runs out of space, if so, adjust the environment variables used in step 2
The dynamo read capacity threshold hits its ceiling, if so increase the read capacity (this can be seen in the Metrics section of the table in Dynamo)

Happy logging!

Log aggregation in AWS – part 2 – keeping your index under control

June 6, 2017 by boro

This is the second part in the series as a follow on to /log-aggregation-aws-part-1/

Hopefully by this point you’ve now got kibana up and running, gathering all the logs from each of your desired CloudWatch groups. Over time the amount of data being stored in the index will constantly be growing so we need to keep things under control.

Here is a good view of the issue. We introduced our cleanup lambda on the 30th, if we hadn’t I reckon we’d have about 2 days more uptime before the disks ran out. The oscillating items from the 31st onward are exactly what we’d want to see – we delete indices older than 10 days every day.

Initially this was done via a scheduled task from a box we host – it worked but wasn’t ideal as it relies on the box running, potentially user creds and lots more. What seemed a better fit was to use AWS Lambda to keep our index under control.

Getting setup

Luckily you don’t need to setup much for this. One AWS Lambda, a trigger and some role permissions and you should be up and running.

Create a new lambda function based off the script shown below
Add 2 environment variables:
1. daysToKeep=10
2. endpoint=elastic search endpoint e.g. search-###-###.eu-west-1.es.amazonaws.com

Create a new role as part of the setup process

Note, these can then be found in the IAM section of AWS e.g. https://console.aws.amazon.com/iam/home?region=eu-west-1#/roles
Update the role to allow Get and Delete access to your index with the policy:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "es:ESHttpGet",
                "es:ESHttpDelete"
            ],
            "Resource": "ARN of elastic search index"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Action": [

"es:ESHttpGet",

"es:ESHttpDelete"

"Resource": "ARN of elastic search index"

}

]

}

Setup a trigger (in CloudWatch -> Events -> Rules)
1. Here you can set the frequency of how often to run e.g. a CRON of
  
  0 2 * * ? *
  
  1
  
  0 2 * * ? *
  
  will run at 2am every night
Test your function, you can always run on demand and then check whether the indices have been removed

And finally the lambda code:

var AWS = require('aws-sdk');

var endpoint; 
var creds = new AWS.EnvironmentCredentials('AWS');

Date.prototype.addDays = function(days) {
	var dat = new Date(this.valueOf());
	dat.setDate(dat.getDate() + days);
	return dat;
}

exports.handler = function(input, context)
{
	endpoint = new AWS.Endpoint(process.env.endpoint);

	let dateBaseline = new Date();

	dateBaseline = dateBaseline.addDays(-parseInt(process.env.daysToKeep));

	console.log("Date baseline: " + dateBaseline.toISOString());

	getIndices(context, function(data)
	{
		data.split('\n').forEach((row) =>
			{
				let parts = row.split(" ");
				
				if (parts.length > 2)
				{
					let indiceName = parts[2];

					if (indiceName.indexOf("cwl") > -1)
					{
						let indiceDate = new Date(indiceName.substr(4, 4), indiceName.substr(9, 2)-1, indiceName.substr(12, 2));
						
						if (indiceDate < dateBaseline)
						{
							console.log("Planning to delete indice: " + indiceName);

							removeIndice("/"+indiceName, context);
						}
					}
				}
			});
	});
}

function removeIndice(indiceName, context) 
{
	makeRequest("DELETE", indiceName, context);
}

function getIndices(context, callback)
{
	makeRequest("GET", '/_cat/indices', context, callback);
}

function makeRequest(method, path, context, callback)
{
	console.log(`Making ${method} call to ${path}`);

	var req = new AWS.HttpRequest(endpoint);

	req.method = method;
	req.path = path;
	req.region = "eu-west-1";
	req.headers['presigned-expires'] = false;
	req.headers['Host'] = endpoint.host;

	var signer = new AWS.Signers.V4(req, 'es');
	signer.addAuthorization(creds, new Date());

	var send = new AWS.NodeHttpClient();
	send.handleRequest(req,
		null,
		function(httpResp)
		{
			var respBody = '';
			httpResp.on('data',
				function(chunk)
				{
					respBody += chunk;
				});
			httpResp.on('end',
				function(chunk)
				{
					if (callback)
					{
						callback(respBody);
					}
					//console.log(respBody);
				});
		},
		function(err)
		{
			console.log('Error: ' + err);
			context.fail('Lambda failed with error ' + err);
		});
}

var AWS = require('aws-sdk');

var endpoint;

var creds = new AWS.EnvironmentCredentials('AWS');

Date.prototype.addDays = function(days) {

var dat = new Date(this.valueOf());

dat.setDate(dat.getDate() + days);

return dat;

}

exports.handler = function(input, context)

{

endpoint = new AWS.Endpoint(process.env.endpoint);

let dateBaseline = new Date();

dateBaseline = dateBaseline.addDays(-parseInt(process.env.daysToKeep));

console.log("Date baseline: " + dateBaseline.toISOString());

getIndices(context, function(data)

{

data.split('\n').forEach((row) =>

{

let parts = row.split(" ");

if (parts.length > 2)

{

let indiceName = parts[2];

if (indiceName.indexOf("cwl") > -1)

{

let indiceDate = new Date(indiceName.substr(4, 4), indiceName.substr(9, 2)-1, indiceName.substr(12, 2));

if (indiceDate < dateBaseline)

{

console.log("Planning to delete indice: " + indiceName);

removeIndice("/"+indiceName, context);

}

});

}

function removeIndice(indiceName, context)

{

makeRequest("DELETE", indiceName, context);

}

function getIndices(context, callback)

{

makeRequest("GET", '/_cat/indices', context, callback);

}

function makeRequest(method, path, context, callback)

{

console.log(`Making ${method} call to ${path}`);

var req = new AWS.HttpRequest(endpoint);

req.method = method;

req.path = path;

req.region = "eu-west-1";

req.headers['presigned-expires'] = false;

req.headers['Host'] = endpoint.host;

var signer = new AWS.Signers.V4(req, 'es');

signer.addAuthorization(creds, new Date());

var send = new AWS.NodeHttpClient();

send.handleRequest(req,

null,

function(httpResp)

{

var respBody = '';

httpResp.on('data',

function(chunk)

{

respBody += chunk;

});

httpResp.on('end',

function(chunk)

{

if (callback)

{

callback(respBody);

}

//console.log(respBody);

});

function(err)

{

console.log('Error: ' + err);

context.fail('Lambda failed with error ' + err);

});

}

Note, if you are running in a different region you will need to tweak req.region = “eu-west-1”;

How does it work?

Elastic search allows you to query the index to find all indices via the url: /_cat/indices. The lambda function makes a web request to this url, parses each row and finds any indices that match the name: cwl-YYYY.MM.dd. If an indice is found that is older than days to keep, a delete request is issued to elasticSearch

Was this the best option?

There are tools available for cleaning up old indices, even ones that Elastic themselves provide: https://github.com/elastic/curator however this requires additional boxes to run hence the choice for keeping it wrapped in a simple lambda.

Happy indexing!

Log aggregation in AWS – part 1

June 5, 2017 by boro

As with most technologies these days you get plenty of options as to how you solve your technical and logistical problems. The following set of posts details one way you can approach solving what I suspect is quite a common problem – how to usefully aggregate large quantities of logs in the cloud.

What to expect from these blog posts

Getting started – how to get log data off each box into a search index (this post)
Keeping the search index under control
Enriching the data you push into your search index

Some background
Our production infrastructure is composed of roughly 30 windows instances, some web boxes and some sql boxes. These are split between 2 regions and within each region are deployed to all the availability zones that AWS provides. We generate roughly 500k log entries in 15 mins which ends up as 20-25GB log data per day.

The first attempt
When we started out on this work we didn’t appreciate quite how much log data we’d be generating. Our initial setup was based around some CloudFormation templates provided by AWSLabs: https://github.com/awslabs/cloudwatch-logs-subscription-consumer. Initially this worked fine however we quickly hit the issue where the index, and hence kibana, would stop working. We aren’t elasticSearch, kibana or even linux experts here so troubleshooting was taking more time than the benefits we got from the tools.

Getting log data off your instances
As much as the first attempt for querying and displaying log entries didn’t quite work out, we did make good progress as to how we pooled all the log data generated across the infrastructure. You have a few options here – we chose to push everything to CloudWatch and then stream onto other tools.

Note, CloudWatch is a great way of aggregating all your logs however searching across large numbers of log groups and log streams isn’t particularly simple or quick.

To push data into CloudWatch you have a few options:
– Write log entries directly to a known group – e.g. setup a log4net appender that writes directly to CloudWatch
– If you are generating physical log files on disk, use EC2Config (an ootb solution provided by AWS) which streams the data from log files into CloudWatch.

Note, this needs configuring to specify which folders contain the source log files. See http://docs.aws.amazon.com/AWSEC2/latest/WindowsGuide/UsingConfig_WinAMI.html for more info.

Provided things have gone to plan, you should now start to see log entries show up in CloudWatch:

CloudWatch log group subscribers
CloudWatch allows you to wire up subscribers to log groups – these forward on any log entries to the respective subscriber. Subscribers can be multiple things, either: kinesis streams or to a lambda function. Via the web ui you can select a log group, choose actions then e.g. ‘Stream to AWS Lambda’.

AWS Lambda functions
Lambda functions can be used for many things – in these examples we use them to:
– Transform cloudformation log entries into a format we want to index in ElasticSearch
– Run a nightly cleanup to kill off old search indices
– Run an hourly job to scrape meta data out of EC2 and store into Dynamo
Note, we chose to use the NodeJS runtime – see http://docs.aws.amazon.com/AWSJavaScriptSDK/latest/top-level-namespace.html for the API documentation

ElasticSearch as a service
Our first attempt, ie self hosting, did provide good insight into how things should work but the failure rate was too high. We found we were needing to rebuild the stack every couple weeks. Hence, SAAS was a lot more appealing an option. Let the experts handle the setup.

Note, Troy Hunt has written some good posts on the benefits of pushing as much to SAAS – https://www.troyhunt.com/heres-how-i-deal-with-managed-platform-outages/

Setting up ElasticSearch
This is the cool bit as it can all be done through the AWS UI! The steps to follow are:

Create an Elasticsearch domain
1. Ensure to pick a large enough volume size. We opted for 500GB in our production account
2. Select a suitable access policy. We whitelisted our office IP
3. This takes a while to rev up so wait until it goes green
4. One neat thing is you now get Kibana automatically available. The UI will provide the kibana url.
Setup a CloudWatch group subscriber
1. Find the group you want to push to the index, then ‘Actions’ -> ‘Stream to Amazon Elasticsearch Service’
2. Select ‘Other’ for the Log Format
3. Complete the wizard, which ultimately will create you a Lambda function
Start testing things out
1. If you now push items into CloudWatch, you should see indices created in ElasticSearch
2. Within Kibana you need to let Kibana know how the data looks:
  1. Visit ‘Management’ -> ‘Index Patterns’ -> ‘Add new’
  2. The log format is [cwl-]YYYY.MM.DD

Next up we’ll go through:

How to prune old indices in order to keep a decent level of disk space left
How to transform the data from CloudWatch, through a Lambda function, into the ElasticSearch index

blog.boro2g .co.uk

Some ideas about coding, dev and all things online.

Author Archives: boro

PUB SUB in AWS Lambda via SNS using C#

AWS Lambda now supports Serverless applications including WebApi

Copying large files between S3 buckets

Serving personalized content as JSON from Sitecore

Who makes the tea? Why not ask Alexa?

AlexaCore – a c# diversion into writing Alexa skills

Make request with fiddler based off a timer

Log aggregation in AWS – part 3 – enriching the data sent into ElasticSearch

Log aggregation in AWS – part 2 – keeping your index under control

Log aggregation in AWS – part 1