使用 CloudWatch LogssUbscridfilters - Amazon CloudWatch Logs
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

使用 CloudWatch LogssUbscridfilters

您可以将订阅筛选器与 Kinesis、Lambda 或 Kinesis Data Firehose 一起使用。通过订阅筛选器发送到接收服务的日志使用 gzip 格式进行 Base64 编码和压缩。

示例 1:订阅filters 与 Kinesis

以下示例将订阅筛选器与日志组关联,该日志组包含Amazon CloudTrail事件使每个记录的活动都由 “Root”Amazon凭据传送到名为 “RootAccess” 的 Kinesis 力数据流。有关如何将更多信息发送Amazon CloudTrail事件添加到 CloudWatch Logs 中,请参阅将 CloudTrail 事件发送到 CloudWatch Logs中的Amazon CloudTrail用户指南

注意

在创建 Kinesis 流之前,请计算将生成日志数据的卷。请确保创建具有足够分片的 Kinesis 流来处理此卷。如果流没有足够的分片,日志流将受限制。有关 Kinesis Stream 卷限制的更多信息,请参阅。Amazon Kinesis Data Streams 限制

为 Kinesis 创建订阅筛选条件

  1. 使用以下命令创建目标 Kinesis 流:

    $ C:\> aws kinesis create-stream --stream-name "RootAccess" --shard-count 1
  2. 请耐心等待,直到 Kinesis 流变为活动状态 (这可能需要一两分钟)。您可以使用以下 Kinesis描述流命令检查流描述。流状态属性。此外,请记下 StreamDescription.StreamARN 值,因为后面的步骤中将会用到它:

    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" } } ] } }
  3. 创建 IAM 角色,该角色将向授予 CloudWatch Logs 将数据放入 Kinesis 流的权限。首先,您需要在文件 (例如 ~/TrustPolicyForCWL.json) 中创建信任策略。使用文本编辑器创建此策略。请勿使用 IAM 控制台来创建策略。

    { "Statement": { "Effect": "Allow", "Principal": { "Service": "logs.region.amazonaws.com" }, "Action": "sts:AssumeRole" } }
  4. 使用create-role命令创建 IAM 角色,并指定信任策略文件。请记下返回的 Role.Arn 值,因为后面的步骤中将会用到它:

    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.region.amazonaws.com" } } }, "RoleId": "AAOIIAH450GAB4HC5F431", "CreateDate": "2015-05-29T13:46:29.431Z", "RoleName": "CWLtoKinesisRole", "Path": "/", "Arn": "arn:aws:iam::123456789012:role/CWLtoKinesisRole" } }
  5. 创建权限策略以定义 CCloudWatch Logs 可对您的账户执行的操作。首先,您将在文件 (例如 ~/PermissionsForCWL.json) 中创建权限策略。使用文本编辑器创建此策略。请勿使用 IAM 控制台来创建策略。

    { "Statement": [ { "Effect": "Allow", "Action": "kinesis:PutRecord", "Resource": "arn:aws:kinesis:region:123456789012:stream/RootAccess" } ] }
  6. 使用以下 put-role-policy 命令将权限策略与角色关联:

    aws iam put-role-policy --role-name CWLtoKinesisRole --policy-name Permissions-Policy-For-CWL --policy-document file://~/PermissionsForCWL.json
  7. 之后 Kinesis 流是在处于活动状态状态并且您已创建 IAM 角色,您便可以创建 CloudWatch Logs 订阅筛选条件。订阅筛选条件将立即启动从所选日志组到您的 Kinesis 流的实时日志数据流动:

    aws logs put-subscription-filter \ --log-group-name "CloudTrail" \ --filter-name "RootAccess" \ --filter-pattern "{$.userIdentity.type = Root}" \ --destination-arn "arn:aws:kinesis:region:123456789012:stream/RootAccess" \ --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisRole"
  8. 当您设置订阅筛选条件后,CloudWatch Logs 将与筛选条件模式匹配的所有传入事件日志转发到您的 Kinesis 流。您可以抓取 Kinesis 分片迭代器并使用 Kinesis get-记录命令获取一些 Kinesis 记录以验证此事件是否发生:

    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"

    请注意,您可能需要在 Kinesis 开始返回数据之前调用几次此命令。

    您将看到包含一组记录的响应。这些区域有:数据属性使用 Base64 编码并使用 gzip 格式进行压缩。您可使用以下 Unix 命令检查命令行中的原始数据:

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

    Base64 解码和解压缩数据被格式化为 JSON 并具有以下结构:

    { "owner": "111111111111", "logGroup": "CloudTrail", "logStream": "111111111111_CloudTrail_us-east-1", "subscriptionFilters": [ "Destination" ], "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\"}" } ] }

    上述数据架构中的关键元素如下:

    owner

    这些区域有:Amazon原始日志数据的账户 ID。

    logGroup

    原始日志数据的日志组名称。

    logStream

    原始日志数据的日志流名称。

    subscriptionFilters

    与原始日志数据匹配的订阅筛选条件名称的列表。

    messageType

    数据消息将使用“DATA_MESSAGE”类型。有时候,CloudWatch Logs 可能发出类型为 “CONTROL_MESSAGE” 的动力记录,这主要是为了检查目标是否可访问。

    logEvents

    表示为一组日志事件记录的实际日志数据。“id”属性是每个日志事件的唯一标识符。

示例 2:订阅f易尔特Amazon Lambda

在此示例中,您将创建 CloudWatch Logs 订阅筛选条件,该筛选条件将日志数据发送到Amazon Lambdafunction.

注意

在创建 Lambda 函数之前,请计算将生成日志数据的卷。请确保创建一个处理此卷的函数。如果函数没有足够的卷,日志流将受限制。有关 Lambda 限制的更多信息,请参阅。Amazon Lambda限制

为 Lambda 创建订阅筛选条件

  1. 创建 Amazon Lambda 函数。

    确保您已设置 Lambda 执行角色。有关更多信息,请参阅 。步骤 2.2:创建 IAM 角色(执行角色)中的Amazon Lambda开发人员指南

  2. 打开文本编辑器,并用以下内容创建名为 helloWorld.js 的文件:

    var zlib = require('zlib'); exports.handler = function(input, context) { var payload = Buffer.from(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. 压缩 helloWorld.js 文件,并用名称 helloWorld.zip 保存它。

  4. 使用以下命令,其中角色是您在第一步中设置的 Lambda 执行角色:

    aws lambda create-function \ --function-name helloworld \ --zip-file fileb://file-path/helloWorld.zip \ --role lambda-execution-role-arn \ --handler helloWorld.handler \ --runtime nodejs12.x
  5. 授予 CloudWatch Logs 执行您的函数的权限。使用以下命令,将占位符账户替换为您自己的账户,将占位符日志组替换为要处理的日志组:

    aws lambda add-permission \ --function-name "helloworld" \ --statement-id "helloworld" \ --principal "logs.region.amazonaws.com" \ --action "lambda:InvokeFunction" \ --source-arn "arn:aws:logs:region:123456789123:log-group:TestLambda:*" \ --source-account "123456789012"
  6. 使用以下命令创建订阅筛选条件,将占位符账户替换为您自己的账户,将占位符日志组替换为要处理的日志组:

    aws logs put-subscription-filter \ --log-group-name myLogGroup \ --filter-name demo \ --filter-pattern "" \ --destination-arn arn:aws:lambda:region:123456789123:function:helloworld
  7. (可选) 使用样本日志事件进行测试。在出现命令提示符时,运行以下命令,以将一条简单的日志消息放入已订阅的流中。

    要查看 Lambda 函数的输出,请导航至 Lambda 函数,在此处可以查看 /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\"}]"

    您将看到包含 Lambda 数组的响应。这些区域有:数据属性使用 Base64 编码并使用 gzip 格式进行压缩。Lambda 接收的实际有效载荷采用以下格式{ "awslogs": {"data": "BASE64ENCODED_GZIP_COMPRESSED_DATA"} }您可以使用以下 Unix 命令检查命令行中的原始数据:

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

    Base64 解码和解压缩数据被格式化为 JSON 并具有以下结构:

    { "owner": "123456789012", "logGroup": "CloudTrail", "logStream": "123456789012_CloudTrail_us-east-1", "subscriptionFilters": [ "Destination" ], "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\"}" } ] }

    上述数据架构中的关键元素如下:

    owner

    这些区域有:Amazon原始日志数据的账户 ID。

    logGroup

    原始日志数据的日志组名称。

    logStream

    原始日志数据的日志流名称。

    subscriptionFilters

    与原始日志数据匹配的订阅筛选条件名称的列表。

    messageType

    数据消息将使用“DATA_MESSAGE”类型。有时候,CloudWatch Logs 可能发出类型为 “CONTROL_MESSAGE” 的 Lambda 记录,这主要是为了检查目标是否可访问。

    logEvents

    表示为一组日志事件记录的实际日志数据。“id”属性是每个日志事件的唯一标识符。

示例 3:订阅fAmazon Kinesis Data Firehose 使用

在此示例中,您将创建一个 CloudWatch Logs 订阅,用于将与所定义筛选条件匹配的所有传入日志事件发送到您的 Amazon Kinesis Data Firehose 传输流。已经使用 gzip 6 级压缩对从 CloudWatch Logs 发送到 Amazon Kinesis Data Firehose 的数据进行压缩,因此您无需在 Kinesis Data Firehose 传输流中使用压缩。

注意

在创建 Kinesis Data Firehose 流之前,请计算将生成日志数据的卷。请确保创建一个 Kinesis Data Firehose 流来处理此卷。如果流无法处理卷,则日志流将受限制。有关 Kinesis Data Firehose 流卷限制的更多信息,请参阅Amazon Kinesis Data Firehose 数据限制

为 Kinesis Data Firehose 创建订阅筛选条件

  1. 创建 Amazon Simple Storage Service (Amazon S3) 存储桶。我们建议您使用专为 CloudWatch Logs 创建的存储桶。但是,如果要使用现有存储桶,请跳至第 2 步。

    运行以下命令,将占用符 Region 替换为要使用的区域:

    aws s3api create-bucket --bucket my-bucket --create-bucket-configuration LocationConstraint=region

    下面是示例输出:

    { "Location": "/my-bucket" }
  2. 创建 IAM 角色,该角色将向授予 Amazon Kinesis Data Firehose 权限。

    有关更多信息,请参阅 。使用 Amazon Kinesis Data Firehose 控制访问中的Amazon Kinesis Data Firehose 开发人员指南

    首先,使用文本编辑器在文件中创建信任策略~/TrustPolicyForFirehose.json如下所示,将account-id与您的Amazon账户 ID:

    { "Statement": { "Effect": "Allow", "Principal": { "Service": "firehose.amazonaws.com" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId":"account-id" } } } }
  3. 使用create-role命令创建 IAM 角色,并指定信任策略文件。请记下返回的 Role.Arn 值,因为后面的步骤中将会用到它:

    aws iam create-role \ --role-name FirehosetoS3Role \ --assume-role-policy-document file://~/TrustPolicyForFirehose.json { "Role": { "AssumeRolePolicyDocument": { "Statement": { "Action": "sts:AssumeRole", "Effect": "Allow", "Principal": { "Service": "firehose.amazonaws.com" } } }, "RoleId": "AAOIIAH450GAB4HC5F431", "CreateDate": "2015-05-29T13:46:29.431Z", "RoleName": "FirehosetoS3Role", "Path": "/", "Arn": "arn:aws:iam::123456789012:role/FirehosetoS3Role" } }
  4. 创建权限策略以定义 Kinesis Data Firehose 可对您的账户执行的操作。首先,使用文本编辑器在文件 ~/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. 使用以下 put-role-policy 命令将权限策略与角色关联:

    aws iam put-role-policy --role-name FirehosetoS3Role --policy-name Permissions-Policy-For-Firehose --policy-document file://~/PermissionsForFirehose.json
  6. 如下所示创建目标 Kinesis Data Firehose 传输流,将RoleARNBucketARN使用您创建的角色和存储桶 ARN:

    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"}'

    请注意,Kinesis Data Firehose 对所传输的 Amazon S3 对象自动使用 YYYY/M/DD/HH UTC 时间格式的前缀。您可以指定要添加在该时间格式前缀前面的额外前缀。如果前缀以正斜杠 (/) 结束,则在 Amazon S3 存储桶中显示为文件夹。

  7. 请耐心等待,直到流变为活动状态 (这可能需要几分钟时间)。您可以使用 Kinesis Data Firehose描述-交付流命令检查交付流描述。交付流状态属性。此外,请记下 DeliveryStreamDescription.DeliveryStreamARN 值,因为后面的步骤中将会用到它:

    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 } } } ] } }
  8. 创建 IAM 角色,该角色将向授予 CloudWatch Logs 权限,将数据放入 Kinesis Data Firehose 传输流的权限。首先,使用文本编辑器在文件 ~/TrustPolicyForCWL.json 中创建信任策略:

    { "Statement": { "Effect": "Allow", "Principal": { "Service": "logs.region.amazonaws.com" }, "Action": "sts:AssumeRole" } }
  9. 使用create-role命令创建 IAM 角色,并指定信任策略文件。请记下返回的 Role.Arn 值,因为后面的步骤中将会用到它:

    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.region.amazonaws.com" } } }, "RoleId": "AAOIIAH450GAB4HC5F431", "CreateDate": "2015-05-29T13:46:29.431Z", "RoleName": "CWLtoKinesisFirehoseRole", "Path": "/", "Arn": "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole" } }
  10. 创建权限策略以定义 CCloudWatch Logs 可对您的账户执行的操作。首先,使用文本编辑器创建权限策略文件 (例如 ~/PermissionsForCWL.json):

    { "Statement":[ { "Effect":"Allow", "Action":["firehose:*"], "Resource":["arn:aws:firehose:region:123456789012:*"] } ] }
  11. 使用 put-role-policy 命令将权限策略与角色关联:

    aws iam put-role-policy --role-name CWLtoKinesisFirehoseRole --policy-name Permissions-Policy-For-CWL --policy-document file://~/PermissionsForCWL.json
  12. 在 Amazon Kinesis Data Firehose 传输流进入活动状态并且您已创建 IAM 角色后,您便可以创建 CloudWatch Logs 订阅筛选条件。订阅筛选条件将立即让实时日志数据开始从所选日志组向您的 Amazon Kinesis Data Firehose 传输流的流动:

    aws logs put-subscription-filter \ --log-group-name "CloudTrail" \ --filter-name "Destination" \ --filter-pattern "{$.userIdentity.type = Root}" \ --destination-arn "arn:aws:firehose:region:123456789012:deliverystream/my-delivery-stream" \ --role-arn "arn:aws:iam::123456789012:role/CWLtoKinesisFirehoseRole"
  13. 当您设置订阅筛选条件后,CloudWatch Logs 将与筛选条件模式匹配的所有传入事件日志转发到您的 Amazon Kinesis Data Firehose 传输流。根据 Amazon Kinesis 数据防火管传输流上设置的时间缓冲间隔,数据将开始显示在您的 Amazon S3 中。经过足够的时间后,您可以通过检查您的 Amazon S3 存储桶验证您的数据。

    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": {} }

    Amazon S3 对象中的数据以 gzip 格式压缩。您可以使用以下 Unix 命令检查命令行中的原始数据:

    zcat testfile.gz