Use persistent connection to postgresql database (#2701)

CHANGELOG.md

@@ -63,6 +63,10 @@
   `use_random_partitionkey` options has been deprecated in favor of the
   `partition` subtable.  This allows for more flexible methods to set the
   partition key such as by metric name or by tag.
+- `postgresql` plugins will now default to using a persistent connection to the database.
+  `Important` In environments TCP connections are terminated when idle for periods shorter than 15 minutes
+  and the collection interval is longer than the termination period then max_lifetime
+  should be set to be less than the collection interval to pervent errors when collecting metrics.
 
 - With the release of the new improved `jolokia2` input, the legacy `jolokia`
   plugin is deprecated and will be removed in a future release.  Users of this

plugins/inputs/postgresql/postgresql.go

@@ -2,26 +2,21 @@ package postgresql
 
 import (
 	"bytes"
-	"database/sql"
 	"fmt"
-	"regexp"
-	"sort"
 	"strings"
 
 	// register in driver.
 	_ "github.com/jackc/pgx/stdlib"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/inputs"
 )
 
 type Postgresql struct {
-	Address          string
+	Service
 	Databases        []string
 	IgnoredDatabases []string
-	OrderedColumns   []string
-	AllColumns       []string
-	sanitizedAddress string
 }
 
 var ignoredColumns = map[string]bool{"stats_reset": true}
@@ -41,6 +36,15 @@ var sampleConfig = `
   ## to grab metrics for.
   ##
   address = "host=localhost user=postgres sslmode=disable"
+  ## A custom name for the database that will be used as the "server" tag in the
+  ## measurement output. If not specified, a default one generated from
+  ## the connection address is used.
+  # outputaddress = "db01"
+
+  ## connection configuration.
+  ## maxlifetime - specify the maximum lifetime of a connection.
+  ## default is forever (0s)
+  max_lifetime = "0s"
 
   ## A  list of databases to explicitly ignore.  If not specified, metrics for all
   ## databases are gathered.  Do NOT use with the 'databases' option.
@@ -63,24 +67,13 @@ func (p *Postgresql) IgnoredColumns() map[string]bool {
 	return ignoredColumns
 }
 
-var localhost = "host=localhost sslmode=disable"
-
 func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
 	var (
-		err   error
-		db    *sql.DB
-		query string
+		err     error
+		query   string
+		columns []string
 	)
 
-	if p.Address == "" || p.Address == "localhost" {
-		p.Address = localhost
-	}
-
-	if db, err = sql.Open("pgx", p.Address); err != nil {
-		return err
-	}
-	defer db.Close()
-
 	if len(p.Databases) == 0 && len(p.IgnoredDatabases) == 0 {
 		query = `SELECT * FROM pg_stat_database`
 	} else if len(p.IgnoredDatabases) != 0 {
@@ -91,93 +84,67 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
 			strings.Join(p.Databases, "','"))
 	}
 
-	rows, err := db.Query(query)
+	rows, err := p.DB.Query(query)
 	if err != nil {
 		return err
 	}
 
 	defer rows.Close()
 
 	// grab the column information from the result
-	p.OrderedColumns, err = rows.Columns()
-	if err != nil {
+	if columns, err = rows.Columns(); err != nil {
 		return err
-	} else {
-		p.AllColumns = make([]string, len(p.OrderedColumns))
-		copy(p.AllColumns, p.OrderedColumns)
 	}
 
 	for rows.Next() {
-		err = p.accRow(rows, acc)
+		err = p.accRow(rows, acc, columns)
 		if err != nil {
 			return err
 		}
 	}
 
 	query = `SELECT * FROM pg_stat_bgwriter`
 
-	bg_writer_row, err := db.Query(query)
+	bg_writer_row, err := p.DB.Query(query)
 	if err != nil {
 		return err
 	}
 
 	defer bg_writer_row.Close()
 
 	// grab the column information from the result
-	p.OrderedColumns, err = bg_writer_row.Columns()
-	if err != nil {
+	if columns, err = bg_writer_row.Columns(); err != nil {
 		return err
-	} else {
-		for _, v := range p.OrderedColumns {
-			p.AllColumns = append(p.AllColumns, v)
-		}
 	}
 
 	for bg_writer_row.Next() {
-		err = p.accRow(bg_writer_row, acc)
+		err = p.accRow(bg_writer_row, acc, columns)
 		if err != nil {
 			return err
 		}
 	}
-	sort.Strings(p.AllColumns)
+
 	return bg_writer_row.Err()
 }
 
 type scanner interface {
 	Scan(dest ...interface{}) error
 }
 
-var passwordKVMatcher, _ = regexp.Compile("password=\\S+ ?")
-
-func (p *Postgresql) SanitizedAddress() (_ string, err error) {
-	var canonicalizedAddress string
-	if strings.HasPrefix(p.Address, "postgres://") || strings.HasPrefix(p.Address, "postgresql://") {
-		canonicalizedAddress, err = ParseURL(p.Address)
-		if err != nil {
-			return p.sanitizedAddress, err
-		}
-	} else {
-		canonicalizedAddress = p.Address
-	}
-	p.sanitizedAddress = passwordKVMatcher.ReplaceAllString(canonicalizedAddress, "")
-
-	return p.sanitizedAddress, err
-}
-
-func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {
+func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator, columns []string) error {
 	var columnVars []interface{}
 	var dbname bytes.Buffer
 
 	// this is where we'll store the column name with its *interface{}
 	columnMap := make(map[string]*interface{})
 
-	for _, column := range p.OrderedColumns {
+	for _, column := range columns {
 		columnMap[column] = new(interface{})
 	}
 
 	// populate the array of interface{} with the pointers in the right order
 	for i := 0; i < len(columnMap); i++ {
-		columnVars = append(columnVars, columnMap[p.OrderedColumns[i]])
+		columnVars = append(columnVars, columnMap[columns[i]])
 	}
 
 	// deconstruct array of variables and send to Scan
@@ -215,6 +182,14 @@ func (p *Postgresql) accRow(row scanner, acc telegraf.Accumulator) error {
 
 func init() {
 	inputs.Add("postgresql", func() telegraf.Input {
-		return &Postgresql{}
+		return &Postgresql{
+			Service: Service{
+				MaxIdle: 1,
+				MaxOpen: 1,
+				MaxLifetime: internal.Duration{
+					Duration: 0,
+				},
+			},
+		}
 	})
 }