第2章 审计日志格式与采集
做数据库审计，首先要搞定日志。
很多人一上来就查异常，结果连日志长什么样子都没看清楚。我个人习惯是，先摸清三个东西：日志从哪来、长什么样、怎么收。
这一章咱们就掰开看看主流的审计日志格式，再聊聊采集工具怎么搭。你看懂了格式，后续的排查才能踩到点上。
2.1 主流数据库审计日志格式
格式不统一，这是最头疼的。我最早接触 MySQL 的时候，觉得 general log 挺好读的；后来切到 Oracle，一看 audit trail 的数据字典，嗯，差点想骂人。但格式再乱，核心信息就那么几样：谁、什么时间、从哪连的、干了什么、结果如何。
2.1.1 MySQL general log
MySQL 的通用查询日志默认是纯文本格式。你打开一看，大概长这样：
2025/03/15 10:32:14 12345 Connect root@10.0.0.5 on db1
2025/03/15 10:32:16 12345 Query SELECT * FROM users WHERE id=1
2025/03/15 10:32:18 12345 Quit每一行都包含时间戳、线程 ID、命令类型、详细信息。说白了，它把所有 SQL 都记下来了，包括连接和断开。优点是全，缺点是全 —— 生产环境开 general log 很容易把磁盘写爆。
我建议：只在排查阶段短时开启，或者配合 slow query log 一起看。
SET GLOBAL general_log = ON 临时开启，用完马上关。记得审计完之后清理日志文件。
2.1.2 Oracle audit trail
Oracle 的审计日志存储在这几个视图里：DBA_AUDIT_TRAIL、DBA_FGA_AUDIT_TRAIL。格式是关系型表格，字段很多，我挑几个最重要的列出来：
| 字段名 | 含义 | 典型值 |
|---|---|---|
| OS_USERNAME | 操作系统用户名 | oracle |
| USERNAME | 数据库用户名 | SCOTT |
| TERMINAL | 客户端主机名 | app-server-01 |
| TIMESTAMP | 审计时间 | 15-MAR-25 10:32:14 |
| ACTION_NAME | 操作类型 | SELECT |
| OBJECT_NAME | 操作对象 | USERS |
| SQL_TEXT | 实际执行的SQL（部分版本） | SELECT * FROM users WHERE id=1 |
Oracle 的审计粒度可以控制得很细，通过 AUDIT 命令精确到特定表或特定操作。我记得有一次排查一个“疑似删表”的案子，就是靠 DBA_AUDIT_TRAIL 里的 ACTION_NAME = 'DROP TABLE' 直接锁定了人。
audit_trail=DB_EXTENDED 参数。我曾经吃过这个亏，查了半天没找到 SQL，最后发现参数没改。
2.1.3 SQL Server Audit
SQL Server 的审计机制比较现代，它把日志写到文件或 Windows 事件日志里，也可以通过 sys.fn_get_audit_file 函数查询。一个典型的审计记录字段如下：
| 字段 | 说明 | |
|---|---|---|
| event_time | 事件发生时间 | 2025-03-15 10:32:14.123 |
| sequence_number | 同一事务内的序号 | 1 |
| action_id | 操作ID（如 SELECT=42） | 42 |
| session_server_principal_name | 登录名 | sa |
| database_name | 数据库名 | db1 |
| schema_name | 架构名 | dbo |
| object_name | 对象名 | users |
| statement | T-SQL 语句 | SELECT * FROM users WHERE id=1 |
SQL Server Audit 有一个好用的特性：可以按 审计规范 分组，把需要监控的数据库、操作、用户打包成一个规范，然后启用。你想想看，如果只审计对某个敏感表的 DELETE 操作，直接配一个规范就行，非常灵活。
2.2 日志采集工具
格式看完了，接下来是怎么把日志从不同的数据库服务器上采集到统一平台。这块我踩过的坑不少，最典型的就是采集工具掉线、日志重复、或者丢数据。
2.2.1 Filebeat
Filebeat 是 Elastic 生态里的轻量级采集器。配置简单，资源占用低，适合采集文本格式的审计日志。
它的核心配置就几个要素：输入路径、输出目标、多行合并规则。举个例子，采集 MySQL general log：
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/mysql/mysql.log
multiline:
pattern: '^\d{4}/\d{2}/\d{2}'
negate: true
match: after
output.elasticsearch:
hosts: ["192.168.1.100:9200"]
index: "mysql-audit-%{+yyyy.MM.dd}"注意上面那个 multiline 配置。为什么要有它？因为有些 SQL 语句跨行，如果不合并，一条语句会被切碎成好几条记录。我记得有一次排查慢查询，就是因为没配多行合并，查了半天总觉得少了几条 SQL，其实它们都被拆散到不同记录了。
2.2.2 Flume
Flume 是 Apache 的项目，比 Filebeat 重一些，但功能更强大。它可以处理复杂的流式数据，比如从 Oracle 的 SYS.AUD$ 表里用 JDBC 拉取日志。
Flume 的基本三件套：Source → Channel → Sink。一个典型的口语化架构：
# 定义 source
agent.sources = oracleAudit
agent.sources.oracleAudit.type = org.keedio.flume.source.SQLSource
agent.sources.oracleAudit.hibernate.connection.url = jdbc:oracle:thin:@//dbhost:1521/auditdb
agent.sources.oracleAudit.hibernate.connection.user = audit_user
agent.sources.oracleAudit.hibernate.connection.password = ******
agent.sources.oracleAudit.query = SELECT * FROM dba_audit_trail WHERE timestamp > ? AND timestamp < ?
agent.sources.oracleAudit.status.file.path = /var/lib/flume/status
agent.sources.oracleAudit.status.file.name = oracle.status
# 定义 channel
agent.channels = memChannel
agent.channels.memChannel.type = memory
agent.channels.memChannel.capacity = 10000
# 定义 sink（写入 HDFS）
agent.sinks = hdfsSink
agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = /audit/oracle/%Y%m%d
agent.sinks.hdfsSink.hdfs.fileType = DataStreamFlume 的优势在于：它能用 SQL 增量拉取数据，而且自带事务保证。不过配置相对复杂，调试起来有点费劲。我个人的建议是：如果团队熟悉 Hadoop 生态，Flume 是上选；如果是个小团队、只想快速看到日志，Filebeat 更友好。
2.3 采集架构与注意事项
不管用哪种工具，架构上通常建议采用 Agent → Collector → Storage 的分层设计。每层都可以做简单的过滤和缓冲，避免单点故障。
| 组件 | 职责 | 常见产品 |
|---|---|---|
| Agent（代理） | 部署在数据库服务器上，读取本地日志文件或查询审计表 | Filebeat、Flume Agent |
| Collector（收集层） | 汇聚多个 Agent 的数据，做初步清洗和路由 | Logstash、Flume Collector |
| Storage（存储层） | 持久化原始日志，供检索和分析 | Elasticsearch、HDFS、Kafka |
最后啰嗦一句：日志格式和采集方案定了之后，务必做一次端到端验证。从数据库操作到最终存储平台，查一条记录，确保字段没有被截断、时间戳时区正确、编码不乱。
嗯，格式和采集是审计的“地基”，地基不牢，后面排查异常全是空中楼阁。下一层我们就要玩真的了——怎么在这些日志里找出异常行为。