The Definitive Crash Course on Serverless with AWS: Centralized logging with Kinesis and Lambda

Don’t you just hate it when APIs are failing and you have absolutely no clue why? Now imagine you don’t have access to the VM, cluster or container where your software is running. Want me to continue with this nightmare? Yes, that’s what debugging AWS Lambda functions tends to seem like. A horrid nightmare of not knowing what is happening nor why things are failing. This article will show you a way of logging function invocations. Letting you track and monitor failures and errors, while also giving you a nice structure for logging info and debug logs for when you need to troubleshoot behavior. The key is to send all logs to a central location where you can later group, filter and make sense of them. is a full-stack observability solution for your entire software stack. Meaning you can implement function logging alongside any existing infrastructure, like Kubernetes clusters and containers. Sematext Ready? Let’s get started! Using CloudWatch for Logs is the default solution for showing AWS Lambda logs. CloudWatch CloudWatch collects monitoring and operational data in the form of logs, metrics, and events, providing you with a unified view of AWS resources, applications and services that run on AWS, and on-premises servers. — AWS Documentation In layman’s terms, it’s an AWS service for showing your logs across all AWS services. We’re interested in knowing how it handles AWS Lambda logs. When a Lambda function executes, whatever you write out to the console, a fmt.printf() in Go or console.log() in Node.js, will be sent to CloudWatch asynchronously in the background. Lucky for us, it won’t add any overhead to the function execution time. Using logging agents in the function runtime will add overhead to the execution and add unnecessary latency. We want to avoid that, and process the logs after they get added to CloudWatch. Below you can see sample log events that get generated from a generic function. Hello World Let’s take a step back and look at the bigger picture. Every function will create something called a in CloudWatch. Click on a particular Log Group. Log Group These log groups will contain that are literally equivalent of log events coming from particular function instances. Log Streams This is hardly a good enough solution for system insight and having proper overview of what your software is doing. Because of its structure, it’s incredibly hard to see and distinguish logs. Using a central location for your logs makes more sense. You can use your own Elasticsearch or a hosted setup. Sematext gives you full-stack observability for every part of your infrastructure and exposes an . Let me show you how easy it is to create CloudWatch log processing of your AWS Lambda functions and pipe them to a Sematext . Elasticsearch API Logs App Creating a centralized logging solution By using CloudWatch log group subscriptions and Kinesis you can funnel all of your Lambda logs to a dedicated function that will ship them to Sematext’s Elasticsearch API. There you have a central location for all your logs. You can search and filter logs for all functions and with little effort have insight into the behavior and health of your functions. I’ll demo . It’s built with the and Node.js. But, you can feel free to use or , and any programming language you want. The concept will stay the same. how to build a one-command-deploy solution you can use for yourself Serverless Framework AWS SAM Terraform Here’s what it will look like in the end. Much prettier than CloudWatch, and you can actually find what you’re looking for! Setting up the Serverless project First of all install the Serverless Framework, configure your IAM user, and create a new project. Full guide can be found . here $ npm install -g serverless $ sls config credentials \ --provider aws \ --key xxxxxxxxxxxxxx \ --secret xxxxxxxxxxxxxx $ sls create --template aws-nodejs --path lambda-cwlogs-to-logsene $ cd lambda-cwlogs-to-logsene $ npm init -y $ npm i logsene-js zlib serverless-iam-roles-per- function Sweet! now move on to the serverless.yml. Configuring resources Open up the directory in a code editor and check out the serverless.yml. Feel free to delete everything and paste this in. lambda-cwlogs-to-logsene # serverless.yml service: lambda-cwlogs-to-logsene plugins: - serverless-iam-roles-per- {opt:stage, :provider.stage} secrets: ${file(secrets.json)} provider: name: aws runtime: nodejs8 stage: dev region: ${ :custom.secrets.REGION, } versionFunctions: functions: shipper: handler: shipper.handler description: Sends CloudWatch logs Kinesis to Sematext Elastic Search API memorySize: timeout: events: - stream: type: kinesis arn: Fn::GetAtt: - LogsKinesisStream - Arn batchSize: ${ :custom.secrets.BATCH_SIZE} startingPosition: LATEST enabled: environment: LOGS_TOKEN: ${ :custom.secrets.LOGS_TOKEN} LOGS_BULK_SIZE: LOG_INTERVAL: subscriber: handler: subscriber.handler description: Subscribe all CloudWatch log groups to Kinesis memorySize: timeout: events: - http: path: subscribe method: get - cloudwatchEvent: event: source: - aws.logs detail-type: - AWS API Call via CloudTrail detail: eventSource: - logs.amazonaws.com eventName: - CreateLogGroup - schedule: rate: rate( minutes) iamRoleStatements: - Effect: Action: - - - - - - - Resource: environment: filterName: ${ :custom.stage}-${ :provider.region} region: ${ :provider.region} shipperFunctionName: subscriberFunctionName: prefix: retentionDays: ${ :custom.secrets.LOG_GROUP_RETENTION_IN_DAYS} kinesisArn: Fn::GetAtt: - LogsKinesisStream - Arn roleArn: Fn::GetAtt: - CloudWatchLogsRole - Arn resources: Resources: LogsKinesisStream: Type: AWS::Kinesis::Stream Properties: Name: ${ :service}-${ :custom.stage}-logs ShardCount: ${ :custom.secrets.KINESIS_SHARD_COUNT} RetentionPeriodHours: ${ :custom.secrets.KINESIS_RETENTION_IN_HOURS} CloudWatchLogsRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: Statement: - Effect: Allow Principal: Service: - logs.amazonaws.com Action: - sts:AssumeRole Policies: - PolicyName: root PolicyDocument: Version: Statement: - Effect: Allow Action: - kinesis:PutRecords - kinesis:PutRecord Resource: Fn::GetAtt: - LogsKinesisStream - Arn RoleName: ${ :service}-${ :custom.stage}-cloudwatchrole : : function custom stage $ self .10 self 'us-east-1' false from 128 3 self true self 100 2000 128 30 60 "Allow" "iam:PassRole" "sts:AssumeRole" "logs:PutSubscriptionFilter" "logs:DeleteSubscriptionFilter" "logs:DescribeSubscriptionFilters" "logs:DescribeLogGroups" "logs:PutRetentionPolicy" "*" self self self "shipper" "subscriber" "/aws/lambda" self self self self self "2012-10-17" "2012-10-17" self self Let’s break it down piece by piece. The shipper function will be triggered by a Kinesis stream, and it has some environment variables for configuring . The Kinesis stream itself is defined at the bottom, in the resources section, and referenced in the function events by using its ARN. Sematext Logs Moving on to the subscriber function. It can be triggered in three ways. It’s up to you to choose. If you have a lot of existing Log Groups, you may want to hit the HTTP endpoint to initially subscribe them all. Otherwise, having it trigger every once in a while, or only when a new Log Group is created, would be fine. The LogsKinesisStream is the Kinesis stream to where we’re subscribing Log Groups, and CloudWatchLogsRole is the IAM Role which will allow CloudWatch to put records into Kinesis. With that out of the way, you can now see we’re missing a secrets.json file. But, before we continue, jump over to , and create a . Press the tiny green button to add a Logs App. Sematext log in Logs App After adding the name of the App and some basic info, you’ll see a screen pop up. Press the and copy your token. waiting for data integrations guide Now you can paste the token in the file. secrets.json { : , : , : , : , : , : } "LOGS_TOKEN" "your-token" "REGION" "us-east-1" "BATCH_SIZE" 1000 "LOG_GROUP_RETENTION_IN_DAYS" 1 "KINESIS_RETENTION_IN_HOURS" 24 "KINESIS_SHARD_COUNT" 1 Adding the subscriber function I like saying Kinesis is a simpler version of Kafka. It’s basically a pipe. You subscribe data to be sent into it and tell it to trigger a Lambda function as an event, once it satisfies a certain batch size. The purpose of having a subscriber function is to subscribe all Log Groups to a Kinesis stream. Ideally they should be subscribed upon creation, and of course, initially when you want to subscribe all existing Log Groups to a new Kinesis stream. As a fallback, I also like to have an HTTP endpoint for when I want to manually trigger the subscriber. In your code editor, create a new file and name it . Paste this snippet in. subscriber.js AWS = ( ) AWS.config.region = process.env.region cloudWatchLogs = AWS.CloudWatchLogs() prefix = process.env.prefix kinesisArn = process.env.kinesisArn roleArn = process.env.roleArn filterName = process.env.filterName retentionDays = process.env.retentionDays shipperFunctionName = process.env.shipperFunctionName filterPattern = setRetentionPolicy = (logGroupName) => { params = { : logGroupName, : retentionDays } cloudWatchLogs.putRetentionPolicy(params).promise() } listLogGroups = (acc, nextToken) => { req = { : , : prefix, : nextToken } res = cloudWatchLogs.describeLogGroups(req).promise() newAcc = acc.concat(res.logGroups.map( logGroup.logGroupName)) (res.nextToken) { listLogGroups(newAcc, res.nextToken) } { newAcc } } upsertSubscriptionFilter = (options) => { .log( ) { subscriptionFilters } = cloudWatchLogs.describeSubscriptionFilters({ : options.logGroupName }).promise() { filterName, filterPattern } = subscriptionFilters[ ] (filterName !== options.filterName || filterPattern !== options.filterPattern) { cloudWatchLogs.deleteSubscriptionFilter({ : filterName, : options.logGroupName }).promise() cloudWatchLogs.putSubscriptionFilter(options).promise() } } subscribe = (logGroupName) => { options = { : kinesisArn, : logGroupName, : filterName, : filterPattern, : roleArn, : } { cloudWatchLogs.putSubscriptionFilter(options).promise() } (err) { .log( ) .error( .stringify(err)) upsertSubscriptionFilter(options) } } subscribeAll = (logGroups) => { .all( logGroups.map( logGroupName => { (logGroupName.endsWith(shipperFunctionName)) { .log( ) } .log( ) subscribe(logGroupName) .log( ) setRetentionPolicy(logGroupName) }) ) } processAll = () => { logGroups = listLogGroups([]) subscribeAll(logGroups) } exports.handler = () => { .log( ) processAll() .log( ) { : , : .stringify({ : }) } } // subscriber.js const require 'aws-sdk' const new const const const const const const const '' const async const logGroupName retentionInDays await const async const limit 50 logGroupNamePrefix nextToken const await const => logGroup if return else return const async console 'UPSERTING...' const await logGroupName const 0 if await filterName logGroupName await const async const destinationArn logGroupName filterName filterPattern roleArn distribution 'ByLogStream' try await catch console `FAILED TO SUBSCRIBE [ ]` ${logGroupName} console JSON await const async await Promise async if console `SKIPPING [ ] BECAUSE IT WILL CREATE CYCLIC EVENTS FROM IT'S OWN LOGS` ${logGroupName} return console `SUBSCRIBING [ ]` ${logGroupName} await console `UPDATING RETENTION POLICY TO [ DAYS] FOR [ ]` ${retentionDays} ${logGroupName} await const async const await await async console 'subscriber start' await console 'subscriber done' return statusCode 200 body JSON message `Subscription successful!` Check out the function. It'll grab all from CloudWatch which match the prefix, and put them in an easily accessible array. You'll then pass them to a function, which will map through them while subscribing them to the Kinesis stream you defined in the . processAll() Log Groups subscribeAll() serverless.yml Another cool thing is setting the retention policy to 7 days. You’ll rarely need more than that and it’ll cut the cost of keeping logs in your AWS account. Keep in mind you can also edit the by which logs will get ingested. For now, I’ve chosen to keep it blank and not filter out anything. But, based on your needs you can match it with what kind of pattern your logger of choice creates. filterPattern Sweet, with that done, let’s move on to shipping some logs! Adding the shipper function After the Kinesis stream receives logs from CloudWatch, it’ll trigger a Lambda function dedicated to sending the logs to an Elasticsearch endpoint. For this example we’ll use as the log shipper. It’s rather simple if you break it down. A batch of records will be sent in the event parameter to the shipper function. You parse the logs, giving them your desired structure, and ship them to Sematext. Here’s what it looks like. Create a new file, name it shipper.js and paste this code in. LogseneJS Zlib = ( ) Logsene = ( ) logger = Logsene(process.env.LOGS_TOKEN) errorPatterns = [ ] configurationErrorPatterns = [ , ] timeoutErrorPatterns = [ , ] structuredLogPattern = regexError = (errorPatterns.join( ), ) regexConfigurationError = (configurationErrorPatterns.join( ), ) regexTimeoutError = (timeoutErrorPatterns.join( ), ) regexStructuredLog = (structuredLogPattern) lambdaVersion = logStream.substring(logStream.indexOf( ) + , logStream.indexOf( )) lambdaName = logGroup.split( ).reverse()[ ] checkLogError = { (log.message.match(regexError)) { log.severity = log.error = { : } } (log.message.match(regexConfigurationError)) { log.severity = log.error = { : } } (log.message.match(regexTimeoutError)) { log.severity = log.error = { : } } log } splitStructuredLog = { parts = message.split( , ) { : parts[ ], : parts[ ], : parts[ ] } } parseLog = { ( message.startsWith( ) || message.startsWith( ) || message.startsWith( ) ) { } (message.match(regexStructuredLog)) { { timestamp, requestId, msg } = splitStructuredLog(message) checkLogError({ : msg, : functionName, : functionVersion, : awsRegion, : , : , : timestamp, : requestId }) } { checkLogError({ : message, : functionName, : functionVersion, : awsRegion, : , : }) } } parseLogs = { logs = [] event.Records.forEach( { payload = Buffer.from(record.kinesis.data, ) json = (Zlib.gunzipSync(payload)).toString( ) data = .parse(json) (data.messageType === ) { } functionName = lambdaName(data.logGroup) functionVersion = lambdaVersion(data.logStream) awsRegion = record.awsRegion data.logEvents.forEach( { log = parseLog(functionName, functionVersion, logEvent.message, awsRegion) (!log) { } logs.push(log) }) }) logs } shipLogs = (logs) => { ( { (!logs.length) { resolve( ) } logs.forEach( logger.log(log.severity, , log)) logger.send( resolve( )) }) } exports.handler = (event) => { { res = shipLogs(parseLogs(event)) .log(res) } (err) { .log(err) err } } // shipper.js const require 'zlib' const require 'logsene-js' const new const 'error' const 'module initialization error' 'unable to import module' const 'task timed out' 'process exited before completing' /** * Sample of a structured log * *************************************************************************** * Timestamp RequestId Message * 2019-03-08T15:58:45.736Z 53499d7f-60f1-476a-adc8-1e6c6125a67c Hello World! * *************************************************************************** */ const '[0-9]{4}-(0[1-9]|1[0-2])-(0[1-9]|[1-2][0-9]|3[0-1])T(2[0-3]|[01][0-9]):[0-5][0-9]:[0-5][0-9].[0-9][0-9][0-9]Z([ \t])[a-zA-Z0-9]{8}-[a-zA-Z0-9]{4}-[a-zA-Z0-9]{4}-[a-zA-Z0-9]{4}-[a-zA-Z0-9]{12}([ \t])(.*)' const new RegExp '|' 'gi' const new RegExp '|' 'gi' const new RegExp '|' 'gi' const new RegExp const ( ) => logStream '[' 1 ']' const ( ) => logGroup '/' 0 const ( ) => log if 'error' type 'runtime' else if 'error' type 'configuration' else if 'error' type 'timeout' return const ( ) => message const '\t' 3 return timestamp 0 requestId 1 msg 2 /** * Create payload for Logsene API */ const ( ) => functionName, functionVersion, message, awsRegion if 'START RequestId' 'END RequestId' 'REPORT RequestId' return // if log is structured if const return message function version region type 'lambda' severity 'debug' timestamp requestId else // when log is NOT structured return message function version region type 'lambda' severity 'debug' const ( ) => event const => record const 'base64' const 'utf8' const JSON if 'CONTROL_MESSAGE' return const const const => logEvent const if return return const async return new Promise ( ) => resolve if return 'No logs to ship.' => log 'LogseneJS' => () 'Logs shipped successfully!' async try const await console catch console return return 'shipper done' The heart of the shipper Lambda lies in the and functions. The former will take the event parameter, extract all log events, parse them, add them to an array, and return that array. While the latter will take that same logs array, add every single log event to the buffer, and send them all in one go. The location is the Logs App you created above. parseLogs() shipLogs() LogseneJS Do you remember the image from the beginning of the article where you saw log events of a typical function invocation? There you can see it generates 4 different types of log events. START RequestId ... END RequestId REPORT RequestId They can start with any of these three patterns, where the ellipsis represents any type of string that is printed to stdout in the function runtime ( in Node.js). console.log() The function will skip the START, END, and REPORT log events entirely, and only return user-defined log events as either debug or error based on if they’re user-defined stdout or any type of error in the function runtime, configuration or duration. parseLog() The log message itself can be structured by default, but not always. By default in the Node.js runtime, it has a structure that looks like this. Timestamp RequestId Message T15: : Z d7f f1 a-adc8 c6125a67c Hello World! 2019 -03 -08 58 45.736 53499 -60 -476 -1e6 The code in the shipper is configured to work with the structure above or with a structure that only has the message part. If you’re using another runtime, I’d advise you to use structured logging to have a common structure for your log events. With the coding part done, you’re ready to deploy and test your custom log shipper. Deploy and test your centralized logging solution The beauty of using an infrastructure as code solution like the Serverless Framework is how simple deployments are. You can push everything to the cloud with one command. Jump back to your terminal and in the directory of your project run: $ sls deploy You’ll see output get printed to the console. [output] Serverless: Packaging service... Serverless: Excluding development dependencies... Serverless: Uploading CloudFormation file to S3... Serverless: Uploading artifacts... Serverless: Uploading service .zip file to S3 ( MB)... Serverless: Validating template... Serverless: Updating Stack... Serverless: Checking Stack update progress... ............ Serverless: Stack update finished... Service Information service: lambda-cwlogs-to-logsene stage: dev region: us-east stack: lambda-cwlogs-to-logsene-dev api keys: None endpoints: GET - https: functions: shipper: lambda-cwlogs-to-logsene-dev-shipper subscriber: lambda-cwlogs-to-logsene-dev-subscriber layers: None Serverless: Removing old service artifacts S3… 2.15 -1 //.execute-api.us-east-1.amazonaws.com/dev/subscribe from That’s it. You now have a setup for shipping all logs from your Lambda functions into . Make sure to trigger the subscriber function to subscribe the Log Groups to the Kinesis stream. After triggering the subscriber you’ll see the logs the subscriber generated in Sematext, and you can rest assured it works. Sematext Cloud Above you can see how I added severity filtering. You can easily choose which value to filter by, giving you an easy way to track errors, timeouts and debug logs. What about costs? The cost of having a setup like this in your AWS account is rather cheap. The flat cost of a single shard Kinesis stream is roughly for the amount of data streamed. The single shard has an ingest capacity of 1MB/sec or 1000 records/sec, which is fine for most users. $14/month with additional costs The Kinesis cost is split into and the size of 25KB. One shard costs $0.36 per day, while one million PUT Payload Units cost $0.014. Hypothetically, if you have one shard and 100 PUT payload units per second that’ll end up costing you . shard hours PUT payload units $10.8 for the shard and $3.6288 for the payload units during a 30 day period The Lambda functions are configured to use the minimum amount of memory possible, 128MB, meaning the costs will often stay in the free tier during moderate use. That’s the least of your worries. Wrapping up Having a central location for your logs is crucial. Even though CloudWatch is useful in its own way, it lacks in the sense of overview. By using a central location you don’t need to switch contexts for debugging different types of applications. Sematext can monitor your whole software stack. Having your logs, logs and Lambda logs in where you can easily keep track of everything is a major benefit. Kubernetes container Sematext Logs If you need to check out the code once again, , give it a star if you want more people to see it on GitHub. You can also clone the repo and deploy it right away. Don’t forget to add you Logs App token first. here’s the repo If you need an observability solution for your software stack, check out . We’re pushing to and make an impact. Sematext open source our products Hope you guys and girls enjoyed reading this as much as I enjoyed writing it. If you liked it, slap that tiny share button so more people will see this tutorial. Until next time, be curious and have fun. Originally published at sematext.com on March 15, 2019.