const (
EnvUnknown = "UNKNOWN"
EnvAzureSingle = "AZURE_SINGLE" //discontinued
EnvAzureFlexible = "AZURE_FLEXIBLE"
EnvGoogle = "GOOGLE"
)
const (
dcsTypeEtcd = "etcd"
dcsTypeZookeeper = "zookeeper"
dcsTypeConsul = "consul"
)
NewConn and NewConnWithConfig are wrappers to allow testing
var (
NewConn = db.New
NewConnWithConfig = db.NewWithConfig
)
var ErrSourceExists = errors.New("source already exists")
var ErrSourceNotFound = errors.New("source not found")
var Kinds = []Kind{ SourcePostgres, SourcePostgresDiscovery, SourcePgBouncer, SourcePgPool, SourcePatroniDiscovery, SourcePrometheus, }
make sure *dbSourcesReaderWriter implements the Migrator interface
var _ db.Migrator = (*dbSourcesReaderWriter)(nil)
defaultResolver backs the convenience methods on Source and Sources so that existing callers keep a single process-wide fallback cache without having to thread a Resolver through.
var defaultResolver = NewResolver()
var logger log.Logger = log.FallbackLogger
VersionToInt parses a given version and returns an integer or an error if unable to parse the version. Only parses valid semantic versions. Performs checking that can find errors within the version. Examples: v1.2 -> 01_02_00, v9.6.3 -> 09_06_03, v11 -> 11_00_00
var regVer = regexp.MustCompile(`(\d+).?(\d*).?(\d*)`)
func RedactURL(rawURL string) string
RedactURL replaces the password in URL userinfo with "xxxxx". If rawURL cannot be parsed or has no password, it is returned unchanged.
func VersionToInt(version string) (v int)
func getTransport(conf HostConfig) (*tls.Config, error)
func jsonTextToStringMap(jsonText string) (map[string]string, error)
SourceOpts specifies the sources related command-line options
type CmdOpts struct {
Sources string `short:"s" long:"sources" mapstructure:"config" description:"Postgres URI, file or folder of YAML files containing info on which DBs to monitor" env:"PW_SOURCES"`
Refresh int `long:"refresh" mapstructure:"refresh" description:"How frequently to resync sources and metrics" env:"PW_REFRESH" default:"120"`
Groups []string `short:"g" long:"group" mapstructure:"group" description:"Groups for filtering which databases to monitor. By default all are monitored" env:"PW_GROUP"`
MinDbSizeMB int64 `long:"min-db-size-mb" mapstructure:"min-db-size-mb" description:"Smaller size DBs will be ignored and not monitored until they reach the threshold." env:"PW_MIN_DB_SIZE_MB" default:"0"`
MaxParallelConnectionsPerDb int `long:"max-parallel-connections-per-db" mapstructure:"max-parallel-connections-per-db" description:"Max parallel metric fetches per DB. Note the multiplication effect on multi-DB instances" env:"PW_MAX_PARALLEL_CONNECTIONS_PER_DB" default:"4"`
TryCreateListedExtsIfMissing string `long:"try-create-listed-exts-if-missing" mapstructure:"try-create-listed-exts-if-missing" description:"Try creating the listed extensions (comma sep.) on first connect for all monitored DBs when missing. Main usage - pg_stat_statements" env:"PW_TRY_CREATE_LISTED_EXTS_IF_MISSING" default:""`
CreateHelpers bool `long:"create-helpers" mapstructure:"create-helpers" description:"Create helper database objects from metric definitions" env:"PW_CREATE_HELPERS"`
}
DbConn represents a single connection to monitor. Unlike source, it contains a database connection. Continuous discovery sources (postgres-continuous-discovery, patroni-continuous-discovery, patroni-namespace-discovery) will produce multiple monitored databases structs based on the discovered databases.
type DbConn struct {
Source
Conn db.PgxPoolIface
ConnConfig *pgxpool.Config
RuntimeInfo
lastCheckedNs atomic.Int64 // nanoseconds of last successful FetchRuntimeInfo; 0 = never
sync.RWMutex
}
func NewDbConn(s Source) *DbConn
func (md *DbConn) ActiveMetrics() metrics.MetricIntervals
ActiveMetrics returns a snapshot of the currently active metric intervals based on the connection's recovery state: standby config wins when the source is in recovery and a standby config is defined, otherwise the primary config is used. The caller receives a cloned copy safe to iterate without holding the lock.
func (md *DbConn) Close()
Close closes the connection if it is not nil.
func (md *DbConn) Connect(ctx context.Context, opts CmdOpts) (err error)
Connect will establish a connection to the database if it's not already connected. If the connection is already established, it pings the server to ensure it's still alive.
func (md *DbConn) DiscoverPlatform(ctx context.Context) error
DiscoverPlatform tries to discover the platform based on the database version string and some special settings that are only available on certain platforms. Populates md.ExecEnv.
func (md *DbConn) FetchApproxSize(ctx context.Context) error
FetchApproxSize fetches the approximate size of the database in bytes and populates md.ApproxDbSize.
func (md *DbConn) FetchControlInfo(ctx context.Context) error
FetchControlInfo queries pg_control_system() and populates the core RuntimeInfo fields.
func (md *DbConn) FetchExtensions(ctx context.Context) error
FetchExtensions queries pg_extension and populates md.Extensions with the installed extension versions.
func (md *DbConn) FetchRuntimeInfo(ctx context.Context, forceRefetch bool) (err error)
func (md *DbConn) FetchVersion(ctx context.Context, kind Kind) (err error)
func (md *DbConn) FunctionExists(ctx context.Context, functionName string) (exists bool)
FunctionExists checks if a function exists in the database
func (md *DbConn) GetClusterIdentifier() string
GetClusterIdentifier returns a unique identifier for the host assuming SysId is the same for primary and all replicas but connection information is different
func (md *DbConn) GetDatabaseName() string
GetDatabaseName returns the database name from the connection string
func (md *DbConn) GetMetricInterval(name string) time.Duration
GetMetricInterval returns the metric interval for the connection
func (md *DbConn) GetSource() Source
GetSource returns a copy of the embedded Source.
func (md *DbConn) IsClientOnSameHost() bool
IsClientOnSameHost checks if the pgwatch client is running on the same host as the PostgreSQL server
func (md *DbConn) IsPostgresSource() bool
func (md *DbConn) ParseConfig() (err error)
ParseConfig will parse the connection string and store the result in the connection config
func (md *DbConn) Ping(ctx context.Context) (err error)
Ping will try to ping the server to ensure the connection is still alive
func (md *DbConn) SetDatabaseName(name string)
SetDatabaseName sets the database name in the connection config for resolved databases
func (md *DbConn) SetMetricIntervals(main, standby metrics.MetricIntervals)
SetMetricIntervals atomically sets metric intervals; nil means "no change".
func (md *DbConn) TryCreateMetricsHelpers(ctx context.Context, getSQLFn func(string) string) (err error)
TryCreateMetricsHelpers should be called once on daemon startup to try to create "metric fetching helper" functions automatically
func (md *DbConn) TryCreateMissingExtensions(ctx context.Context, extensions []string) (string, error)
TryCreateMissingExtensions should be called once on daemon startup if some commonly wanted extension (most notably pg_stat_statements) is missing.
type HostConfig struct {
DcsType string `yaml:"dcs_type"`
DcsEndpoints []string `yaml:"dcs_endpoints"`
Path string
Username string
Password string
CAFile string `yaml:"ca_file"`
CertFile string `yaml:"cert_file"`
KeyFile string `yaml:"key_file"`
}
func NewHostConfig(URI string) (hc HostConfig, err error)
func (hc HostConfig) IsScopeSpecified() bool
type Kind string
const (
SourcePostgres Kind = "postgres"
SourcePostgresDiscovery Kind = "postgres-continuous-discovery"
SourcePgBouncer Kind = "pgbouncer"
SourcePgPool Kind = "pgpool"
SourcePatroniDiscovery Kind = "patroni"
SourcePrometheus Kind = "prometheus"
)
func (k Kind) IsValid() bool
type PatroniClusterMember struct {
Scope string
Name string
ConnURL string `yaml:"conn_url"`
Role string
}
func getConsulClusterMembers(Source) ([]PatroniClusterMember, error)
func getZookeeperClusterMembers(Source, HostConfig) ([]PatroniClusterMember, error)
func (pcm PatroniClusterMember) IsPrimary() bool
PromConn represents a Prometheus source connection.
type PromConn struct {
Source
connConfig *promConnConfig
HTTPClient *http.Client
sync.RWMutex
}
func NewPromConn(s Source) *PromConn
func (pc *PromConn) Close()
func (pc *PromConn) Connect(ctx context.Context, _ CmdOpts) error
func (pc *PromConn) FetchRuntimeInfo(_ context.Context, _ bool) error
func (pc *PromConn) GetMetricInterval(name string) time.Duration
func (pc *PromConn) GetSource() Source
func (pc *PromConn) IsPostgresSource() bool
func (pc *PromConn) ParseConfig() error
ParseConfig parses pc.ConnStr once and caches the result in pc.connConfig. Subsequent calls are no-ops. Mirrors DbConn.ParseConfig.
func (pc *PromConn) Ping(ctx context.Context) error
func (pc *PromConn) Scrape(ctx context.Context) (*http.Response, error)
Scrape executes a single GET request to the source's metrics endpoint with Accept: text/plain and optional Basic Auth from the cached config. The caller is responsible for closing resp.Body. Connect must be called before Scrape.
func (pc *PromConn) SetMetricIntervals(main, _ metrics.MetricIntervals)
type Reader interface {
GetSources() (Sources, error)
}
type ReaderWriter interface {
Reader
Writer
}
func NewPostgresSourcesReaderWriter(ctx context.Context, connstr string) (ReaderWriter, error)
func NewPostgresSourcesReaderWriterConn(ctx context.Context, conn db.PgxPoolIface) (ReaderWriter, error)
func NewYAMLSourcesReaderWriter(ctx context.Context, path string) (ReaderWriter, error)
Resolver discovers the monitored databases behind continuous-monitoring sources (Patroni, Postgres discovery). It owns the last-known-good fallback caches so that a transient DCS/DB outage does not tear down monitoring of already-known databases.
A Resolver is safe for concurrent use. Create independent Resolvers via NewResolver when isolated cache state is desired (e.g. in tests or when running several unrelated resolution pipelines in one process).
type Resolver struct {
mu sync.Mutex // guards the fallback caches below
// lastFoundClusterMembers is needed for cases where DCS is temporarily down;
// we don't want to immediately remove monitoring of DBs. Keyed by source name.
lastFoundClusterMembers map[string][]PatroniClusterMember
// lastFoundDatabases is keyed by the source's identity (name + conn string +
// include/exclude patterns) so a reconfigured source never inherits the
// previous target's database list.
lastFoundDatabases map[postgresDiscoveryKey]SourceConns
}
func NewResolver() *Resolver
NewResolver returns a Resolver with freshly initialized, empty caches.
func (r *Resolver) ResolveDatabase(s Source) (SourceConns, error)
ResolveDatabase returns a slice of found databases for a single continuous monitoring source, e.g. patroni.
func (r *Resolver) ResolveDatabases(srcs Sources, onError func(string)) (_ SourceConns, err error)
ResolveDatabases updates the list of monitored objects from continuous monitoring sources, e.g. patroni. Each source is resolved concurrently so that a slow or unreachable source does not block the others.
func (r *Resolver) ResolveDatabasesFromPatroni(source Source) (SourceConns, error)
func (r *Resolver) ResolveDatabasesFromPostgres(s Source) (resolvedDbs SourceConns, err error)
ResolveDatabasesFromPostgres reads all the databases from the given cluster, additionally matching/not matching specified regex patterns.
On any helper error (pool create or discovery query) and the presence of a previously-cached successful result for the same source identity, the cached list is returned with a nil error: a transient discovery failure (DNS hiccup, connect timeout, discovery-SQL permission error) must not tear down monitoring of already-known databases. The accepted trade-off is that a database dropped server-side while discovery keeps failing stays monitored from cache (surfaced as per-cycle connect errors reported as down) until the next successful resolution replaces the entry.
func (r *Resolver) getEtcdClusterMembers(s Source, hc HostConfig) ([]PatroniClusterMember, error)
type RuntimeInfo struct {
IsInRecovery bool
VersionStr string
Version int
RealDbname string
SystemIdentifier string
IsSuperuser bool
Extensions map[string]int
ExecEnv string
ApproxDbSize int64
ChangeState map[string]map[string]string // ["category"][object_identifier] = state
}
Source represents a configuration how to get databases to monitor. It can be a single database, a group of databases in postgres cluster, a group of databases in HA patroni cluster. pgbouncer and pgpool kinds are purely to indicate that the monitored database connection is made through a connection pooler, which supports its own additional metrics. If one is not interested in those additional metrics, it is ok to specify the connection details as a regular postgres source.
type Source struct {
Name string `yaml:"name" db:"name"`
Group string `yaml:"group" db:"group"`
ConnStr string `yaml:"conn_str" db:"connstr"`
Metrics metrics.MetricIntervals `yaml:"custom_metrics" db:"config"`
MetricsStandby metrics.MetricIntervals `yaml:"custom_metrics_standby" db:"config_standby"`
Kind Kind `yaml:"kind" db:"dbtype"`
IncludePattern string `yaml:"include_pattern" db:"include_pattern"`
ExcludePattern string `yaml:"exclude_pattern" db:"exclude_pattern"`
PresetMetrics string `yaml:"preset_metrics" db:"preset_config"`
PresetMetricsStandby string `yaml:"preset_metrics_standby" db:"preset_config_standby"`
IsEnabled bool `yaml:"is_enabled" db:"is_enabled"`
CustomTags map[string]string `yaml:"custom_tags" db:"custom_tags"`
OnlyIfMaster bool `yaml:"only_if_master" db:"only_if_master"`
}
func (s *Source) Clone() *Source
func (s Source) Equal(s2 Source) bool
func (s *Source) GetDatabaseName() string
func (s Source) ResolveDatabases() (SourceConns, error)
ResolveDatabases() return a slice of found databases for continuous monitoring sources, e.g. patroni. It delegates to the package-wide defaultResolver.
SourceConn is the interface that all monitored source connection types must implement.
type SourceConn interface {
Connect(ctx context.Context, opts CmdOpts) error
Ping(ctx context.Context) error
IsPostgresSource() bool
GetSource() Source
GetMetricInterval(name string) time.Duration
SetMetricIntervals(main, standby metrics.MetricIntervals)
Close()
}
compile-time assertions
var _ SourceConn = (*DbConn)(nil)
var _ SourceConn = (*PromConn)(nil)
func NewSourceConn(s Source) SourceConn
NewSourceConn is a factory dispatcher that returns a SourceConn interface.
DbConn represents a single connection to monitor. Unlike source, it contains a database connection. Continuous discovery sources (postgres-continuous-discovery, patroni-continuous-discovery, patroni-namespace-discovery) will produce multiple monitored databases structs based on the discovered databases.
type SourceConns []SourceConn
func resolveDatabasesFromPostgres(s Source) (resolvedDbs SourceConns, err error)
resolveDatabasesFromPostgres is the inner helper that actually runs the discovery query.
func (mds SourceConns) GetMonitoredDatabase(DBUniqueName string) SourceConn
type Sources []Source
func (srcs Sources) ResolveDatabases(onError func(string)) (SourceConns, error)
ResolveDatabases() updates list of monitored objects from continuous monitoring sources, e.g. patroni. Each source is resolved concurrently so that a slow or unreachable source does not block the others. It delegates to the package-wide defaultResolver.
func (srcs Sources) Validate() (Sources, error)
type Writer interface {
WriteSources(Sources) error
DeleteSource(string) error
UpdateSource(md Source) error
CreateSource(md Source) error
}
type dbSourcesReaderWriter struct {
ctx context.Context
configDb db.PgxIface
}
func (r *dbSourcesReaderWriter) CreateSource(md Source) error
func (r *dbSourcesReaderWriter) DeleteSource(name string) error
func (r *dbSourcesReaderWriter) GetSources() (Sources, error)
func (r *dbSourcesReaderWriter) Migrate() error
func (r *dbSourcesReaderWriter) NeedsMigration() (bool, error)
func (r *dbSourcesReaderWriter) UpdateSource(md Source) error
func (r *dbSourcesReaderWriter) WriteSources(dbs Sources) error
func (r *dbSourcesReaderWriter) createSource(conn db.PgxIface, md Source) (err error)
func (r *dbSourcesReaderWriter) updateSource(conn db.PgxIface, md Source) (err error)
type fileSourcesReaderWriter struct {
ctx context.Context
path string
sync.Mutex
}
func (fcr *fileSourcesReaderWriter) CreateSource(md Source) error
CreateSource creates a new source if it doesn't already exist, then writes the updated sources back to file
func (fcr *fileSourcesReaderWriter) DeleteSource(name string) error
DeleteSource deletes a source by name and writes the updated sources back to file
func (fcr *fileSourcesReaderWriter) GetSources() (dbs Sources, err error)
GetSources reads sources from file with locking
func (fcr *fileSourcesReaderWriter) UpdateSource(md Source) error
UpdateSource updates an existing source or creates it if it doesn't exist, then writes the updated sources back to file
func (fcr *fileSourcesReaderWriter) WriteSources(mds Sources) error
WriteSources writes sources to file with locking
func (fcr *fileSourcesReaderWriter) expandEnvVars(md Source) Source
func (fcr *fileSourcesReaderWriter) getSources() (dbs Sources, err error)
getSources reads sources from file without locking (internal use only)
func (fcr *fileSourcesReaderWriter) loadSourcesFromFile(configFilePath string) (dbs Sources, err error)
loadSourcesFromFile reads sources from a single YAML file, expands environment variables, and returns them
func (fcr *fileSourcesReaderWriter) writeSources(mds Sources) error
writeSources writes sources to file without locking (internal use only)
postgresDiscoveryKey uniquely identifies a Postgres discovery source for caching purposes. It MUST incorporate every field that would change the set of discovered databases; otherwise a reconfigured source could be served the previous target's last-known-good list.
type postgresDiscoveryKey struct {
name string
connStr string
includePattern string
excludePattern string
}
func postgresDiscoveryKeyOf(s Source) postgresDiscoveryKey
promConnConfig holds the parsed Prometheus source connection parameters. It is populated once by ParseConfig and reused by Connect and Ping.
type promConnConfig struct {
URL string
Userinfo *url.Userinfo
TLSConfig *tls.Config
}