Telegraf PostgreSQL Output 插件实战指南自动建表、模板化 Schema 与数据类型映射【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf本指南围绕 Telegraf 的 PostgreSQL 输出插件outputs.postgresql自 Telegraf v1.24.0 起可用展开讲解如何将采集到的指标写入 PostgreSQL 及其兼容数据库如 TimescaleDB涵盖连接配置、自动建表与补列机制、tags/fields 的两种存储模型、类型映射、并发写入、SQL 模板定制以及错误处理。读完本文你将掌握该插件的全部配置项、底层实现原理并能结合模板定制出适配 TimescaleDB、只读表结构等场景的生产级落库方案。插件概述与适用场景outputs.postgresql插件把 Telegraf 指标写入 PostgreSQL或兼容服务器其核心能力是自动管理数据库 schema当指标中出现数据库里不存在的列时插件会自动执行建表、补列等 DDL 操作无需人工预先维护表结构。每个 measurement 对应一张表指标的 field 与 tag 以列的形式落库时间戳作为time列。该插件属于datastore类别支持所有平台。典型应用场景包括将时序指标直接写入 PostgreSQL供 Grafana 等工具直接查询配合 TimescaleDB 扩展获得 hypertable 与压缩能力见后文模板示例以 JSONB 方式灵活存储动态 tag/field避免频繁改表。全局配置、启动错误行为与 Secret 支持与其他 Telegraf 插件一致本插件支持指标改名、tag/field 过滤、别名与插件排序等全局配置详见 docs/CONFIGURATION.md。启动错误行为startup_error_behavior插件支持通过startup_error_behavior指定启动失败时的处理策略可选值如下取值行为error启动失败时 Telegraf 停止并退出默认行为ignore忽略启动错误禁用本插件但其他插件继续运行retry每次 gather/write 周期重试启动成功前插件保持禁用probe探测插件功能若支持探测失败则禁用不支持探测时按ignore处理Secret store 支持connection选项支持从 Secret Store 读取连接串使用方法参见 docs/CONFIGURATION.md 中的 Secret 文档。基础配置详解以下为插件完整配置对应仓库中的 plugins/outputs/postgresql/sample.confREADME 中亦有完整示例# Publishes metrics to a postgresql database [[outputs.postgresql]] ## Specify connection address via the standard libpq connection string: ## host... user... password... sslmode... dbname... ## Or a URL: ## postgres://[user[:password]]localhost[/dbname]?sslmode[disable|verify-ca|verify-full] ## ## All connection parameters are optional. Environment vars are also supported. ## e.g. PGPASSWORD, PGHOST, PGUSER, PGDATABASE ## ## Non-standard parameters: ## pool_max_conns (default: 1) - Maximum size of connection pool for parallel (per-batch per-table) inserts. ## pool_min_conns (default: 0) - Minimum size of connection pool. ## pool_max_conn_lifetime (default: 0s) - Maximum connection age before closing. ## pool_max_conn_idle_time (default: 0s) - Maximum idle time of a connection before closing. ## pool_health_check_period (default: 0s) - Duration between health checks on idle connections. # connection ## Postgres schema to use. # schema public ## Store tags as foreign keys in the metrics table. Default is false. # tags_as_foreign_keys false ## Suffix to append to table name (measurement name) for the foreign tag table. # tag_table_suffix _tag ## Deny inserting metrics if the foreign tag cant be inserted. # foreign_tag_constraint false ## Store all tags as a JSONB object in a single tags column. # tags_as_jsonb false ## Store all fields as a JSONB object in a single fields column. # fields_as_jsonb false ## Name of the timestamp column ## NOTE: Some tools (e.g. Grafana) require the default name so be careful! # timestamp_column_name time ## Type of the timestamp column ## Currently, timestamp without time zone and timestamp with time zone ## are supported # timestamp_column_type timestamp without time zone ## Templated statements to execute when creating a new table. # create_templates [ # CREATE TABLE {{ .table }} ({{ .columns }}), # ] ## Templated statements to execute when adding columns to a table. ## Set to an empty list to disable. Points containing tags for which there is ## no column will be skipped. Points containing fields for which there is ## no column will have the field omitted. # add_column_templates [ # ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join , ADD COLUMN IF NOT EXISTS }}, # ] ## Templated statements to execute when creating a new tag table. # tag_table_create_templates [ # CREATE TABLE {{ .table }} ({{ .columns }}, PRIMARY KEY (tag_id)), # ] ## Templated statements to execute when adding columns to a tag table. ## Set to an empty list to disable. Points containing tags for which there is ## no column will be skipped. # tag_table_add_column_templates [ # ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join , ADD COLUMN IF NOT EXISTS }}, # ] ## The postgres data type to use for storing unsigned 64-bit integer values ## (Postgres does not have a native unsigned 64-bit integer type). ## The value can be one of: ## numeric - Uses the PostgreSQL numeric data type. ## uint8 - Requires pguint extension # uint64_type numeric ## When using pool_max_conns 1, and a temporary error occurs, the query is ## retried with an incremental backoff. This controls the maximum duration. # retry_max_backoff 15s ## Approximate number of tag IDs to store in in-memory cache (when using ## tags_as_foreign_keys). This is an optimization to skip inserting known ## tag IDs. Each entry consumes approximately 34 bytes of memory. # tag_cache_size 100000 ## Cut column names at the given length to not exceed PostgreSQLs ## identifier length limit (default: no limit) ## Be careful to not create duplicate column names! # column_name_length_limit 0 ## Enable set the log level for the Postgres driver. # log_level warn # trace, debug, info, warn, error, none连接串connection连接串支持两种写法libpq 标准连接串host... user... password... sslmode... dbname...URL 形式postgres://[user[:password]]localhost[/dbname]?sslmode[disable|verify-ca|verify-full]。所有连接参数均可选并支持环境变量如PGPASSWORD、PGHOST、PGUSER、PGDATABASE。除标准 libpq 参数外还支持若干非标准连接池参数pool_max_conns默认 1控制并发写入、pool_min_conns默认 0、pool_max_conn_lifetime默认 0s、pool_max_conn_idle_time默认 0s、pool_health_check_period默认 0s。从源码看plugins/outputs/postgresql/postgresql.go 的Init/Connect方法连接管理基于pgx/pgxpool实现pgx默认连接池上限是 4但插件在未显式指定pool_max_conns时会强制改为 1同时若未指定application_name插件会将其设为telegraf方便在数据库侧识别来源连接。log_level可设为trace、debug、info、warn、error、none之一默认warn用于控制 pgx 驱动的日志级别。关键默认值在newPostgresql()plugins/outputs/postgresql/postgresql.go中可以看到各选项默认值schemapublic、tag_table_suffix_tag、tag_cache_size100000、uint64_typenumeric、retry_max_backoff15s、log_levelwarn四组模板均有默认 SQL见下文。Init()阶段还会校验tag_cache_size0、时间戳列类型仅允许timestamp without time zone/timestamp with time zone两种、uint64_type仅允许numeric/uint8。并发写入单连接串行与连接池并行默认情况下插件不启用并发此时重试、缓冲等可靠性由 Telegraf 核心处理。若要提升写入吞吐将连接串中的pool_max_conns设为大于 1 即可启用并发模式此时重试与缓冲需由插件自行负责。启用并发后的行为见 plugins/outputs/postgresql/postgresql.go 的Write/writeSequential/writeConcurrent/writeWorker到达的批次会按 measurement表名拆分为多个子批次每个表一个TableSource插件启动与pool_max_conns等量的 worker goroutine通过 channel 分发子批次并行写入若前一批尚未完成又有新批次到来新批次同样走并发通道若连接池被占满后续批次会在 Telegraf 核心内排队缓冲。串行模式下多表批次会放入同一个事务并为每个子批次开启 savepoint遇到永久性错误时只回滚该子批次、丢弃该子批次数据其余表的数据照常提交仅有一个表时直接返回由 Telegraf 整体重试。Foreign Tags把 tag 存成外键默认tags_as_foreign_keysfalse情况下tag 作为普通列写进指标表开启后tag 会写入一张独立的 tag 表该表包含tag_id列bigint每个唯一的 tag 值组合即每个 series在 tag 表中有唯一的一行与其tag_id指标表中仅存放tag_id外键列。表名由 measurement 名加tag_table_suffix默认_tag构成。写入流程见 plugins/outputs/postgresql/postgresql.go 的writeTagTable在事务内创建临时表LIKE目标 tag 表通过COPY将待插入的 tag 集合批量写入临时表执行INSERT INTO ... SELECT * FROM ... ORDER BY tag_id ON CONFLICT (tag_id) DO NOTHING合并进正式 tag 表。此外插件为每个表维护一个tagHashSalt基于 measurement 名的 FNV-64a 哈希与 tag ID 相加作为内存缓存键plugins/outputs/postgresql/table_source.go 的TagTableSource。缓存由freecache实现plugins/outputs/postgresql/postgresql.go 的Connect大小由tag_cache_size控制每条目约占用 34 字节内存用于跳过已插入过的 tag ID显著降低重复写入开销。foreign_tag_constrainttrue时若 tag 无法写入 tag 表则拒绝插入对应指标为false时仅记录错误并继续tag 值稳定问题修复后可随后续指标补上。数据类型映射插件默认将 Influx 数据类型映射为如下 PostgreSQL 类型Influx 类型PostgreSQL 类型floatdouble precisionintegerbigintuintegernumeric *stringtextbooleanbooleanunix timestamptimestamp其中uinteger无符号 64 位整数映射为numeric因为其取值范围可能超过bigint若直接使用bigint会在插入时出错numeric是任意精度十进制类型效率低于bigint这是为了覆盖完整取值范围的必要取舍。实际映射逻辑在 plugins/outputs/postgresql/datatypes.go 的derivePgDatatype中实现比上表更细致int64/int/uint/uint32→bigint、int32→integer、int16/int8→smallint、float64→double precision、float32→real。注意在非 JSONB 模式下各列的最终类型由该批指标中该列首个被写入的值推导得出。pguint原生无符号整数PostgreSQL 本身没有原生无符号 64 位整数类型社区扩展 pguint在registerUint8中通过查询pg_type表获取uint8的 OID 并注册到 pgx 的类型映射从而在二进制/文本协议下编码解码uint64值其编解码底层复用了numeric的线上传输格式因此无需为uint8编写全新编解码器。SQL 模板化完全掌控 Schema插件使用Go text/template生成建表与加列的 SQL 语句用户可通过模板完全控制 schema 结构。模板引擎实现在 plugins/outputs/postgresql/sqltemplate/template.go默认模板等价于[outputs.postgresql] create_templates [ CREATE TABLE {{.table}} ({{.columns}}), ] add_column_templates [ ALTER TABLE {{.table}} ADD COLUMN IF NOT EXISTS {{.columns|join , ADD COLUMN IF NOT EXISTS }}, ] tag_table_create_templates [ CREATE TABLE {{.table}} ({{.columns}}, PRIMARY KEY (tag_id)) ] tag_table_add_column_templates [ ALTER TABLE {{.table}} ADD COLUMN IF NOT EXISTS {{.columns|join , ADD COLUMN IF NOT EXISTS }}, ]模板变量所有模板执行时均可使用以下变量table— 当前正在创建/修改的表对象columns— 新添加的列新建表时为全部列加列时为新增列allColumns— 表的全部列新旧都包含新建表时等同于columnsmetricTable— 存放 field 的表对象使用tags_as_foreign_keys且当前模板作用于 tag 表时它指向使用该 tag 表的指标表tagTable— 存放 tag 的表对象使用tags_as_foreign_keys且当前模板作用于指标表时它指向对应的 tag 表。对象直接插值时会调用其String()方法自动转字符串。Table对象支持WithSchema、WithName、WithSuffix等方法派生新表引用Columns支持Definitions、Identifiers、Selectors、Tags、Fields、Keys、Concat、Sorted、Hash等便捷方法例如Columns.Hash可生成基于列名的 base32 哈希最长 7 字符常用于“不可变表 重命名 UNION 视图”场景。模板函数除 Sprig 库的全部函数外模板内还额外提供quoteIdentifier— 将字符串作为 PostgreSQL 标识符加双引号转义quoteLiteral— 将字符串作为 PostgreSQL 字面量加单引号转义。模板解析时开启了missingkeyerror引用了不存在的变量会在渲染阶段直接报错避免生成静默错误的 SQL。Schema 同步的底层机制schema变更由TableManagerplugins/outputs/postgresql/table_manager.go负责其关键流程EnsureStructure可概括为先查内存中缓存的表结构map[tableName]map[columnName]Column缺列时再查询information_schema.columns确认表不存在且配置了create_templates时执行建表模板否则无法建表已存在的表缺少列时执行add_column_templates补列模板列表为空即禁用对应操作指标中缺失的 tag 列会导致该指标被跳过缺失的 field 列会导致该 field 被省略所有 DDL 在事务内先执行SELECT pg_advisory_xact_lock(...)锁 ID 为常量5705450890675909945保证多个 Telegraf 进程并发时 schema 修改串行化、避免死锁建列后通过COMMENT ON COLUMN ... IS tag在列注释中标记 tag 角色仅读取注释第一个单词避免独占用户注释空间以便后续读取表结构时区分 tag 列与 field 列。此外TableManager内置了 63 字节的标识符长度上限常量maxIdentifierLength 63作为兜底校验与 PostgreSQL 的限制一致。长列名处理PostgreSQL 对列标识符长度有限制见官方 limits 文档该限制可在服务端调整因此 Telegraf 默认不强制截断——截断还可能造成“仅截断后部分不同”的列名冲突。[!WARNING] 设置column_name_length_limit前请务必确认不会造成列名冲突如有疑虑建议先用 regexp 等 processor 显式缩短 field 与 tag 名。从源码看EnsureStructure会先按column_name_length_limit截断列名对每个超长列名记录一次 warn 日志随后校验列名长度tag 列名过长会直接报错返回field 列名过长则记录 error 并忽略该列。实战模板示例TimescaleDB 单节点开启tags_as_foreign_keys通过模板在建表后创建 hypertable按 7 天分块并启用压缩tags_as_foreign_keys true create_templates [ CREATE TABLE {{ .table }} ({{ .columns }}), SELECT create_hypertable({{ .table|quoteLiteral }}, time, chunk_time_interval INTERVAL 7d), ALTER TABLE {{ .table }} SET (timescaledb.compress, timescaledb.compress_segmentby tag_id), ]TimescaleDB 多节点使用create_distributed_hypertable将数据分布到多个数据节点tags_as_foreign_keys true create_templates [ CREATE TABLE {{ .table }} ({{ .columns }}), SELECT create_distributed_hypertable({{ .table|quoteLiteral }}, time, partitioning_column tag_id, number_partitions (SELECT count(*) FROM timescaledb_information.data_nodes)::integer, replication_factor 2, chunk_time_interval INTERVAL 7d), ALTER TABLE {{ .table }} SET (timescaledb.compress, timescaledb.compress_segmentby tag_id), ]Tag 表 视图开启tags_as_foreign_keys后用视图自动 JOIN 指标表与 tag 表。指标表与 tag 表存放在telegrafschema视图放在publicschematags_as_foreign_keys true schema telegraf create_templates [ CREATE TABLE {{ .table }} ({{ .columns }}), CREATE VIEW {{ .table.WithSchema public }} AS SELECT time, {{ (.tagTable.Columns.Tags.Concat .allColumns.Fields).Identifiers | join , }} FROM {{ .table }} t, {{ .tagTable }} tt WHERE t.tag_id tt.tag_id, ] add_column_templates [ ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join , ADD COLUMN IF NOT EXISTS }}, DROP VIEW IF EXISTS {{ .table.WithSchema public }}, CREATE VIEW {{ .table.WithSchema public }} AS SELECT time, {{ (.tagTable.Columns.Tags.Concat .allColumns.Fields).Identifiers | join , }} FROM {{ .table }} t, {{ .tagTable }} tt WHERE t.tag_id tt.tag_id, ] tag_table_add_column_templates [ ALTER TABLE {{.table}} ADD COLUMN IF NOT EXISTS {{.columns|join , ADD COLUMN IF NOT EXISTS }}, DROP VIEW IF EXISTS {{ .metricTable.WithSchema public }}, CREATE VIEW {{ .metricTable.WithSchema public }} AS SELECT time, {{ (.allColumns.Tags.Concat .metricTable.Columns.Fields).Identifiers | join , }} FROM {{ .metricTable }} t, {{ .tagTable }} tt WHERE t.tag_id tt.tag_id, ]不可变数据表部分 PostgreSQL 兼容数据库不允许在创建后修改表结构。该示例通过“建新表 视图 UNION”绕开限制加列时把旧表重命名附带基于旧列名的哈希后缀再建一张包含全部列的新表并用UNION ALL视图把新旧表数据合并tags_as_foreign_keys true schema telegraf create_templates [ CREATE TABLE {{ .table }} ({{ .allColumns }}), SELECT create_hypertable({{ .table|quoteLiteral }}, time, chunk_time_interval INTERVAL 7d), ALTER TABLE {{ .table }} SET (timescaledb.compress, timescaledb.compress_segmentby tag_id), SELECT add_compression_policy({{ .table|quoteLiteral }}, INTERVAL 14d), CREATE VIEW {{ .table.WithSuffix _data }} AS SELECT {{ .allColumns.Selectors | join , }} FROM {{ .table }}, CREATE VIEW {{ .table.WithSchema public }} AS SELECT time, {{ (.tagTable.Columns.Tags.Concat .allColumns.Fields).Identifiers | join , }} FROM {{ .table.WithSuffix _data }} t, {{ .tagTable }} tt WHERE t.tag_id tt.tag_id, ] add_column_templates [ ALTER TABLE {{ .table }} RENAME TO {{ (.table.WithSuffix _ .table.Columns.Hash).WithSchema }}, ALTER VIEW {{ .table.WithSuffix _data }} RENAME TO {{ (.table.WithSuffix _ .table.Columns.Hash _data).WithSchema }}, DROP VIEW {{ .table.WithSchema public }}, CREATE TABLE {{ .table }} ({{ .allColumns }}), SELECT create_hypertable({{ .table|quoteLiteral }}, time, chunk_time_interval INTERVAL 7d), ALTER TABLE {{ .table }} SET (timescaledb.compress, timescaledb.compress_segmentby tag_id), SELECT add_compression_policy({{ .table|quoteLiteral }}, INTERVAL 14d), CREATE VIEW {{ .table.WithSuffix _data }} AS SELECT {{ .allColumns.Selectors | join , }} FROM {{ .table }} UNION ALL SELECT {{ (.allColumns.Union .table.Columns).Selectors | join , }} FROM {{ .table.WithSuffix _ .table.Columns.Hash _data }}, CREATE VIEW {{ .table.WithSchema public }} AS SELECT time, {{ (.tagTable.Columns.Tags.Concat .allColumns.Fields).Identifiers | join , }} FROM {{ .table.WithSuffix _data }} t, {{ .tagTable }} tt WHERE t.tag_id tt.tag_id, ] tag_table_add_column_templates [ ALTER TABLE {{ .table }} ADD COLUMN IF NOT EXISTS {{ .columns|join , ADD COLUMN IF NOT EXISTS }}, DROP VIEW {{ .metricTable.WithSchema public }}, CREATE VIEW {{ .metricTable.WithSchema public }} AS SELECT time, {{ (.allColumns.Tags.Concat .metricTable.Columns.Fields).Identifiers | join , }} FROM {{ .metricTable.WithSuffix _data }} t, {{ .table }} tt WHERE t.tag_id tt.tag_id, ]索引为 time 与 tag 列创建 btree 索引以加速查询create_templates [ CREATE TABLE {{ .table }} ({{ .columns }}), CREATE INDEX ON {{ .table }} USING btree({{ .columns.Keys.Identifiers | join , }}) ]错误处理临时错误与永久错误写入数据库出错时插件会判断错误是临时性还是永久性逻辑见 plugins/outputs/postgresql/postgresql.go 的isTempError临时错误重试可能成功例如连接中断、死锁等永久错误如非法数据类型、权限不足等重试无意义。临时错误判定基于 PostgreSQL 错误码SQLSTATE的前两位错误类含义判定23xxx完整性约束冲突默认永久但23505 unique_violation且错误信息含pg_type_typname_nsp_index并发建表竞争判为临时25xxx非法事务状态临时可恢复的 bug 场景40xxx事务回滚40P01 deadlock_detected判为临时42xxx语法/权限错误42701 duplicate_column、42P07 duplicate_table判为临时并发 DDL 竞争53xxx资源不足临时57xxx操作者干预57014 query_cancelled、57P04 database_dropped判为永久其余判为临时对于实现Temporary()接口的错误直接采用其返回值其余未识别的错误一律视为永久。这样设计是为了避免陷入“反复重试永远不可能成功的数据”的循环防止缓冲被无用数据占满导致好数据被丢弃。处理策略writeRetry/writeSequential/writeWorker串行模式下子批次遇到临时错误则整体返回错误由 Telegraf 核心重试整个批次并发模式下临时错误按指数退避重试初始 250ms之后每次翻倍上限由retry_max_backoff默认 15s控制重试前会Reset()子批次游标永久错误则丢弃当前子批次即写入同一张表的批次部分不影响其他表的子批次写入并记录 error 日志。小结outputs.postgresql插件把“指标落 PostgreSQL”这件事做成了近乎零维护的操作schema 自动演进、tag 外键化去重、JSONB 弹性存储、SQL 模板全定制、连接池并发与分级错误处理一应俱全。建议读者在落地时重点把握三点根据查询习惯选择 tag 存储模型普通列 / 外键表 / JSONB、在非默认 PostgreSQL 环境下仔细设计四组 SQL 模板、以及为高吞吐场景显式配置pool_max_conns与retry_max_backoff。相关源码与测试可继续阅读 plugins/outputs/postgresql/postgresql.go、plugins/outputs/postgresql/table_manager.go、plugins/outputs/postgresql/table_source.go、plugins/outputs/postgresql/sqltemplate/template.go 及其配套*_test.go文件。【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考