feat(postgresql): support PostgreSQL 17+ metrics compatibility (#1404)

kongfei605 · web-flow · commit 6d7b398f68dd · 2026-02-12T12:46:39.000+08:00
* feat(postgresql): support PostgreSQL 17+ metrics compatibility

* feat(postgresql): optimize
diff --git a/inputs/postgresql/README.md b/inputs/postgresql/README.md
@@ -36,10 +36,41 @@ address = ""
 ## 显式指定采集哪些db 
 # databases = ["app_production", "testing"]
 
+## Whether to collect statement-level metrics.
+## Requires extension pg_stat_statements enabled, see https://www.postgresql.org/docs/current/pgstatstatements.html
+# enable_statement_metrics = false
+
+## Max number of statements to collect
+## applies only when enable_statement_metrics=true
+## 0 means no limit
+# statement_metrics_limit = 100
+
 ## Whether to use prepared statements when connecting to the database.
 ## This should be set to false when connecting through a PgBouncer instance
 ## with pool_mode set to transaction.
 ## 是否使用prepared statements 连接数据库
 # prepared_statements = true
 ```
 ![dashboard](./postgresql.png)
+
+## metrics
+
+The following metrics are emitted when `enable_statement_metrics = true` (requires `pg_stat_statements`). All are cumulative counters (may reset on PostgreSQL restart or when stats are reset).
+
+| Metric | Labels | Type |
+| --- | --- | --- |
+| `postgresql_statements_calls_total` | `server`, `db`, `user`, `datname`, `query` | counter |
+| `postgresql_statements_exec_milliseconds_total` | `server`, `db`, `user`, `datname`, `query`  | counter (milliseconds) |
+| `postgresql_statements_rows_total` | `server`, `db`, `user`, `datname`, `query`  | counter (rows) |
+| `postgresql_statements_block_read_milliseconds_total` | `server`, `db`, `user`, `datname`, `query` | counter (milliseconds) |
+| `postgresql_statements_block_write_milliseconds_total` | `server`, `db`, `user`, `datname`, `query`  | counter (milliseconds) |
+
+Notes:
+- `query` is normalized by replacing newlines/tabs with spaces.
+- For PostgreSQL ≥ 13, the exporter adapts to renamed columns; metric names above remain unchanged.
+- For PostgreSQL ≥ 17, `pg_stat_bgwriter` was split into `pg_stat_bgwriter` and `pg_stat_checkpointer`. The plugin automatically queries the new view and maps the columns to the old metric names to preserve backward compatibility.
+    - `pg_stat_checkpointer.num_timed` -> `checkpoints_timed`
+    - `pg_stat_checkpointer.num_requested` -> `checkpoints_req`
+    - `pg_stat_checkpointer.write_time` -> `checkpoint_write_time`
+    - `pg_stat_checkpointer.sync_time` -> `checkpoint_sync_time`
+    - `pg_stat_checkpointer.buffers_written` -> `buffers_checkpoint`
diff --git a/inputs/postgresql/postgresql.go b/inputs/postgresql/postgresql.go
@@ -72,20 +72,24 @@ type MetricConfig struct {
 type Instance struct {
 	config.InstanceConfig
 
-	Address            string          `toml:"address"`
-	MaxLifetime        config.Duration `toml:"max_lifetime"`
-	IsPgBouncer        bool            `toml:"-"`
-	OutputAddress      string          `toml:"outputaddress"`
-	Databases          []string        `toml:"databases"`
-	IgnoredDatabases   []string        `toml:"ignored_databases"`
-	PreparedStatements bool            `toml:"prepared_statements"`
-	Metrics            []MetricConfig  `toml:"metrics"`
+	Address                string          `toml:"address"`
+	MaxLifetime            config.Duration `toml:"max_lifetime"`
+	IsPgBouncer            bool            `toml:"-"`
+	OutputAddress          string          `toml:"outputaddress"`
+	Databases              []string        `toml:"databases"`
+	IgnoredDatabases       []string        `toml:"ignored_databases"`
+	PreparedStatements     bool            `toml:"prepared_statements"`
+	EnableStatementMetrics bool            `toml:"enable_statement_metrics"`
+	StatementMetricsLimit  int             `toml:"statement_metrics_limit"`
+	Metrics                []MetricConfig  `toml:"metrics"`
+	TrimServerTagSpace     bool            `toml:"trim_server_tag_space"`
 
 	MaxIdle int
 	MaxOpen int
 	db      *sql.DB
 
 	connConfig string
+	Version    int
 }
 
 var ignoredColumns = map[string]bool{"stats_reset": true}
@@ -196,35 +200,112 @@ func (ins *Instance) Gather(slist *types.SampleList) {
 	}
 
 	for rows.Next() {
-		err = ins.accRow(rows, slist, columns)
+		err = ins.accRow(rows, slist, "", columns, columns, nil)
 		if err != nil {
 			log.Println("E! failed to get row data:", err)
 			return
 		}
 	}
 
-	query = `SELECT * FROM pg_stat_bgwriter`
-
-	bgWriterRow, err := ins.db.Query(query)
-	if err != nil {
-		log.Println("E! failed to execute Query:", err)
-		return
+	// Check Postgres Version
+	if ins.Version == 0 {
+		var version int
+		err = ins.db.QueryRow("SELECT current_setting('server_version_num')::int").Scan(&version)
+		if err != nil {
+			log.Println("E! failed to query current version:", err)
+			return
+		}
+		ins.Version = version
 	}
 
-	defer bgWriterRow.Close()
+	if ins.Version < 170000 {
+		query = `SELECT * FROM pg_stat_bgwriter`
+		bgWriterRow, err := ins.db.Query(query)
+		if err != nil {
+			log.Println("E! failed to execute Query:", err)
+			return
+		}
+
+		defer bgWriterRow.Close()
 
-	// grab the column information from the result
-	if columns, err = bgWriterRow.Columns(); err != nil {
-		log.Println("E! failed to grab column info:", err)
-		return
-	}
+		// grab the column information from the result
+		if columns, err = bgWriterRow.Columns(); err != nil {
+			log.Println("E! failed to grab column info:", err)
+			return
+		}
 
-	for bgWriterRow.Next() {
-		err = ins.accRow(bgWriterRow, slist, columns)
+		for bgWriterRow.Next() {
+			err = ins.accRow(bgWriterRow, slist, "", columns, columns, nil)
+			if err != nil {
+				log.Println("E! failed to get row data:", err)
+				return
+			}
+		}
+	} else {
+		// PG 17+ split pg_stat_bgwriter into pg_stat_bgwriter and pg_stat_checkpointer
+
+		// 1. Query pg_stat_bgwriter (remaining columns)
+		query = `SELECT * FROM pg_stat_bgwriter`
+		bgWriterRow, err := ins.db.Query(query)
+		if err != nil {
+			log.Println("E! failed to execute Query pg_stat_bgwriter:", err)
+			return
+		}
+		defer bgWriterRow.Close()
+
+		if columns, err = bgWriterRow.Columns(); err != nil {
+			log.Println("E! failed to grab column info for pg_stat_bgwriter:", err)
+			return
+		}
+
+		for bgWriterRow.Next() {
+			err = ins.accRow(bgWriterRow, slist, "", columns, columns, nil)
+			if err != nil {
+				log.Println("E! failed to get row data from pg_stat_bgwriter:", err)
+				return
+			}
+		}
+
+		// 2. Query pg_stat_checkpointer (moved columns, aliased to old names for compatibility)
+		// num_timed -> checkpoints_timed
+		// num_requested -> checkpoints_req
+		// write_time -> checkpoint_write_time
+		// sync_time -> checkpoint_sync_time
+		// buffers_written -> buffers_checkpoint
+		query = `SELECT 
+			num_timed AS checkpoints_timed,
+			num_requested AS checkpoints_req,
+			write_time AS checkpoint_write_time,
+			sync_time AS checkpoint_sync_time,
+			buffers_written AS buffers_checkpoint,
+			restartpoints_timed,
+			restartpoints_req,
+			restartpoints_done
+			FROM pg_stat_checkpointer`
+
+		checkpointerRow, err := ins.db.Query(query)
 		if err != nil {
 			log.Println("E! failed to get row data:", err)
 			return
 		}
+		defer checkpointerRow.Close()
+
+		if columns, err = checkpointerRow.Columns(); err != nil {
+			log.Println("E! failed to grab column info for pg_stat_checkpointer:", err)
+			return
+		}
+
+		for checkpointerRow.Next() {
+			err = ins.accRow(checkpointerRow, slist, "", columns, columns, nil)
+			if err != nil {
+				log.Println("E! failed to get row data from pg_stat_checkpointer:", err)
+				return
+			}
+		}
+	}
+
+	if ins.EnableStatementMetrics {
+		ins.getStatementMetrics(slist, ins.Version)
 	}
 
 	waitMetrics := new(sync.WaitGroup)
@@ -239,6 +320,84 @@ func (ins *Instance) Gather(slist *types.SampleList) {
 	waitMetrics.Wait()
 }
 
+func (ins *Instance) getStatementMetrics(slist *types.SampleList, version int) {
+	var query string
+
+	limit := func() string {
+		if ins.StatementMetricsLimit > 0 {
+			return fmt.Sprintf(" LIMIT %d", ins.StatementMetricsLimit)
+		}
+		return ""
+	}()
+
+	query = `SELECT
+		pg_get_userbyid(userid) as user,
+		pg_database.datname,
+		regexp_replace(pg_stat_statements.query, E'[\\r\\n\\t]+', ' ', 'g') AS query,
+		pg_stat_statements.calls as calls_total,
+		pg_stat_statements.total_time as exec_milliseconds_total,
+		pg_stat_statements.rows as rows_total,
+		pg_stat_statements.blk_read_time as block_read_milliseconds_total,
+		pg_stat_statements.blk_write_time as block_write_milliseconds_total
+		FROM pg_stat_statements
+	JOIN pg_database
+		ON pg_database.oid = pg_stat_statements.dbid
+	ORDER BY exec_milliseconds_total DESC`
+
+	if version >= 170000 {
+		query = `SELECT
+		pg_get_userbyid(userid) as user,
+		pg_database.datname,
+		regexp_replace(pg_stat_statements.query, E'[\\r\\n\\t]+', ' ', 'g') AS query,
+		pg_stat_statements.calls as calls_total,
+		pg_stat_statements.total_exec_time as exec_milliseconds_total,
+		pg_stat_statements.rows as rows_total,
+		pg_stat_statements.shared_blk_read_time as block_read_milliseconds_total,
+		pg_stat_statements.shared_blk_write_time as block_write_milliseconds_total
+		FROM pg_stat_statements
+	JOIN pg_database
+		ON pg_database.oid = pg_stat_statements.dbid
+	ORDER BY exec_milliseconds_total DESC`
+	} else if version >= 130000 {
+		query = `SELECT
+		pg_get_userbyid(userid) as user,
+		pg_database.datname,
+		regexp_replace(pg_stat_statements.query, E'[\\r\\n\\t]+', ' ', 'g') AS query,
+		pg_stat_statements.calls as calls_total,
+		pg_stat_statements.total_exec_time as exec_milliseconds_total,
+		pg_stat_statements.rows as rows_total,
+		pg_stat_statements.blk_read_time as block_read_milliseconds_total,
+		pg_stat_statements.blk_write_time as block_write_milliseconds_total
+		FROM pg_stat_statements
+	JOIN pg_database
+		ON pg_database.oid = pg_stat_statements.dbid
+	ORDER BY exec_milliseconds_total DESC`
+	}
+
+	statements, err := ins.db.Query(query + limit)
+	if err != nil {
+		log.Println("E! failed to query stat statements:", err.Error())
+		return
+	}
+	defer statements.Close()
+
+	columns, err := statements.Columns()
+	if err != nil {
+		log.Println("E! failed to grab column info:", err.Error())
+		return
+	}
+
+	labelColumns := []string{"user", "query"}
+	valueColumns := []string{"calls_total", "exec_milliseconds_total", "rows_total", "block_read_milliseconds_total", "block_write_milliseconds_total"}
+	for statements.Next() {
+		err := ins.accRow(statements, slist, "statements", columns, valueColumns, labelColumns)
+		if err != nil {
+			log.Println("E! failed to get row data:", err.Error())
+			return
+		}
+	}
+}
+
 func (ins *Instance) scrapeMetric(waitMetrics *sync.WaitGroup, slist *types.SampleList, metricConf MetricConfig, tags map[string]string) {
 	defer waitMetrics.Done()
 
@@ -346,7 +505,7 @@ type scanner interface {
 	Scan(dest ...interface{}) error
 }
 
-func (ins *Instance) accRow(row scanner, slist *types.SampleList, columns []string) error {
+func (ins *Instance) accRow(row scanner, slist *types.SampleList, prefix string, columns, valueColumns, labelColumns []string) error {
 	var columnVars []interface{}
 	var dbname bytes.Buffer
 
@@ -398,11 +557,25 @@ func (ins *Instance) accRow(row scanner, slist *types.SampleList, columns []stri
 
 	tags := map[string]string{"server": tagAddress, "db": dbname.String()}
 
+	for _, labelKey := range labelColumns {
+		if columnMap[labelKey] != nil && !ignoredColumns[labelKey] {
+			tags[labelKey] = fmt.Sprint(*columnMap[labelKey])
+		}
+	}
+
 	fields := make(map[string]interface{})
-	for col, val := range columnMap {
-		_, ignore := ignoredColumns[col]
-		if !ignore {
-			fields[col] = *val
+
+	for _, valueKey := range valueColumns {
+		if columnMap[valueKey] != nil && !ignoredColumns[valueKey] {
+			metricsName := valueKey
+			if prefix != "" {
+				if strings.HasPrefix(valueKey, "_") {
+					metricsName = prefix + valueKey
+				} else {
+					metricsName = prefix + "_" + valueKey
+				}
+			}
+			fields[metricsName] = *columnMap[valueKey]
 		}
 	}
 	// acc.AddFields("postgresql", fields, tags)
@@ -480,6 +653,9 @@ func (ins *Instance) SanitizedAddress() (sanitizedAddress string, err error) {
 	}
 
 	sanitizedAddress = kvMatcher.ReplaceAllString(canonicalizedAddress, "")
+	if ins.TrimServerTagSpace {
+		sanitizedAddress = strings.TrimSpace(sanitizedAddress)
+	}
 
 	return sanitizedAddress, err
-}
+}
