sources

package
v6.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 11, 2026 License: BSD-3-Clause Imports: 32 Imported by: 0

Documentation

Overview

Provides functionality to read monitored data from different sources.

Sources defines how to get the information for the monitored databases. At the moment, sources definitions support two storages: * PostgreSQL database * YAML file

* `postgres.go` files cover the functionality for the PostgreSQL database. * `yaml.go` files cover the functionality for the YAML file. * `resolver.go` implements continuous discovery from patroni and postgres cluster. * `types.go` defines the types and interfaces. * `sample.sources.yaml` is a sample configuration file.

Index

Constants

View Source 
const (
	EnvUnknown       = "UNKNOWN"
	EnvAzureSingle   = "AZURE_SINGLE" //discontinued
	EnvAzureFlexible = "AZURE_FLEXIBLE"
	EnvGoogle        = "GOOGLE"
)

Variables

View Source 
var (
	NewConn           = db.New
	NewConnWithConfig = db.NewWithConfig
)

NewConn and NewConnWithConfig are wrappers to allow testing

View Source 
var ErrSourceExists = errors.New("source already exists")
View Source 
var ErrSourceNotFound = errors.New("source not found")
View Source 
var Kinds = []Kind{
	SourcePostgres,
	SourcePostgresDiscovery,
	SourcePgBouncer,
	SourcePgPool,
	SourcePatroniDiscovery,
	SourcePrometheus,
}

Functions

func RedactURL 

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 

func VersionToInt(version string) (v int)

Types

type CmdOpts 

type CmdOpts struct {
	Sources                      string   `` /* 160-byte string literal not displayed */
	Refresh                      int      `` /* 127-byte string literal not displayed */
	Groups                       []string `` /* 150-byte string literal not displayed */
	MinDbSizeMB                  int64    `` /* 184-byte string literal not displayed */
	MaxParallelConnectionsPerDb  int      `` /* 241-byte string literal not displayed */
	TryCreateListedExtsIfMissing string   `` /* 290-byte string literal not displayed */
	CreateHelpers                bool     `` /* 144-byte string literal not displayed */
}

SourceOpts specifies the sources related command-line options

type DbConn 

type DbConn struct {
	Source
	Conn       db.PgxPoolIface
	ConnConfig *pgxpool.Config
	RuntimeInfo

	sync.RWMutex
	// contains filtered or unexported fields
}

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.

func NewDbConn 

func NewDbConn(s Source) *DbConn

func (*DbConn) ActiveMetrics 

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 (*DbConn) Close 

func (md *DbConn) Close()

Close closes the connection if it is not nil.

func (*DbConn) Connect 

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 (*DbConn) DiscoverPlatform 

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 (*DbConn) FetchApproxSize 

func (md *DbConn) FetchApproxSize(ctx context.Context) error

FetchApproxSize fetches the approximate size of the database in bytes and populates md.ApproxDbSize.

func (*DbConn) FetchControlInfo 

func (md *DbConn) FetchControlInfo(ctx context.Context) error

FetchControlInfo queries pg_control_system() and populates the core RuntimeInfo fields.

func (*DbConn) FetchExtensions 

func (md *DbConn) FetchExtensions(ctx context.Context) error

FetchExtensions queries pg_extension and populates md.Extensions with the installed extension versions.

func (*DbConn) FetchRuntimeInfo 

func (md *DbConn) FetchRuntimeInfo(ctx context.Context, forceRefetch bool) (err error)

func (*DbConn) FetchVersion 

func (md *DbConn) FetchVersion(ctx context.Context, kind Kind) (err error)

func (*DbConn) FunctionExists 

func (md *DbConn) FunctionExists(ctx context.Context, functionName string) (exists bool)

FunctionExists checks if a function exists in the database

func (*DbConn) GetClusterIdentifier 

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 (*DbConn) GetDatabaseName 

func (md *DbConn) GetDatabaseName() string

GetDatabaseName returns the database name from the connection string

func (*DbConn) GetMetricInterval 

func (md *DbConn) GetMetricInterval(name string) time.Duration

GetMetricInterval returns the metric interval for the connection

func (*DbConn) GetSource 

func (md *DbConn) GetSource() Source

GetSource returns a copy of the embedded Source.

func (*DbConn) IsClientOnSameHost 

func (md *DbConn) IsClientOnSameHost() bool

IsClientOnSameHost checks if the pgwatch client is running on the same host as the PostgreSQL server

func (*DbConn) IsPostgresSource 

func (md *DbConn) IsPostgresSource() bool

func (*DbConn) ParseConfig 

func (md *DbConn) ParseConfig() (err error)

ParseConfig will parse the connection string and store the result in the connection config

func (*DbConn) Ping 

func (md *DbConn) Ping(ctx context.Context) (err error)

Ping will try to ping the server to ensure the connection is still alive

func (*DbConn) SetDatabaseName 

func (md *DbConn) SetDatabaseName(name string)

SetDatabaseName sets the database name in the connection config for resolved databases

func (*DbConn) SetMetricIntervals 

func (md *DbConn) SetMetricIntervals(main, standby metrics.MetricIntervals)

SetMetricIntervals atomically sets metric intervals; nil means "no change".

func (*DbConn) TryCreateMetricsHelpers 

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 (*DbConn) TryCreateMissingExtensions 

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 

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 

func NewHostConfig(URI string) (hc HostConfig, err error)

func (HostConfig) IsScopeSpecified 

func (hc HostConfig) IsScopeSpecified() bool

type Kind 

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 (Kind) IsValid 

func (k Kind) IsValid() bool

type PatroniClusterMember 

type PatroniClusterMember struct {
	Scope   string
	Name    string
	ConnURL string `yaml:"conn_url"`
	Role    string
}

func (PatroniClusterMember) IsPrimary 

func (pcm PatroniClusterMember) IsPrimary() bool

type PromConn 

type PromConn struct {
	Source

	HTTPClient *http.Client
	sync.RWMutex
	// contains filtered or unexported fields
}

PromConn represents a Prometheus source connection.

func NewPromConn 

func NewPromConn(s Source) *PromConn

func (*PromConn) Close 

func (pc *PromConn) Close()

func (*PromConn) Connect 

func (pc *PromConn) Connect(ctx context.Context, _ CmdOpts) error

func (*PromConn) FetchRuntimeInfo 

func (pc *PromConn) FetchRuntimeInfo(_ context.Context, _ bool) error

func (*PromConn) GetMetricInterval 

func (pc *PromConn) GetMetricInterval(name string) time.Duration

func (*PromConn) GetSource 

func (pc *PromConn) GetSource() Source

func (*PromConn) IsPostgresSource 

func (pc *PromConn) IsPostgresSource() bool

func (*PromConn) ParseConfig 

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 (*PromConn) Ping 

func (pc *PromConn) Ping(ctx context.Context) error

func (*PromConn) Scrape 

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 (*PromConn) SetMetricIntervals 

func (pc *PromConn) SetMetricIntervals(main, _ metrics.MetricIntervals)

type Reader 

type Reader interface {
	GetSources() (Sources, error)
}

type ReaderWriter 

type ReaderWriter interface {
	Reader
	Writer
}

func NewPostgresSourcesReaderWriter 

func NewPostgresSourcesReaderWriter(ctx context.Context, connstr string) (ReaderWriter, error)

func NewPostgresSourcesReaderWriterConn 

func NewPostgresSourcesReaderWriterConn(ctx context.Context, conn db.PgxPoolIface) (ReaderWriter, error)

func NewYAMLSourcesReaderWriter 

func NewYAMLSourcesReaderWriter(ctx context.Context, path string) (ReaderWriter, error)

type Resolver 

type Resolver struct {
	// contains filtered or unexported fields
}

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).

func NewResolver 

func NewResolver() *Resolver

NewResolver returns a Resolver with freshly initialized, empty caches.

func (*Resolver) ResolveDatabase 

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 (*Resolver) ResolveDatabases 

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 (*Resolver) ResolveDatabasesFromPatroni 

func (r *Resolver) ResolveDatabasesFromPatroni(source Source) (SourceConns, error)

func (*Resolver) ResolveDatabasesFromPostgres 

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.

type RuntimeInfo 

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
}

type 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"`
}

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.

func (*Source) Clone 

func (s *Source) Clone() *Source

func (Source) Equal 

func (s Source) Equal(s2 Source) bool

func (*Source) GetDatabaseName 

func (s *Source) GetDatabaseName() string

func (Source) ResolveDatabases 

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.

type SourceConn 

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()
}

SourceConn is the interface that all monitored source connection types must implement.

func NewSourceConn 

func NewSourceConn(s Source) SourceConn

NewSourceConn is a factory dispatcher that returns a SourceConn interface.

type SourceConns 

type SourceConns []SourceConn

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.

func (SourceConns) GetMonitoredDatabase 

func (mds SourceConns) GetMonitoredDatabase(DBUniqueName string) SourceConn

type Sources 

type Sources []Source

func (Sources) ResolveDatabases 

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 (Sources) Validate 

func (srcs Sources) Validate() (Sources, error)

type Writer 

type Writer interface {
	WriteSources(Sources) error
	DeleteSource(string) error
	UpdateSource(md Source) error
	CreateSource(md Source) error
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.