维护与管理

运行状态查看

参数	说明
`domino.decoder_info()`	当前数据库解码器信息。
`domino.decoder_info_all()`	全部数据库解码器信息。
`domino.decoder_workers()`	查询 worker 的 pid，连接的数据库等。
`domino.decoder_metric()`	查看 decoder 的性能计数器。
`domino.job_info()`	查看流的调度性能计数器，以及报错信息。
`domino.strategy_info()`	查看流的调度策略和状态。
`domino.progress_info()`	用于查看流进度，以及 XLog 保留情况。
`domino.ticker_info()`	查看 ticker 是否在运行（或者暂停）。
`domino.ticker_metric()`	用于查看 ticker 是被什么条件触发的。
`domino.read_tuples()`	用于查看 TLog 内容，debug 用。

使用示例

domino.decoder_workers() ：查询 worker 的 pid，连接的数据库等。

=# select * from domino.decoder_workers();
segid | dbid  | dbname |  pid
-------+-------+--------+-------
   0 | 18745 | haoyu  | 20351
   1 | 18745 | haoyu  | 20353
   2 | 18745 | haoyu  | 20352
  -1 | 18745 | haoyu  | 20354
(4 rows)

domino.decoder_metric()：查看decoder的性能计数器。

segid | dbid  | dbname | bytes_in | xacts | spills | tuples_new | tuples_old | snapshots | latency_ms | loops | total_duration_ms | last_duration_ms
-------+-------+--------+----------+-------+--------+------------+------------+-----------+------------+-------+-------------------+------------------
   0 | 18745 | haoyu  |    36048 |     0 |      0 |          0 |          0 |       119 |          0 | 25851 |             26616 |                1
   1 | 18745 | haoyu  |    35872 |     0 |      0 |          0 |          0 |       119 |          0 | 26755 |             27517 |                1
   2 | 18745 | haoyu  |    36048 |     0 |      0 |          0 |          0 |       119 |          0 | 25814 |             26576 |                1
  -1 | 18745 | haoyu  |    36784 |     0 |      0 |          0 |          0 |       119 |          0 | 22781 |             23444 |                1
(4 rows)

domino.job_info()：查看流的调度性能计数器，以及报错信息。


db   | db_name | stream | stream_name | loops | tuples | total_slip_ms | last_slip_ms | last_lag_ms | total_duration_ms | last_duration_ms |           last_run            |          held_until           | has_error | error_message | error_detail | error_hint | error_context
-------+---------+--------+-------------+-------+--------+---------------+--------------+-------------+-------------------+------------------+-------------------------------+-------------------------------+-----------+---------------+--------------+------------+---------------
18745 | haoyu   |  18792 | s1          |     4 |      0 |             6 |            2 |        1709 |                57 |                7 | 2025-05-23 14:14:26.249188+08 | 2025-05-23 14:14:27.242299+08 | f         |               |              |            |
(1 row)

domino.strategy_info()：查看流的调度策略和状态

id |  db   | db_name | stream | stream_name |  strategy  |                 summary
----+-------+---------+--------+-------------+------------+------------------------------------------
0 | 18745 | haoyu   |  18792 | s1          | roundrobin | held until 2025-05-23 14:14:56.279678+08
(1 row)

domino.progress_info()：用于查看流进度，以及 XLog 保留情况。

segid |  db   | db_name | stream | stream_name | restart_lsn | flush_lsn |   tsn   | xmin |  end_lsn  | lag_size | pending_xid | pending_restart | pending_flush | pending_tsn | pending_xmin | wal_status | safe_wal_size
-------+-------+---------+--------+-------------+-------------+-----------+---------+------+-----------+----------+-------------+-----------------+---------------+-------------+--------------+------------+---------------
   0 | 18745 | haoyu   |  18792 | s1          | 0/C246788   | 0/C249420 | 0/1A160 |  645 | 0/C249CE0 |     2240 |         647 | 0/C246788       | 0/C249610     | 0/1A240     |          646 | reserved   |
   1 | 18745 | haoyu   |  18792 | s1          | 0/C2466F0   | 0/C249388 | 0/1A160 |  645 | 0/C249C48 |     2240 |         647 | 0/C2466F0       | 0/C249578     | 0/1A240     |          646 | reserved   |
   2 | 18745 | haoyu   |  18792 | s1          | 0/C246788   | 0/C249420 | 0/1A160 |  645 | 0/C249CE0 |     2240 |         647 | 0/C246788       | 0/C249610     | 0/1A240     |          646 | reserved   |
(3 rows)

domino.ticker_info()：查看 ticker 是否在运行（或者暂停）。

-- 返回值为 (invalid 未启动、running 运行中、paused 暂停)。
ticker_info
-------------
running
(1 row)

domino.ticker_metric()：用于查看 ticker 是被什么条件触发的。

=# SELECT * FROM domino.ticker_metric();
segid | ticks_by_time | ticks_by_size | ticks_by_force
-------+---------------+---------------+----------------
   0 |         18894 |             2 |              1
   1 |         18894 |             2 |              1
   2 |         18894 |             2 |              1
  -1 |             0 |             0 |              0
(4 rows)

信息视图

Domino 插件提供的两个视图：domino.decoder_info 和 domino.stream_info，用于监控逻辑解码器和数据流的状态。

decoder_info

decoder_info 显示每个 segment 上的逻辑解码器运行状态和指标信息。

字段说明

字段名	类型	含义
segid	integer	segment 编号
dbid	oid	数据库 OID
dbname	text	数据库名称
restart_lsn	pg_lsn	依赖的xlog起始位置(按此位置保留xlog)
confirmed_flush	pg_lsn	依赖的xlog flush位置
start_tsn	pg_lsn	生产的最后一条tlog的起点tsn
end_tsn	pg_lsn	生产的最后一条tlog的尾部tsn
prog_xmin	xid	依赖的catalog/维表的xmin(按此xid保留表的数据)
pid	integer	解码器进程 ID，为 0 表示未运行
bytes_in	bigint	消费xlog总数据量
xacts	bigint	消费事务总量
spills	bigint	溢出到磁盘的事务数
tuples_new	bigint	插入的行数
tuples_old	bigint	更新/删除的旧行数
snapshots	bigint	生成的快照数
latency_ms	bigint	上一轮解码延迟
loops	bigint	解码器执行轮数
total_duration_ms	bigint	累计运行耗时
last_duration_ms	bigint	上一轮运行耗时

查询示例

-- 查看解码延迟高的 segment
SELECT segid, dbid, dbname, dbname, latency_ms
FROM domino.decoder_info
WHERE latency_ms > 60 * 1000;

stream_info

stream_info 展示每个逻辑流（stream）的整体进度、状态和调度信息。

字段说明

字段名	类型	含义
segid	integer	segment 编号
db	oid	数据库 OID
db_name	text	数据库名称
stream	oid	流的 OID
stream_name	text	流名称
restart_lsn	pg_lsn	依赖的 xlog 起始位置(按此位置保留 xlog )
flush_lsn	pg_lsn	依赖的的 xlog flush 位置
start_tsn	pg_lsn	消费的最后一条 tlog 的起点 tsn
end_tsn	pg_lsn	消费的最后一条 tlog 的起点 tsn
xmin	xid	依赖的 catalog /维表的 xmin(按此 xid 保留表的数据)
end_lsn	pg_lsn	此 segment 上最新的 lsn
lag_size	bigint	flush_lsn 距离 end_lsn 的大小
pending_xid	xid	尚未提交的的事务 ID
pending_restart	pg_lsn	尚未提交的依赖的 xlog 起始位置(按此位置保留 xlog)
pending_flush	pg_lsn	尚未提交的依赖的 xlog flush 位置
pending_tsn	pg_lsn	尚未提交的依赖的消费的最后一条 tlog 的起点 tsn
pending_xmin	xid	尚未提交的依赖的 catalog /维表的 xmin(按此 xid 保留表的数据)
wal_status	text	WAL 状态描述,有如下几个值： `reserved`：未超出 max_wal_size 限制,不会被清理; `extended`：超出 max_wal_size,但被逻辑槽或者流保护,不会被清理; `unreserved`：超出槽和流的保护范围，将于下次 checkpoint 被清理; `removed：已经被清理
safe_wal_size	bigint	安全的最大 WAL 大小限制,超出此限制的 wal 会被清理
loops	bigint	执行轮数
tuples	bigint	已处理的行数
total_slip_ms	bigint	累计休眠时长（任务调度延迟）
last_slip_ms	bigint	上一次休眠时长
last_lag_ms	bigint	流处理延迟(处理的最后一个 tick 事务到最新 tick 的时间差)
total_duration_ms	bigint	总执行耗时
last_duration_ms	bigint	上次执行耗时
last_run	timestamp with time zone	上次运行时间
held_until	timestamp with time zone	下次被调度的时间(执行出错会休眠一分钟后再执行)
has_error	boolean	是否有错误发生
error_message	text	错误信息
error_detail	text	错误详情
error_hint	text	错误提示
error_context	text	错误上下文
slot_id	integer	占用的 domino worker 槽
strategy	text	调度策略
summary	text	策略描述摘要

查询示例

-- 检查所有有错误的流
SELECT stream_name, error_message
FROM domino.stream_info
WHERE has_error;

-- 查看滞后较大的流
SELECT stream_name, pg_size_pretty(lag_size) AS lag
FROM domino.stream_info
WHERE lag_size > 64*1024*1024;

管理函数

参数	说明
`domino.start()`	一键启动。由于创建插件后，下次重启之前 v2 组件不会自动启动，需要手动启动。
`domino.stop()`	一键停止。
`domino.start_decoder()`	启动解码器。
`domino.stop_decoder()`	关闭解码器。
`domino.ticker_start()`	启动 ticker。
`domino.ticker_stop()`	关闭 ticker。
`domino.ticker_pause()`	暂停 ticker。
`domino.ticker_resume()`	恢复 ticker。
`domino.ticker_reset()`	重置 ticker。
`domino.wait_tick()`	等待下一个 tick。

进度控制

多用于故障排查。

参数	说明
`domino.decoder_sync()`	等待 decoder 进度追上。
`domino.progress_sync()`	等待流进度追上。
`domino.pause_stream()`	暂停某个流。
`domino.pause_all_streams()`	暂停全部流。
`domino.resume_stream()`	恢复某个流。
`domino.resume_all_streams()`	恢复全部流。
`domino.switch_tlog()`	切换 TLog 文件（Segment）。