PostgreSQL触发器监控表数据落表时间是否有延迟

场景描述

• 假如PostgreSQL数据库tzqdb下有tzq schema和tzq用户，下面有表tzq_log_t，表上开启了RLS行级策略，表数据是从上游拉取过来的数据，现在发现有丢失。现在要监控tzq_log_t的数据落表时间，看是上游的表的数据落表有延迟还是数据拉取工具的问题。

一、创建大账号（etl_tzq）监控普通账号（tzq）的表（tzq_log_t）数据。

用超管账号(prostgres、root等，此处用prostgres)，在schema tzq下创建大账号（etl_tzq）监控普通账号（tzq）的表（tzq_log_t）数据。
让etl_tzq用户越过行级策略。
给etl_tzq用户授权scheme tzq的usage的权限。
给etl_tzq用户授权scheme tzq的create的权限。
给etl_tzq用户授权表tzq.tzq_log_t的trigger的权限。

CREATE USER etl_tzq WITH PASSWORD 'Tzq@123456';
alter user etl_mon_to_ins with bypassrls;
grant usage on schema tzq to etl_tzq;
grant create on schema tzq to etl_tzq;
grant trigger on tzq.tzq_log_t to etl_tzq;

二、创建监控数据表：tzq_monitor_data_t

用大账号(etl_tzq)，创建监控数据表（tzq_monitor_data_t）,监控业务表或者配置表（此处为上述的tzq_log_t表）的数据落表时间（creation_date）及更新时间（last_update_date）情况。
创建序列seq_tzq_monitor_data_t。
创建表tzq.tzq_monitor_data_t，主键id设置默认值为序列seq_tzq_monitor_data_t的下一个数值。

create sequence seq_tzq_monitor_data_t;
CREATE TABLE tzq.tzq_monitor_data_t (
  id int8 NOT NULL default nextval('seq_tzq_monitor_data_t'),
  table_name varchar(64),
  primary_id int8,
  log_number VARCHAR(100),
  source_creation_date TIMESTAMP(6),
  source_last_updated_date TIMESTAMP(6),
  operation varchar(100),
  creation_date TIMESTAMP(6) DEFAULT clock_timestamp(),
  PRIMARY KEY (id)
);
COMMENT ON TABLE  tzq.tzq_monitor_data_t IS '监控表数据表：监控表数据的落表时间是否有延迟，延迟多久';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.id IS '监控表数据表id：IT主键，系统自动生成';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.table_name IS '表名';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.primary_id IS '监控表的主键id';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.log_number IS '日志号';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.source_creation_date IS '原表创建时间';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.source_last_updated_date IS '原表更新时间';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.operation IS '操作：插入、更新、删除';
COMMENT ON COLUMN tzq.tzq_monitor_data_t.creation_date IS '实际落表时间';
ALTER TABLE tzq.tzq_monitor_data_t OWNER TO etl_tzq;

三、创建触发器函数：tzq_change_data_trigger()

用大账号(etl_tzq)，创建触发器函数（tzq_change_data_trigger），对tzq_log_t表数据的落表情况记录到数据监控日志表（tzq_monitor_data_t）表里面。

CREATE or replace FUNCTION tzq_change_data_trigger()
  RETURNS trigger
  LANGUAGE 'plpgsql'
AS $$
    BEGIN
      IF TG_OP = 'INSERT' THEN
        INSERT INTO ins.tpc_monitor_data_t (table_name,primary_id,log_number,source_creation_date,source_last_updated_date,operation) values ('tzq_log_t',new.instruction_id,new.instruction_number,new.creation_date,new.last_update_date,'insert');
        RETURN NEW;
      ELSIF TG_OP = 'UPDATE'  THEN
        INSERT INTO ins.tpc_monitor_data_t (table_name,primary_id,log_number,source_creation_date,source_last_updated_date,operation) values ('tzq_log_t',new.instruction_id,new.instruction_number,new.creation_date,new.last_update_date,'update');
        RETURN NEW;
      ELSIF TG_OP = 'DELETE'  THEN
        INSERT INTO ins.tpc_monitor_data_t (table_name,primary_id,log_number,source_creation_date,source_last_updated_date,operation) values ('tzq_log_t',old.instruction_id,old.instruction_number,old.creation_date,old.last_update_date,'delete');
        RETURN OLD; --返回值要与ELSIF平齐,因为先插入后最好才执行返回
      END IF;
END;
$$;

四、创建触发器：tri_data_tzq_log_t，对表（tzq.tzq_log_t）的增删改进行监控

用大账号(etl_tzq)，创建触发器（tri_data_tzq_log_t），对tzq.tzq_log_t表的增删改进行监控，执行函数tzq_change_data_trigger()

CREATE TRIGGER tri_data_tzq_log_t
  BEFORE INSERT OR UPDATE OR DELETE
  ON tzq.tzq_log_t
  FOR EACH ROW
EXECUTE PROCEDURE tzq_change_data_trigger();

五、给普通用户（tzq）授予监控日志表和序列的权限

• 为了防止普通用户（tzq）修改或删除由大账号（etl_tzq）创建的表（tzq.tzq_monitor_data_t）的数据。
• 在大账号（etl_tzq）下给用户（tzq）授权表（tzq.tzq_monitor_data_t）的select,insert权限。
• 在大账号（etl_tzq）下给用户（tzq）授权序列（tzq.seq_tzq_monitor_data_t）的select,update权限。

-- 授权
grant select,insert on tzq.tzq_monitor_data_t to tzq;
grant select,update on ins.seq_tzq_monitor_data_t to tzq;

六、查源端最后更新时间和实际落表时间的相差秒数

查源端（表tzq_log_t）的最后更新时间（last_update_date）和实际落表时间（tzq.tzq_monitor_data_t.creation_date）的相差秒数。如果绝对值过大就说明有问题。

-- 查相差秒数
select to_number(
           date_trunc('second'
                     ,t.creation_date::TIMESTAMP - 
                      t.source_last_updated_date::TIMESTAMP)::text
       ,'9999999999999') as s1
      ,t.* 
  from tzq.tzq_monitor_data_t t
 where primary_id < 100000
   and operation = 'insert'
   -- and log_number = 'LOG2023102200000006'
 order by creation_date desc;

七、根据差异结果排查相关问题及总结

• 这样，通过上面的一系列操作，我们就可以监控表（tzq_log_t）数据的落表时间跟实际落表时间的差异。
• 如果差异过大（比如超过2分钟），我们就可以确定是上游的java代码落表时间有延迟。
• 如果差异很小（5秒内），我们就可以确定是数据集成ETL工具漏数了。

至此，“PostgreSQL触发器监控表数据落表时间是否有延迟”的范例就演示完了，大家有不明白的地方可以给我留言，谢谢！