Validate mysql hosts during -configtest

metricbeat/module/mysql/mysql.go

@@ -5,14 +5,16 @@ package mysql
 
 import (
 	"database/sql"
+	"time"
 
-	// Register the MySQL driver.
-	_ "github.com/go-sql-driver/mysql"
+	"github.com/go-sql-driver/mysql"
+	"github.com/pkg/errors"
 )
 
 // CreateDSN creates a DSN (data source name) string out of hostname, username,
-// and password.
-func CreateDSN(host string, username string, password string) string {
+// password, and timeout. It validates the resulting DSN and returns an error
+// if the DSN is invalid.
+func CreateDSN(host, username, password string, timeout time.Duration) (string, error) {
 	// Example: [username[:password]@][protocol[(address)]]/
 	dsn := host
 
@@ -27,11 +29,30 @@ func CreateDSN(host string, username string, password string) string {
 	if username != "" {
 		dsn = username + dsn
 	}
-	return dsn
+
+	config, err := mysql.ParseDSN(dsn)
+	if err != nil {
+		return "", errors.Wrapf(err, "config error for host '%s'", host)
+	}
+
+	if timeout > 0 {
+		// Add connection timeouts to the DSN.
+		config.Timeout = timeout
+		config.ReadTimeout = timeout
+		config.WriteTimeout = timeout
+	}
+
+	return config.FormatDSN(), nil
 }
 
-// Connect creates a new DB connection. It expects a full MySQL DSN.
+// NewDB returns a new mysql database handle. The dsn value (data source name)
+// must be valid, otherwise an error will be returned.
+//
 // Example DSN: [username[:password]@][protocol[(address)]]/
-func Connect(dsn string) (*sql.DB, error) {
-	return sql.Open("mysql", dsn)
+func NewDB(dsn string) (*sql.DB, error) {
+	db, err := sql.Open("mysql", dsn)
+	if err != nil {
+		return nil, errors.Wrap(err, "sql open failed")
+	}
+	return db, nil
 }

metricbeat/module/mysql/mysql_integration_test.go

@@ -8,8 +8,8 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestConnect(t *testing.T) {
-	db, err := Connect(GetMySQLEnvDSN())
+func TestNewDB(t *testing.T) {
+	db, err := NewDB(GetMySQLEnvDSN())
 	assert.NoError(t, err)
 
 	err = db.Ping()

metricbeat/module/mysql/mysql_test.go

@@ -4,6 +4,7 @@ package mysql
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -13,12 +14,15 @@ func TestCreateDSN(t *testing.T) {
 	username := "root"
 	password := "test"
 
-	dsn := CreateDSN(hostname, username, password)
+	dsn, _ := CreateDSN(hostname, username, password, 0)
 	assert.Equal(t, "root:test@tcp(127.0.0.1:3306)/", dsn)
 
-	dsn = CreateDSN(hostname, username, "")
+	dsn, _ = CreateDSN(hostname, username, "", 0)
 	assert.Equal(t, "root@tcp(127.0.0.1:3306)/", dsn)
 
-	dsn = CreateDSN(hostname, "", "")
+	dsn, _ = CreateDSN(hostname, "", "", 0)
 	assert.Equal(t, "tcp(127.0.0.1:3306)/", dsn)
+
+	dsn, _ = CreateDSN(hostname, "", "", time.Second)
+	assert.Equal(t, "tcp(127.0.0.1:3306)/?readTimeout=1s&timeout=1s&writeTimeout=1s", dsn)
 }

metricbeat/module/mysql/status/status.go

@@ -45,8 +45,9 @@ type MetricSet struct {
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	// Unpack additional configuration options.
 	config := struct {
-		Username string `config:"username"`
-		Password string `config:"password"`
+		Hosts    []string `config:"hosts"    validate:"nonzero,required"`
+		Username string   `config:"username"`
+		Password string   `config:"password"`
 	}{
 		Username: "",
 		Password: "",
@@ -56,8 +57,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
-	// TODO (akroh): Apply validation to the mysql DSN format.
-	dsn := mysql.CreateDSN(base.Host(), config.Username, config.Password)
+	// Create and validate the data source name.
+	dsn, err := mysql.CreateDSN(base.Host(), config.Username, config.Password, base.Module().Config().Timeout)
+	if err != nil {
+		return nil, err
+	}
 
 	return &MetricSet{
 		BaseMetricSet: base,
@@ -69,9 +73,9 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch() (event common.MapStr, err error) {
 	if m.db == nil {
 		var err error
-		m.db, err = mysql.Connect(m.dsn)
+		m.db, err = mysql.NewDB(m.dsn)
 		if err != nil {
-			return nil, errors.Wrap(err, "mysql-status connect to host")
+			return nil, errors.Wrap(err, "mysql-status fetch failed")
 		}
 	}
 
@@ -90,6 +94,7 @@ func (m *MetricSet) loadStatus(db *sql.DB) (map[string]string, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer rows.Close()
 
 	mysqlStatus := map[string]string{}
 

metricbeat/module/mysql/status/status_test.go

@@ -0,0 +1,80 @@
+package status
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/metricbeat/mb"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestConfigValidation validates that the configuration and the DSN are
+// validated when the MetricSet is created.
+func TestConfigValidation(t *testing.T) {
+	tests := []struct {
+		in  interface{}
+		err string
+	}{
+		{
+			// Missing 'hosts'
+			in: map[string]interface{}{
+				"module":     "mysql",
+				"metricsets": []string{"status"},
+			},
+			err: "missing required field accessing config",
+		},
+		{
+			// Invalid DSN
+			in: map[string]interface{}{
+				"module":     "mysql",
+				"metricsets": []string{"status"},
+				"hosts":      []string{"127.0.0.1"},
+			},
+			err: "config error for host '127.0.0.1': invalid DSN: missing the slash separating the database name",
+		},
+		{
+			// Local unix socket
+			in: map[string]interface{}{
+				"module":     "mysql",
+				"metricsets": []string{"status"},
+				"hosts":      []string{"user@unix(/path/to/socket)/"},
+			},
+		},
+		{
+			// TCP on a remote host, e.g. Amazon RDS:
+			in: map[string]interface{}{
+				"module":     "mysql",
+				"metricsets": []string{"status"},
+				"hosts":      []string{"id:password@tcp(your-amazonaws-uri.com:3306)/}"},
+			},
+		},
+		{
+			// TCP on a remote host with user/pass specified separately
+			in: map[string]interface{}{
+				"module":     "mysql",
+				"metricsets": []string{"status"},
+				"hosts":      []string{"tcp(your-amazonaws-uri.com:3306)/}"},
+				"username":   "id",
+				"password":   "mypass",
+			},
+		},
+	}
+
+	for i, test := range tests {
+		c, err := common.NewConfigFrom(test.in)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, err = mb.NewModules([]*common.Config{c}, mb.Registry)
+		if err != nil && test.err == "" {
+			t.Errorf("unexpected error in testcase %d: %v", i, err)
+			continue
+		}
+		if test.err != "" && assert.Error(t, err, "expected '%v' in testcase %d", test.err, i) {
+			assert.Contains(t, err.Error(), test.err, "testcase %d", i)
+			continue
+		}
+	}
+}

metricbeat/module/mysql/testing.go

@@ -2,18 +2,25 @@ package mysql
 
 import (
 	"os"
+
+	"github.com/go-sql-driver/mysql"
 )
 
-// Helper functions for testing used in the mysql metricsets
+// Helper functions for testing used in the mysql MetricSets.
 
 // GetMySQLEnvDSN returns the MySQL server DSN to use for testing. It
 // reads the value from the MYSQL_DSN environment variable and returns
-// a DSN for 127.0.0.1 if it is not set.
+// root@tcp(127.0.0.1:3306)/ if it is not set.
 func GetMySQLEnvDSN() string {
 	dsn := os.Getenv("MYSQL_DSN")
 
 	if len(dsn) == 0 {
-		dsn = CreateDSN("tcp(127.0.0.1:3306)/", "root", "")
+		c := &mysql.Config{
+			Net:  "tcp",
+			Addr: "127.0.0.1:3306",
+			User: "root",
+		}
+		dsn = c.FormatDSN()
 	}
 	return dsn
 }