Telegraf PostgreSQL Output 插件实战指南：自动建表、模板化 Schema 与数据类型映射

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指定启动失败时的处理策略，可选值如下：

Secret store 支持

connection选项支持从 Secret Store 读取连接串，使用方法参见 docs/CONFIGURATION.md 中的 Secret 文档。

基础配置详解

以下为插件完整配置（对应仓库中的 plugins/outputs/postgresql/sample.conf，README 中亦有完整示例）：

# 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 can't 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 PostgreSQL's ## '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）中可以看到各选项默认值：schema="public"、tag_table_suffix="_tag"、tag_cache_size=100000、uint64_type="numeric"、retry_max_backoff=15s、log_level="warn"；四组模板均有默认 SQL（见下文）。Init()阶段还会校验tag_cache_size>=0、时间戳列类型仅允许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_keys=false）情况下，tag 作为普通列写进指标表；开启后，tag 会写入一张独立的 tag 表，该表包含tag_id列（bigint），每个唯一的 tag 值组合（即每个 series）在 tag 表中有唯一的一行与其tag_id，指标表中仅存放tag_id外键列。表名由 measurement 名加tag_table_suffix（默认_tag）构成。

写入流程（见 plugins/outputs/postgresql/postgresql.go 的writeTagTable）：

1. 在事务内创建临时表（LIKE目标 tag 表）；
2. 通过COPY将待插入的 tag 集合批量写入临时表；
3. 执行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_constraint=true时，若 tag 无法写入 tag 表则拒绝插入对应指标；为false时仅记录错误并继续（tag 值稳定，问题修复后可随后续指标补上）。

数据类型映射

插件默认将 Influx 数据类型映射为如下 PostgreSQL 类型：

其中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— 表的全部列（新旧都包含）；新建表时等同于columns；
• metricTable— 存放 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 字面量加单引号转义。

模板解析时开启了missingkey=error，引用了不存在的变量会在渲染阶段直接报错，避免生成静默错误的 SQL。

Schema 同步的底层机制

schema变更由TableManager（plugins/outputs/postgresql/table_manager.go）负责，其关键流程EnsureStructure可概括为：

1. 先查内存中缓存的表结构（map[tableName]map[columnName]Column），缺列时再查询information_schema.columns确认；
2. 表不存在且配置了create_templates时执行建表模板，否则无法建表；
3. 已存在的表缺少列时执行add_column_templates补列；模板列表为空即禁用对应操作：指标中缺失的 tag 列会导致该指标被跳过，缺失的 field 列会导致该 field 被省略；
4. 所有 DDL 在事务内先执行SELECT pg_advisory_xact_lock(...)（锁 ID 为常量5705450890675909945），保证多个 Telegraf 进程并发时 schema 修改串行化、避免死锁；
5. 建列后通过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，视图放在publicschema：

tags_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）的前两位：

对于实现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