Menu
Amazon CloudWatch Logs
User Guide

Real-time Processing of Log Data with Subscriptions

You can use subscriptions to get access to a real-time feed of log events from CloudWatch Logs and have it delivered to other services such as an Amazon Kinesis stream or AWS Lambda for custom processing, analysis, or loading to other systems. To begin subscribing to log events you will need to create the receiving source, such as an Amazon Kinesis stream, where the events will be delivered. A subscription filter defines the filter pattern to use for filtering which log events get delivered to your AWS resource, as well as information about where to send matching log events to. Each subscription filter is made up of the following key elements:

log group name

The log group to associate the subscription filter with. All log events uploaded to this log group would be subject to the subscription filter and would be delivered to the chosen Amazon Kinesis stream if the filter pattern matches with the log events.

filter pattern

A symbolic description of how CloudWatch Logs should interpret the data in each log event, along with filtering expressions that restrict what gets delivered to the destination AWS resource. For more information about the filter pattern syntax, see Filter and Pattern Syntax.

destination arn

The Amazon Resource Name (ARN) of the Amazon Kinesis stream or Lambda function you want to use as the destination of the subscription feed.

role arn

An IAM role that grants CloudWatch Logs the necessary permissions to put data into the chosen Amazon Kinesis stream. This role is not needed for Lambda destinations because CloudWatch Logs can get the necessary permissions from access control settings on the Lambda function itself.

Using Subscription Filters

This section provides the following examples for using subscription filters with Amazon Kinesis, Lambda, and Firehose.

Note

Currently, only one subscription filter is supported per log group.

Example 1: Amazon Kinesis

In the following example, a subscription filter gets associated with a log group containing AWS CloudTrail events to have every logged activity made by "Root" AWS credentials delivered to an Amazon Kinesis stream called "RootAccess." For more information about how to send AWS CloudTrail events to CloudWatch Logs, see Sending CloudTrail Events to CloudWatch Logs in the AWS CloudTrail User Guide.

  1. Create a destination Amazon Kinesis stream. At a command prompt, type:

     aws kinesis create-stream --stream-name "RootAccess" --shard-count 1

  2. Wait until the Amazon Kinesis stream becomes Active. You can use the Amazon Kinesis describe-stream command to check StreamDescription.StreamStatus property. In addition, take note of the StreamDescription.StreamARN value as that will be passed to CloudWatch Logs later:

    aws kinesis describe-stream --stream-name "RootAccess"

    {
        "StreamDescription": {
            "StreamStatus": "ACTIVE",
            "StreamName": "RootAccess",
            "StreamARN": "arn:aws:kinesis:us-east-1:123456789012:stream/RootAccess",
            "Shards": [
                {
                    "ShardId": "shardId-000000000000",
                    "HashKeyRange": {
                        "EndingHashKey": "340282366920938463463374607431768211455",
                        "StartingHashKey": "0"
                    },
                    "SequenceNumberRange": {
                        "StartingSequenceNumber":
                        "49551135218688818456679503831981458784591352702181572610"
                    }
                }
            ]
        }
    }

    It may take a minute or two for your stream to show as Active.

  3. Create the IAM role that will grant CloudWatch Logs permission to put data into your Amazon Kinesis stream. First, you'll need to create a trust policy in a file ~/TrustPolicyForCWL.json:

    {
      "Statement": {
        "Effect": "Allow",
        "Principal": { "Service": "logs.us-east-1.amazonaws.com" },
        "Action": "sts:AssumeRole"
      }
    }
  4. Use the create-role command to create the IAM role, specifying the trust policy file. Take note of the returned Role.Arn value as that will also be passed to CloudWatch Logs later:

    aws iam create-role --role-name CWLtoKinesisRole --assume-role-policy-document file://~/TrustPolicyForCWL.json

    {
        "Role": {
            "AssumeRolePolicyDocument": {
                "Statement": {
                    "Action": "sts:AssumeRole",
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "logs.us-east-1.amazonaws.com"
                    }
                }
            },
            "RoleId": "AAOIIAH450GAB4HC5F431",
            "CreateDate": "2015-05-29T13:46:29.431Z",
            "RoleName": "CWLtoKinesisRole",
            "Path": "/",
            "Arn": "arn:aws:iam::123456789012:role/CWLtoKinesisRole"
        }
    }
  5. Create a permissions policy to define what actions CloudWatch Logs can do on your account. First, you'll create a permissions policy in a file ~/PermissionsForCWL.json:

    {
      "Statement": [
        {
          "Effect": "Allow",
          "Action": "kinesis:PutRecord",
          "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/RootAccess"
        },
        {
          "Effect": "Allow",
          "Action": "iam:PassRole",
          "Resource": "arn:aws:iam::123456789012:role/CWLtoKinesisRole"
        }
      ]
    }
  6. Associate the permissions policy with the role using the put-role-policy command:

     aws iam put-role-policy --role-name CWLtoKinesisRole --policy-name Permissions-Policy-For-CWL --policy-document file://~/PermissionsForCWL.json
  7. After the Amazon Kinesis stream is in Active state and you have created the IAM role, you can create the CloudWatch Logs subscription filter. The subscription filter immediately starts the flow of real-time log data from the chosen log group to your Amazon Kinesis stream:

    aws logs put-subscription-filter \
        --log-group-name "CloudTrail" \
        --filter-name "RootAccess" \
        --filter-pattern "{$.userIdentity.type = Root}" \
        --destination-arn "arn:aws:kinesis:us-east-1:123456789012:stream/RootAccess" \
        --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisRole"
  8. After you set up the subscription filter, CloudWatch Logs will forward all the incoming log events that match the filter pattern to your Amazon Kinesis stream. You can verify that this is happening by grabbing an Amazon Kinesis shard iterator and using the Amazon Kinesis get-records command to fetch some Amazon Kinesis records:

    aws kinesis get-shard-iterator --stream-name RootAccess --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON
    {
        "ShardIterator":
        "AAAAAAAAAAFGU/kLvNggvndHq2UIFOw5PZc6F01s3e3afsSscRM70JSbjIefg2ub07nk1y6CDxYR1UoGHJNP4m4NFUetzfL+wev+e2P4djJg4L9wmXKvQYoE+rMUiFq+p4Cn3IgvqOb5dRA0yybNdRcdzvnC35KQANoHzzahKdRGb9v4scv+3vaq+f+OIK8zM5My8ID+g6rMo7UKWeI4+IWiK2OSh0uP"
    }
    aws kinesis get-records --limit 10 --shard-iterator "AAAAAAAAAAFGU/kLvNggvndHq2UIFOw5PZc6F01s3e3afsSscRM70JSbjIefg2ub07nk1y6CDxYR1UoGHJNP4m4NFUetzfL+wev+e2P4djJg4L9wmXKvQYoE+rMUiFq+p4Cn3IgvqOb5dRA0yybNdRcdzvnC35KQANoHzzahKdRGb9v4scv+3vaq+f+OIK8zM5My8ID+g6rMo7UKWeI4+IWiK2OSh0uP"

    Note

    You may need to iterate on the get-records command a few times before Amazon Kinesis starts to return data.

    You should expect to see a response with an array of Amazon Kinesis. The Data attribute in the Amazon Kinesis record is Base64 encoded and compressed with the gzip format. You can examine the raw data from the command line using the following Unix commands:

    echo -n "<Content of Data>" | base64 -d | zcat

    The Base64 decoded and decompressed data is formatted as JSON with the following structure:

    {
        "owner": "123456789012",
        "logGroup": "CloudTrail",
        "logStream": "123456789012_CloudTrail_us-east-1",
        "subscriptionFilters": [
            "RootAccess"
        ],
        "messageType": "DATA_MESSAGE",
        "logEvents": [
            {
                "id": "31953106606966983378809025079804211143289615424298221568",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            },
            {
                "id": "31953106606966983378809025079804211143289615424298221569",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            },
            {
                "id": "31953106606966983378809025079804211143289615424298221570",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            }
        ]
    }

    The key elements in the above data structure are the following:

    owner

    The AWS Account ID of the originating log data.

    logGroup

    The log group name of the originating log data.

    logStream

    The log stream name of the originating log data.

    subscriptionFilters

    The list of subscription filter names that matched with the originating log data.

    messageType

    Data messages will use the "DATA_MESSAGE" type. Sometimes CloudWatch Logs may emit Amazon Kinesis records with a "CONTROL_MESSAGE" type, mainly for checking if the destination is reachable.

    logEvents

    The actual log data, represented as an array of log event records. The "id" property is a unique identifier for every log event.

Example 2: AWS Lambda

In this example, you'll create a CloudWatch Logs subscription filter that sends log data to your AWS Lambda function.

  1. Create the AWS Lambda function.

    Ensure that you have set up the Lambda execution role. For more information, see Step 2.2: Create an IAM Role (execution role) in the AWS Lambda Developer Guide.

  2. Open a text editor and create a file named helloWorld.js with the following contents:

    var zlib = require('zlib');
    exports.handler = function(input, context) {
        var payload = new Buffer(input.awslogs.data, 'base64');
        zlib.gunzip(payload, function(e, result) {
            if (e) { 
                context.fail(e);
            } else {
                result = JSON.parse(result.toString('ascii'));
                console.log("Event Data:", JSON.stringify(result, null, 2));
                context.succeed();
            }
        });
    };
  3. Zip the file helloWorld.js and save it with the name helloWorld.zip.

  4. At a command prompt, run the following command, where role-arn is the Lambda execution role set up in the first step, found in the IAM console under Roles:

    aws lambda create-function \
        --function-name helloworld \
        --zip-file file://file-path/helloWorld.zip \
        --role lambda-execution-role-arn \
        --handler helloWorld.handler \
        --runtime nodejs
  5. Grant CloudWatch Logs the permission to execute your function. At a command prompt, run the following command and substitute account 123456789123 with your own and change the log-group to be the log group you want to process:

    aws lambda add-permission \
        --function-name "helloworld" \
        --statement-id "helloworld" \
        --principal "logs.us-east-1.amazonaws.com" \
        --action "lambda:InvokeFunction" \
        --source-arn "arn:aws:logs:us-east-1:123456789123:log-group:TestLambda:*" \
        --source-account "123456789123"
  6. Create a subscription filter. At a command prompt, run the following command and substitute account 123456789123 with your own and change the log-group-name to be the log group you want to process:

    aws logs put-subscription-filter \
        --log-group-name myLogGroup \
        --filter-name demo \
        --filter-pattern "" \
        --destination-arn arn:aws:lambda:us-east-1:123456789123:function:helloworld
  7. (Optional) Test using a sample log event. At a command prompt, run the following command, which will put a simple log message into the subscribed stream.

    To see the output of your Lambda function, navigate to the Lambda function where you will see the output in /aws/lambda/helloworld:

    aws logs put-log-events --log-group-name myLogGroup --log-stream-name stream1 --log-events "[{\"timestamp\":<CURRENT TIMESTAMP MILLIS> , \"message\": \"Simple Lambda Test\"}]"

    You should expect to see a response with an array of Lambda. The Data attribute in the Lambda record is Base64 encoded and compressed with the gzip format. The actual payload that Lambda receives is in the following format { "awslogs": {"data": "BASE64ENCODED_GZIP_COMPRESSED_DATA"} } You can examine the raw data from the command line using the following Unix commands:

    echo -n "<BASE64ENCODED_GZIP_COMPRESSED_DATA>" | base64 -d | zcat

    The Base64 decoded and decompressed data is formatted as JSON with the following structure:

    {
        "owner": "123456789012",
        "logGroup": "CloudTrail",
        "logStream": "123456789012_CloudTrail_us-east-1",
        "subscriptionFilters": [
            "RootAccess"
        ],
        "messageType": "DATA_MESSAGE",
        "logEvents": [
            {
                "id": "31953106606966983378809025079804211143289615424298221568",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            },
            {
                "id": "31953106606966983378809025079804211143289615424298221569",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            },
            {
                "id": "31953106606966983378809025079804211143289615424298221570",
                "timestamp": 1432826855000,
                "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
            }
        ]
    }

    The key elements in the above data structure are the following:

    owner

    The AWS Account ID of the originating log data.

    logGroup

    The log group name of the originating log data.

    logStream

    The log stream name of the originating log data.

    subscriptionFilters

    The list of subscription filter names that matched with the originating log data.

    messageType

    Data messages will use the "DATA_MESSAGE" type. Sometimes CloudWatch Logs may emit Lambda records with a "CONTROL_MESSAGE" type, mainly for checking if the destination is reachable.

    logEvents

    The actual log data, represented as an array of log event records. The "id" property is a unique identifier for every log event.

Example 3: Amazon Kinesis Firehose

In this example, you'll create a CloudWatch Logs subscription that sends any incoming log events that match your defined filters to your Amazon Kinesis Firehose delivery system. Data sent from CloudWatch Logs to Amazon Kinesis Firehose is already compressed with gzip level 6 compression, so you do not need to use compression within your Firehose delivery stream.

  1. Create an Amazon Simple Storage Service (Amazon S3) bucket. We recommend that you use a bucket that was created specifically for CloudWatch Logs. If you want to use an existing bucket, skip to step 2.

    At a command prompt, type:

    aws s3api create-bucket --bucket my-bucket --create-bucket-configuration LocationConstraint=us-east-1

    Replace us-east-1 with the region that you want to use.

    If the bucket was successfully created, the AWS CLI displays the location and name of the bucket:

    {
        "Location": "/my-bucket"
    }
  2. Create the IAM role that will grant Amazon Kinesis Firehose permission to put data into your Amazon S3 bucket.

    For more information, see Controlling Access with Amazon Kinesis Firehose in the Amazon Kinesis Firehose Developer Guide.

    First you'll need to create a trust policy in a file ~/TrustPolicyForFirehose.json:

    {
      "Statement": {
        "Effect": "Allow",
        "Principal": { "Service": "firehose.amazonaws.com" },
        "Action": "sts:AssumeRole",
        "Condition": { "StringEquals": { "sts:ExternalId":"account-ID" } }
      }
      }

    Note

    Replace account-ID with your account ID.
  3. Use the create-role command to create the IAM role, specifying the trust policy file. Take note of the returned Role.Arn value as that will also be passed to Amazon Kinesis Firehose later:

    aws iam create-role \
          --role-name FirehosetoS3Role \
          --assume-role-policy-document file://~/TrustPolicyForFirehose.json
    
    {
        "Role": {
            "AssumeRolePolicyDocument": {
                "Statement": {
                    "Action": "sts:AssumeRole",
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "logs.us-east-1.amazonaws.com"
                    }
                }
            },
            "RoleId": "AAOIIAH450GAB4HC5F431",
            "CreateDate": "2015-05-29T13:46:29.431Z",
            "RoleName": "FirehosetoS3Role",
            "Path": "/",
            "Arn": "arn:aws:iam::123456789012:role/FirehosetoS3Role"
        }
    }
  4. Create a permissions policy to define what actions Firehose can do on your account. First, you'll create a permissions policy in a file ~/PermissionsForFirehose.json:

    {
      "Statement": [
        {
          "Effect": "Allow",
          "Action": [ "s3:AbortMultipartUpload", "s3:GetBucketLocation", "s3:GetObject", "s3:ListBucket", "s3:ListBucketMultipartUploads", "s3:PutObject" ],
          "Resource": [ "arn:aws:s3:::my-bucket", "arn:aws:s3:::my-bucket/*" ]
        }
      ]
    }
  5. Associate the permissions policy with the role using the put-role-policy command:

    aws iam put-role-policy --role-name FirehosetoS3Role --policy-name Permissions-Policy-For-Firehose --policy-document file://~/PermissionsForFirehose.json
  6. Create a destination Firehose delivery stream:

    aws firehose create-delivery-stream \
    --delivery-stream-name 'my-delivery-stream' \
    --s3-destination-configuration \
    RoleARN='arn:aws:iam::123456789012:role/FirehosetoS3Role',BucketARN='arn:aws:s3:::my-bucket'
                        

    Replace RoleARN and BucketARN with the role and bucket ARN that you created.

    Note

    Firehose automatically uses a prefix in YYYY/MM/DD/HH UTC time format for delivered Amazon S3 objects. You can specify an extra prefix to be added in front of the time format prefix. If the prefix ends with a forward slash (/), it appears as a folder in the Amazon S3 bucket.

  7. Wait until the stream becomes active. You can use the Firehose describe-delivery-stream command to check the DeliveryStreamDescription.DeliveryStreamStatus property. In addition, take note of the DeliveryStreamDescription.DeliveryStreamARN value as that will be passed to CloudWatch Logs later:

    aws firehose describe-delivery-stream --delivery-stream-name "my-delivery-stream"
    {
        "DeliveryStreamDescription": {
            "HasMoreDestinations": false,
            "VersionId": "1",
            "CreateTimestamp": 1446075815.822,
            "DeliveryStreamARN": "arn:aws:firehose:us-east-1:123456789012:deliverystream/my-delivery-stream",
            "DeliveryStreamStatus": "ACTIVE",
            "DeliveryStreamName": "my-delivery-stream",
            "Destinations": [
                {
                    "DestinationId": "destinationId-000000000001",
                    "S3DestinationDescription": {
                        "CompressionFormat": "UNCOMPRESSED",
                        "EncryptionConfiguration": {
                            "NoEncryptionConfig": "NoEncryption"
                        },
                        "RoleARN": "delivery-stream-role",
                        "BucketARN": "arn:aws:s3:::my-bucket",
                        "BufferingHints": {
                            "IntervalInSeconds": 300,
                            "SizeInMBs": 5
                        }
                    }
                }
            ]
        }
        }

    It may take a few minutes for your stream to show as active.

  8. Create the IAM role that will grant CloudWatch Logs permission to put data into your Firehose delivery stream. First, you'll need to create a trust policy in a file ~/TrustPolicyForCWL.json:

    {
      "Statement": {
        "Effect": "Allow",
        "Principal": { "Service": "logs.us-east-1.amazonaws.com" },
        "Action": "sts:AssumeRole"
      }
      }

  9. Use the create-role command to create the IAM role, specifying the trust policy file. Take note of the returned Role.Arn value as that will also be passed to CloudWatch Logs later:

    aws iam create-role \
          --role-name CWLtoKinesisFirehoseRole \
          --assume-role-policy-document file://~/TrustPolicyForCWL.json
    
    {
        "Role": {
            "AssumeRolePolicyDocument": {
                "Statement": {
                    "Action": "sts:AssumeRole",
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "logs.us-east-1.amazonaws.com"
                    }
                }
            },
            "RoleId": "AAOIIAH450GAB4HC5F431",
            "CreateDate": "2015-05-29T13:46:29.431Z",
            "RoleName": "CWLtoKinesisFirehoseRole",
            "Path": "/",
            "Arn": "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
        }
    }
  10. Create a permissions policy to define what actions CloudWatch Logs can do on your account. First, you'll create a permissions policy in a file ~/PermissionsForCWL.json:

    {
        "Statement":[
          {
            "Effect":"Allow",
            "Action":["firehose:*"],
            "Resource":["arn:aws:firehose:us-east-1:123456789012:*"]
          },
          {
            "Effect":"Allow",
            "Action":["iam:PassRole"],
            "Resource":["arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"]
          }
        ]
    }
  11. Associate the permissions policy with the role using the put-role-policy command:

    aws iam put-role-policy --role-name CWLtoKinesisFirehoseRole --policy-name Permissions-Policy-For-CWL --policy-document file://~/PermissionsForCWL.json"
  12. After the Amazon Kinesis Firehose delivery stream is in active state and you have created the IAM role, you can create the CloudWatch Logs subscription filter. The subscription filter immediately starts the flow of real-time log data from the chosen log group to your Amazon Kinesis Firehose delivery stream:

    aws logs put-subscription-filter \
        --log-group-name "CloudTrail" \
        --filter-name "RootAccess" \
        --filter-pattern "{$.userIdentity.type = Root}" \
        --destination-arn "arn:aws:firehose:us-east-1:123456789012:deliverystream/my-delivery-stream" \
        --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
    
  13. After you set up the subscription filter, CloudWatch Logs will forward all the incoming log events that match the filter pattern to your Amazon Kinesis Firehose delivery stream. Your data will start appearing in your Amazon S3 based on the time buffer interval set on your Amazon Kinesis Firehose delivery stream. Once enough time has passed, you can verify your data by checking your Amazon S3 Bucket.

    aws s3api list-objects --bucket 'my-bucket' --prefix 'firehose/'
    
    {
        "Contents": [
            {
                "LastModified": "2015-10-29T00:01:25.000Z",
                "ETag": "\"a14589f8897f4089d3264d9e2d1f1610\"",
                "StorageClass": "STANDARD",
                "Key": "firehose/2015/10/29/00/my-delivery-stream-2015-10-29-00-01-21-a188030a-62d2-49e6-b7c2-b11f1a7ba250",
                "Owner": {
                    "DisplayName": "cloudwatch-logs",
                    "ID": "1ec9cf700ef6be062b19584e0b7d84ecc19237f87b5"
                },
                "Size": 593
            },
            {
                "LastModified": "2015-10-29T00:35:41.000Z",
                "ETag": "\"a7035b65872bb2161388ffb63dd1aec5\"",
                "StorageClass": "STANDARD",
                "Key": "firehose/2015/10/29/00/my-delivery-stream-2015-10-29-00-35-40-7cc92023-7e66-49bc-9fd4-fc9819cc8ed3",
                "Owner": {
                    "DisplayName": "cloudwatch-logs",
                    "ID": "1ec9cf700ef6be062b19584e0b7d84ecc19237f87b6"
                },
                "Size": 5752
            }
        ]
        }
    aws s3api get-object --bucket 'my-bucket' --key 'firehose/2015/10/29/00/my-delivery-stream-2015-10-29-00-01-21-a188030a-62d2-49e6-b7c2-b11f1a7ba250' testfile.gz
    
    {
        "AcceptRanges": "bytes",
        "ContentType": "application/octet-stream",
        "LastModified": "Thu, 29 Oct 2015 00:07:06 GMT",
        "ContentLength": 593,
        "Metadata": {}
        }

    The data in the Amazon S3 object is compressed with the gzip format. You can examine the raw data from the command line using the following Unix commands:

    zcat testfile.gz