Remove datasource option from SQL module and add tests (elastic#15686)

Remove datasource option from SQL module. This option was
intended to set the DSN of a database connection, and we were
ignoring the hosts setting. In other SQL modules we are using
the values in hosts as DSNs, do here the same for consistency.
Host is redacted when we cannot parse it as it can contain passwords.

StandardizeEvent is exposed in mbtest.Fetcher interface so we can
more easily check contents of events in tests.

Add integration tests of the module with MySQL and PostgreSQL.

Add real data.json with data from MySQL and PostgreSQL.

(cherry picked from commit 8c71abc)

metricbeat/docs/modules/sql.asciidoc

@@ -8,7 +8,7 @@ This file is generated! See scripts/mage/docs_collector.go
 
 beta[]
 
-This is the sql module that fetches metrics from a SQL database. You can define driver, datasource and SQL query.
+This is the sql module that fetches metrics from a SQL database. You can define driver and SQL query.
 
 
 
@@ -26,10 +26,9 @@ metricbeat.modules:
   metricsets:
     - query
   period: 10s
-  hosts: ["localhost"]
+  hosts: ["user=myuser password=mypassword dbname=mydb sslmode=disable"]
 
   driver: "postgres"
-  datasource: "user=myuser password=mypassword dbname=mydb sslmode=disable"
   sql_query: "select now()"
 
 ----

metricbeat/mb/testing/fetcher.go

@@ -20,6 +20,7 @@ package testing
 import (
 	"testing"
 
+	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/metricbeat/mb"
 )
@@ -32,6 +33,7 @@ type Fetcher interface {
 	FetchEvents() ([]mb.Event, []error)
 	WriteEvents(testing.TB, string)
 	WriteEventsCond(testing.TB, string, func(common.MapStr) bool)
+	StandardizeEvent(mb.Event, ...mb.EventModifier) beat.Event
 }
 
 // NewFetcher returns a test fetcher from a Metricset configuration
@@ -73,6 +75,10 @@ func (f *reportingMetricSetV2Fetcher) WriteEventsCond(t testing.TB, path string,
 	}
 }
 
+func (f *reportingMetricSetV2Fetcher) StandardizeEvent(event mb.Event, modifiers ...mb.EventModifier) beat.Event {
+	return StandardizeEvent(f, event, modifiers...)
+}
+
 type reportingMetricSetV2FetcherError struct {
 	mb.ReportingMetricSetV2Error
 }
@@ -96,6 +102,10 @@ func (f *reportingMetricSetV2FetcherError) WriteEventsCond(t testing.TB, path st
 	}
 }
 
+func (f *reportingMetricSetV2FetcherError) StandardizeEvent(event mb.Event, modifiers ...mb.EventModifier) beat.Event {
+	return StandardizeEvent(f, event, modifiers...)
+}
+
 type reportingMetricSetV2FetcherWithContext struct {
 	mb.ReportingMetricSetV2WithContext
 }
@@ -118,3 +128,7 @@ func (f *reportingMetricSetV2FetcherWithContext) WriteEventsCond(t testing.TB, p
 		t.Fatal("writing events", err)
 	}
 }
+
+func (f *reportingMetricSetV2FetcherWithContext) StandardizeEvent(event mb.Event, modifiers ...mb.EventModifier) beat.Event {
+	return StandardizeEvent(f, event, modifiers...)
+}

x-pack/metricbeat/metricbeat.reference.yml

@@ -935,10 +935,9 @@ metricbeat.modules:
   metricsets:
     - query
   period: 10s
-  hosts: ["localhost"]
+  hosts: ["user=myuser password=mypassword dbname=mydb sslmode=disable"]
 
   driver: "postgres"
-  datasource: "user=myuser password=mypassword dbname=mydb sslmode=disable"
   sql_query: "select now()"
 
 

x-pack/metricbeat/module/sql/_meta/config.yml

@@ -2,9 +2,8 @@
   metricsets:
     - query
   period: 10s
-  hosts: ["localhost"]
+  hosts: ["user=myuser password=mypassword dbname=mydb sslmode=disable"]
 
   driver: "postgres"
-  datasource: "user=myuser password=mypassword dbname=mydb sslmode=disable"
   sql_query: "select now()"
 

x-pack/metricbeat/module/sql/_meta/docs.asciidoc

@@ -1,3 +1,3 @@
-This is the sql module that fetches metrics from a SQL database. You can define driver, datasource and SQL query.
+This is the sql module that fetches metrics from a SQL database. You can define driver and SQL query.
 
 

x-pack/metricbeat/module/sql/docker-compose.yml

@@ -0,0 +1,12 @@
+version: '2.3'
+
+services:
+  mysql:
+    extends:
+      file: ../../../../metricbeat/module/mysql/docker-compose.yml
+      service: mysql
+
+  postgresql:
+    extends:
+      file: ../../../../metricbeat/module/postgresql/docker-compose.yml
+      service: postgresql

x-pack/metricbeat/module/sql/query/_meta/data.json

@@ -1,26 +1,30 @@
 {
-    "@timestamp":"2016-05-23T08:05:34.853Z",
-    "beat":{
-        "hostname":"beathost",
-        "name":"beathost"
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "event": {
+        "dataset": "sql.query",
+        "duration": 115000,
+        "module": "sql"
     },
-    "metricset":{
-        "host":"localhost",
-        "module":"sql",
-        "name":"query",
-        "rtt":44269
+    "metricset": {
+        "name": "query",
+        "period": 10000
     },
-    "sql":{
-        "metrics":{
-           "numeric":{
-              "mynumericfield":1
-           },
-           "string":{
-              "mystringfield":"abc"
-           }
+    "service": {
+        "address": "172.22.0.3:3306",
+        "type": "sql"
+    },
+    "sql": {
+        "driver": "mysql",
+        "metrics": {
+            "numeric": {
+                "table_rows": 6
+            },
+            "string": {
+                "engine": "InnoDB",
+                "table_name": "sys_config",
+                "table_schema": "sys"
+            }
         },
-        "driver":"postgres",
-        "query":"select * from mytable"
-     },
-    "type":"metricsets"
+        "query": "select table_schema, table_name, engine, table_rows from information_schema.tables where table_rows \u003e 0;"
+    }
 }

x-pack/metricbeat/module/sql/query/_meta/data_postgres.json

@@ -0,0 +1,45 @@
+{
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "event": {
+        "dataset": "sql.query",
+        "duration": 115000,
+        "module": "sql"
+    },
+    "metricset": {
+        "name": "query",
+        "period": 10000
+    },
+    "service": {
+        "address": "172.22.0.2:5432",
+        "type": "sql"
+    },
+    "sql": {
+        "driver": "postgres",
+        "metrics": {
+            "numeric": {
+                "blk_read_time": 0,
+                "blk_write_time": 0,
+                "blks_hit": 1923,
+                "blks_read": 111,
+                "conflicts": 0,
+                "datid": 12379,
+                "deadlocks": 0,
+                "numbackends": 1,
+                "temp_bytes": 0,
+                "temp_files": 0,
+                "tup_deleted": 0,
+                "tup_fetched": 1249,
+                "tup_inserted": 0,
+                "tup_returned": 1356,
+                "tup_updated": 0,
+                "xact_commit": 18,
+                "xact_rollback": 0
+            },
+            "string": {
+                "datname": "postgres",
+                "stats_reset": "2020-01-21 11:23:56.53"
+            }
+        },
+        "query": "select * from pg_stat_database"
+    }
+}

x-pack/metricbeat/module/sql/query/dsn.go

@@ -0,0 +1,42 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package query
+
+import (
+	"net/url"
+
+	"github.com/go-sql-driver/mysql"
+
+	"github.com/elastic/beats/metricbeat/mb"
+)
+
+// ParseDSN tries to parse the host
+func ParseDSN(mod mb.Module, host string) (mb.HostData, error) {
+	// TODO: Add support for `username` and `password` as module options
+
+	sanitized := sanitize(host)
+
+	return mb.HostData{
+		URI:          host,
+		SanitizedURI: sanitized,
+		Host:         sanitized,
+	}, nil
+}
+
+func sanitize(host string) string {
+	// Host is a standard URL
+	if url, err := url.Parse(host); err == nil && len(url.Host) > 0 {
+		return url.Host
+	}
+
+	// Host is a MySQL DSN
+	if config, err := mysql.ParseDSN(host); err == nil {
+		return config.Addr
+	}
+
+	// TODO: Add support for PostgreSQL connection strings and other formats
+
+	return "(redacted)"
+}

x-pack/metricbeat/module/sql/query/query.go

@@ -10,21 +10,22 @@ import (
 	"strings"
 	"time"
 
+	"github.com/jmoiron/sqlx"
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
 	"github.com/elastic/beats/metricbeat/mb"
-
-	"github.com/jmoiron/sqlx"
 )
 
 // init registers the MetricSet with the central registry as soon as the program
 // starts. The New function will be called later to instantiate an instance of
 // the MetricSet for each host defined in the module's configuration. After the
 // MetricSet has been created then Fetch will begin to be called periodically.
 func init() {
-	mb.Registry.MustAddMetricSet("sql", "query", New)
+	mb.Registry.MustAddMetricSet("sql", "query", New,
+		mb.WithHostParser(ParseDSN),
+	)
 }
 
 // MetricSet holds any configuration or state information. It must implement
@@ -33,9 +34,8 @@ func init() {
 // interface methods except for Fetch.
 type MetricSet struct {
 	mb.BaseMetricSet
-	Driver     string
-	Datasource string
-	Query      string
+	Driver string
+	Query  string
 }
 
 // New creates a new instance of the MetricSet. New is responsible for unpacking
@@ -44,9 +44,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	cfgwarn.Beta("The sql query metricset is beta.")
 
 	config := struct {
-		Driver     string `config:"driver"`
-		Datasource string `config:"datasource"`
-		Query      string `config:"sql_query"`
+		Driver string `config:"driver"`
+		Query  string `config:"sql_query"`
 	}{}
 
 	if err := base.Module().UnpackConfig(&config); err != nil {
@@ -56,7 +55,6 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		Driver:        config.Driver,
-		Datasource:    config.Datasource,
 		Query:         config.Query,
 	}, nil
 }
@@ -65,7 +63,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // format. It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
-	db, err := sqlx.Open(m.Driver, m.Datasource)
+	db, err := sqlx.Open(m.Driver, m.HostData().URI)
 	if err != nil {
 		return errors.Wrap(err, "error opening connection")
 	}