fix tests

Author: James Lawrence

plugins/inputs/postgresql/postgresql.go

@@ -13,7 +13,8 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
 
-	_ "github.com/jackc/pgx/stdlib"
+	"github.com/jackc/pgx"
+	"github.com/jackc/pgx/stdlib"
 )
 
 type Postgresql struct {
@@ -68,7 +69,7 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
 		p.Address = localhost
 	}
 
-	db, err := sql.Open("pgx", p.Address)
+	db, err := connect(p.Address)
 	if err != nil {
 		return err
 	}
@@ -277,3 +278,21 @@ 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_test.go

@@ -28,6 +28,7 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 	for _, col := range p.AllColumns {
 		availableColumns[col] = true
 	}
+
 	intMetrics := []string{
 		"xact_commit",
 		"xact_rollback",
@@ -42,7 +43,6 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		"temp_files",
 		"temp_bytes",
 		"deadlocks",
-		"numbackends",
 		"buffers_alloc",
 		"buffers_backend",
 		"buffers_backend_fsync",
@@ -53,9 +53,20 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		"maxwritten_clean",
 	}
 
+	int32Metrics := []string{
+		"numbackends",
+	}
+
 	floatMetrics := []string{
 		"blk_read_time",
 		"blk_write_time",
+		"checkpoint_write_time",
+		"checkpoint_sync_time",
+	}
+
+	stringMetrics := []string{
+		"datname",
+		"datid",
 	}
 
 	metricsCounted := 0
@@ -68,6 +79,14 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		}
 	}
 
+	for _, metric := range int32Metrics {
+		_, ok := availableColumns[metric]
+		if ok {
+			assert.True(t, acc.HasInt32Field("postgresql", metric))
+			metricsCounted++
+		}
+	}
+
 	for _, metric := range floatMetrics {
 		_, ok := availableColumns[metric]
 		if ok {
@@ -76,8 +95,16 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		}
 	}
 
+	for _, metric := range stringMetrics {
+		_, ok := availableColumns[metric]
+		if ok {
+			assert.True(t, acc.HasStringField("postgresql", metric))
+			metricsCounted++
+		}
+	}
+
 	assert.True(t, metricsCounted > 0)
-	//assert.Equal(t, len(availableColumns)-len(p.IgnoredColumns()), metricsCounted)
+	assert.Equal(t, len(availableColumns)-len(p.IgnoredColumns()), metricsCounted)
 }
 
 func TestPostgresqlTagsMetricsWithDatabaseName(t *testing.T) {

plugins/inputs/postgresql_extensible/postgresql_extensible.go

@@ -14,7 +14,8 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
 
-	_ "github.com/jackc/pgx/stdlib"
+	"github.com/jackc/pgx"
+	"github.com/jackc/pgx/stdlib"
 )
 
 type Postgresql struct {
@@ -128,7 +129,7 @@ func (p *Postgresql) Gather(acc telegraf.Accumulator) error {
 		p.Address = localhost
 	}
 
-	db, err := sql.Open("pgx", p.Address)
+	db, err := connect(p.Address)
 	if err != nil {
 		return err
 	}
@@ -286,7 +287,7 @@ COLUMN:
 				tags[col] = v
 			case []byte:
 				tags[col] = string(v)
-			case int64:
+			case int64, int32, int:
 				tags[col] = fmt.Sprintf("%d", v)
 			default:
 				log.Println("failed to add additional tag", col)
@@ -376,3 +377,21 @@ 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_extensible/postgresql_extensible_test.go

@@ -33,6 +33,7 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 	for _, col := range p.AllColumns {
 		availableColumns[col] = true
 	}
+
 	intMetrics := []string{
 		"xact_commit",
 		"xact_rollback",
@@ -47,6 +48,9 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		"temp_files",
 		"temp_bytes",
 		"deadlocks",
+	}
+
+	int32Metrics := []string{
 		"numbackends",
 	}
 
@@ -55,6 +59,11 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		"blk_write_time",
 	}
 
+	stringMetrics := []string{
+		"datname",
+		"datid",
+	}
+
 	metricsCounted := 0
 
 	for _, metric := range intMetrics {
@@ -65,6 +74,14 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		}
 	}
 
+	for _, metric := range int32Metrics {
+		_, ok := availableColumns[metric]
+		if ok {
+			assert.True(t, acc.HasInt32Field("postgresql", metric))
+			metricsCounted++
+		}
+	}
+
 	for _, metric := range floatMetrics {
 		_, ok := availableColumns[metric]
 		if ok {
@@ -73,6 +90,14 @@ func TestPostgresqlGeneratesMetrics(t *testing.T) {
 		}
 	}
 
+	for _, metric := range stringMetrics {
+		_, ok := availableColumns[metric]
+		if ok {
+			assert.True(t, acc.HasStringField("postgresql", metric))
+			metricsCounted++
+		}
+	}
+
 	assert.True(t, metricsCounted > 0)
 	assert.Equal(t, len(availableColumns)-len(p.IgnoredColumns()), metricsCounted)
 }

testutil/accumulator.go

@@ -188,7 +188,7 @@ func (a *Accumulator) AssertContainsFields(
 	assert.Fail(t, msg)
 }
 
-// HasIntValue returns true if the measurement has an Int value
+// HasIntField returns true if the measurement has an Int value
 func (a *Accumulator) HasIntField(measurement string, field string) bool {
 	a.Lock()
 	defer a.Unlock()
@@ -206,6 +206,42 @@ func (a *Accumulator) HasIntField(measurement string, field string) bool {
 	return false
 }
 
+// HasInt32Field returns true if the measurement has an Int value
+func (a *Accumulator) HasInt32Field(measurement string, field string) bool {
+	a.Lock()
+	defer a.Unlock()
+	for _, p := range a.Metrics {
+		if p.Measurement == measurement {
+			for fieldname, value := range p.Fields {
+				if fieldname == field {
+					_, ok := value.(int32)
+					return ok
+				}
+			}
+		}
+	}
+
+	return false
+}
+
+// HasStringField returns true if the measurement has an Int value
+func (a *Accumulator) HasStringField(measurement string, field string) bool {
+	a.Lock()
+	defer a.Unlock()
+	for _, p := range a.Metrics {
+		if p.Measurement == measurement {
+			for fieldname, value := range p.Fields {
+				if fieldname == field {
+					_, ok := value.(string)
+					return ok
+				}
+			}
+		}
+	}
+
+	return false
+}
+
 // HasUIntValue returns true if the measurement has a UInt value
 func (a *Accumulator) HasUIntField(measurement string, field string) bool {
 	a.Lock()