| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package postgres |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "database/sql" |
| 8 | _ "embed" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "sync" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/jackc/pgx/v5/stdlib" |
| 15 | "github.com/netdata/netdata/go/plugins/pkg/confopt" |
| 16 | "github.com/netdata/netdata/go/plugins/pkg/matcher" |
| 17 | "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi" |
| 18 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/cloudauth" |
| 19 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/cloudauth/sqladapter" |
| 20 | "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/oldmetrix" |
| 21 | ) |
| 22 | |
| 23 | //go:embed "config_schema.json" |
| 24 | var configSchema string |
| 25 | |
| 26 | func init() { |
| 27 | collectorapi.Register("postgres", collectorapi.Creator{ |
| 28 | JobConfigSchema: configSchema, |
| 29 | Create: func() collectorapi.CollectorV1 { return New() }, |
| 30 | Config: func() any { return &Config{} }, |
| 31 | Methods: pgMethods, |
| 32 | MethodHandler: pgFunctionHandler, |
| 33 | }) |
| 34 | } |
| 35 | |
| 36 | func New() *Collector { |
| 37 | return &Collector{ |
| 38 | Config: Config{ |
| 39 | Timeout: confopt.Duration(time.Second * 2), |
| 40 | DSN: "postgres://postgres:postgres@127.0.0.1:5432/postgres", |
| 41 | XactTimeHistogram: []float64{.1, .5, 1, 2.5, 5, 10}, |
| 42 | QueryTimeHistogram: []float64{.1, .5, 1, 2.5, 5, 10}, |
| 43 | // charts: 20 x table, 4 x index. |
| 44 | // https://discord.com/channels/847502280503590932/1022693928874549368 |
| 45 | MaxDBTables: 50, |
| 46 | MaxDBIndexes: 250, |
| 47 | Functions: FunctionsConfig{ |
| 48 | TopQueries: TopQueriesConfig{ |
| 49 | Limit: 500, |
| 50 | }, |
| 51 | }, |
| 52 | }, |
| 53 | charts: baseCharts.Copy(), |
| 54 | dbConns: make(map[string]*dbConn), |
| 55 | mx: &pgMetrics{ |
| 56 | dbs: make(map[string]*dbMetrics), |
| 57 | indexes: make(map[string]*indexMetrics), |
| 58 | tables: make(map[string]*tableMetrics), |
| 59 | replApps: make(map[string]*replStandbyAppMetrics), |
| 60 | replSlots: make(map[string]*replSlotMetrics), |
| 61 | }, |
| 62 | recheckSettingsEvery: time.Minute * 30, |
| 63 | doSlowEvery: time.Minute * 5, |
| 64 | addXactQueryRunningTimeChartsOnce: &sync.Once{}, |
| 65 | addWALFilesChartsOnce: &sync.Once{}, |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | type Config struct { |
| 70 | Vnode string `yaml:"vnode,omitempty" json:"vnode"` |
| 71 | UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"` |
| 72 | AutoDetectionRetry int `yaml:"autodetection_retry,omitempty" json:"autodetection_retry"` |
| 73 | DSN string `yaml:"dsn" json:"dsn"` |
| 74 | Timeout confopt.Duration `yaml:"timeout,omitempty" json:"timeout"` |
| 75 | CloudAuth cloudauth.Config `yaml:"cloud_auth" json:"cloud_auth"` |
| 76 | DBSelector string `yaml:"collect_databases_matching,omitempty" json:"collect_databases_matching"` |
| 77 | XactTimeHistogram []float64 `yaml:"transaction_time_histogram,omitempty" json:"transaction_time_histogram"` |
| 78 | QueryTimeHistogram []float64 `yaml:"query_time_histogram,omitempty" json:"query_time_histogram"` |
| 79 | MaxDBTables int64 `yaml:"max_db_tables" json:"max_db_tables"` |
| 80 | MaxDBIndexes int64 `yaml:"max_db_indexes" json:"max_db_indexes"` |
| 81 | Functions FunctionsConfig `yaml:"functions,omitempty" json:"functions"` |
| 82 | } |
| 83 | |
| 84 | type FunctionsConfig struct { |
| 85 | TopQueries TopQueriesConfig `yaml:"top_queries,omitempty" json:"top_queries"` |
| 86 | } |
| 87 | |
| 88 | type TopQueriesConfig struct { |
| 89 | Disabled bool `yaml:"disabled" json:"disabled"` |
| 90 | Timeout confopt.Duration `yaml:"timeout,omitempty" json:"timeout"` |
| 91 | Limit int `yaml:"limit,omitempty" json:"limit"` |
| 92 | } |
| 93 | |
| 94 | func (c Config) topQueriesTimeout() time.Duration { |
| 95 | if c.Functions.TopQueries.Timeout == 0 { |
| 96 | return c.Timeout.Duration() |
| 97 | } |
| 98 | return c.Functions.TopQueries.Timeout.Duration() |
| 99 | } |
| 100 | |
| 101 | func (c Config) topQueriesLimit() int { |
| 102 | if c.Functions.TopQueries.Limit <= 0 { |
| 103 | return 500 |
| 104 | } |
| 105 | return c.Functions.TopQueries.Limit |
| 106 | } |
| 107 | |
| 108 | type ( |
| 109 | Collector struct { |
| 110 | collectorapi.Base |
| 111 | Config `yaml:",inline" json:""` |
| 112 | |
| 113 | charts *collectorapi.Charts |
| 114 | addXactQueryRunningTimeChartsOnce *sync.Once |
| 115 | addWALFilesChartsOnce *sync.Once |
| 116 | |
| 117 | db *sql.DB |
| 118 | dbConns map[string]*dbConn |
| 119 | |
| 120 | superUser *bool |
| 121 | canExecutePgLsDir *bool |
| 122 | pgIsInRecovery *bool |
| 123 | pgVersion int |
| 124 | pgStatStatementsAvail bool // cached positive result only |
| 125 | pgStatStatementsColumns map[string]bool // cached column names from pg_stat_statements |
| 126 | pgStatMonitorAvail bool // cached positive result only |
| 127 | pgStatMonitorColumns map[string]bool // cached column names from pg_stat_monitor |
| 128 | queryStatsSource string // "pg_stat_monitor" or "pg_stat_statements" (auto-detected) |
| 129 | pgStatStatementsMu sync.RWMutex // protects pgStatStatements*/pgStatMonitor* fields for concurrent access |
| 130 | dbSr matcher.Matcher |
| 131 | recheckSettingsTime time.Time |
| 132 | recheckSettingsEvery time.Duration |
| 133 | doSlowTime time.Time |
| 134 | doSlowEvery time.Duration |
| 135 | |
| 136 | azureTokenProvider *cloudauth.TokenProvider |
| 137 | |
| 138 | mx *pgMetrics |
| 139 | |
| 140 | funcRouter *funcRouter |
| 141 | } |
| 142 | dbConn struct { |
| 143 | db *sql.DB |
| 144 | connStr string |
| 145 | connErrors int |
| 146 | } |
| 147 | ) |
| 148 | |
| 149 | func (c *Collector) Configuration() any { |
| 150 | return c.Config |
| 151 | } |
| 152 | |
| 153 | func (c *Collector) Init(context.Context) error { |
| 154 | err := c.validateConfig() |
| 155 | if err != nil { |
| 156 | return fmt.Errorf("config validation: %v", err) |
| 157 | } |
| 158 | if err := c.CloudAuth.Validate(); err != nil { |
| 159 | return fmt.Errorf("config validation: %v", err) |
| 160 | } |
| 161 | if c.CloudAuth.IsEnabled() { |
| 162 | cred, err := c.CloudAuth.NewCredential() |
| 163 | if err != nil { |
| 164 | return fmt.Errorf("config validation: creating cloud auth credential: %v", err) |
| 165 | } |
| 166 | provider, err := cloudauth.NewTokenProvider( |
| 167 | cred, |
| 168 | []string{sqladapter.AzurePostgreSQLAADScope}, |
| 169 | cloudauth.DefaultTokenRefreshMargin, |
| 170 | ) |
| 171 | if err != nil { |
| 172 | return fmt.Errorf("config validation: creating cloud auth token provider: %v", err) |
| 173 | } |
| 174 | c.azureTokenProvider = provider |
| 175 | } |
| 176 | |
| 177 | sr, err := c.initDBSelector() |
| 178 | if err != nil { |
| 179 | return fmt.Errorf("config validation: %v", err) |
| 180 | } |
| 181 | c.dbSr = sr |
| 182 | |
| 183 | c.mx.xactTimeHist = oldmetrix.NewHistogramWithRangeBuckets(c.XactTimeHistogram) |
| 184 | c.mx.queryTimeHist = oldmetrix.NewHistogramWithRangeBuckets(c.QueryTimeHistogram) |
| 185 | |
| 186 | c.funcRouter = newFuncRouter(c) |
| 187 | |
| 188 | return nil |
| 189 | } |
| 190 | |
| 191 | func (c *Collector) Check(context.Context) error { |
| 192 | mx, err := c.collect() |
| 193 | if err != nil { |
| 194 | return err |
| 195 | } |
| 196 | if len(mx) == 0 { |
| 197 | return errors.New("no metrics collected") |
| 198 | } |
| 199 | return nil |
| 200 | } |
| 201 | |
| 202 | func (c *Collector) Charts() *collectorapi.Charts { |
| 203 | return c.charts |
| 204 | } |
| 205 | |
| 206 | func (c *Collector) Collect(context.Context) map[string]int64 { |
| 207 | mx, err := c.collect() |
| 208 | if err != nil { |
| 209 | c.Error(err) |
| 210 | } |
| 211 | |
| 212 | if len(mx) == 0 { |
| 213 | return nil |
| 214 | } |
| 215 | return mx |
| 216 | } |
| 217 | |
| 218 | func (c *Collector) Cleanup(ctx context.Context) { |
| 219 | if c.funcRouter != nil { |
| 220 | c.funcRouter.Cleanup(ctx) |
| 221 | } |
| 222 | if c.db == nil { |
| 223 | return |
| 224 | } |
| 225 | if err := c.db.Close(); err != nil { |
| 226 | c.Warningf("cleanup: error on closing the Postgres database [%s]: %v", c.DSN, err) |
| 227 | } |
| 228 | c.db = nil |
| 229 | |
| 230 | for dbname, conn := range c.dbConns { |
| 231 | delete(c.dbConns, dbname) |
| 232 | if conn.connStr != "" { |
| 233 | stdlib.UnregisterConnConfig(conn.connStr) |
| 234 | } |
| 235 | if conn.db != nil { |
| 236 | _ = conn.db.Close() |
| 237 | } |
| 238 | } |
| 239 | } |