如何在RDS各Schema的增删改操作时触发AWS Lambda或SQS？

针对你的需求（记录所有增删改操作到审计表用于分析），以下是几种可行的实现方式，同时解决你之前plpgsql触发器失败的问题：

方案一：PostgreSQL触发器 + AWS扩展调用Lambda/SQS #

这是最直接的方式，通过数据库触发器捕获变更，再调用AWS服务。

前置准备 #

1. 给RDS PostgreSQL实例关联IAM角色，该角色需具备lambda:InvokeFunction（调用Lambda）或sqs:SendMessage（发送SQS）权限，且信任策略允许rds.amazonaws.com。
2. 在RDS参数组中设置shared_preload_libraries包含aws_lambda，重启实例生效。

具体步骤 #

1. 创建AWS Lambda扩展（用于直接在PostgreSQL中调用Lambda）：

CREATE EXTENSION IF NOT EXISTS aws_lambda;

2. 编写通用审计触发器函数，捕获Schema、表名及变更数据：

CREATE OR REPLACE FUNCTION audit_trigger_func()
RETURNS TRIGGER AS $$
DECLARE
    event_data JSONB;
BEGIN
    -- 按操作类型封装变更数据
    CASE TG_OP
        WHEN 'INSERT' THEN
            event_data = jsonb_build_object(
                'operation', 'INSERT',
                'schema', TG_TABLE_SCHEMA,
                'table', TG_TABLE_NAME,
                'timestamp', now(),
                'data', row_to_json(NEW)
            );
        WHEN 'UPDATE' THEN
            event_data = jsonb_build_object(
                'operation', 'UPDATE',
                'schema', TG_TABLE_SCHEMA,
                'table', TG_TABLE_NAME,
                'timestamp', now(),
                'old_data', row_to_json(OLD),
                'new_data', row_to_json(NEW)
            );
        WHEN 'DELETE' THEN
            event_data = jsonb_build_object(
                'operation', 'DELETE',
                'schema', TG_TABLE_SCHEMA,
                'table', TG_TABLE_NAME,
                'timestamp', now(),
                'data', row_to_json(OLD)
            );
    END CASE;

    -- 调用指定Lambda函数，替换为你的Lambda ARN
    PERFORM aws_lambda.invoke(
        'arn:aws:lambda:us-east-1:123456789012:function:your-audit-processor',
        event_data
    );

    -- 若要发送到SQS，可改用pg_http扩展调用SQS API（需先安装pg_http）
    -- PERFORM http_post('https://sqs.us-east-1.amazonaws.com/123456789012/your-audit-queue', event_data::text);

    RETURN NULL; -- AFTER触发器返回值不影响原操作
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

3. 批量给指定Schema的所有表添加触发器：

CREATE OR REPLACE FUNCTION add_audit_triggers(target_schema TEXT)
RETURNS VOID AS $$
DECLARE
    table_rec RECORD;
BEGIN
    FOR table_rec IN SELECT tablename FROM pg_tables WHERE schemaname = target_schema LOOP
        EXECUTE format(
            'CREATE TRIGGER audit_trigger AFTER INSERT OR UPDATE OR DELETE ON %I.%I FOR EACH ROW EXECUTE FUNCTION audit_trigger_func()',
            target_schema, table_rec.tablename
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

-- 示例：给public schema下所有表添加审计触发器
SELECT add_audit_triggers('public');

常见失败原因排查 #

• 未启用aws_lambda或pg_http扩展，导致无法调用外部服务。
• RDS实例未关联具备对应权限的IAM角色，调用AWS服务时权限不足。
• 触发器函数中OLD/NEW变量使用错误（DELETE操作仅能访问OLD，INSERT仅能访问NEW）。

方案二：AWS DMS CDC（变更数据捕获） #

若不想修改数据库代码，或高并发场景下担心触发器性能影响，可使用DMS的CDC功能捕获RDS变更。

步骤 #

1. 启用RDS PostgreSQL的逻辑复制：在参数组中设置rds.logical_replication=1，重启实例。
2. 创建DMS复制实例，源端点配置为你的RDS实例，目标端点选择Lambda或SQS。
3. 创建DMS任务，配置为CDC模式，指定要捕获的Schema/表，将变更事件转发到Lambda/SQS。
4. 在Lambda中处理事件数据，写入你的审计表。

优势 #

• 无侵入式，无需修改数据库代码。
• 适合大规模、高并发的变更捕获，性能影响极小。

方案三：审计表直接写入（无需Lambda/SQS） #

如果仅需记录到RDS内部的审计表，可直接在触发器中写入，跳过中间服务：

-- 先创建审计表
CREATE TABLE IF NOT EXISTS audit_logs (
    id SERIAL PRIMARY KEY,
    operation VARCHAR(10) NOT NULL,
    schema_name VARCHAR(64) NOT NULL,
    table_name VARCHAR(64) NOT NULL,
    event_timestamp TIMESTAMPTZ DEFAULT now(),
    old_data JSONB,
    new_data JSONB
);

-- 修改触发器函数，直接写入审计表
CREATE OR REPLACE FUNCTION audit_trigger_func()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO audit_logs (operation, schema_name, table_name, old_data, new_data)
    VALUES (
        TG_OP,
        TG_TABLE_SCHEMA,
        TG_TABLE_NAME,
        CASE WHEN TG_OP = 'UPDATE' OR TG_OP = 'DELETE' THEN row_to_json(OLD)::JSONB ELSE NULL END,
        CASE WHEN TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN row_to_json(NEW)::JSONB ELSE NULL END
    );
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

内容的提问来源于stack exchange，提问作者Atul Kumar