使用事件触发器实现DDL回收站、防火墙、增量订阅同步

DDL回收站

1. 执行如下命令创建表、函数和相关触发器。

   /* external/rds_ddl_pulication/rds_ddl_pulication--1.0.sql */
   
   --create schema
   CREATE SCHEMA IF NOT EXISTS dts_audit;
   --create table for ddl record
   CREATE TABLE IF NOT EXISTS dts_audit.dts_tb_ddl_command ( event text,
   tag text, classid oid, objid oid, objsubid int,
   object_type text, schema_name text, object_identity text, is_extension bool, query text,
   username text default current_user, datname text default current_database(), client_addr inet default inet_client_addr(), crt_time timestamp default now()
   );
   -- create function for event triggers
   create or replace function dts_audit.dts_func_ddl_command() returns event_trigger as $$
   declare v1 text;
   is_superuser bool = false;
   r record; 
   begin
       -- we don't record  ddl command from superusers 
       select u.rolsuper into is_superuser from pg_catalog.pg_roles u where u.rolname = SESSION_USER;
       if is_superuser then
           return;
       end if;
       select query into v1 from pg_stat_activity where pid=pg_backend_pid();
       -- RAISE NOTICE 'ddl event:%, command:%', tg_event, tg_tag;
       -- NB:since ddl_command_end cannot collect the details of the drop statement, we use sql_drop
       if TG_EVENT='ddl_command_end' then
           SELECT * into r FROM pg_event_trigger_ddl_commands(); 
           if r.classid > 0 then
               insert into dts_audit.dts_tb_ddl_command(event, tag, classid, objid, objsubid, object_type, schema_name, object_identity, is_extension, query)
               values(TG_EVENT, TG_TAG, r.classid, r.objid, r.objsubid, r.object_type, r.schema_name, r.object_identity, r.in_extension, v1);
            end if; 
       end if;
       if TG_EVENT='sql_drop' then
           -- To avoid repeated collection, we filtered 'ALTER TABLE' and 'ALTER FOREIGN TABLE'
           if TG_TAG != 'ALTER TABLE' and TG_TAG != 'ALTER FOREIGN TABLE' then
               SELECT * into r FROM pg_event_trigger_dropped_objects();
               insert into dts_audit.dts_tb_ddl_command(event, tag, classid, objid, objsubid, object_type, schema_name, object_identity, is_extension, query)
               values(TG_EVENT, TG_TAG, r.classid, r.objid, r.objsubid, r.object_type, r.schema_name, r.object_identity, 'f', v1);
           end if; 
       end if;
   end;
   $$ language plpgsql strict;
   -- ddl_command_end event trigger
   CREATE EVENT TRIGGER pg_get_ddl_command on ddl_command_end EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();
   -- pg_get_ddl_drop event trigger
   CREATE EVENT TRIGGER pg_get_ddl_drop on sql_drop EXECUTE PROCEDURE dts_audit.dts_func_ddl_command();
   -- grant privileges to all user
   GRANT USAGE ON SCHEMA dts_audit TO PUBLIC;
   GRANT SELECT, INSERT ON TABLE dts_audit.dts_tb_ddl_command TO PUBLIC;

   说明 执行以上命令后，您的DDL语句就会记录在表dts_audit.dts_tb_ddl_command中。
2. 执行一个DDL语句，测试能否记录变更。

   create table tb_test(id int);
   select * from dts_audit.dts_tb_ddl_command;

   

DDL防火墙

您可以根据业务需求创建事件触发器，使用ddl_command_start事件类型，可以阻止相应的DDL语句执行。

1. 创建触发器函数。

   CREATE OR REPLACE FUNCTION abort1()
     RETURNS event_trigger
    LANGUAGE plpgsql
     AS $$
   BEGIN
     if current_user = 'test1' then
       RAISE EXCEPTION 'event:%, command:%', tg_event, tg_tag;
     end if;
    END;
   $$;

2. 创建触发器阻止创建和删除表的语句。

   create event trigger b on ddl_command_start when TAG IN ('CREATE TABLE', 'DROP TABLE') execute procedure abort1();

3. 使用对应的用户test1登录实例后创建表，测试能否创建。
   

   说明 DDL语句被成功阻止。

DDL增量订阅同步

在发布端我们将已经执行了的DDL语句存储在dts_audit.dts_tb_ddl_command中。订阅端可以读取记录进行同步。

1. 在发布端执行发布命令。

   CREATE PUBLICATION my_ddl_publication FOR TABLE ONLY dts_audit.dts_tb_ddl_command;

2. 在订阅端创建相同的表。

   CREATE SCHEMA IF NOT EXISTS dts_audit;
   CREATE TABLE IF NOT EXISTS dts_audit.dts_tb_ddl_command ( event text,
   tag text, classid oid, objid oid, objsubid int,
   object_type text, schema_name text, object_identity text, is_extension bool, query text,
   username text, datname text, client_addr inet , crt_time timestamp );

3. 在订阅端创建订阅。

   CREATE SUBSCRIPTION my_ddl_subscriptin CONNECTION 'host=*** port=*** user=*** password=***  dbname=**' PUBLICATION my_ddl_publication;

   说明 需要确保实例参数wal_level的值为logical，您可以在控制台的参数设置页面进行修改，该参数修改后需要重启实例才能生效。详情请参见逻辑订阅。

   示例

   CREATE SUBSCRIPTION my_ddl_subscriptin CONNECTION 'host=pgm-bpxxxxx.pg.rds.aliyuncs.com port=1433 user=test1 password=xxxxx  dbname=testdb1' PUBLICATION my_ddl_publication;

4. 在订阅端针对dts_audit.dts_tb_ddl_command表创建相应的触发器，实现DDL增量同步。