(0.3.0) Postgres plugin: canonicalize & sanitize adress

CHANGELOG.md

@@ -5,6 +5,9 @@
 - **breaking change** Plugin measurements aggregated into a single measurement.
 - **breaking change** `jolokia` plugin: must use global tag/drop/pass parameters
 for configuration.
+- **breaking change** `postgresql` plugin: by default, converts both forms of address
+from config to the key=value format for tag, and drops password. Both behaviors can be
+suppressed with the new `verbatim_address` config for that plugin.
 - **breaking change** `procstat` plugin has `*cpu*` fields renamed to
 `*cpu_time*`
 - `twemproxy` plugin: `prefix` option removed.

plugins/postgresql/postgresql.go

@@ -4,17 +4,21 @@ import (
 	"bytes"
 	"database/sql"
 	"fmt"
+	"regexp"
 	"strings"
 
 	"github.com/influxdb/telegraf/plugins"
 
-	_ "github.com/lib/pq"
+	"github.com/lib/pq"
 )
 
 type Postgresql struct {
 	Address        string
 	Databases      []string
 	OrderedColumns []string
+
+	VerbatimAddress  bool
+	sanitizedAddress string
 }
 
 var ignoredColumns = map[string]bool{"datid": true, "datname": true, "stats_reset": true}
@@ -34,6 +38,16 @@ var sampleConfig = `
   #
   address = "host=localhost user=postgres sslmode=disable"
 
+  # Starting in 0.3.0 the default behavior is to convert the above given address to the
+  # key value form and, for security, remove the password before using it to tag the
+  # collected data.
+  #
+  # If you are using the URL form and/or have existing tooling matching against a previous
+  # value, you might want to prevent this transformation / sanitization. Set the following
+  # to true to leave it as entered for the tag.
+
+  # verbatim_address = true
+
   # A list of databases to pull metrics about. If not specified, metrics for all
   # databases are gathered.
   # databases = ["app_production", "testing"]
@@ -101,6 +115,27 @@ type scanner interface {
 	Scan(dest ...interface{}) error
 }
 
+var passwordKVMatcher, _ = regexp.Compile("password=\\S+ ?")
+
+func (p *Postgresql) SanitizedAddress() (_ string, err error) {
+	var canonicalizedAddress string
+
+	if p.sanitizedAddress == "" {
+		if strings.HasPrefix(p.Address, "postgres://") || strings.HasPrefix(p.Address, "postgresql://") {
+			canonicalizedAddress, err = pq.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 plugins.Accumulator) error {
 	var columnVars []interface{}
 	var dbname bytes.Buffer
@@ -130,7 +165,17 @@ func (p *Postgresql) accRow(row scanner, acc plugins.Accumulator) error {
 		dbname.WriteString(string(dbnameChars[i]))
 	}
 
-	tags := map[string]string{"server": p.Address, "db": dbname.String()}
+	var tagAddress string
+	if p.VerbatimAddress {
+		tagAddress = p.Address
+	} else {
+		tagAddress, err = p.SanitizedAddress()
+		if err != nil {
+			return err
+		}
+	}
+
+	tags := map[string]string{"server": tagAddress, "db": dbname.String()}
 
 	fields := make(map[string]interface{})
 	for col, val := range columnMap {

plugins/postgresql/postgresql_test.go

@@ -96,6 +96,104 @@ func TestPostgresqlTagsMetricsWithDatabaseName(t *testing.T) {
 	assert.Equal(t, "postgres", point.Tags["db"])
 }
 
+func TestPostgresqlCanonicalizesAndSanitizesURLServerName(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	p := &Postgresql{
+		Address: fmt.Sprintf("postgres://postgres:swordfish@%s?sslmode=disable",
+			testutil.GetLocalHost()),
+		Databases: []string{"postgres"},
+	}
+
+	var acc testutil.Accumulator
+
+	err := p.Gather(&acc)
+	require.NoError(t, err)
+
+	point, ok := acc.Get("postgresql")
+	require.True(t, ok)
+
+	assert.Equal(t,
+		fmt.Sprintf("host=%s sslmode=disable user=postgres", testutil.GetLocalHost()),
+		point.Tags["server"])
+}
+
+func TestPostgresqlSanitizesKVServerName(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	p := &Postgresql{
+		Address: fmt.Sprintf("host=%s user=postgres password=swordfish sslmode=disable",
+			testutil.GetLocalHost()),
+		Databases: []string{"postgres"},
+	}
+
+	var acc testutil.Accumulator
+
+	err := p.Gather(&acc)
+	require.NoError(t, err)
+
+	point, ok := acc.Get("postgresql")
+	require.True(t, ok)
+
+	assert.Equal(t,
+		fmt.Sprintf("host=%s user=postgres sslmode=disable", testutil.GetLocalHost()),
+		point.Tags["server"])
+}
+
+func TestPostgresqlMaintainsVerbatimKVServerNameWhenRequested(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	p := &Postgresql{
+		Address: fmt.Sprintf("host=%s user=postgres password=swordfish sslmode=disable",
+			testutil.GetLocalHost()),
+		VerbatimAddress: true,
+		Databases:       []string{"postgres"},
+	}
+
+	var acc testutil.Accumulator
+
+	err := p.Gather(&acc)
+	require.NoError(t, err)
+
+	point, ok := acc.Get("postgresql")
+	require.True(t, ok)
+
+	assert.Equal(t,
+		fmt.Sprintf("host=%s user=postgres password=swordfish sslmode=disable", testutil.GetLocalHost()),
+		point.Tags["server"])
+}
+
+func TestPostgresqlMaintainsVerbatimURLServerNameWhenRequested(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	p := &Postgresql{
+		Address: fmt.Sprintf("postgres://postgres:swordfish@%s?sslmode=disable",
+			testutil.GetLocalHost()),
+		VerbatimAddress: true,
+		Databases:       []string{"postgres"},
+	}
+
+	var acc testutil.Accumulator
+
+	err := p.Gather(&acc)
+	require.NoError(t, err)
+
+	point, ok := acc.Get("postgresql")
+	require.True(t, ok)
+
+	assert.Equal(t,
+		fmt.Sprintf("postgres://postgres:swordfish@%s?sslmode=disable", testutil.GetLocalHost()),
+		point.Tags["server"])
+}
+
 func TestPostgresqlDefaultsToAllDatabases(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping integration test in short mode")