Merge pull request #552 from sysadmind/refactor-files

Refactor code into logical files

cmd/postgres_exporter/datasource.go

@@ -0,0 +1,168 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/url"
+	"os"
+	"regexp"
+	"strings"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+func (e *Exporter) discoverDatabaseDSNs() []string {
+	// connstring syntax is complex (and not sure if even regular).
+	// we don't need to parse it, so just superficially validate that it starts
+	// with a valid-ish keyword pair
+	connstringRe := regexp.MustCompile(`^ *[a-zA-Z0-9]+ *= *[^= ]+`)
+
+	dsns := make(map[string]struct{})
+	for _, dsn := range e.dsn {
+		var dsnURI *url.URL
+		var dsnConnstring string
+
+		if strings.HasPrefix(dsn, "postgresql://") {
+			var err error
+			dsnURI, err = url.Parse(dsn)
+			if err != nil {
+				level.Error(logger).Log("msg", "Unable to parse DSN as URI", "dsn", loggableDSN(dsn), "err", err)
+				continue
+			}
+		} else if connstringRe.MatchString(dsn) {
+			dsnConnstring = dsn
+		} else {
+			level.Error(logger).Log("msg", "Unable to parse DSN as either URI or connstring", "dsn", loggableDSN(dsn))
+			continue
+		}
+
+		server, err := e.servers.GetServer(dsn)
+		if err != nil {
+			level.Error(logger).Log("msg", "Error opening connection to database", "dsn", loggableDSN(dsn), "err", err)
+			continue
+		}
+		dsns[dsn] = struct{}{}
+
+		// If autoDiscoverDatabases is true, set first dsn as master database (Default: false)
+		server.master = true
+
+		databaseNames, err := queryDatabases(server)
+		if err != nil {
+			level.Error(logger).Log("msg", "Error querying databases", "dsn", loggableDSN(dsn), "err", err)
+			continue
+		}
+		for _, databaseName := range databaseNames {
+			if contains(e.excludeDatabases, databaseName) {
+				continue
+			}
+
+			if len(e.includeDatabases) != 0 && !contains(e.includeDatabases, databaseName) {
+				continue
+			}
+
+			if dsnURI != nil {
+				dsnURI.Path = databaseName
+				dsn = dsnURI.String()
+			} else {
+				// replacing one dbname with another is complicated.
+				// just append new dbname to override.
+				dsn = fmt.Sprintf("%s dbname=%s", dsnConnstring, databaseName)
+			}
+			dsns[dsn] = struct{}{}
+		}
+	}
+
+	result := make([]string, len(dsns))
+	index := 0
+	for dsn := range dsns {
+		result[index] = dsn
+		index++
+	}
+
+	return result
+}
+
+func (e *Exporter) scrapeDSN(ch chan<- prometheus.Metric, dsn string) error {
+	server, err := e.servers.GetServer(dsn)
+
+	if err != nil {
+		return &ErrorConnectToServer{fmt.Sprintf("Error opening connection to database (%s): %s", loggableDSN(dsn), err.Error())}
+	}
+
+	// Check if autoDiscoverDatabases is false, set dsn as master database (Default: false)
+	if !e.autoDiscoverDatabases {
+		server.master = true
+	}
+
+	// Check if map versions need to be updated
+	if err := e.checkMapVersions(ch, server); err != nil {
+		level.Warn(logger).Log("msg", "Proceeding with outdated query maps, as the Postgres version could not be determined", "err", err)
+	}
+
+	return server.Scrape(ch, e.disableSettingsMetrics)
+}
+
+// try to get the DataSource
+// DATA_SOURCE_NAME always wins so we do not break older versions
+// reading secrets from files wins over secrets in environment variables
+// DATA_SOURCE_NAME > DATA_SOURCE_{USER|PASS}_FILE > DATA_SOURCE_{USER|PASS}
+func getDataSources() ([]string, error) {
+	var dsn = os.Getenv("DATA_SOURCE_NAME")
+	if len(dsn) != 0 {
+		return strings.Split(dsn, ","), nil
+	}
+
+	var user, pass, uri string
+
+	dataSourceUserFile := os.Getenv("DATA_SOURCE_USER_FILE")
+	if len(dataSourceUserFile) != 0 {
+		fileContents, err := ioutil.ReadFile(dataSourceUserFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed loading data source user file %s: %s", dataSourceUserFile, err.Error())
+		}
+		user = strings.TrimSpace(string(fileContents))
+	} else {
+		user = os.Getenv("DATA_SOURCE_USER")
+	}
+
+	dataSourcePassFile := os.Getenv("DATA_SOURCE_PASS_FILE")
+	if len(dataSourcePassFile) != 0 {
+		fileContents, err := ioutil.ReadFile(dataSourcePassFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed loading data source pass file %s: %s", dataSourcePassFile, err.Error())
+		}
+		pass = strings.TrimSpace(string(fileContents))
+	} else {
+		pass = os.Getenv("DATA_SOURCE_PASS")
+	}
+
+	ui := url.UserPassword(user, pass).String()
+	dataSrouceURIFile := os.Getenv("DATA_SOURCE_URI_FILE")
+	if len(dataSrouceURIFile) != 0 {
+		fileContents, err := ioutil.ReadFile(dataSrouceURIFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed loading data source URI file %s: %s", dataSrouceURIFile, err.Error())
+		}
+		uri = strings.TrimSpace(string(fileContents))
+	} else {
+		uri = os.Getenv("DATA_SOURCE_URI")
+	}
+
+	dsn = "postgresql://" + ui + "@" + uri
+
+	return []string{dsn}, nil
+}

cmd/postgres_exporter/main.go

@@ -0,0 +1,129 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"net/http"
+	"os"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/prometheus/common/promlog"
+	"github.com/prometheus/common/promlog/flag"
+	"github.com/prometheus/common/version"
+	"github.com/prometheus/exporter-toolkit/web"
+	webflag "github.com/prometheus/exporter-toolkit/web/kingpinflag"
+	"gopkg.in/alecthomas/kingpin.v2"
+)
+
+var (
+	listenAddress          = kingpin.Flag("web.listen-address", "Address to listen on for web interface and telemetry.").Default(":9187").Envar("PG_EXPORTER_WEB_LISTEN_ADDRESS").String()
+	webConfig              = webflag.AddFlags(kingpin.CommandLine)
+	metricPath             = kingpin.Flag("web.telemetry-path", "Path under which to expose metrics.").Default("/metrics").Envar("PG_EXPORTER_WEB_TELEMETRY_PATH").String()
+	disableDefaultMetrics  = kingpin.Flag("disable-default-metrics", "Do not include default metrics.").Default("false").Envar("PG_EXPORTER_DISABLE_DEFAULT_METRICS").Bool()
+	disableSettingsMetrics = kingpin.Flag("disable-settings-metrics", "Do not include pg_settings metrics.").Default("false").Envar("PG_EXPORTER_DISABLE_SETTINGS_METRICS").Bool()
+	autoDiscoverDatabases  = kingpin.Flag("auto-discover-databases", "Whether to discover the databases on a server dynamically.").Default("false").Envar("PG_EXPORTER_AUTO_DISCOVER_DATABASES").Bool()
+	queriesPath            = kingpin.Flag("extend.query-path", "Path to custom queries to run.").Default("").Envar("PG_EXPORTER_EXTEND_QUERY_PATH").String()
+	onlyDumpMaps           = kingpin.Flag("dumpmaps", "Do not run, simply dump the maps.").Bool()
+	constantLabelsList     = kingpin.Flag("constantLabels", "A list of label=value separated by comma(,).").Default("").Envar("PG_EXPORTER_CONSTANT_LABELS").String()
+	excludeDatabases       = kingpin.Flag("exclude-databases", "A list of databases to remove when autoDiscoverDatabases is enabled").Default("").Envar("PG_EXPORTER_EXCLUDE_DATABASES").String()
+	includeDatabases       = kingpin.Flag("include-databases", "A list of databases to include when autoDiscoverDatabases is enabled").Default("").Envar("PG_EXPORTER_INCLUDE_DATABASES").String()
+	metricPrefix           = kingpin.Flag("metric-prefix", "A metric prefix can be used to have non-default (not \"pg\") prefixes for each of the metrics").Default("pg").Envar("PG_EXPORTER_METRIC_PREFIX").String()
+	logger                 = log.NewNopLogger()
+)
+
+// Metric name parts.
+const (
+	// Namespace for all metrics.
+	namespace = "pg"
+	// Subsystems.
+	exporter = "exporter"
+	// The name of the exporter.
+	exporterName = "postgres_exporter"
+	// Metric label used for static string data thats handy to send to Prometheus
+	// e.g. version
+	staticLabelName = "static"
+	// Metric label used for server identification.
+	serverLabelName = "server"
+)
+
+func main() {
+	kingpin.Version(version.Print(exporterName))
+	promlogConfig := &promlog.Config{}
+	flag.AddFlags(kingpin.CommandLine, promlogConfig)
+	kingpin.HelpFlag.Short('h')
+	kingpin.Parse()
+	logger = promlog.New(promlogConfig)
+
+	// landingPage contains the HTML served at '/'.
+	// TODO: Make this nicer and more informative.
+	var landingPage = []byte(`<html>
+	<head><title>Postgres exporter</title></head>
+	<body>
+	<h1>Postgres exporter</h1>
+	<p><a href='` + *metricPath + `'>Metrics</a></p>
+	</body>
+	</html>
+	`)
+
+	if *onlyDumpMaps {
+		dumpMaps()
+		return
+	}
+
+	dsn, err := getDataSources()
+	if err != nil {
+		level.Error(logger).Log("msg", "Failed reading data sources", "err", err.Error())
+		os.Exit(1)
+	}
+
+	if len(dsn) == 0 {
+		level.Error(logger).Log("msg", "Couldn't find environment variables describing the datasource to use")
+		os.Exit(1)
+	}
+
+	opts := []ExporterOpt{
+		DisableDefaultMetrics(*disableDefaultMetrics),
+		DisableSettingsMetrics(*disableSettingsMetrics),
+		AutoDiscoverDatabases(*autoDiscoverDatabases),
+		WithUserQueriesPath(*queriesPath),
+		WithConstantLabels(*constantLabelsList),
+		ExcludeDatabases(*excludeDatabases),
+		IncludeDatabases(*includeDatabases),
+	}
+
+	exporter := NewExporter(dsn, opts...)
+	defer func() {
+		exporter.servers.Close()
+	}()
+
+	prometheus.MustRegister(version.NewCollector(exporterName))
+
+	prometheus.MustRegister(exporter)
+
+	http.Handle(*metricPath, promhttp.Handler())
+	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/html; charset=UTF-8") // nolint: errcheck
+		w.Write(landingPage)                                       // nolint: errcheck
+	})
+
+	level.Info(logger).Log("msg", "Listening on address", "address", *listenAddress)
+	srv := &http.Server{Addr: *listenAddress}
+	if err := web.ListenAndServe(srv, *webConfig, logger); err != nil {
+		level.Error(logger).Log("msg", "Error running HTTP server", "err", err)
+		os.Exit(1)
+	}
+}

cmd/postgres_exporter/namespace.go

@@ -0,0 +1,264 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/blang/semver"
+	"github.com/go-kit/kit/log/level"
+	"github.com/lib/pq"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Query within a namespace mapping and emit metrics. Returns fatal errors if
+// the scrape fails, and a slice of errors if they were non-fatal.
+func queryNamespaceMapping(server *Server, namespace string, mapping MetricMapNamespace) ([]prometheus.Metric, []error, error) {
+	// Check for a query override for this namespace
+	query, found := server.queryOverrides[namespace]
+
+	// Was this query disabled (i.e. nothing sensible can be queried on cu
+	// version of PostgreSQL?
+	if query == "" && found {
+		// Return success (no pertinent data)
+		return []prometheus.Metric{}, []error{}, nil
+	}
+
+	// Don't fail on a bad scrape of one metric
+	var rows *sql.Rows
+	var err error
+
+	if !found {
+		// I've no idea how to avoid this properly at the moment, but this is
+		// an admin tool so you're not injecting SQL right?
+		rows, err = server.db.Query(fmt.Sprintf("SELECT * FROM %s;", namespace)) // nolint: gas
+	} else {
+		rows, err = server.db.Query(query)
+	}
+	if err != nil {
+		return []prometheus.Metric{}, []error{}, fmt.Errorf("Error running query on database %q: %s %v", server, namespace, err)
+	}
+	defer rows.Close() // nolint: errcheck
+
+	var columnNames []string
+	columnNames, err = rows.Columns()
+	if err != nil {
+		return []prometheus.Metric{}, []error{}, errors.New(fmt.Sprintln("Error retrieving column list for: ", namespace, err))
+	}
+
+	// Make a lookup map for the column indices
+	var columnIdx = make(map[string]int, len(columnNames))
+	for i, n := range columnNames {
+		columnIdx[n] = i
+	}
+
+	var columnData = make([]interface{}, len(columnNames))
+	var scanArgs = make([]interface{}, len(columnNames))
+	for i := range columnData {
+		scanArgs[i] = &columnData[i]
+	}
+
+	nonfatalErrors := []error{}
+
+	metrics := make([]prometheus.Metric, 0)
+
+	for rows.Next() {
+		err = rows.Scan(scanArgs...)
+		if err != nil {
+			return []prometheus.Metric{}, []error{}, errors.New(fmt.Sprintln("Error retrieving rows:", namespace, err))
+		}
+
+		// Get the label values for this row.
+		labels := make([]string, len(mapping.labels))
+		for idx, label := range mapping.labels {
+			labels[idx], _ = dbToString(columnData[columnIdx[label]])
+		}
+
+		// Loop over column names, and match to scan data. Unknown columns
+		// will be filled with an untyped metric number *if* they can be
+		// converted to float64s. NULLs are allowed and treated as NaN.
+		for idx, columnName := range columnNames {
+			var metric prometheus.Metric
+			if metricMapping, ok := mapping.columnMappings[columnName]; ok {
+				// Is this a metricy metric?
+				if metricMapping.discard {
+					continue
+				}
+
+				if metricMapping.histogram {
+					var keys []float64
+					err = pq.Array(&keys).Scan(columnData[idx])
+					if err != nil {
+						return []prometheus.Metric{}, []error{}, errors.New(fmt.Sprintln("Error retrieving", columnName, "buckets:", namespace, err))
+					}
+
+					var values []int64
+					valuesIdx, ok := columnIdx[columnName+"_bucket"]
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Missing column: ", namespace, columnName+"_bucket")))
+						continue
+					}
+					err = pq.Array(&values).Scan(columnData[valuesIdx])
+					if err != nil {
+						return []prometheus.Metric{}, []error{}, errors.New(fmt.Sprintln("Error retrieving", columnName, "bucket values:", namespace, err))
+					}
+
+					buckets := make(map[float64]uint64, len(keys))
+					for i, key := range keys {
+						if i >= len(values) {
+							break
+						}
+						buckets[key] = uint64(values[i])
+					}
+
+					idx, ok = columnIdx[columnName+"_sum"]
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Missing column: ", namespace, columnName+"_sum")))
+						continue
+					}
+					sum, ok := dbToFloat64(columnData[idx])
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Unexpected error parsing column: ", namespace, columnName+"_sum", columnData[idx])))
+						continue
+					}
+
+					idx, ok = columnIdx[columnName+"_count"]
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Missing column: ", namespace, columnName+"_count")))
+						continue
+					}
+					count, ok := dbToUint64(columnData[idx])
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Unexpected error parsing column: ", namespace, columnName+"_count", columnData[idx])))
+						continue
+					}
+
+					metric = prometheus.MustNewConstHistogram(
+						metricMapping.desc,
+						count, sum, buckets,
+						labels...,
+					)
+				} else {
+					value, ok := dbToFloat64(columnData[idx])
+					if !ok {
+						nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Unexpected error parsing column: ", namespace, columnName, columnData[idx])))
+						continue
+					}
+					// Generate the metric
+					metric = prometheus.MustNewConstMetric(metricMapping.desc, metricMapping.vtype, value, labels...)
+				}
+			} else {
+				// Unknown metric. Report as untyped if scan to float64 works, else note an error too.
+				metricLabel := fmt.Sprintf("%s_%s", namespace, columnName)
+				desc := prometheus.NewDesc(metricLabel, fmt.Sprintf("Unknown metric from %s", namespace), mapping.labels, server.labels)
+
+				// Its not an error to fail here, since the values are
+				// unexpected anyway.
+				value, ok := dbToFloat64(columnData[idx])
+				if !ok {
+					nonfatalErrors = append(nonfatalErrors, errors.New(fmt.Sprintln("Unparseable column type - discarding: ", namespace, columnName, err)))
+					continue
+				}
+				metric = prometheus.MustNewConstMetric(desc, prometheus.UntypedValue, value, labels...)
+			}
+			metrics = append(metrics, metric)
+		}
+	}
+	return metrics, nonfatalErrors, nil
+}
+
+// Iterate through all the namespace mappings in the exporter and run their
+// queries.
+func queryNamespaceMappings(ch chan<- prometheus.Metric, server *Server) map[string]error {
+	// Return a map of namespace -> errors
+	namespaceErrors := make(map[string]error)
+
+	scrapeStart := time.Now()
+
+	for namespace, mapping := range server.metricMap {
+		level.Debug(logger).Log("msg", "Querying namespace", "namespace", namespace)
+
+		if mapping.master && !server.master {
+			level.Debug(logger).Log("msg", "Query skipped...")
+			continue
+		}
+
+		// check if the query is to be run on specific database server version range or not
+		if len(server.runonserver) > 0 {
+			serVersion, _ := semver.Parse(server.lastMapVersion.String())
+			runServerRange, _ := semver.ParseRange(server.runonserver)
+			if !runServerRange(serVersion) {
+				level.Debug(logger).Log("msg", "Query skipped for this database version", "version", server.lastMapVersion.String(), "target_version", server.runonserver)
+				continue
+			}
+		}
+
+		scrapeMetric := false
+		// Check if the metric is cached
+		server.cacheMtx.Lock()
+		cachedMetric, found := server.metricCache[namespace]
+		server.cacheMtx.Unlock()
+		// If found, check if needs refresh from cache
+		if found {
+			if scrapeStart.Sub(cachedMetric.lastScrape).Seconds() > float64(mapping.cacheSeconds) {
+				scrapeMetric = true
+			}
+		} else {
+			scrapeMetric = true
+		}
+
+		var metrics []prometheus.Metric
+		var nonFatalErrors []error
+		var err error
+		if scrapeMetric {
+			metrics, nonFatalErrors, err = queryNamespaceMapping(server, namespace, mapping)
+		} else {
+			metrics = cachedMetric.metrics
+		}
+
+		// Serious error - a namespace disappeared
+		if err != nil {
+			namespaceErrors[namespace] = err
+			level.Info(logger).Log("err", err)
+		}
+		// Non-serious errors - likely version or parsing problems.
+		if len(nonFatalErrors) > 0 {
+			for _, err := range nonFatalErrors {
+				level.Info(logger).Log("err", err)
+			}
+		}
+
+		// Emit the metrics into the channel
+		for _, metric := range metrics {
+			ch <- metric
+		}
+
+		if scrapeMetric {
+			// Only cache if metric is meaningfully cacheable
+			if mapping.cacheSeconds > 0 {
+				server.cacheMtx.Lock()
+				server.metricCache[namespace] = cachedMetrics{
+					metrics:    metrics,
+					lastScrape: scrapeStart,
+				}
+				server.cacheMtx.Unlock()
+			}
+		}
+	}
+
+	return namespaceErrors
+}

cmd/postgres_exporter/queries.go

@@ -0,0 +1,303 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/blang/semver"
+	"github.com/go-kit/kit/log/level"
+	"gopkg.in/yaml.v2"
+)
+
+// UserQuery represents a user defined query
+type UserQuery struct {
+	Query        string    `yaml:"query"`
+	Metrics      []Mapping `yaml:"metrics"`
+	Master       bool      `yaml:"master"`        // Querying only for master database
+	CacheSeconds uint64    `yaml:"cache_seconds"` // Number of seconds to cache the namespace result metrics for.
+	RunOnServer  string    `yaml:"runonserver"`   // Querying to run on which server version
+}
+
+// UserQueries represents a set of UserQuery objects
+type UserQueries map[string]UserQuery
+
+// OverrideQuery 's are run in-place of simple namespace look ups, and provide
+// advanced functionality. But they have a tendency to postgres version specific.
+// There aren't too many versions, so we simply store customized versions using
+// the semver matching we do for columns.
+type OverrideQuery struct {
+	versionRange semver.Range
+	query        string
+}
+
+// Overriding queries for namespaces above.
+// TODO: validate this is a closed set in tests, and there are no overlaps
+var queryOverrides = map[string][]OverrideQuery{
+	"pg_locks": {
+		{
+			semver.MustParseRange(">0.0.0"),
+			`SELECT pg_database.datname,tmp.mode,COALESCE(count,0) as count
+			FROM
+				(
+				  VALUES ('accesssharelock'),
+				         ('rowsharelock'),
+				         ('rowexclusivelock'),
+				         ('shareupdateexclusivelock'),
+				         ('sharelock'),
+				         ('sharerowexclusivelock'),
+				         ('exclusivelock'),
+				         ('accessexclusivelock'),
+					 ('sireadlock')
+				) AS tmp(mode) CROSS JOIN pg_database
+			LEFT JOIN
+			  (SELECT database, lower(mode) AS mode,count(*) AS count
+			  FROM pg_locks WHERE database IS NOT NULL
+			  GROUP BY database, lower(mode)
+			) AS tmp2
+			ON tmp.mode=tmp2.mode and pg_database.oid = tmp2.database ORDER BY 1`,
+		},
+	},
+
+	"pg_stat_replication": {
+		{
+			semver.MustParseRange(">=10.0.0"),
+			`
+			SELECT *,
+				(case pg_is_in_recovery() when 't' then null else pg_current_wal_lsn() end) AS pg_current_wal_lsn,
+				(case pg_is_in_recovery() when 't' then null else pg_wal_lsn_diff(pg_current_wal_lsn(), pg_lsn('0/0'))::float end) AS pg_current_wal_lsn_bytes,
+				(case pg_is_in_recovery() when 't' then null else pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn)::float end) AS pg_wal_lsn_diff
+			FROM pg_stat_replication
+			`,
+		},
+		{
+			semver.MustParseRange(">=9.2.0 <10.0.0"),
+			`
+			SELECT *,
+				(case pg_is_in_recovery() when 't' then null else pg_current_xlog_location() end) AS pg_current_xlog_location,
+				(case pg_is_in_recovery() when 't' then null else pg_xlog_location_diff(pg_current_xlog_location(), replay_location)::float end) AS pg_xlog_location_diff
+			FROM pg_stat_replication
+			`,
+		},
+		{
+			semver.MustParseRange("<9.2.0"),
+			`
+			SELECT *,
+				(case pg_is_in_recovery() when 't' then null else pg_current_xlog_location() end) AS pg_current_xlog_location
+			FROM pg_stat_replication
+			`,
+		},
+	},
+
+	"pg_replication_slots": {
+		{
+			semver.MustParseRange(">=9.4.0 <10.0.0"),
+			`
+			SELECT slot_name, database, active, pg_xlog_location_diff(pg_current_xlog_location(), restart_lsn)
+			FROM pg_replication_slots
+			`,
+		},
+		{
+			semver.MustParseRange(">=10.0.0"),
+			`
+			SELECT slot_name, database, active, pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
+			FROM pg_replication_slots
+			`,
+		},
+	},
+
+	"pg_stat_archiver": {
+		{
+			semver.MustParseRange(">=0.0.0"),
+			`
+			SELECT *,
+				extract(epoch from now() - last_archived_time) AS last_archive_age
+			FROM pg_stat_archiver
+			`,
+		},
+	},
+
+	"pg_stat_activity": {
+		// This query only works
+		{
+			semver.MustParseRange(">=9.2.0"),
+			`
+			SELECT
+				pg_database.datname,
+				tmp.state,
+				COALESCE(count,0) as count,
+				COALESCE(max_tx_duration,0) as max_tx_duration
+			FROM
+				(
+				  VALUES ('active'),
+				  		 ('idle'),
+				  		 ('idle in transaction'),
+				  		 ('idle in transaction (aborted)'),
+				  		 ('fastpath function call'),
+				  		 ('disabled')
+				) AS tmp(state) CROSS JOIN pg_database
+			LEFT JOIN
+			(
+				SELECT
+					datname,
+					state,
+					count(*) AS count,
+					MAX(EXTRACT(EPOCH FROM now() - xact_start))::float AS max_tx_duration
+				FROM pg_stat_activity GROUP BY datname,state) AS tmp2
+				ON tmp.state = tmp2.state AND pg_database.datname = tmp2.datname
+			`,
+		},
+		{
+			semver.MustParseRange("<9.2.0"),
+			`
+			SELECT
+				datname,
+				'unknown' AS state,
+				COALESCE(count(*),0) AS count,
+				COALESCE(MAX(EXTRACT(EPOCH FROM now() - xact_start))::float,0) AS max_tx_duration
+			FROM pg_stat_activity GROUP BY datname
+			`,
+		},
+	},
+}
+
+// Convert the query override file to the version-specific query override file
+// for the exporter.
+func makeQueryOverrideMap(pgVersion semver.Version, queryOverrides map[string][]OverrideQuery) map[string]string {
+	resultMap := make(map[string]string)
+	for name, overrideDef := range queryOverrides {
+		// Find a matching semver. We make it an error to have overlapping
+		// ranges at test-time, so only 1 should ever match.
+		matched := false
+		for _, queryDef := range overrideDef {
+			if queryDef.versionRange(pgVersion) {
+				resultMap[name] = queryDef.query
+				matched = true
+				break
+			}
+		}
+		if !matched {
+			level.Warn(logger).Log("msg", "No query matched override, disabling metric space", "name", name)
+			resultMap[name] = ""
+		}
+	}
+
+	return resultMap
+}
+
+func parseUserQueries(content []byte) (map[string]intermediateMetricMap, map[string]string, error) {
+	var userQueries UserQueries
+
+	err := yaml.Unmarshal(content, &userQueries)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Stores the loaded map representation
+	metricMaps := make(map[string]intermediateMetricMap)
+	newQueryOverrides := make(map[string]string)
+
+	for metric, specs := range userQueries {
+		level.Debug(logger).Log("msg", "New user metric namespace from YAML metric", "metric", metric, "cache_seconds", specs.CacheSeconds)
+		newQueryOverrides[metric] = specs.Query
+		metricMap, ok := metricMaps[metric]
+		if !ok {
+			// Namespace for metric not found - add it.
+			newMetricMap := make(map[string]ColumnMapping)
+			metricMap = intermediateMetricMap{
+				columnMappings: newMetricMap,
+				master:         specs.Master,
+				cacheSeconds:   specs.CacheSeconds,
+			}
+			metricMaps[metric] = metricMap
+		}
+		for _, metric := range specs.Metrics {
+			for name, mappingOption := range metric {
+				var columnMapping ColumnMapping
+				tmpUsage, _ := stringToColumnUsage(mappingOption.Usage)
+				columnMapping.usage = tmpUsage
+				columnMapping.description = mappingOption.Description
+
+				// TODO: we should support cu
+				columnMapping.mapping = nil
+				// Should we support this for users?
+				columnMapping.supportedVersions = nil
+
+				metricMap.columnMappings[name] = columnMapping
+			}
+		}
+	}
+	return metricMaps, newQueryOverrides, nil
+}
+
+// Add queries to the builtinMetricMaps and queryOverrides maps. Added queries do not
+// respect version requirements, because it is assumed that the user knows
+// what they are doing with their version of postgres.
+//
+// This function modifies metricMap and queryOverrideMap to contain the new
+// queries.
+// TODO: test code for all cu.
+// TODO: the YAML this supports is "non-standard" - we should move away from it.
+func addQueries(content []byte, pgVersion semver.Version, server *Server) error {
+	metricMaps, newQueryOverrides, err := parseUserQueries(content)
+	if err != nil {
+		return err
+	}
+	// Convert the loaded metric map into exporter representation
+	partialExporterMap := makeDescMap(pgVersion, server.labels, metricMaps)
+
+	// Merge the two maps (which are now quite flatteend)
+	for k, v := range partialExporterMap {
+		_, found := server.metricMap[k]
+		if found {
+			level.Debug(logger).Log("msg", "Overriding metric from user YAML file", "metric", k)
+		} else {
+			level.Debug(logger).Log("msg", "Adding new metric from user YAML file", "metric", k)
+		}
+		server.metricMap[k] = v
+	}
+
+	// Merge the query override map
+	for k, v := range newQueryOverrides {
+		_, found := server.queryOverrides[k]
+		if found {
+			level.Debug(logger).Log("msg", "Overriding query override from user YAML file", "query_override", k)
+		} else {
+			level.Debug(logger).Log("msg", "Adding new query override from user YAML file", "query_override", k)
+		}
+		server.queryOverrides[k] = v
+	}
+	return nil
+}
+
+func queryDatabases(server *Server) ([]string, error) {
+	rows, err := server.db.Query("SELECT datname FROM pg_database WHERE datallowconn = true AND datistemplate = false AND datname != current_database()")
+	if err != nil {
+		return nil, fmt.Errorf("Error retrieving databases: %v", err)
+	}
+	defer rows.Close() // nolint: errcheck
+
+	var databaseName string
+	result := make([]string, 0)
+	for rows.Next() {
+		err = rows.Scan(&databaseName)
+		if err != nil {
+			return nil, errors.New(fmt.Sprintln("Error retrieving rows:", err))
+		}
+		result = append(result, databaseName)
+	}
+
+	return result, nil
+}

cmd/postgres_exporter/server.go

@@ -0,0 +1,190 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/blang/semver"
+	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Server describes a connection to Postgres.
+// Also it contains metrics map and query overrides.
+type Server struct {
+	db          *sql.DB
+	labels      prometheus.Labels
+	master      bool
+	runonserver string
+
+	// Last version used to calculate metric map. If mismatch on scrape,
+	// then maps are recalculated.
+	lastMapVersion semver.Version
+	// Currently active metric map
+	metricMap map[string]MetricMapNamespace
+	// Currently active query overrides
+	queryOverrides map[string]string
+	mappingMtx     sync.RWMutex
+	// Currently cached metrics
+	metricCache map[string]cachedMetrics
+	cacheMtx    sync.Mutex
+}
+
+// ServerOpt configures a server.
+type ServerOpt func(*Server)
+
+// ServerWithLabels configures a set of labels.
+func ServerWithLabels(labels prometheus.Labels) ServerOpt {
+	return func(s *Server) {
+		for k, v := range labels {
+			s.labels[k] = v
+		}
+	}
+}
+
+// NewServer establishes a new connection using DSN.
+func NewServer(dsn string, opts ...ServerOpt) (*Server, error) {
+	fingerprint, err := parseFingerprint(dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	db, err := sql.Open("postgres", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
+
+	level.Info(logger).Log("msg", "Established new database connection", "fingerprint", fingerprint)
+
+	s := &Server{
+		db:     db,
+		master: false,
+		labels: prometheus.Labels{
+			serverLabelName: fingerprint,
+		},
+		metricCache: make(map[string]cachedMetrics),
+	}
+
+	for _, opt := range opts {
+		opt(s)
+	}
+
+	return s, nil
+}
+
+// Close disconnects from Postgres.
+func (s *Server) Close() error {
+	return s.db.Close()
+}
+
+// Ping checks connection availability and possibly invalidates the connection if it fails.
+func (s *Server) Ping() error {
+	if err := s.db.Ping(); err != nil {
+		if cerr := s.Close(); cerr != nil {
+			level.Error(logger).Log("msg", "Error while closing non-pinging DB connection", "server", s, "err", cerr)
+		}
+		return err
+	}
+	return nil
+}
+
+// String returns server's fingerprint.
+func (s *Server) String() string {
+	return s.labels[serverLabelName]
+}
+
+// Scrape loads metrics.
+func (s *Server) Scrape(ch chan<- prometheus.Metric, disableSettingsMetrics bool) error {
+	s.mappingMtx.RLock()
+	defer s.mappingMtx.RUnlock()
+
+	var err error
+
+	if !disableSettingsMetrics && s.master {
+		if err = querySettings(ch, s); err != nil {
+			err = fmt.Errorf("error retrieving settings: %s", err)
+		}
+	}
+
+	errMap := queryNamespaceMappings(ch, s)
+	if len(errMap) > 0 {
+		err = fmt.Errorf("queryNamespaceMappings returned %d errors", len(errMap))
+	}
+
+	return err
+}
+
+// Servers contains a collection of servers to Postgres.
+type Servers struct {
+	m       sync.Mutex
+	servers map[string]*Server
+	opts    []ServerOpt
+}
+
+// NewServers creates a collection of servers to Postgres.
+func NewServers(opts ...ServerOpt) *Servers {
+	return &Servers{
+		servers: make(map[string]*Server),
+		opts:    opts,
+	}
+}
+
+// GetServer returns established connection from a collection.
+func (s *Servers) GetServer(dsn string) (*Server, error) {
+	s.m.Lock()
+	defer s.m.Unlock()
+	var err error
+	var ok bool
+	errCount := 0 // start at zero because we increment before doing work
+	retries := 1
+	var server *Server
+	for {
+		if errCount++; errCount > retries {
+			return nil, err
+		}
+		server, ok = s.servers[dsn]
+		if !ok {
+			server, err = NewServer(dsn, s.opts...)
+			if err != nil {
+				time.Sleep(time.Duration(errCount) * time.Second)
+				continue
+			}
+			s.servers[dsn] = server
+		}
+		if err = server.Ping(); err != nil {
+			delete(s.servers, dsn)
+			time.Sleep(time.Duration(errCount) * time.Second)
+			continue
+		}
+		break
+	}
+	return server, nil
+}
+
+// Close disconnects from all known servers.
+func (s *Servers) Close() {
+	s.m.Lock()
+	defer s.m.Unlock()
+	for _, server := range s.servers {
+		if err := server.Close(); err != nil {
+			level.Error(logger).Log("msg", "Failed to close connection", "server", server, "err", err)
+		}
+	}
+}

cmd/postgres_exporter/util.go

@@ -0,0 +1,218 @@
+// Copyright 2021 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"fmt"
+	"math"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-kit/kit/log/level"
+	"github.com/lib/pq"
+)
+
+func contains(a []string, x string) bool {
+	for _, n := range a {
+		if x == n {
+			return true
+		}
+	}
+	return false
+}
+
+// convert a string to the corresponding ColumnUsage
+func stringToColumnUsage(s string) (ColumnUsage, error) {
+	var u ColumnUsage
+	var err error
+	switch s {
+	case "DISCARD":
+		u = DISCARD
+
+	case "LABEL":
+		u = LABEL
+
+	case "COUNTER":
+		u = COUNTER
+
+	case "GAUGE":
+		u = GAUGE
+
+	case "HISTOGRAM":
+		u = HISTOGRAM
+
+	case "MAPPEDMETRIC":
+		u = MAPPEDMETRIC
+
+	case "DURATION":
+		u = DURATION
+
+	default:
+		err = fmt.Errorf("wrong ColumnUsage given : %s", s)
+	}
+
+	return u, err
+}
+
+// Convert database.sql types to float64s for Prometheus consumption. Null types are mapped to NaN. string and []byte
+// types are mapped as NaN and !ok
+func dbToFloat64(t interface{}) (float64, bool) {
+	switch v := t.(type) {
+	case int64:
+		return float64(v), true
+	case float64:
+		return v, true
+	case time.Time:
+		return float64(v.Unix()), true
+	case []byte:
+		// Try and convert to string and then parse to a float64
+		strV := string(v)
+		result, err := strconv.ParseFloat(strV, 64)
+		if err != nil {
+			level.Info(logger).Log("msg", "Could not parse []byte", "err", err)
+			return math.NaN(), false
+		}
+		return result, true
+	case string:
+		result, err := strconv.ParseFloat(v, 64)
+		if err != nil {
+			level.Info(logger).Log("msg", "Could not parse string", "err", err)
+			return math.NaN(), false
+		}
+		return result, true
+	case bool:
+		if v {
+			return 1.0, true
+		}
+		return 0.0, true
+	case nil:
+		return math.NaN(), true
+	default:
+		return math.NaN(), false
+	}
+}
+
+// Convert database.sql types to uint64 for Prometheus consumption. Null types are mapped to 0. string and []byte
+// types are mapped as 0 and !ok
+func dbToUint64(t interface{}) (uint64, bool) {
+	switch v := t.(type) {
+	case uint64:
+		return v, true
+	case int64:
+		return uint64(v), true
+	case float64:
+		return uint64(v), true
+	case time.Time:
+		return uint64(v.Unix()), true
+	case []byte:
+		// Try and convert to string and then parse to a uint64
+		strV := string(v)
+		result, err := strconv.ParseUint(strV, 10, 64)
+		if err != nil {
+			level.Info(logger).Log("msg", "Could not parse []byte", "err", err)
+			return 0, false
+		}
+		return result, true
+	case string:
+		result, err := strconv.ParseUint(v, 10, 64)
+		if err != nil {
+			level.Info(logger).Log("msg", "Could not parse string", "err", err)
+			return 0, false
+		}
+		return result, true
+	case bool:
+		if v {
+			return 1, true
+		}
+		return 0, true
+	case nil:
+		return 0, true
+	default:
+		return 0, false
+	}
+}
+
+// Convert database.sql to string for Prometheus labels. Null types are mapped to empty strings.
+func dbToString(t interface{}) (string, bool) {
+	switch v := t.(type) {
+	case int64:
+		return fmt.Sprintf("%v", v), true
+	case float64:
+		return fmt.Sprintf("%v", v), true
+	case time.Time:
+		return fmt.Sprintf("%v", v.Unix()), true
+	case nil:
+		return "", true
+	case []byte:
+		// Try and convert to string
+		return string(v), true
+	case string:
+		return v, true
+	case bool:
+		if v {
+			return "true", true
+		}
+		return "false", true
+	default:
+		return "", false
+	}
+}
+
+func parseFingerprint(url string) (string, error) {
+	dsn, err := pq.ParseURL(url)
+	if err != nil {
+		dsn = url
+	}
+
+	pairs := strings.Split(dsn, " ")
+	kv := make(map[string]string, len(pairs))
+	for _, pair := range pairs {
+		splitted := strings.SplitN(pair, "=", 2)
+		if len(splitted) != 2 {
+			return "", fmt.Errorf("malformed dsn %q", dsn)
+		}
+		kv[splitted[0]] = splitted[1]
+	}
+
+	var fingerprint string
+
+	if host, ok := kv["host"]; ok {
+		fingerprint += host
+	} else {
+		fingerprint += "localhost"
+	}
+
+	if port, ok := kv["port"]; ok {
+		fingerprint += ":" + port
+	} else {
+		fingerprint += ":5432"
+	}
+
+	return fingerprint, nil
+}
+
+func loggableDSN(dsn string) string {
+	pDSN, err := url.Parse(dsn)
+	if err != nil {
+		return "could not parse DATA_SOURCE_NAME"
+	}
+	// Blank user info if not nil
+	if pDSN.User != nil {
+		pDSN.User = url.UserPassword(pDSN.User.Username(), "PASSWORD_REMOVED")
+	}
+
+	return pDSN.String()
+}