使用 CloudWatch Logs，Kinesis Firehose，Athena 和 Quicksight 实时分析 Amazon Aurora 数据库审计日志

Aurora 高级审计功能介绍

关系数据库管理系统（RDBMS）通常支撑着最重要的联机交易类应用，存放着最重要的数据资产，所以在用户 IT 系统里占据着非常核心的位置。也正因为如此，不管是出于内在的数据保护的需求还是外部监管及合规的要求，数据库用户都需要对数据库的操作记录进行审计。传统的商业数据库管理系统（比如 Oracle，SQL Server）也都提供了审计的功能，开源数据库 MySQL 也可以通过 MariaDB 审计插件提供此项功能。但现实情况往往是审计功能虽然使用并不复杂，但是鲜见有用户打开此功能。主要原因在于审计功能的对数据库性能的负面影响太大，特别是在业务负载较大的时候。这种情况下，数据库的审计功能成为鸡肋也就不奇怪了。

Amazon Aurora MySQL 1.10.1及其以后版本提供了高性能的高级审计功能，即使在开启审计功能的情况下依然能够保持数据库的高性能。那么 Aurora 是如何做到这一点的呢？我们可以对比一下 MariaDB 审计插件和 Aurora 高级审计功能的实现机制来找寻答案。

MariaDB 审计插件使用单线程和互斥锁来记录每个审计事件，这种方式虽然能够严格保持事件的顺序，但是写日志产生的瓶颈对性能的影响也较大。为了维持 Aurora 数据库的高性能，在设计审计功能的时候使用了 latch-free 的队列方式存储审计事件，从而解耦了日志写和实际的业务操作。通过对队列使用多线程消费，写入到多个审计日志文件中。

MariaDB audit plugin 和 Aurora 审计机制对比

在 8xlarge 实例上对 MySQL5.7 和 Aurora 审计开关分别进行 sysbench 测试，MySQL5.7 开启审计情况下对吞吐量有65%的影响，而 Aurora 仅有15%

MySQL5.7、Aurora 审计对性能影响对比

但是 Aurora 这种实现方式也存在一些不便之处，比如：

审计文件中日志中事件的顺序可能不是严格按照时间顺序排列，审计事件存放在多个审计文件中，需要后续进一步处理。
在 console 上虽然可以查看日志文件，但是不方便检索关键字。
审计文件存放在 Aurora 实例的本地盘上，遇到实例 failure 的情况，可能会丢失审计日志。
系统会根据空间使用情况以及日志文件的存放时间进行清理以释放空间，因此需要考虑备份和持久保存的问题。

幸运的是，在 AWS 上有很多服务不但可以解决以上问题，同时还可以很容易地构建一个自动化的无服务器架构的审计日志分析 pipeline。本 blog 的以下部分将分别介绍：

使用 CloudWatch Logs 收集和管理审计日志
使用 CloudWatch Logs subscription filter 发送审计日志到 Kinesis Firehose 进行数据格式转换
使用 Athena 和 QuickSight 进行日志查询和可视化分析

Aurora 审计日志实时分析架构设计

使用 CloudWatch Logs 管理 Aurora 审计日志

要使用 CloudWatch Logs 管理 Aurora 审计日志，需要首先打开数据库的审计功能，然后赋予数据库集群发送审计日志到 CloudWatch Logs 的权限。

Step 1: 打开集群的审计功能

登陆 AWS 管理控制台，打开 RDS 控制台，在左边的导航面板点击 “Parameter Groups”，接着点击你的 Aurora 集群关联的参数组，然后选择 “Edit Parameters”。

Aurora集群参数组管理

在过滤器里输入“server_audit”，设置 server_audit_logging 为1，设置 server_audit_events, 输入你要审计的事件类型，类型之间用逗号分隔, 其余参数设置及详细说明参见 Amazon Aurora MySQL 数据库群集使用高级审计。

修改 Aurora 集群参数打开审计功能

以上设置打开了集群的审计功能，此时你可以导航到 “Instances” 查看 Logs：

Aurora 实例 Log 文件管理

Step 2: 赋予 Aurora 集群向 CloudWatch Logs 发送日志的权限，并打开发送开关

我们首先需要创建一个 service role，你可以参照文档“将审核日志数据从 Amazon Aurora 发布到 Amazon CloudWatch Logs”，手工创建 policy 和 role，也可以使用系统自带的 AWSServiceRoleForRDS 角色，它 attach 了 AWS managed policy: AmazonRDSServiceRolePolicy，其中包含 CloudWatch Logs 发送日志需要的权限，在这里我们可以直接使用。

为 Aurora 集群创建 CloudWatch Logs role

创建成果 role 之后，记下 Role ARN，打开 cluster parameter，编辑 aws_default_logs_role，设置参数值为记下的 Role ARN，同时将 server_audit_upload 参数设置为 1，该参数是 upload 的开关。

设置 Aurora 集群 upload CloudWatch Logs

现在 Aurora 开始向 CloudWatch Logs 发送日志，查看日志请打开 CloudWatch 的 console，点击 Logs，找到 /aws/rds/cluster/<your cluster’s name>/audit 日志组

CloudWatch Logs 中查看审计日志

日志上传到 CloudWatch Logs 之后你可以：

查询和搜索日志
定义日志保存日期（默认永久保存）
自定义 metric filter，比如对 delete 和 drop table 的 SQL 操作语句定义 metric filter
根据自定义 metric 设置 alarm，比如你可以定义 drop table 的时候自动发送告警短信或者 email 到管理员

metric filter 设置

为审计日志设置 Alarm

使用 CloudWatch Logs 订阅过滤器（subscription filter）将日志发送到 Firehose 进一步处理

CloudWatch Logs 订阅可以从 CloudWatch Logs 中访问日志事件的实时源并将其传输到其他服务 (如 Amazon Kinesis 流、Amazon Kinesis Data Firehose 流或 AWS Lambda)，以进行自定义处理、分析或加载到其他系统中。

Step 1: 为 Kinesis Firehose 建立 subscription filter

该操作暂时不支持 console，只能通过 AWS CLI 或者 API 进行，详细操作步骤请参考 Amazon Kinesis Data Firehose 订阅筛选器，主要步骤如下：

为 Firehose 建立 S3 bucket 用于存放日志文件

创建 IAM 角色，该角色将向 Amazon Kinesis Data Firehose 授予将数据放入您的 Amazon S3 存储桶的权限
创建目标 Kinesis Data Firehose 传输流
创建 IAM 角色，该角色将向 CloudWatch Logs 授予将数据放入 Kinesis Data Firehose 传输流的权限
在 Amazon Kinesis Data Firehose 传输流进入活动状态并且您已创建 IAM 角色后，您便可以创建 CloudWatch Logs 订阅筛选器。订阅筛选器将立即启动从所选日志组到您的 Amazon Kinesis Data Firehose 传输流的实时日志数据流动：

aws logs put-subscription-filter \

–log-group-name “/aws/rds/cluster/aurora-1/audit” \

–filter-name “FirehoseDestination” \

–filter-pattern “” \

–destination-arn “arn:aws:firehose:us-west-1:1234567890:deliverystream/my-delivery-stream” \

–role-arn “arn:aws:iam::1234567890:role/CWLtoKinesisFirehoseRole” \

–region us-west-1

Step 2: 使用 Lambda 对日志进行格式化

Step 1完成了 CloudWatch Logs 实时发送到 Kinesis Firehose，传送到 S3 的过程，但是这个日志是数据库实例产生的原始日志，日志字段之间使用逗号分隔没有问题，但是在 “event” 字段记录着数据库用户提交的 SQL 语句，由于语句内部可能存在逗号（比如 select id, name from student），如果要使用 Athena 直接分析 csv 文件的话，会出现我们所不期望的将 event 字段分隔成多个 field 的情况，因此我们需要将 event 字段内部的逗号替换成空格字符。Kinesis Firehose 可以调用 Lambda 函数完成该转换工作。

Kinesis Firehose 提供了 “Kinesis Firehose CloudWatch Logs Processor” 的 Lambda blueprint，该 blueprint 可以完成对 CloudWatch Logs subscription filter 发送过来的 log event 的解析，我们只要修改其中的transformLogEvent 函数，就可以完成类似 “select id,name from student” 转换为 “select id name from student” 的功能。修改后的代码如下：

def transformLogEvent(log_event):

“””Transform each log event.

The default implementation below just extracts the message and appends a newline to it.

Args:log_event (dict): The original log event. Structure is {“id”: str, “timestamp”: long, “message”: str}

Returns:

str: The transformed log event.

“””

#return log_event[‘message’] + ‘\n’ this statement is replaced by below to replace the “,” in sql with space

converted=””

temp=log_event[‘message’]

#print(temp)

templist=temp.split(‘\”)

for i in range(0,len(templist)):

if i%2==1:

a=templist[i]

templist[i]=a.replace(‘,’,’ ‘)

if i < len(templist)-1:

converted=converted + templist[i]+’\”

converted=converted + templist[len(templist)-1]

return converted + ‘\n’

Step 3: 使用 Athena 对转换后的日志文件进行分析

在 Athena 控制台为日志文件创建外部表

CREATE EXTERNAL TABLE IF NOT EXISTS auroralog.audit_records (

`timestamp` bigint,`serverhost` string,`username` string,`host` string,

`connectionid` string,

`queryid` int,

`operation` string,

`mydatabase` string,

`object` string,

`retcode` int)

ROW FORMAT SERDE ‘org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe’

WITH SERDEPROPERTIES (

‘serialization.format’ = ‘,’,

‘field.delim’ = ‘,’

) LOCATION ‘s3://auroraauditlog/’

TBLPROPERTIES (‘has_encrypted_data’=’false’);

Athena console 中创建 database 及 external table

例如，我们可以查询某个时间段包含 drop 操作的审计事件：

select FROM_UNIXTIME(timestamp/1000000) as timestamp1,

username,

host,

operation,

object

from audit_records

where FROM_UNIXTIME(timestamp/1000000) >timestamp ‘2018-03-20 06:04:00’ and upper(object) like ‘%DROP%’

Step 4: 使用 Quicksight 可视化分析审计日志

现在我们有了审计日志在 S3 和 Athena 外部表，就可以很方便的通过几步点击使用 QuickSight 生成可视图表：

打开Amazon QuickSight 仪表板，选择 “New analysis”，接着选择 “New data set”，选择 Athena 作为数据源，并为数据源输入一个名字，接着选择 “aurora” 数据库和 “audit_records” 表，选择 Select。选择 SPICE 或者Direct query，点击 “Edit/Preview data”，输入以下 SQL 查询而不是使用原表格，为 Query 输入一个名字，点击 “Finish”，最后点击顶部栏的 “Save & visualize”

select distinct queryid,

date_trunc(‘day’,FROM_UNIXTIME(timestamp/1000000)) as event_time,

username,

host,

connectionid,

operation,

mydatabase,

object,

retcode

from auroralog.audit_records

order by event_time desc

建立好 analysis 之后，就可以选择不同类型的 visual type 创建可视化分析，例如我们可以分析不同用户每天操作的次数：

折线图

不同用户不同类型的操作次数：

水平条形图

总结

Aurora MySQL 数据库原生的高级审计功能因其独辟蹊径的实现方式，较传统数据库审计对数据库的性能影响显著降低，在开启审计功能的情况下依然能保持高性能，从而使得用户不必在合规、数据安全和性能之间做痛苦的妥协。同时，通过和 AWS 的其他服务集成，可以非常方便的构建一个无服务器架构的审计日志实时分析的工作流。当然以上架构并非唯一选择，用户可以根据实际情况予以选择取舍，如果你不需要实时分析，也可以通过批量的方式从 CloudWatch Logs 拉取 log event 到 S3，采用自己熟悉的分析工具进行处理。

亚马逊AWS官方博客