search
数据采集 采集源配置 PostgreSQL数据采集

PostgreSQL数据采集

简介

采集 PostgreSQL 数据指标上报到 DataFlux 中。

前置条件

配置

PostgreSQL 配置

PostgreSQL 需要开启 replication slots 功能,配置文件通常存放在 /etc/postgresql/12/main 目录下,这里的 12 是指大版本号,比如 9.49

修改postgresql.conf文件,搜索wal_levelmax_replication_slots项,将其取消注释,然后配置为:

wal_level = logical;
max_replication_slots = 5; #该值要大于1

打开 pb_hba.conf 文件,在其末尾添加以下内容,为用户 repl 添加复制权限(用户 repl 在下一步创建)。

host replication test_rep all md5`,

登录 PostgreSQL 数据库,创建有 REPLICATION 权限的用户 repl,并授权该用户连接 test_database 数据库。 具体命令如下:

CREATE ROLE repl LOGIN  ENCRYPTED PASSWORD 'xxxxxxxx' REPLICATION;

GRANT CONNECT ON DATABASE test_database to repl;

注意,修改配置完成后需要重启PostgreSQL数据库

DataKit 配置

进入 DataKit 安装目录下的 conf.d/db 目录,复制 postgresql_replication.conf.sample 并命名为 postgresql_replication.openldapconf。示例如下:

[[inputs.postgresql_replication]]
    # PG 主机 ip
    # 必填
    host="127.0.0.1"
    
    # PG 端口
    # 必填
    port=25432
    
    # 登录用户,需要有复制权限
    # 必填
    user="<your-user>"
    
    # 用户对应密码
    # 必填
    password="<your-password>"
    
    # 采集的数据库
    # 必填
    database="<your-database>"

    # 指定采集数据的表,如果为空,则默认采集所有表的数据
    table="<your-table>"
    
    # 数据上传类型,可选: "metric", "logging"
    # 如果为空,则默认上传到 "metric"
    categories = ["metric"]
    
    # 注册监听事件,事件包括"INSERT","UPDATE","DELETE"
    # 必填,注意是大写英文
    events=["INSERT"]
    
    # 采集该字段数据且类型为 tags,可以为空
    tag_colunms=[]
    
    # 采集该字段数据且类型为 fieldss
    # 必填
    field_colunms=["fieldName"]
    
    # 自定义 tags
    # [inputs.postgresql_replication.tags]
    # tags1="value1"

只要是该数据库中的新数据都会被采集,跟表无关

指标说明:

tags 以用户配置为第一优先;

采集器会补充 database 字段,其值为配置文件中的 database 参数;

采集器会补充 relation 字段,其值为当前数据的 <schema>.<table>,例如 public.test_table

常见问题:

报错信息: failed to connect, err: failed to create replication slot: ERROR: all replication slots are in use (SQLSTATE 53400)

此情况为 postgresql 数据库的 slot 已经达到上限,无法创建新的 replication slot,需要登录数据手动删除,或者将上限调高。

删除 replication slot 命令:

datakit_test_db=# select * from  pg_replication_slots;
            slot_name             |    plugin     | slot_type | datoid |    database     | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn
----------------------------------+---------------+-----------+--------+-----------------+-----------+--------+------------+------+--------------+-------------+---------------------
 datakit_slot_1596110300190109741 | test_decoding | logical   |  16385 | datakit_test_db | f         | f      |            |      |          575 | 0/1684630   | 0/1684668
 pg_slot_1                        |               | physical  |        |                 | f         | f      |            |      |              |             |
 datakit_slot_1596110399865334476 | test_decoding | logical   |  16385 | datakit_test_db | f         | f      |            |      |          577 | 0/1684F48   | 0/1684F80
(3 rows)

datakit_test_db=# SELECT pg_drop_replication_slot('datakit_slot_1596110300190109741');
 pg_drop_replication_slot
--------------------------

(1 row)

datakit_test_db=# SELECT pg_drop_replication_slot('datakit_slot_1596110399865334476');
 pg_drop_replication_slot
--------------------------
(1 row)

采集器会创建以 datakit_slot_ 加时间戳的为名的 slot。采集器正常退出会删除自己创建的 slot。

注意不要删除其他可能正在使用的 slot。

示例输出

postgresql_replication,database=datakit_test_db,relation=public.test_table,test=DATAKIT name="dk" 1596186225306949666