fix postgresql connection leak (#2611)

plugins/inputs/postgresql/connect.go

@@ -1,15 +1,11 @@
 package postgresql
 
 import (
-	"database/sql"
 	"fmt"
 	"net"
 	"net/url"
 	"sort"
 	"strings"
-
-	"github.com/jackc/pgx"
-	"github.com/jackc/pgx/stdlib"
 )
 
 // pulled from lib/pq
@@ -79,21 +75,3 @@ func ParseURL(uri string) (string, error) {
 	sort.Strings(kvs) // Makes testing easier (not a performance concern)
 	return strings.Join(kvs, " "), nil
 }
-
-func Connect(address string) (*sql.DB, error) {
-	if strings.HasPrefix(address, "postgres://") || strings.HasPrefix(address, "postgresql://") {
-		return sql.Open("pgx", address)
-	}
-
-	config, err := pgx.ParseDSN(address)
-	if err != nil {
-		return nil, err
-	}
-
-	pool, err := pgx.NewConnPool(pgx.ConnPoolConfig{ConnConfig: config})
-	if err != nil {
-		return nil, err
-	}
-
-	return stdlib.OpenFromConnPool(pool)
-}

plugins/inputs/postgresql/postgresql.go

@@ -2,11 +2,15 @@ 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/plugins/inputs"
 )
@@ -62,17 +66,19 @@ func (p *Postgresql) IgnoredColumns() map[string]bool {
 var localhost = "host=localhost sslmode=disable"
 
 func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
-	var query string
+	var (
+		err   error
+		db    *sql.DB
+		query string
+	)
 
 	if p.Address == "" || p.Address == "localhost" {
 		p.Address = localhost
 	}
 
-	db, err := Connect(p.Address)
-	if err != nil {
+	if db, err = sql.Open("pgx", p.Address); err != nil {
 		return err
 	}
-
 	defer db.Close()
 
 	if len(p.Databases) == 0 && len(p.IgnoredDatabases) == 0 {
@@ -107,7 +113,7 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
 			return err
 		}
 	}
-	//return rows.Err()
+
 	query = `SELECT * FROM pg_stat_bgwriter`
 
 	bg_writer_row, err := db.Query(query)

plugins/inputs/postgresql_extensible/postgresql_extensible.go

@@ -2,11 +2,15 @@ package postgresql_extensible
 
 import (
 	"bytes"
+	"database/sql"
 	"fmt"
 	"log"
 	"regexp"
 	"strings"
 
+	// register in driver.
+	_ "github.com/jackc/pgx/stdlib"
+
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/inputs/postgresql"
@@ -112,23 +116,24 @@ func (p *Postgresql) IgnoredColumns() map[string]bool {
 var localhost = "host=localhost sslmode=disable"
 
 func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
-
-	var sql_query string
-	var query_addon string
-	var db_version int
-	var query string
-	var tag_value string
-	var meas_name string
+	var (
+		err         error
+		db          *sql.DB
+		sql_query   string
+		query_addon string
+		db_version  int
+		query       string
+		tag_value   string
+		meas_name   string
+	)
 
 	if p.Address == "" || p.Address == "localhost" {
 		p.Address = localhost
 	}
 
-	db, err := postgresql.Connect(p.Address)
-	if err != nil {
+	if db, err = sql.Open("pgx", p.Address); err != nil {
 		return err
 	}
-
 	defer db.Close()
 
 	// Retreiving the database version