feat(go.d.plugin/postgres): add running-queries and pg_stat_monitor support (#21656)
Co-authored-by: ilyam8 <ilya@netdata.cloud>
Costa Tsaousis committed
Jan 28, 2026 at 11:19 UTC
ed49ed27427b5c1ab840d212498732c636dd8fe0
15 files changed
+1264
-165
docs/functions/databases.md
+5
-3
@@ -25,7 +25,7 @@ Database query functions provide deep visibility into SQL and NoSQL database per
25
| MariaDB | ✅ | - | ✅ | ✅* | [MariaDB](/src/go/plugin/go.d/collector/mysql/integrations/mariadb.md) |
26
| Percona Server | ✅ | - | ✅ | ✅* | [Percona](/src/go/plugin/go.d/collector/mysql/integrations/percona_mysql.md) |
27
| Oracle Database | ✅ | ✅ | - | - | [Oracle](/src/go/plugin/go.d/collector/oracledb/integrations/oracle_db.md) |
28
-| PostgreSQL | ✅ | - | - | - | [PostgreSQL](/src/go/plugin/go.d/collector/postgres/integrations/postgresql.md) |
28
+| PostgreSQL | ✅ | ✅ | - | ✅** | [PostgreSQL](/src/go/plugin/go.d/collector/postgres/integrations/postgresql.md) |
29
| ProxySQL | ✅ | - | - | - | [ProxySQL](/src/go/plugin/go.d/collector/proxysql/integrations/proxysql.md) |
30
| Redis | ✅ | - | - | - | [Redis](/src/go/plugin/go.d/collector/redis/integrations/redis.md) |
31
| RethinkDB | - | ✅ | - | - | [RethinkDB](/src/go/plugin/go.d/collector/rethinkdb/integrations/rethinkdb.md) |
@@ -33,6 +33,8 @@ Database query functions provide deep visibility into SQL and NoSQL database per
33
34
*\* Error Info is integrated directly into Top Queries results—each query row shows its associated errors.*
35
36
+*\*\* PostgreSQL error info requires [pg_stat_monitor](https://docs.percona.com/pg-stat-monitor/) (Percona). The collector auto-detects and uses it when available.*
37
+
38
## Function Types
39
40
### Top Queries
@@ -54,7 +56,7 @@ The number of queries returned is configurable (default: 500). This is a two-sta
56
57
Shows **currently executing queries** at the moment of request. Essential for diagnosing stuck queries, long-running transactions, or unexpected load.
58
57
-**Supported**: CockroachDB, Oracle, RethinkDB, YugabyteDB
59
+**Supported**: CockroachDB, Oracle, PostgreSQL, RethinkDB, YugabyteDB
60
61
### Deadlock Info
62
@@ -73,7 +75,7 @@ Information provided:
75
76
Shows **recent SQL errors** from the database's error history. Error attribution is embedded directly in Top Queries results—each query row includes error details when available.
77
76
-**Supported**: MySQL/MariaDB/Percona, Microsoft SQL Server
78
+**Supported**: MySQL/MariaDB/Percona, Microsoft SQL Server, PostgreSQL (with pg_stat_monitor)
79
80
Attribution status values:
81
- `enabled` — Error details available for this query
src/go/plugin/go.d/agent/jobmgr/funcshandler.go
+7
-2
@@ -122,7 +122,11 @@ func (m *Manager) makeMethodFuncHandler(moduleName, methodID string) func(functi
122
}
123
124
// Core injects required_params into the response before sending
125
- m.respondWithParams(fn, moduleName, dataResp, methodParams)
125
+ updateEvery := 1
126
+ if methodCfg.UpdateEvery > 1 {
127
+ updateEvery = methodCfg.UpdateEvery
128
+ }
129
+ m.respondWithParams(fn, moduleName, dataResp, methodParams, updateEvery)
130
}
131
}
132
@@ -161,7 +165,7 @@ func (m *Manager) handleMethodFuncInfo(moduleName, methodID string, fn functions
165
}
166
167
// respondWithParams wraps the module's data response with current required_params
164
-func (m *Manager) respondWithParams(fn functions.Function, moduleName string, dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig) {
168
+func (m *Manager) respondWithParams(fn functions.Function, moduleName string, dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
169
// Nil guard: if module returns nil, treat as internal error
170
if dataResp == nil {
171
m.respondError(fn, 500, "internal error: module returned nil response")
@@ -182,6 +186,7 @@ func (m *Manager) respondWithParams(fn functions.Function, moduleName string, da
186
// Use dynamic sort options from response if provided (reflects actual DB capabilities)
187
resp := map[string]any{
188
"v": 3,
189
+ "update_every": updateEvery,
190
"status": dataResp.Status,
191
"type": "table",
192
"has_history": false,
src/go/plugin/go.d/collector/postgres/collect.go
+1
@@ -19,6 +19,7 @@ const (
19
pgVersion10 = 10_00_00
20
pgVersion11 = 11_00_00
21
pgVersion13 = 13_00_00
22
+ pgVersion14 = 14_00_00
23
pgVersion17 = 17_00_00
24
)
25
src/go/plugin/go.d/collector/postgres/collector.go
+4
-1
@@ -120,7 +120,10 @@ type (
120
pgVersion int
121
pgStatStatementsAvail bool // cached positive result only
122
pgStatStatementsColumns map[string]bool // cached column names from pg_stat_statements
123
- pgStatStatementsMu sync.RWMutex // protects pgStatStatements* fields for concurrent access
123
+ pgStatMonitorAvail bool // cached positive result only
124
+ pgStatMonitorColumns map[string]bool // cached column names from pg_stat_monitor
125
+ queryStatsSource string // "pg_stat_monitor" or "pg_stat_statements" (auto-detected)
126
+ pgStatStatementsMu sync.RWMutex // protects pgStatStatements*/pgStatMonitor* fields for concurrent access
127
dbSr matcher.Matcher
128
recheckSettingsTime time.Time
129
recheckSettingsEvery time.Duration
src/go/plugin/go.d/collector/postgres/func_router.go
+2
@@ -23,6 +23,7 @@ func newFuncRouter(c *Collector) *funcRouter {
23
handlers: make(map[string]funcapi.MethodHandler),
24
}
25
r.handlers[topQueriesMethodID] = newFuncTopQueries(r)
26
+ r.handlers[runningQueriesMethodID] = newFuncRunningQueries(r)
27
return r
28
}
29
@@ -52,6 +53,7 @@ func (r *funcRouter) Cleanup(ctx context.Context) {
53
func pgMethods() []funcapi.MethodConfig {
54
return []funcapi.MethodConfig{
55
topQueriesMethodConfig(),
56
+ runningQueriesMethodConfig(),
57
}
58
}
59
src/go/plugin/go.d/collector/postgres/func_running_queries.go
new
+310
@@ -0,0 +1,310 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package postgres
4
+
5
+import (
6
+ "context"
7
+ "database/sql"
8
+ "fmt"
9
+ "strings"
10
+
11
+ "github.com/netdata/netdata/go/plugins/pkg/funcapi"
12
+ "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/strmutil"
13
+)
14
+
15
+const (
16
+ runningQueriesMethodID = "running-queries"
17
+ runningQueriesMaxTextLength = 4096
18
+)
19
+
20
+func runningQueriesMethodConfig() funcapi.MethodConfig {
21
+ return funcapi.MethodConfig{
22
+ ID: runningQueriesMethodID,
23
+ Name: "Running Queries",
24
+ UpdateEvery: 10,
25
+ Help: "Currently executing queries from pg_stat_activity. WARNING: Query text may contain unmasked literals (potential PII).",
26
+ RequireCloud: true,
27
+ RequiredParams: []funcapi.ParamConfig{funcapi.BuildSortParam(runningQueriesColumns)},
28
+ }
29
+}
30
+
31
+// runningQueriesColumn defines metadata for a pg_stat_activity column.
32
+type runningQueriesColumn struct {
33
+ funcapi.ColumnMeta
34
+
35
+ // DBColumn is the database column expression
36
+ DBColumn string
37
+ // sortOpt indicates whether this column appears in the sort dropdown
38
+ sortOpt bool
39
+ // sortLbl is the label shown in the sort dropdown
40
+ sortLbl string
41
+ // defaultSort indicates whether this is the default sort column
42
+ defaultSort bool
43
+ // minVersion is the minimum PostgreSQL version (0 means all versions)
44
+ minVersion int
45
+}
46
+
47
+// funcapi.SortableColumn interface implementation.
48
+func (c runningQueriesColumn) IsSortOption() bool { return c.sortOpt }
49
+func (c runningQueriesColumn) SortLabel() string { return c.sortLbl }
50
+func (c runningQueriesColumn) IsDefaultSort() bool { return c.defaultSort }
51
+func (c runningQueriesColumn) ColumnName() string { return c.Name }
52
+func (c runningQueriesColumn) SortColumn() string { return "" }
53
+
54
+// runningQueriesColumns defines ALL columns from pg_stat_activity.
55
+// Order matters - this determines column display order. Visible columns first, then hidden.
56
+var runningQueriesColumns = []runningQueriesColumn{
57
+ // === VISIBLE COLUMNS (most important first) ===
58
+
59
+ // Key metrics
60
+ {ColumnMeta: funcapi.ColumnMeta{Name: "durationMs", Tooltip: "Query duration in milliseconds (since query_start)", Type: funcapi.FieldTypeDuration, Units: "milliseconds", Visible: true, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformDuration, DecimalPoints: 2, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax}, DBColumn: "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - query_start)) * 1000", sortOpt: true, sortLbl: "Query Duration", defaultSort: true},
61
+ {ColumnMeta: funcapi.ColumnMeta{Name: "query", Tooltip: "Query text (may be truncated at track_activity_query_size)", Type: funcapi.FieldTypeString, Visible: true, Sortable: false, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sticky: true, FullWidth: true, Wrap: true, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "query"},
62
+
63
+ // Context: who/where
64
+ {ColumnMeta: funcapi.ColumnMeta{Name: "datname", Tooltip: "Name of the database", Type: funcapi.FieldTypeString, Visible: true, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "datname"},
65
+ {ColumnMeta: funcapi.ColumnMeta{Name: "usename", Tooltip: "Name of the user logged into this backend", Type: funcapi.FieldTypeString, Visible: true, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "usename"},
66
+ {ColumnMeta: funcapi.ColumnMeta{Name: "applicationName", Tooltip: "Name of the application connected to this backend", Type: funcapi.FieldTypeString, Visible: true, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "application_name"},
67
+ {ColumnMeta: funcapi.ColumnMeta{Name: "clientAddr", Tooltip: "IP address of the client (NULL for Unix socket or internal process)", Type: funcapi.FieldTypeString, Visible: true, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "client_addr::text"},
68
+
69
+ // Status
70
+ {ColumnMeta: funcapi.ColumnMeta{Name: "waitEvent", Tooltip: "Specific wait event name if backend is currently waiting", Type: funcapi.FieldTypeString, Visible: true, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "wait_event"},
71
+
72
+ // Process ID (useful for pg_terminate_backend)
73
+ {ColumnMeta: funcapi.ColumnMeta{Name: "pid", Tooltip: "Process ID of this backend", Type: funcapi.FieldTypeInteger, Visible: true, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformNumber, UniqueKey: true, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "pid"},
74
+
75
+ // === HIDDEN COLUMNS ===
76
+
77
+ // Additional status info
78
+ {ColumnMeta: funcapi.ColumnMeta{Name: "waitEventType", Tooltip: "Type of event the backend is waiting for (Activity, BufferPin, Client, Extension, IO, IPC, Lock, LWLock, Timeout)", Type: funcapi.FieldTypeString, Visible: false, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "wait_event_type"},
79
+ {ColumnMeta: funcapi.ColumnMeta{Name: "state", Tooltip: "Current state: active, idle, idle in transaction, idle in transaction (aborted), fastpath function call, disabled", Type: funcapi.FieldTypeString, Visible: false, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount, Visualization: funcapi.FieldVisualPill}, DBColumn: "state"},
80
+ {ColumnMeta: funcapi.ColumnMeta{Name: "backendType", Tooltip: "Type of backend: client backend, autovacuum worker, parallel worker, walsender, walreceiver, etc.", Type: funcapi.FieldTypeString, Visible: false, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "backend_type", minVersion: pgVersion10},
81
+
82
+ // Timestamps
83
+ {ColumnMeta: funcapi.ColumnMeta{Name: "queryStart", Tooltip: "Time when current/last query started", Type: funcapi.FieldTypeTimestamp, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformDatetime, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax}, DBColumn: "query_start", sortOpt: true, sortLbl: "Query Start Time"},
84
+ {ColumnMeta: funcapi.ColumnMeta{Name: "xactStart", Tooltip: "Time when current transaction started (NULL if no transaction)", Type: funcapi.FieldTypeTimestamp, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformDatetime, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax}, DBColumn: "xact_start"},
85
+ {ColumnMeta: funcapi.ColumnMeta{Name: "backendStart", Tooltip: "Time when this process/connection started", Type: funcapi.FieldTypeTimestamp, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformDatetime, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax}, DBColumn: "backend_start"},
86
+ {ColumnMeta: funcapi.ColumnMeta{Name: "stateChange", Tooltip: "Time when state was last changed", Type: funcapi.FieldTypeTimestamp, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformDatetime, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax}, DBColumn: "state_change"},
87
+
88
+ // Query identification
89
+ {ColumnMeta: funcapi.ColumnMeta{Name: "queryId", Tooltip: "Query identifier (requires compute_query_id or extension)", Type: funcapi.FieldTypeString, Visible: false, Sortable: false, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "query_id::text", minVersion: pgVersion14},
90
+
91
+ // Session IDs
92
+ {ColumnMeta: funcapi.ColumnMeta{Name: "leaderPid", Tooltip: "Process ID of parallel group leader (NULL if this is leader or not parallel)", Type: funcapi.FieldTypeInteger, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformNumber, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "leader_pid", minVersion: pgVersion13},
93
+ {ColumnMeta: funcapi.ColumnMeta{Name: "datid", Tooltip: "OID of the database", Type: funcapi.FieldTypeInteger, Visible: false, Sortable: false, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformNumber, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "datid"},
94
+ {ColumnMeta: funcapi.ColumnMeta{Name: "usesysid", Tooltip: "OID of the user", Type: funcapi.FieldTypeInteger, Visible: false, Sortable: false, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformNumber, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "usesysid"},
95
+
96
+ // Client details
97
+ {ColumnMeta: funcapi.ColumnMeta{Name: "clientHostname", Tooltip: "Hostname of the client via reverse DNS (only if log_hostname enabled)", Type: funcapi.FieldTypeString, Visible: false, Sortable: true, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "client_hostname"},
98
+ {ColumnMeta: funcapi.ColumnMeta{Name: "clientPort", Tooltip: "TCP port of client (-1 for Unix socket, NULL for internal process)", Type: funcapi.FieldTypeInteger, Visible: false, Sortable: true, Filter: funcapi.FieldFilterRange, Transform: funcapi.FieldTransformNumber, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "client_port"},
99
+
100
+ // Transaction IDs
101
+ {ColumnMeta: funcapi.ColumnMeta{Name: "backendXid", Tooltip: "Top-level transaction identifier of this backend", Type: funcapi.FieldTypeString, Visible: false, Sortable: false, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "backend_xid::text"},
102
+ {ColumnMeta: funcapi.ColumnMeta{Name: "backendXmin", Tooltip: "Backend's xmin horizon", Type: funcapi.FieldTypeString, Visible: false, Sortable: false, Filter: funcapi.FieldFilterMultiselect, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount}, DBColumn: "backend_xmin::text"},
103
+}
104
+
105
+// funcRunningQueries handles the running-queries function.
106
+type funcRunningQueries struct {
107
+ router *funcRouter
108
+}
109
+
110
+func newFuncRunningQueries(r *funcRouter) *funcRunningQueries {
111
+ return &funcRunningQueries{router: r}
112
+}
113
+
114
+// Compile-time interface check.
115
+var _ funcapi.MethodHandler = (*funcRunningQueries)(nil)
116
+
117
+func (f *funcRunningQueries) MethodParams(ctx context.Context, method string) ([]funcapi.ParamConfig, error) {
118
+ return []funcapi.ParamConfig{funcapi.BuildSortParam(f.getColumnsForVersion())}, nil
119
+}
120
+
121
+func (f *funcRunningQueries) Cleanup(ctx context.Context) {}
122
+
123
+func (f *funcRunningQueries) Handle(ctx context.Context, method string, params funcapi.ResolvedParams) *funcapi.FunctionResponse {
124
+ if f.router.collector.db == nil {
125
+ return funcapi.UnavailableResponse("collector is still initializing, please retry in a few seconds")
126
+ }
127
+
128
+ cols := f.getColumnsForVersion()
129
+ sortColumn := f.resolveSortColumn(params.Column("__sort"), cols)
130
+ if sortColumn == "" {
131
+ return funcapi.InternalErrorResponse("no sortable columns available")
132
+ }
133
+
134
+ // Build the query
135
+ query := f.buildQuery(cols, sortColumn)
136
+
137
+ rows, err := f.router.collector.db.QueryContext(ctx, query)
138
+ if err != nil {
139
+ if ctx.Err() == context.DeadlineExceeded {
140
+ return funcapi.ErrorResponse(504, "query timed out")
141
+ }
142
+ return funcapi.InternalErrorResponse("running queries query failed: %v", err)
143
+ }
144
+ defer rows.Close()
145
+
146
+ data, err := f.scanRows(rows, cols)
147
+ if err != nil {
148
+ return funcapi.InternalErrorResponse("%s", err)
149
+ }
150
+
151
+ cs := f.columnSet(cols)
152
+ if len(data) == 0 {
153
+ return &funcapi.FunctionResponse{
154
+ Status: 200,
155
+ Message: "No active queries found.",
156
+ Help: "Currently executing queries from pg_stat_activity",
157
+ Columns: cs.BuildColumns(),
158
+ Data: [][]any{},
159
+ DefaultSortColumn: sortColumn,
160
+ RequiredParams: []funcapi.ParamConfig{funcapi.BuildSortParam(cols)},
161
+ }
162
+ }
163
+
164
+ return &funcapi.FunctionResponse{
165
+ Status: 200,
166
+ Help: "Currently executing queries from pg_stat_activity. WARNING: Query text may contain unmasked literals (potential PII).",
167
+ Columns: cs.BuildColumns(),
168
+ Data: data,
169
+ DefaultSortColumn: sortColumn,
170
+ RequiredParams: []funcapi.ParamConfig{funcapi.BuildSortParam(cols)},
171
+ }
172
+}
173
+
174
+func (f *funcRunningQueries) getColumnsForVersion() []runningQueriesColumn {
175
+ version := f.router.collector.pgVersion
176
+ if version == 0 {
177
+ version = pgVersion14 // Assume recent version if not detected
178
+ }
179
+
180
+ var cols []runningQueriesColumn
181
+ for _, col := range runningQueriesColumns {
182
+ if col.minVersion == 0 || version >= col.minVersion {
183
+ cols = append(cols, col)
184
+ }
185
+ }
186
+ return cols
187
+}
188
+
189
+func (f *funcRunningQueries) columnSet(cols []runningQueriesColumn) funcapi.ColumnSet[runningQueriesColumn] {
190
+ return funcapi.Columns(cols, func(c runningQueriesColumn) funcapi.ColumnMeta { return c.ColumnMeta })
191
+}
192
+
193
+func (f *funcRunningQueries) resolveSortColumn(requested string, cols []runningQueriesColumn) string {
194
+ for _, col := range cols {
195
+ if col.IsSortOption() && col.Name == requested {
196
+ return col.Name
197
+ }
198
+ }
199
+ // Return default sort column
200
+ for _, col := range cols {
201
+ if col.IsDefaultSort() {
202
+ return col.Name
203
+ }
204
+ }
205
+ // Fallback to first sortable column
206
+ for _, col := range cols {
207
+ if col.IsSortOption() {
208
+ return col.Name
209
+ }
210
+ }
211
+ return ""
212
+}
213
+
214
+func (f *funcRunningQueries) buildSelectClause(cols []runningQueriesColumn) string {
215
+ var parts []string
216
+ for _, col := range cols {
217
+ parts = append(parts, col.DBColumn)
218
+ }
219
+ return strings.Join(parts, ", ")
220
+}
221
+
222
+func (f *funcRunningQueries) buildQuery(cols []runningQueriesColumn, sortColumn string) string {
223
+ // Find the actual DB column for sorting
224
+ sortExpr := "query_start"
225
+ for _, col := range cols {
226
+ if col.Name == sortColumn {
227
+ sortExpr = col.DBColumn
228
+ break
229
+ }
230
+ }
231
+
232
+ return fmt.Sprintf(`
233
+SELECT %s
234
+FROM pg_stat_activity
235
+WHERE state = 'active'
236
+ AND pid != pg_backend_pid()
237
+ AND query NOT LIKE '%%pg_stat_activity%%'
238
+ORDER BY %s DESC NULLS LAST
239
+LIMIT 500
240
+`, f.buildSelectClause(cols), sortExpr)
241
+}
242
+
243
+func (f *funcRunningQueries) scanRows(rows *sql.Rows, cols []runningQueriesColumn) ([][]any, error) {
244
+ var result [][]any
245
+
246
+ for rows.Next() {
247
+ values := make([]any, len(cols))
248
+ valuePtrs := make([]any, len(cols))
249
+
250
+ for i := range values {
251
+ valuePtrs[i] = &values[i]
252
+ }
253
+
254
+ if err := rows.Scan(valuePtrs...); err != nil {
255
+ return nil, fmt.Errorf("scanning row: %w", err)
256
+ }
257
+
258
+ row := make([]any, len(cols))
259
+ for i, col := range cols {
260
+ row[i] = f.formatValue(values[i], col)
261
+ }
262
+ result = append(result, row)
263
+ }
264
+
265
+ if err := rows.Err(); err != nil {
266
+ return nil, fmt.Errorf("iterating rows: %w", err)
267
+ }
268
+
269
+ return result, nil
270
+}
271
+
272
+func (f *funcRunningQueries) formatValue(v any, col runningQueriesColumn) any {
273
+ if v == nil {
274
+ return nil
275
+ }
276
+
277
+ switch col.Type {
278
+ case funcapi.FieldTypeString:
279
+ s := fmt.Sprintf("%v", v)
280
+ // Truncate long strings (like query text)
281
+ if col.Name == "query" && len(s) > runningQueriesMaxTextLength {
282
+ s = strmutil.TruncateText(s, runningQueriesMaxTextLength)
283
+ }
284
+ return s
285
+ case funcapi.FieldTypeInteger:
286
+ switch val := v.(type) {
287
+ case int64:
288
+ return val
289
+ case int32:
290
+ return int64(val)
291
+ case int:
292
+ return int64(val)
293
+ default:
294
+ return v
295
+ }
296
+ case funcapi.FieldTypeDuration:
297
+ switch val := v.(type) {
298
+ case float64:
299
+ return val
300
+ case int64:
301
+ return float64(val)
302
+ default:
303
+ return v
304
+ }
305
+ case funcapi.FieldTypeTimestamp:
306
+ return v
307
+ default:
308
+ return v
309
+ }
310
+}
src/go/plugin/go.d/collector/postgres/func_running_queries_test.go
new
+299
@@ -0,0 +1,299 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package postgres
4
+
5
+import (
6
+ "strings"
7
+ "testing"
8
+
9
+ "github.com/netdata/netdata/go/plugins/pkg/funcapi"
10
+ "github.com/stretchr/testify/assert"
11
+)
12
+
13
+func TestRunningQueriesColumns_HasRequiredColumns(t *testing.T) {
14
+ requiredNames := []string{
15
+ "durationMs", "query", "datname", "usename",
16
+ "applicationName", "clientAddr", "waitEvent", "pid",
17
+ "state", "queryStart",
18
+ }
19
+
20
+ colNames := make(map[string]bool)
21
+ for _, col := range runningQueriesColumns {
22
+ colNames[col.Name] = true
23
+ }
24
+
25
+ for _, name := range requiredNames {
26
+ assert.True(t, colNames[name], "column %s should be defined in runningQueriesColumns", name)
27
+ }
28
+}
29
+
30
+func TestRunningQueriesColumns_HasValidMetadata(t *testing.T) {
31
+ for _, col := range runningQueriesColumns {
32
+ assert.NotEmpty(t, col.Name, "column must have Name")
33
+ assert.NotEmpty(t, col.Tooltip, "column %s must have Tooltip", col.Name)
34
+ assert.NotEqual(t, funcapi.FieldTypeNone, col.Type, "column %s must have Type", col.Name)
35
+ assert.NotEmpty(t, col.DBColumn, "column %s must have DBColumn", col.Name)
36
+
37
+ if col.Type == funcapi.FieldTypeDuration {
38
+ assert.NotEmpty(t, col.Units, "duration column %s must have Units", col.Name)
39
+ }
40
+
41
+ if col.sortOpt {
42
+ assert.NotEmpty(t, col.sortLbl, "sort option column %s must have sortLbl", col.Name)
43
+ }
44
+ }
45
+}
46
+
47
+func TestRunningQueriesColumns_HasDefaultSort(t *testing.T) {
48
+ var defaultSortCol string
49
+ for _, col := range runningQueriesColumns {
50
+ if col.defaultSort {
51
+ defaultSortCol = col.Name
52
+ break
53
+ }
54
+ }
55
+ assert.Equal(t, "durationMs", defaultSortCol, "durationMs should be the default sort column")
56
+}
57
+
58
+func TestFuncRunningQueries_getColumnsForVersion(t *testing.T) {
59
+ tests := map[string]struct {
60
+ pgVersion int
61
+ expectCols []string
62
+ notExpectCols []string
63
+ }{
64
+ "PG9 excludes version-gated columns": {
65
+ pgVersion: 9_06_00,
66
+ expectCols: []string{"durationMs", "query", "pid", "state"},
67
+ notExpectCols: []string{"backendType", "leaderPid", "queryId"},
68
+ },
69
+ "PG10 includes backendType": {
70
+ pgVersion: pgVersion10,
71
+ expectCols: []string{"durationMs", "query", "backendType"},
72
+ notExpectCols: []string{"leaderPid", "queryId"},
73
+ },
74
+ "PG13 includes leaderPid": {
75
+ pgVersion: pgVersion13,
76
+ expectCols: []string{"durationMs", "backendType", "leaderPid"},
77
+ notExpectCols: []string{"queryId"},
78
+ },
79
+ "PG14 includes queryId": {
80
+ pgVersion: pgVersion14,
81
+ expectCols: []string{"durationMs", "backendType", "leaderPid", "queryId"},
82
+ },
83
+ "PG0 defaults to PG14 behavior": {
84
+ pgVersion: 0,
85
+ expectCols: []string{"backendType", "leaderPid", "queryId"},
86
+ },
87
+ }
88
+
89
+ for name, tc := range tests {
90
+ t.Run(name, func(t *testing.T) {
91
+ c := &Collector{pgVersion: tc.pgVersion}
92
+ r := &funcRouter{collector: c}
93
+ f := &funcRunningQueries{router: r}
94
+
95
+ cols := f.getColumnsForVersion()
96
+ colNames := make(map[string]bool)
97
+ for _, col := range cols {
98
+ colNames[col.Name] = true
99
+ }
100
+
101
+ for _, expected := range tc.expectCols {
102
+ assert.True(t, colNames[expected], "expected column %s for PG version %d", expected, tc.pgVersion)
103
+ }
104
+ for _, notExpected := range tc.notExpectCols {
105
+ assert.False(t, colNames[notExpected], "did not expect column %s for PG version %d", notExpected, tc.pgVersion)
106
+ }
107
+ })
108
+ }
109
+}
110
+
111
+func TestFuncRunningQueries_resolveSortColumn(t *testing.T) {
112
+ c := &Collector{pgVersion: pgVersion14}
113
+ r := &funcRouter{collector: c}
114
+ f := &funcRunningQueries{router: r}
115
+ cols := f.getColumnsForVersion()
116
+
117
+ tests := map[string]struct {
118
+ input string
119
+ expected string
120
+ }{
121
+ "valid sort option durationMs": {
122
+ input: "durationMs",
123
+ expected: "durationMs",
124
+ },
125
+ "valid sort option queryStart": {
126
+ input: "queryStart",
127
+ expected: "queryStart",
128
+ },
129
+ "invalid column falls back to default": {
130
+ input: "invalid_column",
131
+ expected: "durationMs",
132
+ },
133
+ "empty string falls back to default": {
134
+ input: "",
135
+ expected: "durationMs",
136
+ },
137
+ "SQL injection attempt falls back to default": {
138
+ input: "'; DROP TABLE users;--",
139
+ expected: "durationMs",
140
+ },
141
+ "non-sortable column falls back to default": {
142
+ input: "query", // query column has sortOpt: false
143
+ expected: "durationMs",
144
+ },
145
+ }
146
+
147
+ for name, tc := range tests {
148
+ t.Run(name, func(t *testing.T) {
149
+ result := f.resolveSortColumn(tc.input, cols)
150
+ assert.Equal(t, tc.expected, result)
151
+ })
152
+ }
153
+}
154
+
155
+func TestFuncRunningQueries_buildQuery(t *testing.T) {
156
+ c := &Collector{pgVersion: pgVersion14}
157
+ r := &funcRouter{collector: c}
158
+ f := &funcRunningQueries{router: r}
159
+ cols := f.getColumnsForVersion()
160
+
161
+ query := f.buildQuery(cols, "durationMs")
162
+
163
+ assert.Contains(t, query, "pg_stat_activity")
164
+ assert.Contains(t, query, "WHERE state = 'active'")
165
+ assert.Contains(t, query, "pid != pg_backend_pid()")
166
+ assert.Contains(t, query, "LIMIT 500")
167
+ assert.Contains(t, query, "ORDER BY")
168
+ assert.Contains(t, query, "DESC NULLS LAST")
169
+ // Check that durationMs expression is used for sorting
170
+ assert.Contains(t, query, "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - query_start)) * 1000")
171
+}
172
+
173
+func TestFuncRunningQueries_buildQuery_SortByQueryStart(t *testing.T) {
174
+ c := &Collector{pgVersion: pgVersion14}
175
+ r := &funcRouter{collector: c}
176
+ f := &funcRunningQueries{router: r}
177
+ cols := f.getColumnsForVersion()
178
+
179
+ query := f.buildQuery(cols, "queryStart")
180
+
181
+ // Should use query_start for ORDER BY
182
+ assert.True(t, strings.Contains(query, "ORDER BY query_start DESC"),
183
+ "expected ORDER BY query_start DESC in query")
184
+}
185
+
186
+func TestFuncRunningQueries_buildSelectClause(t *testing.T) {
187
+ c := &Collector{pgVersion: pgVersion14}
188
+ r := &funcRouter{collector: c}
189
+ f := &funcRunningQueries{router: r}
190
+ cols := f.getColumnsForVersion()
191
+
192
+ selectClause := f.buildSelectClause(cols)
193
+
194
+ // Should contain key DB columns
195
+ assert.Contains(t, selectClause, "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - query_start)) * 1000")
196
+ assert.Contains(t, selectClause, "query")
197
+ assert.Contains(t, selectClause, "datname")
198
+ assert.Contains(t, selectClause, "pid")
199
+ assert.Contains(t, selectClause, "backend_type") // PG10+
200
+ assert.Contains(t, selectClause, "query_id::text") // PG14+
201
+}
202
+
203
+func TestFuncRunningQueries_formatValue(t *testing.T) {
204
+ f := &funcRunningQueries{}
205
+
206
+ tests := map[string]struct {
207
+ value any
208
+ col runningQueriesColumn
209
+ expected any
210
+ }{
211
+ "nil returns nil": {
212
+ value: nil,
213
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeString}},
214
+ expected: nil,
215
+ },
216
+ "string passthrough": {
217
+ value: "test_value",
218
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeString}},
219
+ expected: "test_value",
220
+ },
221
+ "int64 passthrough": {
222
+ value: int64(12345),
223
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeInteger}},
224
+ expected: int64(12345),
225
+ },
226
+ "int32 converts to int64": {
227
+ value: int32(123),
228
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeInteger}},
229
+ expected: int64(123),
230
+ },
231
+ "int converts to int64": {
232
+ value: int(456),
233
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeInteger}},
234
+ expected: int64(456),
235
+ },
236
+ "float64 duration passthrough": {
237
+ value: float64(123.456),
238
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeDuration}},
239
+ expected: float64(123.456),
240
+ },
241
+ "int64 duration converts to float64": {
242
+ value: int64(100),
243
+ col: runningQueriesColumn{ColumnMeta: funcapi.ColumnMeta{Type: funcapi.FieldTypeDuration}},
244
+ expected: float64(100),
245
+ },
246
+ }
247
+
248
+ for name, tc := range tests {
249
+ t.Run(name, func(t *testing.T) {
250
+ result := f.formatValue(tc.value, tc.col)
251
+ assert.Equal(t, tc.expected, result)
252
+ })
253
+ }
254
+}
255
+
256
+func TestFuncRunningQueries_formatValue_QueryTruncation(t *testing.T) {
257
+ f := &funcRunningQueries{}
258
+
259
+ // Create a string longer than runningQueriesMaxTextLength (4096)
260
+ longQuery := strings.Repeat("SELECT * FROM very_long_table_name WHERE id = 1; ", 200)
261
+ assert.Greater(t, len(longQuery), runningQueriesMaxTextLength)
262
+
263
+ col := runningQueriesColumn{
264
+ ColumnMeta: funcapi.ColumnMeta{Name: "query", Type: funcapi.FieldTypeString},
265
+ }
266
+
267
+ result := f.formatValue(longQuery, col)
268
+ resultStr := result.(string)
269
+
270
+ assert.LessOrEqual(t, len(resultStr), runningQueriesMaxTextLength+50, // Allow some buffer for truncation marker
271
+ "query text should be truncated")
272
+}
273
+
274
+func TestFuncRunningQueries_formatValue_NonQueryStringNotTruncated(t *testing.T) {
275
+ f := &funcRunningQueries{}
276
+
277
+ // Long string but NOT the query column
278
+ longValue := strings.Repeat("a", 5000)
279
+
280
+ col := runningQueriesColumn{
281
+ ColumnMeta: funcapi.ColumnMeta{Name: "datname", Type: funcapi.FieldTypeString},
282
+ }
283
+
284
+ result := f.formatValue(longValue, col)
285
+ resultStr := result.(string)
286
+
287
+ assert.Equal(t, len(longValue), len(resultStr), "non-query string columns should not be truncated")
288
+}
289
+
290
+func TestRunningQueriesMethodConfig(t *testing.T) {
291
+ config := runningQueriesMethodConfig()
292
+
293
+ assert.Equal(t, "running-queries", config.ID)
294
+ assert.Equal(t, "Running Queries", config.Name)
295
+ assert.Equal(t, 10, config.UpdateEvery)
296
+ assert.True(t, config.RequireCloud)
297
+ assert.NotEmpty(t, config.Help)
298
+ assert.NotEmpty(t, config.RequiredParams)
299
+}
src/go/plugin/go.d/collector/postgres/func_top_queries.go
+298
-40
@@ -18,7 +18,172 @@ const (
18
paramSort = "__sort"
19
)
20
21
-// pgColumn defines metadata for a pg_stat_statements column.
21
+// queryStatsSourceName is the type for query stats source
22
+type queryStatsSourceName string
23
+
24
+const (
25
+ queryStatsSourcePgStatMonitor queryStatsSourceName = "pg_stat_monitor"
26
+ queryStatsSourcePgStatStatements queryStatsSourceName = "pg_stat_statements"
27
+ queryStatsSourceNone queryStatsSourceName = ""
28
+)
29
+
30
+// getQueryStatsSource detects and returns the best available query stats source.
31
+// Prefers pg_stat_monitor if available, falls back to pg_stat_statements.
32
+// Result is cached after first detection.
33
+func (f *funcTopQueries) getQueryStatsSource(ctx context.Context) (queryStatsSourceName, error) {
34
+ c := f.router.collector
35
+
36
+ // Fast path: return cached result
37
+ c.pgStatStatementsMu.RLock()
38
+ source := c.queryStatsSource
39
+ c.pgStatStatementsMu.RUnlock()
40
+ if source != "" {
41
+ return queryStatsSourceName(source), nil
42
+ }
43
+
44
+ // Slow path: detect and cache
45
+ c.pgStatStatementsMu.Lock()
46
+ defer c.pgStatStatementsMu.Unlock()
47
+
48
+ // Double-check after acquiring write lock
49
+ if c.queryStatsSource != "" {
50
+ return queryStatsSourceName(c.queryStatsSource), nil
51
+ }
52
+
53
+ // Check pg_stat_monitor first (preferred)
54
+ var hasPgStatMonitor bool
55
+ query := `SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_monitor')`
56
+ if err := c.db.QueryRowContext(ctx, query).Scan(&hasPgStatMonitor); err != nil {
57
+ return queryStatsSourceNone, fmt.Errorf("failed to check pg_stat_monitor: %v", err)
58
+ }
59
+ if hasPgStatMonitor {
60
+ c.queryStatsSource = string(queryStatsSourcePgStatMonitor)
61
+ c.pgStatMonitorAvail = true
62
+ return queryStatsSourcePgStatMonitor, nil
63
+ }
64
+
65
+ // Fall back to pg_stat_statements
66
+ var hasPgStatStatements bool
67
+ query = `SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_statements')`
68
+ if err := c.db.QueryRowContext(ctx, query).Scan(&hasPgStatStatements); err != nil {
69
+ return queryStatsSourceNone, fmt.Errorf("failed to check pg_stat_statements: %v", err)
70
+ }
71
+ if hasPgStatStatements {
72
+ c.queryStatsSource = string(queryStatsSourcePgStatStatements)
73
+ c.pgStatStatementsAvail = true
74
+ return queryStatsSourcePgStatStatements, nil
75
+ }
76
+
77
+ return queryStatsSourceNone, nil
78
+}
79
+
80
+// detectPgStatStatementsColumns queries the database to find available columns
81
+func (f *funcTopQueries) detectPgStatStatementsColumns(ctx context.Context) (map[string]bool, error) {
82
+ c := f.router.collector
83
+
84
+ // Fast path: return cached result
85
+ c.pgStatStatementsMu.RLock()
86
+ if c.pgStatStatementsColumns != nil {
87
+ cols := c.pgStatStatementsColumns
88
+ c.pgStatStatementsMu.RUnlock()
89
+ return cols, nil
90
+ }
91
+ c.pgStatStatementsMu.RUnlock()
92
+
93
+ // Slow path: query and cache
94
+ c.pgStatStatementsMu.Lock()
95
+ defer c.pgStatStatementsMu.Unlock()
96
+
97
+ // Double-check after acquiring write lock
98
+ if c.pgStatStatementsColumns != nil {
99
+ return c.pgStatStatementsColumns, nil
100
+ }
101
+
102
+ // Query available columns from pg_stat_statements
103
+ query := `
104
+ SELECT column_name
105
+ FROM information_schema.columns
106
+ WHERE table_name = 'pg_stat_statements'
107
+ AND table_schema = 'public'
108
+ `
109
+ rows, err := c.db.QueryContext(ctx, query)
110
+ if err != nil {
111
+ return nil, fmt.Errorf("failed to query columns: %v", err)
112
+ }
113
+ defer rows.Close()
114
+
115
+ cols := make(map[string]bool)
116
+ for rows.Next() {
117
+ var colName string
118
+ if err := rows.Scan(&colName); err != nil {
119
+ return nil, fmt.Errorf("failed to scan column name: %v", err)
120
+ }
121
+ cols[colName] = true
122
+ }
123
+
124
+ if err := rows.Err(); err != nil {
125
+ return nil, fmt.Errorf("rows iteration error: %v", err)
126
+ }
127
+
128
+ // Cache the result
129
+ c.pgStatStatementsColumns = cols
130
+ return cols, nil
131
+}
132
+
133
+// detectPgStatMonitorColumns queries the database to find available columns
134
+func (f *funcTopQueries) detectPgStatMonitorColumns(ctx context.Context) (map[string]bool, error) {
135
+ c := f.router.collector
136
+
137
+ // Fast path: return cached result
138
+ c.pgStatStatementsMu.RLock()
139
+ if c.pgStatMonitorColumns != nil {
140
+ cols := c.pgStatMonitorColumns
141
+ c.pgStatStatementsMu.RUnlock()
142
+ return cols, nil
143
+ }
144
+ c.pgStatStatementsMu.RUnlock()
145
+
146
+ // Slow path: query and cache
147
+ c.pgStatStatementsMu.Lock()
148
+ defer c.pgStatStatementsMu.Unlock()
149
+
150
+ // Double-check after acquiring write lock
151
+ if c.pgStatMonitorColumns != nil {
152
+ return c.pgStatMonitorColumns, nil
153
+ }
154
+
155
+ // Query available columns from pg_stat_monitor
156
+ query := `
157
+ SELECT column_name
158
+ FROM information_schema.columns
159
+ WHERE table_name = 'pg_stat_monitor'
160
+ AND table_schema = 'public'
161
+ `
162
+ rows, err := c.db.QueryContext(ctx, query)
163
+ if err != nil {
164
+ return nil, fmt.Errorf("failed to query columns: %v", err)
165
+ }
166
+ defer rows.Close()
167
+
168
+ cols := make(map[string]bool)
169
+ for rows.Next() {
170
+ var colName string
171
+ if err := rows.Scan(&colName); err != nil {
172
+ return nil, fmt.Errorf("failed to scan column name: %v", err)
173
+ }
174
+ cols[colName] = true
175
+ }
176
+
177
+ if err := rows.Err(); err != nil {
178
+ return nil, fmt.Errorf("rows iteration error: %v", err)
179
+ }
180
+
181
+ // Cache the result
182
+ c.pgStatMonitorColumns = cols
183
+ return cols, nil
184
+}
185
+
186
+// pgColumn defines metadata for a pg_stat_statements/pg_stat_monitor column.
187
// Embeds funcapi.ColumnMeta for UI rendering and adds PG-specific fields.
188
type pgColumn struct {
189
funcapi.ColumnMeta
@@ -31,6 +196,8 @@ type pgColumn struct {
196
SortLabel string
197
// IsDefaultSort indicates whether this is the default sort column
198
IsDefaultSort bool
199
+ // OnlyPgStatMonitor indicates this column only exists in pg_stat_monitor
200
+ OnlyPgStatMonitor bool
201
}
202
203
// pgColumnSet creates a ColumnSet from a slice of pgColumn.
@@ -108,6 +275,20 @@ var pgAllColumns = []pgColumn{
275
// Temp file statistics (PG 15+ only)
276
{ColumnMeta: funcapi.ColumnMeta{Name: "tempBlkReadTime", Tooltip: "Temp Block Read Time", Type: funcapi.FieldTypeDuration, Units: "milliseconds", Visible: false, Transform: funcapi.FieldTransformDuration, DecimalPoints: 2, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummarySum, Filter: funcapi.FieldFilterRange, Sortable: true}, DBColumn: "s.temp_blk_read_time"},
277
{ColumnMeta: funcapi.ColumnMeta{Name: "tempBlkWriteTime", Tooltip: "Temp Block Write Time", Type: funcapi.FieldTypeDuration, Units: "milliseconds", Visible: false, Transform: funcapi.FieldTransformDuration, DecimalPoints: 2, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummarySum, Filter: funcapi.FieldFilterRange, Sortable: true}, DBColumn: "s.temp_blk_write_time"},
278
+
279
+ // pg_stat_monitor-specific columns (only available with pg_stat_monitor extension)
280
+ {ColumnMeta: funcapi.ColumnMeta{Name: "applicationName", Tooltip: "Application Name", Type: funcapi.FieldTypeString, Visible: true, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount, Filter: funcapi.FieldFilterMultiselect, Sortable: true}, DBColumn: "s.application_name", OnlyPgStatMonitor: true},
281
+ {ColumnMeta: funcapi.ColumnMeta{Name: "clientIp", Tooltip: "Client IP Address", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount, Filter: funcapi.FieldFilterMultiselect, Sortable: true}, DBColumn: "s.client_ip::text", OnlyPgStatMonitor: true},
282
+ {ColumnMeta: funcapi.ColumnMeta{Name: "cmdType", Tooltip: "Query Type", Type: funcapi.FieldTypeString, Visible: true, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Summary: funcapi.FieldSummaryCount, Filter: funcapi.FieldFilterMultiselect, Visualization: funcapi.FieldVisualPill, Sortable: true}, DBColumn: "s.cmd_type_text", OnlyPgStatMonitor: true},
283
+ {ColumnMeta: funcapi.ColumnMeta{Name: "comments", Tooltip: "Query Comments", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Sortable: true}, DBColumn: "s.comments", OnlyPgStatMonitor: true},
284
+ {ColumnMeta: funcapi.ColumnMeta{Name: "relations", Tooltip: "Involved Tables", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Sortable: true}, DBColumn: "array_to_string(s.relations, ', ')", OnlyPgStatMonitor: true},
285
+ {ColumnMeta: funcapi.ColumnMeta{Name: "cpuUserTime", Tooltip: "User CPU Time", Type: funcapi.FieldTypeDuration, Units: "milliseconds", Visible: false, Transform: funcapi.FieldTransformDuration, DecimalPoints: 2, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummarySum, Filter: funcapi.FieldFilterRange, Sortable: true}, DBColumn: "s.cpu_user_time", IsSortOption: true, SortLabel: "User CPU Time", OnlyPgStatMonitor: true},
286
+ {ColumnMeta: funcapi.ColumnMeta{Name: "cpuSysTime", Tooltip: "System CPU Time", Type: funcapi.FieldTypeDuration, Units: "milliseconds", Visible: false, Transform: funcapi.FieldTransformDuration, DecimalPoints: 2, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummarySum, Filter: funcapi.FieldFilterRange, Sortable: true}, DBColumn: "s.cpu_sys_time", IsSortOption: true, SortLabel: "System CPU Time", OnlyPgStatMonitor: true},
287
+ {ColumnMeta: funcapi.ColumnMeta{Name: "elevel", Tooltip: "Error Level", Type: funcapi.FieldTypeInteger, Visible: false, Transform: funcapi.FieldTransformNumber, Sort: funcapi.FieldSortDescending, Summary: funcapi.FieldSummaryMax, Filter: funcapi.FieldFilterRange, Sortable: true}, DBColumn: "s.elevel", OnlyPgStatMonitor: true},
288
+ {ColumnMeta: funcapi.ColumnMeta{Name: "sqlcode", Tooltip: "SQL Error Code", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Filter: funcapi.FieldFilterMultiselect, Sortable: true}, DBColumn: "s.sqlcode", OnlyPgStatMonitor: true},
289
+ {ColumnMeta: funcapi.ColumnMeta{Name: "message", Tooltip: "Error Message", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, FullWidth: true, Sortable: true}, DBColumn: "s.message", OnlyPgStatMonitor: true},
290
+ {ColumnMeta: funcapi.ColumnMeta{Name: "toplevel", Tooltip: "Top-level Statement", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortAscending, Filter: funcapi.FieldFilterMultiselect, Sortable: true}, DBColumn: "s.toplevel::text", OnlyPgStatMonitor: true},
291
+ {ColumnMeta: funcapi.ColumnMeta{Name: "bucketStartTime", Tooltip: "Bucket Start Time", Type: funcapi.FieldTypeString, Visible: false, Transform: funcapi.FieldTransformNone, Sort: funcapi.FieldSortDescending, Sortable: true}, DBColumn: "s.bucket_start_time::text", OnlyPgStatMonitor: true},
292
}
293
294
// pgChartGroupDefs defines chart groupings for columns. These are applied at runtime via decoratePgColumns.
@@ -131,12 +312,17 @@ var pgChartGroupDefs = []struct {
312
{key: "JITCounts", title: "JIT Counts", columns: []string{"jitFunctions", "jitInliningCount", "jitOptimizationCount", "jitEmissionCount"}},
313
{key: "JITTime", title: "JIT Time", columns: []string{"jitGenerationTime", "jitInliningTime", "jitOptimizationTime", "jitEmissionTime"}},
314
{key: "TempIOTime", title: "Temp Block I/O Time", columns: []string{"tempBlkReadTime", "tempBlkWriteTime"}},
315
+ // pg_stat_monitor-specific chart groups
316
+ {key: "CPUTime", title: "CPU Time", columns: []string{"cpuUserTime", "cpuSysTime"}},
317
+ {key: "Errors", title: "Error Info", columns: []string{"elevel", "sqlcode", "message"}},
318
}
319
320
// pgLabelColumnIDs defines which columns are available for group-by.
321
var pgLabelColumnIDs = map[string]bool{
138
- "database": true,
139
- "user": true,
322
+ "database": true,
323
+ "user": true,
324
+ "applicationName": true, // pg_stat_monitor only
325
+ "cmdType": true, // pg_stat_monitor only
326
}
327
328
const pgPrimaryLabelID = "database"
@@ -215,40 +401,54 @@ func buildPgSortOptions() []funcapi.ParamOption {
401
return opts
402
}
403
218
-// collectTopQueries queries pg_stat_statements for top queries.
404
+// collectTopQueries queries pg_stat_statements or pg_stat_monitor for top queries.
405
+// It auto-detects pg_stat_monitor and uses it when available, falling back to pg_stat_statements.
406
func (f *funcTopQueries) collectTopQueries(ctx context.Context, sortColumn string) *funcapi.FunctionResponse {
407
c := f.router.collector
408
222
- // Check pg_stat_statements availability (lazy check)
223
- available, err := c.checkPgStatStatements(ctx)
409
+ // Auto-detect best available query stats source
410
+ source, err := f.getQueryStatsSource(ctx)
411
if err != nil {
225
- return funcapi.InternalErrorResponse("failed to check pg_stat_statements availability: %v", err)
412
+ return funcapi.InternalErrorResponse("failed to detect query stats source: %v", err)
413
}
227
- if !available {
228
- return funcapi.UnavailableResponse("pg_stat_statements extension is not installed in this database. " +
229
- "Run 'CREATE EXTENSION pg_stat_statements;' in the database the collector connects to.")
414
+ if source == queryStatsSourceNone {
415
+ return funcapi.UnavailableResponse("No query statistics extension is installed in this database. " +
416
+ "Install pg_stat_monitor (recommended) or pg_stat_statements:\n\n" +
417
+ "For pg_stat_monitor:\n" +
418
+ " ALTER SYSTEM SET shared_preload_libraries = 'pg_stat_monitor';\n" +
419
+ " -- restart PostgreSQL\n" +
420
+ " CREATE EXTENSION pg_stat_monitor;\n\n" +
421
+ "For pg_stat_statements:\n" +
422
+ " ALTER SYSTEM SET shared_preload_libraries = 'pg_stat_statements';\n" +
423
+ " -- restart PostgreSQL\n" +
424
+ " CREATE EXTENSION pg_stat_statements;")
425
}
426
232
- // Detect available columns (lazy detection, cached)
233
- availableCols, err := c.detectPgStatStatementsColumns(ctx)
427
+ // Detect available columns based on source
428
+ var availableCols map[string]bool
429
+ if source == queryStatsSourcePgStatMonitor {
430
+ availableCols, err = f.detectPgStatMonitorColumns(ctx)
431
+ } else {
432
+ availableCols, err = f.detectPgStatStatementsColumns(ctx)
433
+ }
434
if err != nil {
435
return funcapi.InternalErrorResponse("failed to detect available columns: %v", err)
436
}
437
238
- // Build list of columns to query based on what's available
239
- queryCols := f.buildAvailableColumns(availableCols)
438
+ // Build list of columns to query based on what's available and source
439
+ queryCols := f.buildAvailableColumns(availableCols, source)
440
if len(queryCols) == 0 {
241
- return funcapi.InternalErrorResponse("no queryable columns found in pg_stat_statements")
441
+ return funcapi.InternalErrorResponse("no queryable columns found in %s", source)
442
}
443
444
// Map and validate sort column
245
- actualSortCol := f.mapAndValidateSortColumn(sortColumn, availableCols)
445
+ actualSortCol := f.mapAndValidateSortColumn(sortColumn, availableCols, source)
446
447
// Get query limit (default 500)
448
limit := c.topQueriesLimit()
449
450
// Build and execute query
251
- query := f.buildDynamicSQL(queryCols, actualSortCol, limit)
451
+ query := f.buildDynamicSQL(queryCols, actualSortCol, limit, source)
452
rows, err := c.db.QueryContext(ctx, query)
453
if err != nil {
454
if ctx.Err() == context.DeadlineExceeded {
@@ -288,9 +488,15 @@ func (f *funcTopQueries) collectTopQueries(ctx context.Context, sortColumn strin
488
annotatedCols := decoratePgColumns(queryCols)
489
cs := pgColumnSet(annotatedCols)
490
491
+ // Build help message based on source
492
+ helpMsg := "Top SQL queries from pg_stat_statements"
493
+ if source == queryStatsSourcePgStatMonitor {
494
+ helpMsg = "Top SQL queries from pg_stat_monitor (includes application, client IP, CPU time, and error info)"
495
+ }
496
+
497
return &funcapi.FunctionResponse{
498
Status: 200,
293
- Help: "Top SQL queries from pg_stat_statements",
499
+ Help: helpMsg,
500
Columns: cs.BuildColumns(),
501
Data: data,
502
DefaultSortColumn: defaultSort,
@@ -335,26 +541,45 @@ func decoratePgColumns(cols []pgColumn) []pgColumn {
541
return out
542
}
543
338
-// buildAvailableColumns returns column metadata for columns that exist in this PG version.
339
-func (f *funcTopQueries) buildAvailableColumns(availableCols map[string]bool) []pgColumn {
544
+// buildAvailableColumns returns column metadata for columns that exist in this PG version and source.
545
+func (f *funcTopQueries) buildAvailableColumns(availableCols map[string]bool, source queryStatsSourceName) []pgColumn {
546
c := f.router.collector
547
var result []pgColumn
548
+ isPgStatMonitor := source == queryStatsSourcePgStatMonitor
549
550
for _, col := range pgAllColumns {
344
- // Extract the actual column name (remove table prefix and type cast)
551
+ // Skip pg_stat_monitor-only columns when using pg_stat_statements
552
+ if col.OnlyPgStatMonitor && !isPgStatMonitor {
553
+ continue
554
+ }
555
+
556
+ // Extract the actual column name for availability check
557
colName := col.DBColumn
558
+
559
+ // Strip array_to_string wrapper FIRST (before table prefix removal)
560
+ // e.g., "array_to_string(s.relations, ', ')" -> "s.relations"
561
+ if strings.HasPrefix(colName, "array_to_string(") {
562
+ colName = strings.TrimPrefix(colName, "array_to_string(")
563
+ if idx := strings.Index(colName, ","); idx != -1 {
564
+ colName = colName[:idx]
565
+ }
566
+ }
567
+
568
+ // Remove table prefix (e.g., "s.relations" -> "relations")
569
if idx := strings.LastIndex(colName, "."); idx != -1 {
570
colName = colName[idx+1:]
571
}
349
- // Remove PostgreSQL type cast suffix (e.g., "::text")
572
+
573
+ // Remove PostgreSQL type cast suffix (e.g., "queryid::text" -> "queryid")
574
if idx := strings.Index(colName, "::"); idx != -1 {
575
colName = colName[:idx]
576
}
577
354
- // Handle version-specific column names for time columns
355
- // PG 13+ renamed time columns: total_time -> total_exec_time, etc.
578
+ // Handle version-specific column names for time columns.
579
+ // pg_stat_statements PG 13+ and pg_stat_monitor both use: total_exec_time, mean_exec_time, etc.
580
+ // pg_stat_statements < PG 13 uses: total_time, mean_time, etc.
581
actualColName := colName
357
- if c.pgVersion >= pgVersion13 {
582
+ if isPgStatMonitor || c.pgVersion >= pgVersion13 {
583
switch colName {
584
case "total_time":
585
actualColName = "total_exec_time"
@@ -371,7 +596,12 @@ func (f *funcTopQueries) buildAvailableColumns(availableCols map[string]bool) []
596
597
// Check if column exists (either directly or via join)
598
// Join columns (database, user) come from other tables (d.datname, u.usename)
599
+ // pg_stat_monitor has datname directly, pg_stat_statements needs join
600
isJoinCol := col.Name == "database" || col.Name == "user"
601
+ if isPgStatMonitor && col.Name == "database" {
602
+ // pg_stat_monitor has datname directly
603
+ isJoinCol = false
604
+ }
605
if isJoinCol || availableCols[actualColName] {
606
// Create a copy with the actual column name for this version
607
colCopy := col
@@ -381,6 +611,10 @@ func (f *funcTopQueries) buildAvailableColumns(availableCols map[string]bool) []
611
colCopy.DBColumn = "s." + actualColName
612
}
613
}
614
+ // For pg_stat_monitor, database comes from s.datname directly
615
+ if isPgStatMonitor && col.Name == "database" {
616
+ colCopy.DBColumn = "s.datname"
617
+ }
618
result = append(result, colCopy)
619
}
620
}
@@ -389,8 +623,9 @@ func (f *funcTopQueries) buildAvailableColumns(availableCols map[string]bool) []
623
}
624
625
// mapAndValidateSortColumn maps the semantic sort column to actual SQL column.
392
-func (f *funcTopQueries) mapAndValidateSortColumn(sortColumn string, availableCols map[string]bool) string {
626
+func (f *funcTopQueries) mapAndValidateSortColumn(sortColumn string, availableCols map[string]bool, source queryStatsSourceName) string {
627
c := f.router.collector
628
+ isPgStatMonitor := source == queryStatsSourcePgStatMonitor
629
630
// Map column ID back to DBColumn
631
for _, col := range pgAllColumns {
@@ -404,8 +639,10 @@ func (f *funcTopQueries) mapAndValidateSortColumn(sortColumn string, availableCo
639
colName = colName[:idx]
640
}
641
407
- // Handle version-specific mapping
408
- if c.pgVersion >= pgVersion13 {
642
+ // Handle version-specific mapping for time columns.
643
+ // pg_stat_statements PG 13+ and pg_stat_monitor both use: total_exec_time, mean_exec_time, etc.
644
+ // pg_stat_statements < PG 13 uses: total_time, mean_time, etc.
645
+ if isPgStatMonitor || c.pgVersion >= pgVersion13 {
646
switch colName {
647
case "total_time":
648
colName = "total_exec_time"
@@ -428,22 +665,24 @@ func (f *funcTopQueries) mapAndValidateSortColumn(sortColumn string, availableCo
665
}
666
667
// Default fallback
431
- if c.pgVersion >= pgVersion13 {
668
+ if isPgStatMonitor || c.pgVersion >= pgVersion13 {
669
return "total_exec_time"
670
}
671
return "total_time"
672
}
673
674
// buildDynamicSQL builds the SQL query with only available columns.
438
-func (f *funcTopQueries) buildDynamicSQL(cols []pgColumn, sortColumn string, limit int) string {
675
+func (f *funcTopQueries) buildDynamicSQL(cols []pgColumn, sortColumn string, limit int, source queryStatsSourceName) string {
676
c := f.router.collector
677
var selectCols []string
678
679
for _, col := range cols {
680
colExpr := col.DBColumn
681
445
- // Handle version-specific column names
446
- if c.pgVersion >= pgVersion13 {
682
+ // Handle version-specific column names for time columns.
683
+ // pg_stat_statements PG 13+ and pg_stat_monitor both use: total_exec_time, mean_exec_time, etc.
684
+ // pg_stat_statements < PG 13 uses: total_time, mean_time, etc.
685
+ if source == queryStatsSourcePgStatMonitor || c.pgVersion >= pgVersion13 {
686
switch {
687
case strings.HasSuffix(colExpr, ".total_time"):
688
colExpr = strings.Replace(colExpr, ".total_time", ".total_exec_time", 1)
@@ -473,6 +712,19 @@ func (f *funcTopQueries) buildDynamicSQL(cols []pgColumn, sortColumn string, lim
712
selectCols = append(selectCols, fmt.Sprintf("%s AS \"%s\"", colExpr, col.Name))
713
}
714
715
+ // Build query based on source
716
+ if source == queryStatsSourcePgStatMonitor {
717
+ // pg_stat_monitor has datname and username columns directly
718
+ return fmt.Sprintf(`
719
+SELECT %s
720
+FROM pg_stat_monitor s
721
+JOIN pg_user u ON s.userid = u.usesysid
722
+ORDER BY "%s" DESC
723
+LIMIT %d
724
+`, strings.Join(selectCols, ", "), sortColumn, limit)
725
+ }
726
+
727
+ // pg_stat_statements needs joins for database and user names
728
return fmt.Sprintf(`
729
SELECT %s
730
FROM pg_stat_statements s
@@ -588,27 +840,33 @@ func (f *funcTopQueries) topQueriesSortParam(queryCols []pgColumn) (funcapi.Para
840
}
841
842
func (f *funcTopQueries) topQueriesParams(ctx context.Context) ([]funcapi.ParamConfig, error) {
591
- c := f.router.collector
592
-
593
- available, err := c.checkPgStatStatements(ctx)
843
+ // Auto-detect best available query stats source
844
+ source, err := f.getQueryStatsSource(ctx)
845
if err != nil {
846
return nil, err
847
}
597
- if !available {
598
- return nil, fmt.Errorf("pg_stat_statements extension is not installed")
848
+ if source == queryStatsSourceNone {
849
+ return nil, fmt.Errorf("no query statistics extension is installed (pg_stat_monitor or pg_stat_statements)")
850
}
851
601
- availableCols, err := c.detectPgStatStatementsColumns(ctx)
852
+ // Detect available columns based on source
853
+ var availableCols map[string]bool
854
+ if source == queryStatsSourcePgStatMonitor {
855
+ availableCols, err = f.detectPgStatMonitorColumns(ctx)
856
+ } else {
857
+ availableCols, err = f.detectPgStatStatementsColumns(ctx)
858
+ }
859
if err != nil {
860
return nil, err
861
}
862
606
- queryCols := f.buildAvailableColumns(availableCols)
863
+ queryCols := f.buildAvailableColumns(availableCols, source)
864
if len(queryCols) == 0 {
608
- return nil, fmt.Errorf("no queryable columns found in pg_stat_statements")
865
+ return nil, fmt.Errorf("no queryable columns found in %s", source)
866
}
867
868
sortParam, _ := f.topQueriesSortParam(queryCols)
869
+
870
return []funcapi.ParamConfig{sortParam}, nil
871
}
872
src/go/plugin/go.d/collector/postgres/func_top_queries_test.go
renamed
+13
-6
@@ -16,10 +16,12 @@ func TestPgMethods(t *testing.T) {
16
methods := pgMethods()
17
18
require := assert.New(t)
19
- require.Len(methods, 1)
19
+ require.Len(methods, 2)
20
require.Equal("top-queries", methods[0].ID)
21
require.Equal("Top Queries", methods[0].Name)
22
require.NotEmpty(methods[0].RequiredParams)
23
+ require.Equal("running-queries", methods[1].ID)
24
+ require.Equal("Running Queries", methods[1].Name)
25
26
// Verify at least one default sort option exists
27
var sortParam *funcapi.ParamConfig
@@ -131,7 +133,7 @@ func TestFuncTopQueries_mapAndValidateSortColumn(t *testing.T) {
133
c := &Collector{pgVersion: tc.pgVersion}
134
r := &funcRouter{collector: c}
135
f := &funcTopQueries{router: r}
134
- result := f.mapAndValidateSortColumn(tc.input, tc.availableCols)
136
+ result := f.mapAndValidateSortColumn(tc.input, tc.availableCols, queryStatsSourcePgStatStatements)
137
assert.Equal(t, tc.expected, result)
138
})
139
}
@@ -171,7 +173,7 @@ func TestFuncTopQueries_buildAvailableColumns(t *testing.T) {
173
c := &Collector{pgVersion: tc.pgVersion}
174
r := &funcRouter{collector: c}
175
f := &funcTopQueries{router: r}
174
- cols := f.buildAvailableColumns(tc.availableCols)
176
+ cols := f.buildAvailableColumns(tc.availableCols, queryStatsSourcePgStatStatements)
177
cs := pgColumnSet(cols)
178
179
for _, id := range tc.expectCols {
@@ -216,7 +218,7 @@ func TestFuncTopQueries_buildDynamicSQL(t *testing.T) {
218
{ColumnMeta: funcapi.ColumnMeta{Name: "totalTime", Type: funcapi.FieldTypeDuration}, DBColumn: "total_time"},
219
}
220
219
- sql := f.buildDynamicSQL(cols, tc.sortColumn, 500)
221
+ sql := f.buildDynamicSQL(cols, tc.sortColumn, 500, queryStatsSourcePgStatStatements)
222
223
assert.Contains(t, sql, "pg_stat_statements")
224
assert.Contains(t, sql, tc.sortColumn)
@@ -275,11 +277,16 @@ func TestPgMethods_SortOptionsHaveLabels(t *testing.T) {
277
break
278
}
279
}
278
- assert.NotNil(t, sortParam)
280
+ if sortParam == nil {
281
+ continue // Some methods may not have sort params
282
+ }
283
for _, opt := range sortParam.Options {
284
assert.NotEmpty(t, opt.ID, "sort option must have ID")
285
assert.NotEmpty(t, opt.Name, "sort option %s must have Name", opt.ID)
282
- assert.Contains(t, opt.Name, "Top queries by", "label should have standard prefix")
286
+ // Top-queries uses "Top queries by X", running-queries uses plain labels
287
+ if method.ID == "top-queries" {
288
+ assert.Contains(t, opt.Name, "Top queries by", "label should have standard prefix for top-queries")
289
+ }
290
}
291
}
292
}
src/go/plugin/go.d/collector/postgres/functions.go
deleted
-89
@@ -1,89 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-package postgres
4
-
5
-import (
6
- "context"
7
- "fmt"
8
-)
9
-
10
-// detectPgStatStatementsColumns queries the database to find available columns
11
-func (c *Collector) detectPgStatStatementsColumns(ctx context.Context) (map[string]bool, error) {
12
- // Fast path: return cached result
13
- c.pgStatStatementsMu.RLock()
14
- if c.pgStatStatementsColumns != nil {
15
- cols := c.pgStatStatementsColumns
16
- c.pgStatStatementsMu.RUnlock()
17
- return cols, nil
18
- }
19
- c.pgStatStatementsMu.RUnlock()
20
-
21
- // Slow path: query and cache
22
- c.pgStatStatementsMu.Lock()
23
- defer c.pgStatStatementsMu.Unlock()
24
-
25
- // Double-check after acquiring write lock
26
- if c.pgStatStatementsColumns != nil {
27
- return c.pgStatStatementsColumns, nil
28
- }
29
-
30
- // Query available columns from pg_stat_statements
31
- query := `
32
- SELECT column_name
33
- FROM information_schema.columns
34
- WHERE table_name = 'pg_stat_statements'
35
- AND table_schema = 'public'
36
- `
37
- rows, err := c.db.QueryContext(ctx, query)
38
- if err != nil {
39
- return nil, fmt.Errorf("failed to query columns: %v", err)
40
- }
41
- defer rows.Close()
42
-
43
- cols := make(map[string]bool)
44
- for rows.Next() {
45
- var colName string
46
- if err := rows.Scan(&colName); err != nil {
47
- return nil, fmt.Errorf("failed to scan column name: %v", err)
48
- }
49
- cols[colName] = true
50
- }
51
-
52
- if err := rows.Err(); err != nil {
53
- return nil, fmt.Errorf("rows iteration error: %v", err)
54
- }
55
-
56
- // Cache the result
57
- c.pgStatStatementsColumns = cols
58
- return cols, nil
59
-}
60
-
61
-// checkPgStatStatements checks if pg_stat_statements extension is available
62
-// Only positive results are cached - negative results are re-checked each time
63
-// so users don't need to restart after installing the extension
64
-func (c *Collector) checkPgStatStatements(ctx context.Context) (bool, error) {
65
- // Fast path: return cached positive result
66
- c.pgStatStatementsMu.RLock()
67
- avail := c.pgStatStatementsAvail
68
- c.pgStatStatementsMu.RUnlock()
69
- if avail {
70
- return true, nil
71
- }
72
-
73
- // Slow path: query the database
74
- var exists bool
75
- query := `SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_statements')`
76
- err := c.db.QueryRowContext(ctx, query).Scan(&exists)
77
- if err != nil {
78
- return false, err
79
- }
80
-
81
- // Only cache positive results
82
- if exists {
83
- c.pgStatStatementsMu.Lock()
84
- c.pgStatStatementsAvail = true
85
- c.pgStatStatementsMu.Unlock()
86
- }
87
-
88
- return exists, nil
89
-}
src/go/plugin/go.d/collector/postgres/metadata.yaml
+241
-13
@@ -245,14 +245,19 @@ modules:
245
- id: top-queries
246
name: Top Queries
247
description: |
248
- Retrieves aggregated SQL query performance metrics from PostgreSQL [pg_stat_statements](https://www.postgresql.org/docs/current/pgstatstatements.html) extension.
248
+ Retrieves aggregated SQL query performance metrics from PostgreSQL using either [pg_stat_monitor](https://docs.percona.com/pg-stat-monitor/) (preferred) or [pg_stat_statements](https://www.postgresql.org/docs/current/pgstatstatements.html).
249
250
- This function queries `pg_stat_statements` which tracks execution statistics for all SQL statements. Statistics include execution counts, timing metrics, I/O operations, and resource consumption. Columns are dynamically detected based on your PostgreSQL version.
250
+ The collector automatically detects which extension is available:
251
+ - **pg_stat_monitor** (Percona): Enhanced statistics with additional columns like application name, client IP, CPU time, error info, and query classification
252
+ - **pg_stat_statements** (standard): Core execution statistics available in all PostgreSQL installations
253
+
254
+ Statistics include execution counts, timing metrics, I/O operations, and resource consumption. Columns are dynamically detected based on your PostgreSQL version and available extension.
255
256
Use cases:
257
- Identify slow queries consuming the most total execution time
258
- Find queries with high shared block reads for I/O optimization
259
- Analyze temp block usage to detect queries needing memory tuning
260
+ - With pg_stat_monitor: Track queries by application, identify error patterns
261
262
Query text is truncated at 4096 characters for display purposes.
263
parameters:
@@ -264,7 +269,7 @@ modules:
269
default: totalTime
270
options: []
271
returns:
267
- description: Aggregated query statistics from `pg_stat_statements`. Each row represents a unique query pattern with cumulative metrics across all executions.
272
+ description: Aggregated query statistics from `pg_stat_statements` or `pg_stat_monitor`. Each row represents a unique query pattern with cumulative metrics across all executions.
273
columns:
274
- name: Query ID
275
type: string
@@ -463,15 +468,75 @@ modules:
468
unit: "milliseconds"
469
visibility: hidden
470
description: "Time spent writing temp blocks. Available in PostgreSQL 15+. Requires `track_io_timing`."
471
+ - name: Application Name
472
+ type: string
473
+ unit: ""
474
+ description: "Name of the application that executed the query. Available with pg_stat_monitor only."
475
+ - name: Client IP
476
+ type: string
477
+ unit: ""
478
+ visibility: hidden
479
+ description: "IP address of the client that executed the query. Available with pg_stat_monitor only."
480
+ - name: Command Type
481
+ type: string
482
+ unit: ""
483
+ description: "Type of SQL command (SELECT, INSERT, UPDATE, DELETE, etc.). Available with pg_stat_monitor only."
484
+ - name: Comments
485
+ type: string
486
+ unit: ""
487
+ visibility: hidden
488
+ description: "SQL comments extracted from the query. Available with pg_stat_monitor only."
489
+ - name: Relations
490
+ type: string
491
+ unit: ""
492
+ visibility: hidden
493
+ description: "Tables/relations involved in the query. Available with pg_stat_monitor only."
494
+ - name: CPU User Time
495
+ type: duration
496
+ unit: "milliseconds"
497
+ visibility: hidden
498
+ description: "CPU time spent in user mode. Available with pg_stat_monitor only."
499
+ - name: CPU System Time
500
+ type: duration
501
+ unit: "milliseconds"
502
+ visibility: hidden
503
+ description: "CPU time spent in system/kernel mode. Available with pg_stat_monitor only."
504
+ - name: Error Level
505
+ type: integer
506
+ unit: ""
507
+ visibility: hidden
508
+ description: "PostgreSQL error level if query produced an error. Available with pg_stat_monitor only."
509
+ - name: SQL Code
510
+ type: string
511
+ unit: ""
512
+ visibility: hidden
513
+ description: "PostgreSQL SQLSTATE error code if query produced an error. Available with pg_stat_monitor only."
514
+ - name: Error Message
515
+ type: string
516
+ unit: ""
517
+ visibility: hidden
518
+ description: "Error message if query produced an error. Available with pg_stat_monitor only."
519
+ - name: Top Level
520
+ type: string
521
+ unit: ""
522
+ visibility: hidden
523
+ description: "Whether this is a top-level statement (true) or nested (false). Available with pg_stat_monitor only."
524
+ - name: Bucket Start Time
525
+ type: string
526
+ unit: ""
527
+ visibility: hidden
528
+ description: "Start time of the statistics bucket. Available with pg_stat_monitor only."
529
performance: |
467
- Queries `pg_stat_statements` which maintains statistics in shared memory:<br/>• On busy servers with many unique queries, the extension may consume significant memory<br/>• Default limit of 500 rows balances usefulness with performance
530
+ Queries `pg_stat_statements` or `pg_stat_monitor` which maintain statistics in shared memory:<br/>• On busy servers with many unique queries, the extension may consume significant memory<br/>• Default limit of 500 rows balances usefulness with performance<br/>• pg_stat_monitor uses time-based buckets which may have different memory characteristics
531
security: |
532
Query text may contain unmasked literal values including potentially sensitive data:<br/>• Personal information in WHERE clauses or INSERT values<br/>• Business data and internal identifiers<br/>• Access should be restricted to authorized personnel only
533
prerequisites:
534
list:
472
- - title: Enable pg_stat_statements
535
+ - title: Enable pg_stat_statements or pg_stat_monitor
536
description: |
474
- The `pg_stat_statements` extension must be installed and configured.
537
+ Either `pg_stat_statements` (standard) or `pg_stat_monitor` (Percona) must be installed. The collector auto-detects which is available, preferring pg_stat_monitor when both are present.
538
+
539
+ **Option 1: pg_stat_statements (standard PostgreSQL)**
540
541
1. Add to `postgresql.conf`:
542
@@ -485,22 +550,185 @@ modules:
550
CREATE EXTENSION pg_stat_statements;
551
```
552
488
- 3. Verify the extension is working:
553
+ **Option 2: pg_stat_monitor (Percona - recommended)**
554
+
555
+ Provides additional columns: application name, client IP, CPU time, error tracking, and query classification.
556
+
557
+ 1. Install pg_stat_monitor (available in Percona distribution or as separate package)
558
+
559
+ 2. Add to `postgresql.conf`:
560
+
561
+ ```ini
562
+ shared_preload_libraries = 'pg_stat_monitor'
563
+ ```
564
+
565
+ 3. Restart PostgreSQL, then create the extension:
566
567
```sql
491
- SELECT COUNT(*) FROM pg_stat_statements;
568
+ CREATE EXTENSION pg_stat_monitor;
569
```
570
571
:::info
572
496
- - `pg_stat_statements` requires a server restart to load the shared library
497
- - Statistics can be reset with `SELECT pg_stat_statements_reset()`
498
- - The `pg_stat_statements.max` parameter controls maximum tracked statements (default 5000)
499
- - Enable `track_io_timing` for block read/write timing metrics (may add slight overhead)
573
+ - Both extensions require a server restart to load the shared library
574
+ - Statistics can be reset with `SELECT pg_stat_statements_reset()` or `SELECT pg_stat_monitor_reset()`
575
+ - Enable `track_io_timing` for block read/write timing metrics
576
577
:::
578
availability: |
503
- Available when:<br/>• The `pg_stat_statements` extension is installed in the database<br/>• The collector has successfully connected to PostgreSQL<br/>• Returns HTTP 503 if extension is not installed (with instructions to install)<br/>• Returns HTTP 500 if the query fails<br/>• Returns HTTP 504 if the query times out
579
+ Available when:<br/>• Either `pg_stat_statements` or `pg_stat_monitor` extension is installed<br/>• The collector has successfully connected to PostgreSQL<br/>• Returns HTTP 503 if no query statistics extension is installed<br/>• Returns HTTP 500 if the query fails<br/>• Returns HTTP 504 if the query times out
580
+ require_cloud: true
581
+ - id: running-queries
582
+ name: Running Queries
583
+ description: |
584
+ Retrieves currently executing queries from PostgreSQL [pg_stat_activity](https://www.postgresql.org/docs/current/monitoring-stats.html#MONITORING-PG-STAT-ACTIVITY-VIEW) system view.
585
+
586
+ This function queries `pg_stat_activity` which shows real-time information about each server process including the SQL query being executed, wait events, and session state. Unlike Top Queries which shows aggregated historical statistics, Running Queries shows live snapshots of active queries.
587
+
588
+ Use cases:
589
+ - Identify long-running queries that may be blocking other operations
590
+ - Debug stuck transactions or hanging connections
591
+ - Monitor active workload during performance issues
592
+ - Investigate wait events and lock contention in real-time
593
+
594
+ Query text is truncated at 4096 characters for display purposes.
595
+ parameters:
596
+ - id: __sort
597
+ name: Sort By
598
+ description: Select the sort column. Defaults to query duration (longest running first).
599
+ type: select
600
+ required: true
601
+ default: durationMs
602
+ options: []
603
+ returns:
604
+ description: Live query data from `pg_stat_activity`. Each row represents a currently active backend process.
605
+ columns:
606
+ # Visible columns (most important first)
607
+ - name: Duration
608
+ type: duration
609
+ unit: "milliseconds"
610
+ description: "Query duration in milliseconds (since query_start). High values indicate long-running queries."
611
+ - name: Query
612
+ type: string
613
+ unit: ""
614
+ description: "Query text of the currently executing or most recent query. May be truncated at track_activity_query_size."
615
+ - name: Database
616
+ type: string
617
+ unit: ""
618
+ description: "Name of the database this backend is connected to."
619
+ - name: User
620
+ type: string
621
+ unit: ""
622
+ description: "Name of the user logged into this backend."
623
+ - name: Application Name
624
+ type: string
625
+ unit: ""
626
+ description: "Name of the application connected to this backend."
627
+ - name: Client Address
628
+ type: string
629
+ unit: ""
630
+ description: "IP address of the client (NULL for Unix socket or internal process)."
631
+ - name: Wait Event
632
+ type: string
633
+ unit: ""
634
+ description: "Specific wait event name if backend is currently waiting."
635
+ - name: PID
636
+ type: integer
637
+ unit: ""
638
+ description: "Process ID of this backend. Use with pg_terminate_backend() to kill a query."
639
+ # Hidden columns
640
+ - name: Wait Event Type
641
+ type: string
642
+ unit: ""
643
+ visibility: hidden
644
+ description: "Type of event the backend is waiting for (Activity, BufferPin, Client, Extension, IO, IPC, Lock, LWLock, Timeout)."
645
+ - name: State
646
+ type: string
647
+ unit: ""
648
+ visibility: hidden
649
+ description: "Current state: active, idle, idle in transaction, idle in transaction (aborted), fastpath function call, disabled."
650
+ - name: Backend Type
651
+ type: string
652
+ unit: ""
653
+ visibility: hidden
654
+ description: "Type of backend: client backend, autovacuum worker, parallel worker, walsender, walreceiver, etc. Available in PostgreSQL 10+."
655
+ - name: Query Start
656
+ type: timestamp
657
+ unit: ""
658
+ visibility: hidden
659
+ description: "Time when the currently active query was started."
660
+ - name: Transaction Start
661
+ type: timestamp
662
+ unit: ""
663
+ visibility: hidden
664
+ description: "Time when current transaction started (NULL if no transaction)."
665
+ - name: Backend Start
666
+ type: timestamp
667
+ unit: ""
668
+ visibility: hidden
669
+ description: "Time when this process/connection started."
670
+ - name: State Change
671
+ type: timestamp
672
+ unit: ""
673
+ visibility: hidden
674
+ description: "Time when state was last changed."
675
+ - name: Query ID
676
+ type: string
677
+ unit: ""
678
+ visibility: hidden
679
+ description: "Query identifier (requires compute_query_id or extension). Available in PostgreSQL 14+."
680
+ - name: Leader PID
681
+ type: integer
682
+ unit: ""
683
+ visibility: hidden
684
+ description: "Process ID of parallel group leader (NULL if this is leader or not parallel). Available in PostgreSQL 13+."
685
+ - name: Database ID
686
+ type: integer
687
+ unit: ""
688
+ visibility: hidden
689
+ description: "OID of the database this backend is connected to."
690
+ - name: User ID
691
+ type: integer
692
+ unit: ""
693
+ visibility: hidden
694
+ description: "OID of the user logged into this backend."
695
+ - name: Client Hostname
696
+ type: string
697
+ unit: ""
698
+ visibility: hidden
699
+ description: "Hostname of the client via reverse DNS (only if log_hostname enabled)."
700
+ - name: Client Port
701
+ type: integer
702
+ unit: ""
703
+ visibility: hidden
704
+ description: "TCP port of client (-1 for Unix socket, NULL for internal process)."
705
+ - name: Backend Xid
706
+ type: string
707
+ unit: ""
708
+ visibility: hidden
709
+ description: "Top-level transaction identifier of this backend."
710
+ - name: Backend Xmin
711
+ type: string
712
+ unit: ""
713
+ visibility: hidden
714
+ description: "Backend's xmin horizon."
715
+ performance: |
716
+ Queries `pg_stat_activity` which is a live system view:<br/>• Very lightweight query, no impact on database performance<br/>• Returns only active queries by default (state = 'active')<br/>• Limited to 500 rows
717
+ security: |
718
+ Query text contains actual SQL being executed, which may include:<br/>• Personal information in WHERE clauses or INSERT values<br/>• Business data and internal identifiers<br/>• Access should be restricted to authorized personnel only
719
+ prerequisites:
720
+ list:
721
+ - title: Database user permissions
722
+ description: |
723
+ The monitoring user needs `pg_monitor` role to view all sessions:
724
+
725
+ ```sql
726
+ GRANT pg_monitor TO netdata;
727
+ ```
728
+
729
+ Without this role, the user can only see their own sessions.
730
+ availability: |
731
+ Available when:<br/>• The collector has successfully connected to PostgreSQL<br/>• Returns HTTP 503 if collector is still initializing<br/>• Returns HTTP 500 if the query fails<br/>• Returns HTTP 504 if the query times out
732
require_cloud: true
733
metrics:
734
folding:
src/go/tools/functions-validation/docker-compose.yml
+1
-1
@@ -5,7 +5,7 @@ services:
5
POSTGRES_USER: netdata
6
POSTGRES_PASSWORD: netdata
7
POSTGRES_DB: netdata
8
- command: ["postgres", "-c", "shared_preload_libraries=pg_stat_statements"]
8
+ command: ["postgres", "-c", "shared_preload_libraries=${POSTGRES_PRELOAD_LIBRARIES:-pg_stat_statements}"]
9
ports:
10
- "${POSTGRES_PORT:-5432}:5432"
11
volumes:
src/go/tools/functions-validation/e2e/postgres-matrix.sh
+29
-9
@@ -3,18 +3,38 @@ set -euo pipefail
3
4
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
5
6
-# PostgreSQL versions 14-18 (all currently supported versions)
6
+# PostgreSQL versions 14-18 with pg_stat_statements (standard)
7
# Version 13 reached EOL in November 2025
8
POSTGRES_VARIANTS=(
9
- "postgres:14|postgres-14"
10
- "postgres:15|postgres-15"
11
- "postgres:16|postgres-16"
12
- "postgres:17|postgres-17"
13
- "postgres:18|postgres-18"
9
+ "postgres:14|postgres-14|pg_stat_statements"
10
+ "postgres:15|postgres-15|pg_stat_statements"
11
+ "postgres:16|postgres-16|pg_stat_statements"
12
+ "postgres:17|postgres-17|pg_stat_statements"
13
+ "postgres:18|postgres-18|pg_stat_statements"
14
)
15
16
+# pg_stat_monitor variants using Percona distribution
17
+# Note: Percona images include pg_stat_monitor pre-installed
18
+PGSM_VARIANTS=(
19
+ "percona/percona-distribution-postgresql:16|percona-pgsm-16|pg_stat_monitor"
20
+ "percona/percona-distribution-postgresql:17|percona-pgsm-17|pg_stat_monitor"
21
+)
22
+
23
+# Run pg_stat_statements tests
24
for entry in "${POSTGRES_VARIANTS[@]}"; do
17
- IFS='|' read -r image label <<< "$entry"
18
- printf '\n=== Running PostgreSQL collector E2E for %s (%s) ===\n' "$label" "$image" >&2
19
- POSTGRES_IMAGE="$image" POSTGRES_VARIANT="$label" bash "$SCRIPT_DIR/postgres.sh"
25
+ IFS='|' read -r image label ext <<< "$entry"
26
+ printf '\n=== Running PostgreSQL collector E2E for %s (%s) with %s ===\n' "$label" "$image" "$ext" >&2
27
+ POSTGRES_IMAGE="$image" POSTGRES_VARIANT="$label" POSTGRES_STATS_EXT="$ext" bash "$SCRIPT_DIR/postgres.sh"
28
done
29
+
30
+# Run pg_stat_monitor tests
31
+for entry in "${PGSM_VARIANTS[@]}"; do
32
+ IFS='|' read -r image label ext <<< "$entry"
33
+ printf '\n=== Running PostgreSQL collector E2E for %s (%s) with %s ===\n' "$label" "$image" "$ext" >&2
34
+ POSTGRES_IMAGE="$image" POSTGRES_VARIANT="$label" POSTGRES_STATS_EXT="$ext" bash "$SCRIPT_DIR/postgres.sh"
35
+done
36
+
37
+echo ""
38
+echo "=== All PostgreSQL E2E tests passed ==="
39
+echo " - pg_stat_statements: PostgreSQL 14, 15, 16, 17, 18"
40
+echo " - pg_stat_monitor: Percona 16, 17"
src/go/tools/functions-validation/e2e/postgres.sh
+36
-1
@@ -17,12 +17,47 @@ if [ -n "${POSTGRES_IMAGE:-}" ]; then
17
write_env "POSTGRES_IMAGE" "$POSTGRES_IMAGE"
18
fi
19
20
+# Support pg_stat_monitor testing with Percona distribution
21
+# POSTGRES_STATS_EXT can be "pg_stat_statements" (default) or "pg_stat_monitor"
22
+POSTGRES_STATS_EXT="${POSTGRES_STATS_EXT:-pg_stat_statements}"
23
+write_env "POSTGRES_PRELOAD_LIBRARIES" "$POSTGRES_STATS_EXT"
24
+
25
+# Use the appropriate init script based on extension
26
+if [ "$POSTGRES_STATS_EXT" = "pg_stat_monitor" ]; then
27
+ cp "$WORKDIR/seed/postgres/init-pgsm.sql" "$WORKDIR/seed/postgres/init.sql"
28
+fi
29
+
30
compose_up postgres
31
wait_healthy postgres 90
32
33
build_plugin
34
+
35
+# Test top-queries (pg_stat_statements or pg_stat_monitor)
36
run_info postgres
37
run_top_queries postgres
38
assert_column_visibility "$WORKDIR/postgres-top-queries.json" "top-queries"
39
28
-echo "E2E checks passed for ${POSTGRES_VARIANT_LABEL}." >&2
40
+# Verify the correct extension is being used
41
+if [ "$POSTGRES_STATS_EXT" = "pg_stat_monitor" ]; then
42
+ # pg_stat_monitor should have applicationName and cpuUserTime columns
43
+ if ! jq -e '.columns | has("applicationName")' "$WORKDIR/postgres-top-queries.json" > /dev/null 2>&1; then
44
+ echo "ERROR: pg_stat_monitor expected but applicationName column not found" >&2
45
+ exit 1
46
+ fi
47
+ echo "Verified: pg_stat_monitor is being used (applicationName column present)" >&2
48
+else
49
+ # pg_stat_statements should NOT have applicationName column
50
+ if jq -e '.columns | has("applicationName")' "$WORKDIR/postgres-top-queries.json" > /dev/null 2>&1; then
51
+ echo "ERROR: pg_stat_statements expected but applicationName column found" >&2
52
+ exit 1
53
+ fi
54
+ echo "Verified: pg_stat_statements is being used" >&2
55
+fi
56
+
57
+# Test running-queries (pg_stat_activity)
58
+# Note: running-queries may return 0 rows if no active queries at test time
59
+run_info_method postgres running-queries
60
+run_running_queries postgres 0
61
+assert_column_visibility "$WORKDIR/postgres-running-queries.json" "running-queries"
62
+
63
+echo "E2E checks passed for ${POSTGRES_VARIANT_LABEL} with ${POSTGRES_STATS_EXT}." >&2
src/go/tools/functions-validation/seed/postgres/init-pgsm.sql
new
+18
@@ -0,0 +1,18 @@
1
+-- pg_stat_monitor initialization script
2
+CREATE EXTENSION IF NOT EXISTS pg_stat_monitor;
3
+
4
+CREATE TABLE IF NOT EXISTS public.sample (
5
+ id SERIAL PRIMARY KEY,
6
+ name TEXT NOT NULL,
7
+ value INTEGER NOT NULL
8
+);
9
+
10
+INSERT INTO public.sample (name, value)
11
+VALUES
12
+ ('alpha', 10),
13
+ ('beta', 20),
14
+ ('gamma', 30);
15
+
16
+SELECT COUNT(*) FROM public.sample;
17
+SELECT * FROM public.sample WHERE value > 15;
18
+UPDATE public.sample SET value = value + 1 WHERE name = 'alpha';