fix postgresql connection leak (#2611)

CHANGELOG.md

@@ -92,7 +92,7 @@ be deprecated eventually.
 - [#2576](https://github.com/influxdata/telegraf/pull/2576): Add write timeout to Riemann output
 - [#2596](https://github.com/influxdata/telegraf/pull/2596): fix timestamp parsing on prometheus plugin
 - [#2610](https://github.com/influxdata/telegraf/pull/2610): Fix deadlock when output cannot write
-
+- [#2410](https://github.com/influxdata/telegraf/issues/2410): Fix connection leak in postgresql.
 
 ## v1.2.1 [2017-02-01]
 

Godeps

@@ -24,7 +24,7 @@ github.com/hashicorp/consul 63d2fc68239b996096a1c55a0d4b400ea4c2583f
 github.com/influxdata/tail e9ef7e826dafcb3093b40b989fefa90eeb9a8ca1
 github.com/influxdata/toml 5d1d907f22ead1cd47adde17ceec5bda9cacaf8f
 github.com/influxdata/wlog 7c63b0a71ef8300adc255344d275e10e5c3a71ec
-github.com/jackc/pgx c8080fc4a1bfa44bf90383ad0fdce2f68b7d313c
+github.com/jackc/pgx b84338d7d62598f75859b2b146d830b22f1b9ec8
 github.com/kardianos/osext c2c54e542fb797ad986b31721e1baedf214ca413
 github.com/kardianos/service 6d3a0ee7d3425d9d835debc51a0ca1ffa28f4893
 github.com/kballard/go-shellquote d8ec1a69a250a17bb0e419c386eac1f3711dc142

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 db, err = sql.Open("pgx", p.Address); err != nil {
-	if 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 (
-	var sql_query string
+		err         error
-	var query_addon string
+		db          *sql.DB
-	var db_version int
+		sql_query   string
-	var query string
+		query_addon string
-	var tag_value string
+		db_version  int
-	var meas_name string
+		query       string
+		tag_value   string
+		meas_name   string
+	)
 
 	if p.Address == "" || p.Address == "localhost" {
 		p.Address = localhost
 	}
 
-	db, err := postgresql.Connect(p.Address)
+	if db, err = sql.Open("pgx", p.Address); err != nil {
-	if err != nil {
 		return err
 	}
-
 	defer db.Close()
 
 	// Retreiving the database version