conn.go

Documentation: github.com/cybertec-postgresql/pgwatch/v6/internal/sources

     1  package sources
     2  
     3  import (
     4  	"context"
     5  	"crypto/tls"
     6  	"crypto/x509"
     7  	"errors"
     8  	"fmt"
     9  	"io"
    10  	"maps"
    11  	"math"
    12  	"net/http"
    13  	"net/url"
    14  	"os"
    15  	"regexp"
    16  	"slices"
    17  	"strconv"
    18  	"strings"
    19  	"sync"
    20  	"sync/atomic"
    21  	"time"
    22  
    23  	"github.com/cybertec-postgresql/pgwatch/v6/internal/db"
    24  	"github.com/cybertec-postgresql/pgwatch/v6/internal/metrics"
    25  	"github.com/jackc/pgx/v5"
    26  	"github.com/jackc/pgx/v5/pgxpool"
    27  )
    28  
    29  // NewConn and NewConnWithConfig are wrappers to allow testing
    30  var (
    31  	NewConn           = db.New
    32  	NewConnWithConfig = db.NewWithConfig
    33  )
    34  
    35  const (
    36  	EnvUnknown       = "UNKNOWN"
    37  	EnvAzureSingle   = "AZURE_SINGLE" //discontinued
    38  	EnvAzureFlexible = "AZURE_FLEXIBLE"
    39  	EnvGoogle        = "GOOGLE"
    40  )
    41  
    42  // SourceConn is the interface that all monitored source connection types must implement.
    43  type SourceConn interface {
    44  	Connect(ctx context.Context, opts CmdOpts) error
    45  	Ping(ctx context.Context) error
    46  	IsPostgresSource() bool
    47  	GetSource() Source
    48  	GetMetricInterval(name string) time.Duration
    49  	SetMetricIntervals(main, standby metrics.MetricIntervals)
    50  	Close()
    51  }
    52  
    53  // compile-time assertions
    54  var _ SourceConn = (*DbConn)(nil)
    55  var _ SourceConn = (*PromConn)(nil)
    56  
    57  type RuntimeInfo struct {
    58  	IsInRecovery     bool
    59  	VersionStr       string
    60  	Version          int
    61  	RealDbname       string
    62  	SystemIdentifier string
    63  	IsSuperuser      bool
    64  	Extensions       map[string]int
    65  	ExecEnv          string
    66  	ApproxDbSize     int64
    67  	ChangeState      map[string]map[string]string // ["category"][object_identifier] = state
    68  }
    69  
    70  // DbConn represents a single connection to monitor. Unlike source, it contains a database connection.
    71  // Continuous discovery sources (postgres-continuous-discovery, patroni-continuous-discovery, patroni-namespace-discovery)
    72  // will produce multiple monitored databases structs based on the discovered databases.
    73  type (
    74  	DbConn struct {
    75  		Source
    76  		Conn       db.PgxPoolIface
    77  		ConnConfig *pgxpool.Config
    78  		RuntimeInfo
    79  		lastCheckedNs atomic.Int64 // nanoseconds of last successful FetchRuntimeInfo; 0 = never
    80  		sync.RWMutex
    81  	}
    82  
    83  	SourceConns []SourceConn
    84  )
    85  
    86  func NewDbConn(s Source) *DbConn {
    87  	return &DbConn{
    88  		Source: s,
    89  		RuntimeInfo: RuntimeInfo{
    90  			Extensions:  make(map[string]int),
    91  			ChangeState: make(map[string]map[string]string),
    92  		},
    93  	}
    94  }
    95  
    96  // NewSourceConn is a factory dispatcher that returns a SourceConn interface.
    97  func NewSourceConn(s Source) SourceConn {
    98  	switch s.Kind {
    99  	case SourcePrometheus:
   100  		return NewPromConn(s)
   101  	default:
   102  		return NewDbConn(s)
   103  	}
   104  }
   105  
   106  // GetSource returns a copy of the embedded Source.
   107  func (md *DbConn) GetSource() Source {
   108  	return md.Source
   109  }
   110  
   111  // ActiveMetrics returns a snapshot of the currently active metric intervals
   112  // based on the connection's recovery state: standby config wins when the source
   113  // is in recovery and a standby config is defined, otherwise the primary config is used.
   114  // The caller receives a cloned copy safe to iterate without holding the lock.
   115  func (md *DbConn) ActiveMetrics() metrics.MetricIntervals {
   116  	md.RLock()
   117  	defer md.RUnlock()
   118  	if md.IsInRecovery && len(md.MetricsStandby) > 0 {
   119  		return maps.Clone(md.MetricsStandby)
   120  	}
   121  	return maps.Clone(md.Metrics)
   122  }
   123  
   124  // SetMetricIntervals atomically sets metric intervals; nil means "no change".
   125  func (md *DbConn) SetMetricIntervals(main, standby metrics.MetricIntervals) {
   126  	md.Lock()
   127  	defer md.Unlock()
   128  	if main != nil {
   129  		md.Metrics = main
   130  	}
   131  	if standby != nil {
   132  		md.MetricsStandby = standby
   133  	}
   134  }
   135  
   136  // Close closes the connection if it is not nil.
   137  func (md *DbConn) Close() {
   138  	if md.Conn != nil {
   139  		md.Conn.Close()
   140  	}
   141  }
   142  
   143  // Ping will try to ping the server to ensure the connection is still alive
   144  func (md *DbConn) Ping(ctx context.Context) (err error) {
   145  	// Bound the round-trip so a half-open peer or a wedged pool cannot stall
   146  	// the main-loop sweep. The bound is the configured ConnectTimeout plus a
   147  	// small margin. Ping is exported, so ConnConfig may not be populated at
   148  	// all — fall back to a 10 s default total in that case. When only the
   149  	// outer pgxpool.Config is set but the inner ConnConfig is nil (or
   150  	// ConnectTimeout is unconfigured), the effective bound collapses to just
   151  	// PingTimeoutMargin.
   152  	timeout := 10 * time.Second
   153  	if md.ConnConfig != nil {
   154  		ct := time.Duration(0)
   155  		if md.ConnConfig.ConnConfig != nil {
   156  			ct = md.ConnConfig.ConnConfig.ConnectTimeout
   157  		}
   158  		timeout = ct + db.PingTimeoutMargin
   159  	}
   160  	pingCtx, cancel := db.WithOpTimeout(ctx, "ping", timeout)
   161  	defer cancel()
   162  
   163  	if md.Kind == SourcePgBouncer {
   164  		// pgbouncer is very picky about the queries it accepts
   165  		_, err = md.Conn.Exec(pingCtx, "SHOW VERSION")
   166  		return
   167  	}
   168  	return md.Conn.Ping(pingCtx)
   169  }
   170  
   171  // Connect will establish a connection to the database if it's not already connected.
   172  // If the connection is already established, it pings the server to ensure it's still alive.
   173  func (md *DbConn) Connect(ctx context.Context, opts CmdOpts) (err error) {
   174  	if md.Conn == nil {
   175  		if err = md.ParseConfig(); err != nil {
   176  			return err
   177  		}
   178  		if md.Kind == SourcePgBouncer {
   179  			md.ConnConfig.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol
   180  			// pgbouncer rejects pgxpool's `-- ping` liveness check, see Ping() above
   181  			md.ConnConfig.ShouldPing = func(context.Context, pgxpool.ShouldPingParams) bool { return false }
   182  		}
   183  		if opts.MaxParallelConnectionsPerDb > 0 {
   184  			md.ConnConfig.MaxConns = int32(opts.MaxParallelConnectionsPerDb)
   185  		}
   186  		md.Conn, err = NewConnWithConfig(ctx, md.ConnConfig)
   187  		if err != nil {
   188  			return err
   189  		}
   190  	}
   191  	return md.Ping(ctx)
   192  }
   193  
   194  // ParseConfig will parse the connection string and store the result in the connection config
   195  func (md *DbConn) ParseConfig() (err error) {
   196  	if md.ConnConfig == nil {
   197  		md.ConnConfig, err = pgxpool.ParseConfig(md.ConnStr)
   198  		return
   199  	}
   200  	return
   201  }
   202  
   203  // GetClusterIdentifier returns a unique identifier for the host assuming SysId is the same for
   204  // primary and all replicas but connection information is different
   205  func (md *DbConn) GetClusterIdentifier() string {
   206  	if err := md.ParseConfig(); err != nil {
   207  		return ""
   208  	}
   209  	md.RLock()
   210  	defer md.RUnlock()
   211  	return fmt.Sprintf("%s:%s:%d", md.SystemIdentifier, md.ConnConfig.ConnConfig.Host, md.ConnConfig.ConnConfig.Port)
   212  }
   213  
   214  // GetDatabaseName returns the database name from the connection string
   215  func (md *DbConn) GetDatabaseName() string {
   216  	if err := md.ParseConfig(); err != nil {
   217  		return ""
   218  	}
   219  	return md.ConnConfig.ConnConfig.Database
   220  }
   221  
   222  // GetMetricInterval returns the metric interval for the connection
   223  func (md *DbConn) GetMetricInterval(name string) time.Duration {
   224  	md.RLock()
   225  	defer md.RUnlock()
   226  	if md.IsInRecovery && len(md.MetricsStandby) > 0 {
   227  		return time.Duration(md.MetricsStandby[name]) * time.Second
   228  	}
   229  	return time.Duration(md.Metrics[name]) * time.Second
   230  }
   231  
   232  // IsClientOnSameHost checks if the pgwatch client is running on the same host as the PostgreSQL server
   233  func (md *DbConn) IsClientOnSameHost() bool {
   234  	ok, err := db.IsClientOnSameHost(md.Conn)
   235  	return ok && err == nil
   236  }
   237  
   238  // SetDatabaseName sets the database name in the connection config for resolved databases
   239  func (md *DbConn) SetDatabaseName(name string) {
   240  	if err := md.ParseConfig(); err != nil {
   241  		return
   242  	}
   243  	md.ConnStr = "" // unset the connection string to force conn config usage
   244  	md.ConnConfig.ConnConfig.Database = name
   245  }
   246  
   247  func (md *DbConn) IsPostgresSource() bool {
   248  	switch md.Kind {
   249  	case SourcePostgres, SourcePatroniDiscovery, SourcePostgresDiscovery:
   250  		return true
   251  	default:
   252  		return false
   253  	}
   254  }
   255  
   256  // VersionToInt parses a given version and returns an integer  or
   257  // an error if unable to parse the version. Only parses valid semantic versions.
   258  // Performs checking that can find errors within the version.
   259  // Examples: v1.2 -> 01_02_00, v9.6.3 -> 09_06_03, v11 -> 11_00_00
   260  var regVer = regexp.MustCompile(`(\d+).?(\d*).?(\d*)`)
   261  
   262  func VersionToInt(version string) (v int) {
   263  	if matches := regVer.FindStringSubmatch(version); len(matches) > 1 {
   264  		for i, match := range matches[1:] {
   265  			v += func() (m int) { m, _ = strconv.Atoi(match); return }() * int(math.Pow10(4-i*2))
   266  		}
   267  	}
   268  	return
   269  }
   270  
   271  func (md *DbConn) FetchRuntimeInfo(ctx context.Context, forceRefetch bool) (err error) {
   272  	// Fast path: check the atomic timestamp without acquiring any lock.
   273  	// This avoids lock contention when the cached value is still fresh.
   274  	if !forceRefetch && time.Duration(time.Now().UnixNano()-md.lastCheckedNs.Load()) < 5*time.Minute {
   275  		return
   276  	}
   277  	if ctx.Err() != nil {
   278  		return ctx.Err()
   279  	}
   280  
   281  	md.Lock()
   282  	defer md.Unlock()
   283  
   284  	switch md.Kind {
   285  	case SourcePgBouncer, SourcePgPool:
   286  		err = md.FetchVersion(ctx, md.Kind)
   287  	default:
   288  		err = errors.Join(
   289  			md.FetchControlInfo(ctx),
   290  			md.DiscoverPlatform(ctx),
   291  			md.FetchApproxSize(ctx),
   292  			md.FetchExtensions(ctx))
   293  	}
   294  	if err == nil {
   295  		md.lastCheckedNs.Store(time.Now().UnixNano())
   296  	}
   297  	return
   298  }
   299  
   300  // FetchControlInfo queries pg_control_system() and populates the core RuntimeInfo fields.
   301  func (md *DbConn) FetchControlInfo(ctx context.Context) error {
   302  	sql := `select /* pgwatch_generated */
   303  div(current_setting('server_version_num')::int, 10000) as ver,
   304  version(),
   305  pg_is_in_recovery(),
   306  current_database()::TEXT,
   307  system_identifier,
   308  current_setting('is_superuser')::bool
   309  FROM
   310  	pg_control_system()`
   311  	controlCtx, controlCancel := db.WithOpTimeout(ctx, "runtime_info control", db.RuntimeInfoTimeout)
   312  	defer controlCancel()
   313  	return md.Conn.QueryRow(controlCtx, sql).
   314  		Scan(&md.Version, &md.VersionStr,
   315  			&md.IsInRecovery, &md.RealDbname,
   316  			&md.SystemIdentifier, &md.IsSuperuser)
   317  }
   318  
   319  // FetchExtensions queries pg_extension and populates md.Extensions with the installed extension versions.
   320  func (md *DbConn) FetchExtensions(ctx context.Context) error {
   321  	sqlExtensions := `select /* pgwatch_generated */ extname::text, (regexp_matches(extversion, $$\d+\.?\d+?$$))[1]::text as extversion from pg_extension order by 1;`
   322  	extCtx, extCancel := db.WithOpTimeout(ctx, "runtime_info extensions", db.RuntimeInfoTimeout)
   323  	defer extCancel()
   324  	res, err := md.Conn.Query(extCtx, sqlExtensions)
   325  	if err != nil {
   326  		return err
   327  	}
   328  	var ext, ver string
   329  	_, err = pgx.ForEachRow(res, []any{&ext, &ver}, func() error {
   330  		extver := VersionToInt(ver)
   331  		if extver == 0 {
   332  			return fmt.Errorf("unexpected extension %s version input: %s", ext, ver)
   333  		}
   334  		md.Extensions[ext] = extver
   335  		return nil
   336  	})
   337  	return err
   338  }
   339  
   340  func (md *DbConn) FetchVersion(ctx context.Context, kind Kind) (err error) {
   341  	sqls := map[Kind]string{SourcePgBouncer: "SHOW VERSION", SourcePgPool: "SHOW POOL_VERSION"}
   342  	versionCtx, versionCancel := db.WithOpTimeout(ctx, "runtime_info version", db.RuntimeInfoTimeout)
   343  	defer versionCancel()
   344  	if err = md.Conn.QueryRow(versionCtx, sqls[kind], pgx.QueryExecModeSimpleProtocol).Scan(&md.VersionStr); err != nil {
   345  		return
   346  	}
   347  	md.Version = VersionToInt(md.VersionStr)
   348  	return
   349  }
   350  
   351  // DiscoverPlatform tries to discover the platform based on the database version string and some special settings
   352  // that are only available on certain platforms. Populates md.ExecEnv.
   353  func (md *DbConn) DiscoverPlatform(ctx context.Context) error {
   354  	if md.ExecEnv != "" {
   355  		return nil // carry over as not likely to change ever
   356  	}
   357  	platformCtx, platformCancel := db.WithOpTimeout(ctx, "runtime_info platform", db.RuntimeInfoTimeout)
   358  	defer platformCancel()
   359  	sql := `select /* pgwatch_generated */
   360  	case
   361  	  when exists (select * from pg_settings where name = 'pg_qs.host_database' and setting = 'azure_sys') and version() ~* 'compiled by Visual C' then 'AZURE_SINGLE'
   362  	  when exists (select * from pg_settings where name = 'pg_qs.host_database' and setting = 'azure_sys') and version() ~* 'compiled by gcc' then 'AZURE_FLEXIBLE'
   363  	  when exists (select * from pg_settings where name = 'cloudsql.supported_extensions') then 'GOOGLE'
   364  	else
   365  	  'UNKNOWN'
   366  	end as exec_env`
   367  	return md.Conn.QueryRow(platformCtx, sql).Scan(&md.ExecEnv)
   368  }
   369  
   370  // FetchApproxSize fetches the approximate size of the database in bytes and populates md.ApproxDbSize.
   371  func (md *DbConn) FetchApproxSize(ctx context.Context) error {
   372  	sizeCtx, sizeCancel := db.WithOpTimeout(ctx, "runtime_info size", db.RuntimeInfoTimeout)
   373  	defer sizeCancel()
   374  	sqlApproxDBSize := `select /* pgwatch_generated */ current_setting('block_size')::int8 * sum(relpages) from pg_class c where c.relpersistence != 't'`
   375  	return md.Conn.QueryRow(sizeCtx, sqlApproxDBSize).Scan(&md.ApproxDbSize)
   376  }
   377  
   378  // FunctionExists checks if a function exists in the database
   379  func (md *DbConn) FunctionExists(ctx context.Context, functionName string) (exists bool) {
   380  	sql := `select /* pgwatch_generated */ true 
   381  from 
   382  	pg_proc join pg_namespace n on pronamespace = n.oid 
   383  where 
   384  	proname = $1 and n.nspname = 'public'`
   385  	_ = md.Conn.QueryRow(ctx, sql, functionName).Scan(&exists)
   386  	return
   387  }
   388  
   389  // TryCreateMissingExtensions should be called once on daemon startup if some commonly wanted extension (most notably pg_stat_statements) is missing.
   390  func (md *DbConn) TryCreateMissingExtensions(ctx context.Context, extensions []string) (string, error) {
   391  	// Snapshot the already-known extensions under RLock; release before doing any I/O.
   392  	md.RLock()
   393  	knownExts := maps.Clone(md.Extensions)
   394  	md.RUnlock()
   395  
   396  	sqlAvailableExts := `select name::text from pg_available_extensions order by 1`
   397  	createdExts := make([]string, 0)
   398  
   399  	availableCtx, availableCancel := db.WithOpTimeout(ctx, "available extensions", db.RuntimeInfoTimeout)
   400  	data, err := md.Conn.Query(availableCtx, sqlAvailableExts)
   401  	if err != nil {
   402  		availableCancel()
   403  		return "", err
   404  	}
   405  	availableExts, err := pgx.CollectRows(data, pgx.RowTo[string])
   406  	availableCancel()
   407  	if err != nil {
   408  		return "", err
   409  	}
   410  
   411  	for _, extToCreate := range extensions {
   412  		if _, ok := knownExts[extToCreate]; ok {
   413  			continue
   414  		}
   415  		if _, ok := slices.BinarySearch(availableExts, extToCreate); !ok {
   416  			err = errors.Join(err, fmt.Errorf("requested extension %s is not available on instance", extToCreate))
   417  			continue
   418  		}
   419  		if _, e := md.Conn.Exec(ctx, fmt.Sprintf(`create extension if not exists "%s"`, extToCreate)); e != nil {
   420  			err = errors.Join(err, fmt.Errorf("failed to create extension %s: %w", extToCreate, e))
   421  		} else {
   422  			createdExts = append(createdExts, extToCreate)
   423  		}
   424  	}
   425  	return strings.Join(createdExts, ","), err
   426  }
   427  
   428  // TryCreateMetricsHelpers should be called once on daemon startup to try to create "metric fetching helper" functions automatically
   429  func (md *DbConn) TryCreateMetricsHelpers(ctx context.Context, getSQLFn func(string) string) (err error) {
   430  	// Clone the metric map under RLock; release before doing any I/O.
   431  	md.RLock()
   432  	metricsMap := maps.Clone(md.Metrics)
   433  	maps.Insert(metricsMap, maps.All(md.MetricsStandby))
   434  	md.RUnlock()
   435  
   436  	var sql string
   437  	for metricName := range metricsMap {
   438  		if sql = getSQLFn(metricName); sql == "" {
   439  			continue
   440  		}
   441  		if _, e := md.Conn.Exec(ctx, sql); e != nil {
   442  			err = errors.Join(err, fmt.Errorf("failed to create helper for metric %s: %w", metricName, e))
   443  		}
   444  	}
   445  	return
   446  }
   447  
   448  func (mds SourceConns) GetMonitoredDatabase(DBUniqueName string) SourceConn {
   449  	for _, md := range mds {
   450  		if md.GetSource().Name == DBUniqueName {
   451  			return md
   452  		}
   453  	}
   454  	return nil
   455  }
   456  
   457  // RedactURL replaces the password in URL userinfo with "xxxxx".
   458  // If rawURL cannot be parsed or has no password, it is returned unchanged.
   459  func RedactURL(rawURL string) string {
   460  	u, err := url.Parse(rawURL)
   461  	if err != nil {
   462  		return rawURL
   463  	}
   464  	if u.User == nil {
   465  		return rawURL
   466  	}
   467  	if _, hasPass := u.User.Password(); !hasPass {
   468  		return rawURL
   469  	}
   470  	u.User = url.UserPassword(u.User.Username(), "xxxxx")
   471  	return u.String()
   472  }
   473  
   474  // promConnConfig holds the parsed Prometheus source connection parameters.
   475  // It is populated once by ParseConfig and reused by Connect and Ping.
   476  type promConnConfig struct {
   477  	URL       string
   478  	Userinfo  *url.Userinfo
   479  	TLSConfig *tls.Config
   480  }
   481  
   482  // PromConn represents a Prometheus source connection.
   483  type PromConn struct {
   484  	Source
   485  	connConfig *promConnConfig
   486  	HTTPClient *http.Client
   487  	sync.RWMutex
   488  }
   489  
   490  func NewPromConn(s Source) *PromConn {
   491  	return &PromConn{
   492  		Source: s,
   493  	}
   494  }
   495  
   496  // ParseConfig parses pc.ConnStr once and caches the result in pc.connConfig.
   497  // Subsequent calls are no-ops. Mirrors DbConn.ParseConfig.
   498  func (pc *PromConn) ParseConfig() error {
   499  	if pc.connConfig != nil {
   500  		return nil
   501  	}
   502  	u, err := url.Parse(pc.ConnStr)
   503  	if err != nil {
   504  		return fmt.Errorf("parsing prometheus source URL: %w", err)
   505  	}
   506  	userinfo := u.User
   507  	u.User = nil
   508  
   509  	q := u.Query()
   510  	tlsRootCert := q.Get("tlsrootcert")
   511  	tlsSkipVerify := q.Get("tlsskipverify") == "true"
   512  	q.Del("tlsrootcert")
   513  	q.Del("tlsskipverify")
   514  	u.RawQuery = q.Encode()
   515  
   516  	tlsConfig := &tls.Config{InsecureSkipVerify: tlsSkipVerify} //nolint:gosec // intentional per config
   517  	if tlsRootCert != "" {
   518  		caCert, readErr := os.ReadFile(tlsRootCert)
   519  		if readErr != nil {
   520  			return fmt.Errorf("reading tlsrootcert %q: %w", tlsRootCert, readErr)
   521  		}
   522  		pool := x509.NewCertPool()
   523  		pool.AppendCertsFromPEM(caCert)
   524  		tlsConfig.RootCAs = pool
   525  	}
   526  
   527  	pc.connConfig = &promConnConfig{
   528  		URL:       u.String(),
   529  		Userinfo:  userinfo,
   530  		TLSConfig: tlsConfig,
   531  	}
   532  	return nil
   533  }
   534  
   535  func (pc *PromConn) Connect(ctx context.Context, _ CmdOpts) error {
   536  	pc.Lock()
   537  	if pc.HTTPClient == nil {
   538  		if err := pc.ParseConfig(); err != nil {
   539  			pc.Unlock()
   540  			return err
   541  		}
   542  		pc.HTTPClient = &http.Client{
   543  			Transport: &http.Transport{TLSClientConfig: pc.connConfig.TLSConfig},
   544  			Timeout:   30 * time.Second,
   545  			CheckRedirect: func(*http.Request, []*http.Request) error {
   546  				return http.ErrUseLastResponse
   547  			},
   548  		}
   549  	}
   550  	pc.Unlock()
   551  	return pc.Ping(ctx)
   552  }
   553  
   554  // Scrape executes a single GET request to the source's metrics endpoint with
   555  // Accept: text/plain and optional Basic Auth from the cached config.
   556  // The caller is responsible for closing resp.Body.
   557  // Connect must be called before Scrape.
   558  func (pc *PromConn) Scrape(ctx context.Context) (*http.Response, error) {
   559  	pc.RLock()
   560  	client := pc.HTTPClient
   561  	cfg := pc.connConfig
   562  	pc.RUnlock()
   563  
   564  	if client == nil || cfg == nil {
   565  		return nil, errors.New("prometheus source not connected: call Connect first")
   566  	}
   567  
   568  	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
   569  	if err != nil {
   570  		return nil, err
   571  	}
   572  	req.Header.Set("Accept", "text/plain")
   573  	if cfg.Userinfo != nil {
   574  		pass, _ := cfg.Userinfo.Password()
   575  		req.SetBasicAuth(cfg.Userinfo.Username(), pass)
   576  	}
   577  	return client.Do(req)
   578  }
   579  
   580  func (pc *PromConn) Ping(ctx context.Context) error {
   581  	pc.RLock()
   582  	client := pc.HTTPClient
   583  	cfg := pc.connConfig
   584  	pc.RUnlock()
   585  
   586  	if client == nil || cfg == nil {
   587  		return errors.New("prometheus source not connected: call Connect first")
   588  	}
   589  
   590  	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
   591  	if err != nil {
   592  		return err
   593  	}
   594  	if cfg.Userinfo != nil {
   595  		pass, _ := cfg.Userinfo.Password()
   596  		req.SetBasicAuth(cfg.Userinfo.Username(), pass)
   597  	}
   598  
   599  	resp, err := client.Do(req)
   600  	if err != nil {
   601  		return err
   602  	}
   603  	_, _ = io.Copy(io.Discard, resp.Body)
   604  	_ = resp.Body.Close()
   605  
   606  	if resp.StatusCode >= 300 {
   607  		return fmt.Errorf("prometheus ping: unexpected status %s", resp.Status)
   608  	}
   609  	return nil
   610  }
   611  
   612  func (pc *PromConn) IsPostgresSource() bool                           { return false }
   613  func (pc *PromConn) GetSource() Source                                { return pc.Source }
   614  func (pc *PromConn) FetchRuntimeInfo(_ context.Context, _ bool) error { return nil }
   615  func (pc *PromConn) Close()                                           {}
   616  
   617  func (pc *PromConn) GetMetricInterval(name string) time.Duration {
   618  	pc.RLock()
   619  	defer pc.RUnlock()
   620  	return time.Duration(pc.Metrics[name]) * time.Second
   621  }
   622  
   623  func (pc *PromConn) SetMetricIntervals(main, _ metrics.MetricIntervals) {
   624  	pc.Lock()
   625  	defer pc.Unlock()
   626  	if main != nil {
   627  		pc.Metrics = main
   628  	}
   629  }
   630