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  		}
   181  		if opts.MaxParallelConnectionsPerDb > 0 {
   182  			md.ConnConfig.MaxConns = int32(opts.MaxParallelConnectionsPerDb)
   183  		}
   184  		md.Conn, err = NewConnWithConfig(ctx, md.ConnConfig)
   185  		if err != nil {
   186  			return err
   187  		}
   188  	}
   189  	return md.Ping(ctx)
   190  }
   191  
   192  // ParseConfig will parse the connection string and store the result in the connection config
   193  func (md *DbConn) ParseConfig() (err error) {
   194  	if md.ConnConfig == nil {
   195  		md.ConnConfig, err = pgxpool.ParseConfig(md.ConnStr)
   196  		return
   197  	}
   198  	return
   199  }
   200  
   201  // GetClusterIdentifier returns a unique identifier for the host assuming SysId is the same for
   202  // primary and all replicas but connection information is different
   203  func (md *DbConn) GetClusterIdentifier() string {
   204  	if err := md.ParseConfig(); err != nil {
   205  		return ""
   206  	}
   207  	md.RLock()
   208  	defer md.RUnlock()
   209  	return fmt.Sprintf("%s:%s:%d", md.SystemIdentifier, md.ConnConfig.ConnConfig.Host, md.ConnConfig.ConnConfig.Port)
   210  }
   211  
   212  // GetDatabaseName returns the database name from the connection string
   213  func (md *DbConn) GetDatabaseName() string {
   214  	if err := md.ParseConfig(); err != nil {
   215  		return ""
   216  	}
   217  	return md.ConnConfig.ConnConfig.Database
   218  }
   219  
   220  // GetMetricInterval returns the metric interval for the connection
   221  func (md *DbConn) GetMetricInterval(name string) time.Duration {
   222  	md.RLock()
   223  	defer md.RUnlock()
   224  	if md.IsInRecovery && len(md.MetricsStandby) > 0 {
   225  		return time.Duration(md.MetricsStandby[name]) * time.Second
   226  	}
   227  	return time.Duration(md.Metrics[name]) * time.Second
   228  }
   229  
   230  // IsClientOnSameHost checks if the pgwatch client is running on the same host as the PostgreSQL server
   231  func (md *DbConn) IsClientOnSameHost() bool {
   232  	ok, err := db.IsClientOnSameHost(md.Conn)
   233  	return ok && err == nil
   234  }
   235  
   236  // SetDatabaseName sets the database name in the connection config for resolved databases
   237  func (md *DbConn) SetDatabaseName(name string) {
   238  	if err := md.ParseConfig(); err != nil {
   239  		return
   240  	}
   241  	md.ConnStr = "" // unset the connection string to force conn config usage
   242  	md.ConnConfig.ConnConfig.Database = name
   243  }
   244  
   245  func (md *DbConn) IsPostgresSource() bool {
   246  	switch md.Kind {
   247  	case SourcePostgres, SourcePatroniDiscovery, SourcePostgresDiscovery:
   248  		return true
   249  	default:
   250  		return false
   251  	}
   252  }
   253  
   254  // VersionToInt parses a given version and returns an integer  or
   255  // an error if unable to parse the version. Only parses valid semantic versions.
   256  // Performs checking that can find errors within the version.
   257  // Examples: v1.2 -> 01_02_00, v9.6.3 -> 09_06_03, v11 -> 11_00_00
   258  var regVer = regexp.MustCompile(`(\d+).?(\d*).?(\d*)`)
   259  
   260  func VersionToInt(version string) (v int) {
   261  	if matches := regVer.FindStringSubmatch(version); len(matches) > 1 {
   262  		for i, match := range matches[1:] {
   263  			v += func() (m int) { m, _ = strconv.Atoi(match); return }() * int(math.Pow10(4-i*2))
   264  		}
   265  	}
   266  	return
   267  }
   268  
   269  func (md *DbConn) FetchRuntimeInfo(ctx context.Context, forceRefetch bool) (err error) {
   270  	// Fast path: check the atomic timestamp without acquiring any lock.
   271  	// This avoids lock contention when the cached value is still fresh.
   272  	if !forceRefetch && time.Duration(time.Now().UnixNano()-md.lastCheckedNs.Load()) < 5*time.Minute {
   273  		return
   274  	}
   275  	if ctx.Err() != nil {
   276  		return ctx.Err()
   277  	}
   278  
   279  	md.Lock()
   280  	defer md.Unlock()
   281  
   282  	switch md.Kind {
   283  	case SourcePgBouncer, SourcePgPool:
   284  		err = md.FetchVersion(ctx, md.Kind)
   285  	default:
   286  		err = errors.Join(
   287  			md.FetchControlInfo(ctx),
   288  			md.DiscoverPlatform(ctx),
   289  			md.FetchApproxSize(ctx),
   290  			md.FetchExtensions(ctx))
   291  	}
   292  	if err == nil {
   293  		md.lastCheckedNs.Store(time.Now().UnixNano())
   294  	}
   295  	return
   296  }
   297  
   298  // FetchControlInfo queries pg_control_system() and populates the core RuntimeInfo fields.
   299  func (md *DbConn) FetchControlInfo(ctx context.Context) error {
   300  	sql := `select /* pgwatch_generated */
   301  div(current_setting('server_version_num')::int, 10000) as ver,
   302  version(),
   303  pg_is_in_recovery(),
   304  current_database()::TEXT,
   305  system_identifier,
   306  current_setting('is_superuser')::bool
   307  FROM
   308  	pg_control_system()`
   309  	controlCtx, controlCancel := db.WithOpTimeout(ctx, "runtime_info control", db.RuntimeInfoTimeout)
   310  	defer controlCancel()
   311  	return md.Conn.QueryRow(controlCtx, sql).
   312  		Scan(&md.Version, &md.VersionStr,
   313  			&md.IsInRecovery, &md.RealDbname,
   314  			&md.SystemIdentifier, &md.IsSuperuser)
   315  }
   316  
   317  // FetchExtensions queries pg_extension and populates md.Extensions with the installed extension versions.
   318  func (md *DbConn) FetchExtensions(ctx context.Context) error {
   319  	sqlExtensions := `select /* pgwatch_generated */ extname::text, (regexp_matches(extversion, $$\d+\.?\d+?$$))[1]::text as extversion from pg_extension order by 1;`
   320  	extCtx, extCancel := db.WithOpTimeout(ctx, "runtime_info extensions", db.RuntimeInfoTimeout)
   321  	defer extCancel()
   322  	res, err := md.Conn.Query(extCtx, sqlExtensions)
   323  	if err != nil {
   324  		return err
   325  	}
   326  	var ext, ver string
   327  	_, err = pgx.ForEachRow(res, []any{&ext, &ver}, func() error {
   328  		extver := VersionToInt(ver)
   329  		if extver == 0 {
   330  			return fmt.Errorf("unexpected extension %s version input: %s", ext, ver)
   331  		}
   332  		md.Extensions[ext] = extver
   333  		return nil
   334  	})
   335  	return err
   336  }
   337  
   338  func (md *DbConn) FetchVersion(ctx context.Context, kind Kind) (err error) {
   339  	sqls := map[Kind]string{SourcePgBouncer: "SHOW VERSION", SourcePgPool: "SHOW POOL_VERSION"}
   340  	versionCtx, versionCancel := db.WithOpTimeout(ctx, "runtime_info version", db.RuntimeInfoTimeout)
   341  	defer versionCancel()
   342  	if err = md.Conn.QueryRow(versionCtx, sqls[kind], pgx.QueryExecModeSimpleProtocol).Scan(&md.VersionStr); err != nil {
   343  		return
   344  	}
   345  	md.Version = VersionToInt(md.VersionStr)
   346  	return
   347  }
   348  
   349  // DiscoverPlatform tries to discover the platform based on the database version string and some special settings
   350  // that are only available on certain platforms. Populates md.ExecEnv.
   351  func (md *DbConn) DiscoverPlatform(ctx context.Context) error {
   352  	if md.ExecEnv != "" {
   353  		return nil // carry over as not likely to change ever
   354  	}
   355  	platformCtx, platformCancel := db.WithOpTimeout(ctx, "runtime_info platform", db.RuntimeInfoTimeout)
   356  	defer platformCancel()
   357  	sql := `select /* pgwatch_generated */
   358  	case
   359  	  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'
   360  	  when exists (select * from pg_settings where name = 'pg_qs.host_database' and setting = 'azure_sys') and version() ~* 'compiled by gcc' then 'AZURE_FLEXIBLE'
   361  	  when exists (select * from pg_settings where name = 'cloudsql.supported_extensions') then 'GOOGLE'
   362  	else
   363  	  'UNKNOWN'
   364  	end as exec_env`
   365  	return md.Conn.QueryRow(platformCtx, sql).Scan(&md.ExecEnv)
   366  }
   367  
   368  // FetchApproxSize fetches the approximate size of the database in bytes and populates md.ApproxDbSize.
   369  func (md *DbConn) FetchApproxSize(ctx context.Context) error {
   370  	sizeCtx, sizeCancel := db.WithOpTimeout(ctx, "runtime_info size", db.RuntimeInfoTimeout)
   371  	defer sizeCancel()
   372  	sqlApproxDBSize := `select /* pgwatch_generated */ current_setting('block_size')::int8 * sum(relpages) from pg_class c where c.relpersistence != 't'`
   373  	return md.Conn.QueryRow(sizeCtx, sqlApproxDBSize).Scan(&md.ApproxDbSize)
   374  }
   375  
   376  // FunctionExists checks if a function exists in the database
   377  func (md *DbConn) FunctionExists(ctx context.Context, functionName string) (exists bool) {
   378  	sql := `select /* pgwatch_generated */ true 
   379  from 
   380  	pg_proc join pg_namespace n on pronamespace = n.oid 
   381  where 
   382  	proname = $1 and n.nspname = 'public'`
   383  	_ = md.Conn.QueryRow(ctx, sql, functionName).Scan(&exists)
   384  	return
   385  }
   386  
   387  // TryCreateMissingExtensions should be called once on daemon startup if some commonly wanted extension (most notably pg_stat_statements) is missing.
   388  func (md *DbConn) TryCreateMissingExtensions(ctx context.Context, extensions []string) (string, error) {
   389  	// Snapshot the already-known extensions under RLock; release before doing any I/O.
   390  	md.RLock()
   391  	knownExts := maps.Clone(md.Extensions)
   392  	md.RUnlock()
   393  
   394  	sqlAvailableExts := `select name::text from pg_available_extensions order by 1`
   395  	createdExts := make([]string, 0)
   396  
   397  	availableCtx, availableCancel := db.WithOpTimeout(ctx, "available extensions", db.RuntimeInfoTimeout)
   398  	data, err := md.Conn.Query(availableCtx, sqlAvailableExts)
   399  	if err != nil {
   400  		availableCancel()
   401  		return "", err
   402  	}
   403  	availableExts, err := pgx.CollectRows(data, pgx.RowTo[string])
   404  	availableCancel()
   405  	if err != nil {
   406  		return "", err
   407  	}
   408  
   409  	for _, extToCreate := range extensions {
   410  		if _, ok := knownExts[extToCreate]; ok {
   411  			continue
   412  		}
   413  		if _, ok := slices.BinarySearch(availableExts, extToCreate); !ok {
   414  			err = errors.Join(err, fmt.Errorf("requested extension %s is not available on instance", extToCreate))
   415  			continue
   416  		}
   417  		if _, e := md.Conn.Exec(ctx, fmt.Sprintf(`create extension if not exists "%s"`, extToCreate)); e != nil {
   418  			err = errors.Join(err, fmt.Errorf("failed to create extension %s: %w", extToCreate, e))
   419  		} else {
   420  			createdExts = append(createdExts, extToCreate)
   421  		}
   422  	}
   423  	return strings.Join(createdExts, ","), err
   424  }
   425  
   426  // TryCreateMetricsHelpers should be called once on daemon startup to try to create "metric fetching helper" functions automatically
   427  func (md *DbConn) TryCreateMetricsHelpers(ctx context.Context, getSQLFn func(string) string) (err error) {
   428  	// Clone the metric map under RLock; release before doing any I/O.
   429  	md.RLock()
   430  	metricsMap := maps.Clone(md.Metrics)
   431  	maps.Insert(metricsMap, maps.All(md.MetricsStandby))
   432  	md.RUnlock()
   433  
   434  	var sql string
   435  	for metricName := range metricsMap {
   436  		if sql = getSQLFn(metricName); sql == "" {
   437  			continue
   438  		}
   439  		if _, e := md.Conn.Exec(ctx, sql); e != nil {
   440  			err = errors.Join(err, fmt.Errorf("failed to create helper for metric %s: %w", metricName, e))
   441  		}
   442  	}
   443  	return
   444  }
   445  
   446  func (mds SourceConns) GetMonitoredDatabase(DBUniqueName string) SourceConn {
   447  	for _, md := range mds {
   448  		if md.GetSource().Name == DBUniqueName {
   449  			return md
   450  		}
   451  	}
   452  	return nil
   453  }
   454  
   455  // RedactURL replaces the password in URL userinfo with "xxxxx".
   456  // If rawURL cannot be parsed or has no password, it is returned unchanged.
   457  func RedactURL(rawURL string) string {
   458  	u, err := url.Parse(rawURL)
   459  	if err != nil {
   460  		return rawURL
   461  	}
   462  	if u.User == nil {
   463  		return rawURL
   464  	}
   465  	if _, hasPass := u.User.Password(); !hasPass {
   466  		return rawURL
   467  	}
   468  	u.User = url.UserPassword(u.User.Username(), "xxxxx")
   469  	return u.String()
   470  }
   471  
   472  // promConnConfig holds the parsed Prometheus source connection parameters.
   473  // It is populated once by ParseConfig and reused by Connect and Ping.
   474  type promConnConfig struct {
   475  	URL       string
   476  	Userinfo  *url.Userinfo
   477  	TLSConfig *tls.Config
   478  }
   479  
   480  // PromConn represents a Prometheus source connection.
   481  type PromConn struct {
   482  	Source
   483  	connConfig *promConnConfig
   484  	HTTPClient *http.Client
   485  	sync.RWMutex
   486  }
   487  
   488  func NewPromConn(s Source) *PromConn {
   489  	return &PromConn{
   490  		Source: s,
   491  	}
   492  }
   493  
   494  // ParseConfig parses pc.ConnStr once and caches the result in pc.connConfig.
   495  // Subsequent calls are no-ops. Mirrors DbConn.ParseConfig.
   496  func (pc *PromConn) ParseConfig() error {
   497  	if pc.connConfig != nil {
   498  		return nil
   499  	}
   500  	u, err := url.Parse(pc.ConnStr)
   501  	if err != nil {
   502  		return fmt.Errorf("parsing prometheus source URL: %w", err)
   503  	}
   504  	userinfo := u.User
   505  	u.User = nil
   506  
   507  	q := u.Query()
   508  	tlsRootCert := q.Get("tlsrootcert")
   509  	tlsSkipVerify := q.Get("tlsskipverify") == "true"
   510  	q.Del("tlsrootcert")
   511  	q.Del("tlsskipverify")
   512  	u.RawQuery = q.Encode()
   513  
   514  	tlsConfig := &tls.Config{InsecureSkipVerify: tlsSkipVerify} //nolint:gosec // intentional per config
   515  	if tlsRootCert != "" {
   516  		caCert, readErr := os.ReadFile(tlsRootCert)
   517  		if readErr != nil {
   518  			return fmt.Errorf("reading tlsrootcert %q: %w", tlsRootCert, readErr)
   519  		}
   520  		pool := x509.NewCertPool()
   521  		pool.AppendCertsFromPEM(caCert)
   522  		tlsConfig.RootCAs = pool
   523  	}
   524  
   525  	pc.connConfig = &promConnConfig{
   526  		URL:       u.String(),
   527  		Userinfo:  userinfo,
   528  		TLSConfig: tlsConfig,
   529  	}
   530  	return nil
   531  }
   532  
   533  func (pc *PromConn) Connect(ctx context.Context, _ CmdOpts) error {
   534  	pc.Lock()
   535  	if pc.HTTPClient == nil {
   536  		if err := pc.ParseConfig(); err != nil {
   537  			pc.Unlock()
   538  			return err
   539  		}
   540  		pc.HTTPClient = &http.Client{
   541  			Transport: &http.Transport{TLSClientConfig: pc.connConfig.TLSConfig},
   542  			Timeout:   30 * time.Second,
   543  			CheckRedirect: func(*http.Request, []*http.Request) error {
   544  				return http.ErrUseLastResponse
   545  			},
   546  		}
   547  	}
   548  	pc.Unlock()
   549  	return pc.Ping(ctx)
   550  }
   551  
   552  // Scrape executes a single GET request to the source's metrics endpoint with
   553  // Accept: text/plain and optional Basic Auth from the cached config.
   554  // The caller is responsible for closing resp.Body.
   555  // Connect must be called before Scrape.
   556  func (pc *PromConn) Scrape(ctx context.Context) (*http.Response, error) {
   557  	pc.RLock()
   558  	client := pc.HTTPClient
   559  	cfg := pc.connConfig
   560  	pc.RUnlock()
   561  
   562  	if client == nil || cfg == nil {
   563  		return nil, errors.New("prometheus source not connected: call Connect first")
   564  	}
   565  
   566  	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
   567  	if err != nil {
   568  		return nil, err
   569  	}
   570  	req.Header.Set("Accept", "text/plain")
   571  	if cfg.Userinfo != nil {
   572  		pass, _ := cfg.Userinfo.Password()
   573  		req.SetBasicAuth(cfg.Userinfo.Username(), pass)
   574  	}
   575  	return client.Do(req)
   576  }
   577  
   578  func (pc *PromConn) Ping(ctx context.Context) error {
   579  	pc.RLock()
   580  	client := pc.HTTPClient
   581  	cfg := pc.connConfig
   582  	pc.RUnlock()
   583  
   584  	if client == nil || cfg == nil {
   585  		return errors.New("prometheus source not connected: call Connect first")
   586  	}
   587  
   588  	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
   589  	if err != nil {
   590  		return err
   591  	}
   592  	if cfg.Userinfo != nil {
   593  		pass, _ := cfg.Userinfo.Password()
   594  		req.SetBasicAuth(cfg.Userinfo.Username(), pass)
   595  	}
   596  
   597  	resp, err := client.Do(req)
   598  	if err != nil {
   599  		return err
   600  	}
   601  	_, _ = io.Copy(io.Discard, resp.Body)
   602  	_ = resp.Body.Close()
   603  
   604  	if resp.StatusCode >= 300 {
   605  		return fmt.Errorf("prometheus ping: unexpected status %s", resp.Status)
   606  	}
   607  	return nil
   608  }
   609  
   610  func (pc *PromConn) IsPostgresSource() bool                           { return false }
   611  func (pc *PromConn) GetSource() Source                                { return pc.Source }
   612  func (pc *PromConn) FetchRuntimeInfo(_ context.Context, _ bool) error { return nil }
   613  func (pc *PromConn) Close()                                           {}
   614  
   615  func (pc *PromConn) GetMetricInterval(name string) time.Duration {
   616  	pc.RLock()
   617  	defer pc.RUnlock()
   618  	return time.Duration(pc.Metrics[name]) * time.Second
   619  }
   620  
   621  func (pc *PromConn) SetMetricIntervals(main, _ metrics.MetricIntervals) {
   622  	pc.Lock()
   623  	defer pc.Unlock()
   624  	if main != nil {
   625  		pc.Metrics = main
   626  	}
   627  }
   628