database

PostgreSQL 事件触发器与审计日志实现

深入讲解 PostgreSQL 事件触发器（Event Trigger）的底层机制与生产级审计日志实现方案。涵盖 DDL 事件触发器的完整生命周期（ddl_command_start、ddl_command_end、sql_drop、table_rewrite）、自定义审计表记录结构化变更日志、登录审计与 row_security 强制审计、表结构变更追踪策略、pgAudit 扩展的企业级审计能力，以及面向合规要求的完整审计体系设计。

Leeting Yan 2026-09-30 7 分钟阅读 3143 字

数据库审计不只是 DBA “我想知道发生了什么”，更是等保、ISO27001、PCI DSS 等合规框架的硬性要求。PostgreSQL 提供了两套互补的审计机制：**事件触发器（Event Trigger）**处理 DDL（Data Definition Language）变更审计，pgAudit 扩展处理 DML 与 SELECT 的行级审计日志。两者结合，可以实现从登录、表结构变更到数据读写操作的完整追踪链路。

核心认知：审计不是 “事后查日志”，而是 “事先定规则”。没有结构化审计表的野生 pg_log，在审计面前等于没有证据。

一、事件触发器基础

1.1 事件触发器 vs 行级触发器

维度	事件触发器（Event Trigger）	行级触发器（Row Trigger）
触发时机	DDL 执行前后	DML（INSERT/UPDATE/DELETE）
粒度	命令级	行级
可访问数据	tg_event / tg_tag / pg_event_trigger_ddl_commands()	NEW / OLD
用途	审计 DDL、强制命名规范、自动维护	审计数据变更、级联更新、软删除

事件触发器只能捕获 DDL，对 DML 无感知。要审计 DML，需要 pgAudit 或行级触发器。

1.2 四个核心触发事件

事件	触发时机	典型用途
`ddl_command_start`	DDL 命令执行前	权限拦截、阻止危险操作
`ddl_command_end`	DDL 命令执行后	记录变更日志、自动维护
`sql_drop`	DROP 操作执行后	记录被删除的对象
`table_rewrite`	表重写触发时	VACUUM FULL / CLUSTER 监控

-- 创建事件触发器的基本结构
CREATE OR REPLACE FUNCTION audit_ddl()
RETURNS event_trigger AS $$
BEGIN
    -- 通过 tg_event 获取事件信息
    RAISE NOTICE 'DDL event: %, tag: %', tg_event, tg_tag;
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER audit_ddl_trigger
ON ddl_command_end
EXECUTE FUNCTION audit_ddl();

二、DDL 审计表的完整实现

2.1 审计表设计

CREATE TABLE ddl_audit_log (
    id              bigserial PRIMARY KEY,
    event_time      timestamptz DEFAULT NOW(),
    username        text        DEFAULT CURRENT_USER,
    database_name   text        DEFAULT current_database(),
    client_addr     inet        DEFAULT inet_client_addr(),
    application     text        DEFAULT current_setting('application_name', true),
    event_type      text        NOT NULL,  -- ddL_command_end / sql_drop
    tag             text        NOT NULL,  -- CREATE TABLE / ALTER INDEX etc.
    command_text    text,               -- 完整 DDL 语句
    object_type     text,               -- table / index / function 等
    schema_name     text,
    object_name     text,
    object_identity text                -- 完整限定名
);

-- 分区或按时间索引，避免审计表膨胀
CREATE INDEX idx_ddl_audit_time ON ddl_audit_log(event_time DESC);
CREATE INDEX idx_ddl_audit_user  ON ddl_audit_log(username);

2.2 事件触发器函数（ddl_command_end）

CREATE OR REPLACE FUNCTION log_ddl_commands()
RETURNS event_trigger AS $$
DECLARE
    cmd record;
BEGIN
    FOR cmd IN
        SELECT * FROM pg_event_trigger_ddl_commands()
    LOOP
        INSERT INTO ddl_audit_log (
            event_type, tag, command_text,
            object_type, schema_name, object_name, object_identity
        ) VALUES (
            tg_event,
            tg_tag,
            current_query(),
            cmd.object_type,
            cmd.schema_name,
            cmd.object_name,
            cmd.object_identity
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER log_ddl_trigger
ON ddl_command_end
EXECUTE FUNCTION log_ddl_commands();

pg_event_trigger_ddl_commands() 返回以下关键字段：

字段	含义
`classid`	系统目录 OID
`objid`	对象 OID
`object_type`	`table`, `index`, `function`, `trigger` 等
`schema_name`	所属 schema
`object_name`	对象名
`object_identity`	完整限定名
`command_tag`	命令标签

2.3 对象删除审计（sql_drop）

CREATE OR REPLACE FUNCTION log_drop_commands()
RETURNS event_trigger AS $$
DECLARE
    obj record;
BEGIN
    FOR obj IN
        SELECT * FROM pg_event_trigger_dropped_objects()
    LOOP
        INSERT INTO ddl_audit_log (
            event_type, tag, command_text,
            object_type, schema_name, object_name, object_identity
        ) VALUES (
            tg_event,
            tg_tag,
            current_query(),
            obj.object_type,
            obj.schema_name,
            obj.object_name,
            obj.object_identity
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER log_drop_trigger
ON sql_drop
EXECUTE FUNCTION log_drop_commands();

关键差异：sql_drop 触发器中必须使用 pg_event_trigger_dropped_objects()，而非 pg_event_trigger_ddl_commands()。

2.4 表重写监控（table_rewrite）

CREATE OR REPLACE FUNCTION log_table_rewrite()
RETURNS event_trigger AS $$
BEGIN
    INSERT INTO ddl_audit_log (event_type, tag, command_text)
    VALUES (tg_event, tg_tag, current_query());
    RAISE NOTICE 'Table rewrite triggered by: %', current_query();
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER log_rewrite_trigger
ON table_rewrite
EXECUTE FUNCTION log_table_rewrite();

三、登录审计

3.1 数据库连接日志

-- postgresql.conf 配置
log_connections = on
log_disconnections = on
log_line_prefix = '%t [%p]: [%l-1] user=%u,db=%d,app=%a,client=%h '

日志输出：

2026-10-01 14:30:01 UTC [12345]: [1-1] user=alice,db=myapp,app=psql,client=172.16.0.15 LOG:  connection authorized: user=alice database=myapp

3.2 登录失败监控

-- 查看认证失败日志（需 log_min_messages >= LOG）
-- 在日志中搜索: FATAL:  password authentication failed
-- 或用 pgBadger 报告分析

-- 通过 log_statement 捕获所有语句（谨慎开启）
ALTER SYSTEM SET log_statement = 'all';  -- ddl | mod | all
SELECT pg_reload_conf();

3.3 连接频次审计

-- 按用户统计连接数（活跃会话）
SELECT usename, count(*) as active_conn,
       count(*) FILTER (WHERE state = 'idle in transaction') as idle_tx
FROM pg_stat_activity
GROUP BY usename
ORDER BY active_conn DESC;

四、pgAudit 扩展：企业级审计

4.1 安装与配置

-- 在 shared_preload_libraries 中启用 pgaudit
SHOW shared_preload_libraries;
-- 如果不在列表中，修改 postgresql.conf 后重启
ALTER SYSTEM SET shared_preload_libraries = 'pgaudit,pg_stat_statements';
-- 重启 PostgreSQL 后
CREATE EXTENSION IF NOT EXISTS pgaudit;

4.2 pgaudit 核心参数

-- 审计全部 READ/WRITE 操作
ALTER SYSTEM SET pgaudit.log = 'write,read';
ALTER SYSTEM SET pgaudit.log_catalog = off;        -- 不审计系统目录访问
ALTER SYSTEM SET pgaudit.log_parameter = on;       -- 记录绑定参数
ALTER SYSTEM SET pgaudit.log_statement_once = off; -- 每条语句都记录
SELECT pg_reload_conf();

参数	可选值	说明
`pgaudit.log`	`none`, `all`, `ddl`, `mod`, `read`, `write`, `role`	审计类别
`pgaudit.log_relation`	on/off	按关系记录（适合 Object Audit）
`pgaudit.log_parameter`	on/off	记录绑定参数
`pgaudit.log_catalog`	on/off	是否审计 pg_catalog
`pgaudit.role`	角色名	以角色级对象审计

4.3 审计日志输出示例

2026-10-01 14:35:02 UTC [12355]: [2-1] AUDIT: SESSION,2,1,WRITE,INSERT,TABLE,public.orders,INSERT INTO orders (user_id, total) VALUES ($1, $2),<not logged>

4.4 对象级审计（更细粒度）

-- 只对特定表开启审计（而非全库）
CREATE ROLE audit_role;

-- 为角色设置审计策略
ALTER ROLE audit_role SET pgaudit.log = 'write';

-- 将需要审计的表分配给该角色
GRANT audit_role TO app_user;

-- 更精确：使用 pgaudit.log_relation
SET pgaudit.log_relation = on;
-- 审计当前会话中所有涉及特定表的操作

五、row_security 与强制审计

5.1 Row-Level Security 审计

-- 在关键表上启用 RLS，并强制审计策略
ALTER TABLE sensitive_data ENABLE ROW LEVEL SECURITY;
ALTER TABLE sensitive_data FORCE ROW LEVEL SECURITY;  -- 连 owner 跳过

CREATE POLICY audit_all_access
ON sensitive_data
FOR ALL
TO PUBLIC
USING (true)
WITH CHECK (true);

5.2 结合触发器实现行级变更追踪

-- DML 审计（推荐 pgAudit；如需自定义逻辑，可用行级触发器）
CREATE TABLE data_audit_log (
    audit_id      bigserial PRIMARY KEY,
    table_name    text,
    operation     text,  -- INSERT / UPDATE / DELETE
    changed_at    timestamptz DEFAULT NOW(),
    changed_by    text DEFAULT CURRENT_USER,
    old_data      jsonb,
    new_data      jsonb
);

CREATE OR REPLACE FUNCTION audit_data_changes()
RETURNS trigger AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO data_audit_log (table_name, operation, new_data)
        VALUES (TG_TABLE_NAME, 'INSERT', to_jsonb(NEW));
        RETURN NEW;
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO data_audit_log (table_name, operation, old_data, new_data)
        VALUES (TG_TABLE_NAME, 'UPDATE', to_jsonb(OLD), to_jsonb(NEW));
        RETURN NEW;
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO data_audit_log (table_name, operation, old_data)
        VALUES (TG_TABLE_NAME, 'DELETE', to_jsonb(OLD));
        RETURN OLD;
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER audit_sensitive_data
AFTER INSERT OR UPDATE OR DELETE ON sensitive_data
FOR EACH ROW EXECUTE FUNCTION audit_data_changes();

六、完整审计体系设计

6.1 三层审计架构

层级	机制	覆盖范围
连接层	log_connections / log_disconnections / pgaudit	登录/登出/认证失败
DDL 层	Event Trigger	CREATE / ALTER / DROP / TRUNCATE
DML 层	pgAudit / 行级触发器	INSERT / UPDATE / DELETE / SELECT

6.2 日志管理与保留

-- 日志轮转配置（postgresql.conf）
log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'
log_rotation_age = 1d
log_rotation_size = 100MB
log_truncate_on_rotation = off

生产环境建议日志集中采集至 ELK / Loki / CloudWatch，设置保留周期至少 90 天（金融等合规行业 180~365 天）。

6.3 等保/合规审计检查清单

□ 启用 log_connections / log_disconnections
□ 启用 log_statement = 'ddl' 或 'mod'
□ 部署 pgAudit 扩展并配置审计范围
□ Event Trigger 记录所有 DDL 变更至结构化审计表
□ 审计表分区管理，防止无限膨胀
□ 关键表行级变更追踪（触发器或 pgAudit）
□ 日志集中采集，定期备份，不可篡改
□ 定期审计报告（Top 风险操作者、异常 DDL 时段）
□ 审计系统自身访问控制（防止篡改审计数据）

常见问题（FAQ）

Event Trigger 会影响 DDL 性能吗？

普通审计触发器开销很低（一次 DDL 通常只写入 1 行审计记录），但如果在触发器中做复杂查询或远程调用，可能显著拖慢 DDL。审计表自身需要合理索引与分区。

pgAudit 和手动触发器哪个好？

场景	推荐方案
简单 DDL 审计	Event Trigger
全量 DML 审计	pgAudit（性能更好、更标准化）
特定表特定字段变更	行级触发器（灵活自定义）
合规报表	pgAudit + 集中日志采集

审计表膨胀怎么办？

-- 按月分区
CREATE TABLE ddl_audit_log_2026_10 PARTITION OF ddl_audit_log
FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

-- 定期归档旧分区到冷存储
-- 或设置 TTL 策略

能阻止危险 DDL 吗？

-- 用 ddl_command_start 在命令执行前拦截
CREATE OR REPLACE FUNCTION block_dangerous_ddl()
RETURNS event_trigger AS $$
BEGIN
    IF tg_tag IN ('DROP TABLE', 'TRUNCATE TABLE') THEN
        RAISE EXCEPTION 'DDL operation % is blocked by policy', tg_tag;
    END IF;
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER block_dangerous_ddl_trigger
ON ddl_command_start
EXECUTE FUNCTION block_dangerous_ddl();

完整示例（一键复制）

-- ========== 1. DDL 审计表与触发器 ==========

CREATE TABLE IF NOT EXISTS ddl_audit_log (
    id              bigserial PRIMARY KEY,
    event_time      timestamptz DEFAULT NOW(),
    username        text        DEFAULT CURRENT_USER,
    database_name   text        DEFAULT current_database(),
    client_addr     inet        DEFAULT inet_client_addr(),
    application     text        DEFAULT current_setting('application_name', true),
    event_type      text        NOT NULL,
    tag             text        NOT NULL,
    command_text    text,
    object_type     text,
    schema_name     text,
    object_name     text,
    object_identity text
);

CREATE INDEX IF NOT EXISTS idx_ddl_audit_time ON ddl_audit_log(event_time DESC);
CREATE INDEX IF NOT EXISTS idx_ddl_audit_user  ON ddl_audit_log(username);

-- ddl_command_end 审计
CREATE OR REPLACE FUNCTION log_ddl_commands()
RETURNS event_trigger AS $$
DECLARE
    cmd record;
BEGIN
    FOR cmd IN SELECT * FROM pg_event_trigger_ddl_commands()
    LOOP
        INSERT INTO ddl_audit_log (
            event_type, tag, command_text,
            object_type, schema_name, object_name, object_identity
        ) VALUES (
            tg_event, tg_tag, current_query(),
            cmd.object_type, cmd.schema_name,
            cmd.object_name, cmd.object_identity
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS log_ddl_trigger;
CREATE EVENT TRIGGER log_ddl_trigger
ON ddl_command_end
EXECUTE FUNCTION log_ddl_commands();

-- sql_drop 审计
CREATE OR REPLACE FUNCTION log_drop_commands()
RETURNS event_trigger AS $$
DECLARE
    obj record;
BEGIN
    FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects()
    LOOP
        INSERT INTO ddl_audit_log (
            event_type, tag, command_text,
            object_type, schema_name, object_name, object_identity
        ) VALUES (
            tg_event, tg_tag, current_query(),
            obj.object_type, obj.schema_name,
            obj.object_name, obj.object_identity
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS log_drop_trigger;
CREATE EVENT TRIGGER log_drop_trigger
ON sql_drop
EXECUTE FUNCTION log_drop_commands();

-- table_rewrite 审计
CREATE OR REPLACE FUNCTION log_table_rewrite()
RETURNS event_trigger AS $$
BEGIN
    INSERT INTO ddl_audit_log (event_type, tag, command_text)
    VALUES (tg_event, tg_tag, current_query());
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS log_rewrite_trigger;
CREATE EVENT TRIGGER log_rewrite_trigger
ON table_rewrite
EXECUTE FUNCTION log_table_rewrite();

-- ========== 2. 危险 DDL 拦截 ==========
CREATE OR REPLACE FUNCTION block_dangerous_ddl()
RETURNS event_trigger AS $$
BEGIN
    IF tg_tag IN ('DROP TABLE', 'TRUNCATE TABLE') THEN
        RAISE EXCEPTION 'DDL operation % is blocked by policy', tg_tag;
    END IF;
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS block_dangerous_ddl_trigger;
CREATE EVENT TRIGGER block_dangerous_ddl_trigger
ON ddl_command_start
EXECUTE FUNCTION block_dangerous_ddl();

-- ========== 3. DML 行级审计（备用方案） ==========
CREATE TABLE IF NOT EXISTS data_audit_log (
    audit_id      bigserial PRIMARY KEY,
    table_name    text,
    operation     text,
    changed_at    timestamptz DEFAULT NOW(),
    changed_by    text DEFAULT CURRENT_USER,
    old_data      jsonb,
    new_data      jsonb
);

CREATE OR REPLACE FUNCTION audit_data_changes()
RETURNS trigger AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO data_audit_log (table_name, operation, new_data)
        VALUES (TG_TABLE_NAME, 'INSERT', to_jsonb(NEW));
        RETURN NEW;
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO data_audit_log (table_name, operation, old_data, new_data)
        VALUES (TG_TABLE_NAME, 'UPDATE', to_jsonb(OLD), to_jsonb(NEW));
        RETURN NEW;
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO data_audit_log (table_name, operation, old_data)
        VALUES (TG_TABLE_NAME, 'DELETE', to_jsonb(OLD));
        RETURN OLD;
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- ========== 4. pgAudit 配置命令 ==========
-- 需重启 PostgreSQL（shared_preload_libraries 包含 pgaudit）
-- 然后执行：
-- CREATE EXTENSION IF NOT EXISTS pgaudit;
-- ALTER SYSTEM SET pgaudit.log = 'write,read';
-- ALTER SYSTEM SET pgaudit.log_catalog = off;
-- ALTER SYSTEM SET pgaudit.log_parameter = on;
-- SELECT pg_reload_conf();

-- ========== 5. 审计日志查询 ==========
-- 按操作类型统计
SELECT tag, count(*) as cnt,
       max(event_time) as last_time
FROM ddl_audit_log
WHERE event_time > NOW() - INTERVAL '7 days'
GROUP BY tag
ORDER BY cnt DESC;

-- 按用户统计
SELECT username, count(*) as cnt
FROM ddl_audit_log
WHERE event_time > NOW() - INTERVAL '7 days'
GROUP BY username
ORDER BY cnt DESC;

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页