Fix mapping and race condition in sql module (#15738)

Mapping was defined at the metricset level, but it should be defined at
the module level.

Add system test to earlier detect this kind of issues.

Also add validation so driver and sql_query options cannot be empty.

Fetch was reusing some maps when building the event, what was
causing mixed data in events and panics on queries with lots of rows.

Fixes #15736

metricbeat/docs/fields.asciidoc

@@ -30466,25 +30466,12 @@ type: long
 [[exported-fields-sql]]
 == sql fields
 
-sql module
+SQL module fetches metrics from a SQL database
 
 
 
-[float]
-=== sql
-
-`sql` fetches metrics from a SQL database 
-
-
-
-[float]
-=== query
-
-query
-
-
 
-*`sql.query.driver`*::
+*`sql.driver`*::
 +
 --
 Driver used to execute the query.
@@ -30494,7 +30481,7 @@ type: keyword
 
 --
 
-*`sql.query.query`*::
+*`sql.query`*::
 +
 --
 Query executed to collect metrics.
@@ -30504,7 +30491,7 @@ type: keyword
 
 --
 
-*`sql.query.metrics.numeric.*`*::
+*`sql.metrics.numeric.*`*::
 +
 --
 Numeric metrics collected.
@@ -30514,7 +30501,7 @@ type: object
 
 --
 
-*`sql.query.metrics.string.*`*::
+*`sql.metrics.string.*`*::
 +
 --
 Non-numeric values collected.

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

@@ -2,10 +2,26 @@
   title: "sql"
   release: beta
   description: >
-    sql module
+    SQL module fetches metrics from a SQL database
   fields:
     - name: sql
       type: group
-      description: >
-        `sql` fetches metrics from a SQL database 
       fields:
+        - name: driver
+          type: keyword
+          description: >
+            Driver used to execute the query.
+        - name: query
+          type: keyword
+          description: >
+            Query executed to collect metrics.
+        - name: metrics.numeric.*
+          type: object
+          object_type: double
+          description: >
+            Numeric metrics collected.
+        - name: metrics.string.*
+          type: object
+          object_type: keyword
+          description: >
+            Non-numeric values collected.

x-pack/metricbeat/module/sql/query/_meta/fields.yml

@@ -1,24 +1 @@
-- name: query
-  type: group
-  release: beta
-  description: >
-    query
-  fields:
-    - name: driver
-      type: keyword
-      description: >
-        Driver used to execute the query.
-    - name: query
-      type: keyword
-      description: >
-        Query executed to collect metrics.
-    - name: metrics.numeric.*
-      type: object
-      object_type: double
-      description: >
-        Numeric metrics collected.
-    - name: metrics.string.*
-      type: object
-      object_type: keyword
-      description: >
-        Non-numeric values collected.
+- release: beta

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

@@ -44,8 +44,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	cfgwarn.Beta("The sql query metricset is beta.")
 
 	config := struct {
-		Driver string `config:"driver"`
-		Query  string `config:"sql_query"`
+		Driver string `config:"driver" validate:"nonzero,required"`
+		Query  string `config:"sql_query" validate:"nonzero,required"`
 	}{}
 
 	if err := base.Module().UnpackConfig(&config); err != nil {
@@ -96,41 +96,39 @@ func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 		vals[i] = new(interface{})
 	}
 
-	numericValue := common.MapStr{}
-
-	stringValue := common.MapStr{}
-
 	for rows.Next() {
 		err = rows.Scan(vals...)
 		if err != nil {
 			m.Logger().Debug(errors.Wrap(err, "error trying to scan rows"))
 			continue
 		}
 
-		data := common.MapStr{
-			"metrics": common.MapStr{},
-		}
-		data.Put("driver", m.Driver)
-		data.Put("query", m.Query)
+		numericMetrics := common.MapStr{}
+		stringMetrics := common.MapStr{}
 
 		for i := 0; i < len(vals); i++ {
-			dataMetrics := data["metrics"].(common.MapStr)
-
 			value := getValue(vals[i].(*interface{}))
-
 			num, err := strconv.ParseFloat(value, 64)
 			if err == nil {
-				numericValue.Put(cols[i], num)
-				dataMetrics.Put("numeric", numericValue)
+				numericMetrics[cols[i]] = num
 			} else {
-				stringValue.Put(cols[i], value)
-				dataMetrics.Put("string", stringValue)
+				stringMetrics[cols[i]] = value
 			}
 
-			report.Event(mb.Event{
-				RootFields: common.MapStr{"sql": data},
-			})
 		}
+
+		report.Event(mb.Event{
+			RootFields: common.MapStr{
+				"sql": common.MapStr{
+					"driver": m.Driver,
+					"query":  m.Query,
+					"metrics": common.MapStr{
+						"numeric": numericMetrics,
+						"string":  stringMetrics,
+					},
+				},
+			},
+		})
 	}
 
 	if rows.Err() != nil {

x-pack/metricbeat/module/sql/test_sql.py

@@ -0,0 +1,41 @@
+import os
+import sys
+import unittest
+
+sys.path.append(os.path.join(os.path.dirname(__file__), '../../tests/system'))
+from xpack_metricbeat import XPackTest, metricbeat
+
+
+class Test(XPackTest):
+
+    COMPOSE_SERVICES = ['mysql']
+
+    @unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")
+    def test_query(self):
+        """
+        sql custom query test
+        """
+        self.render_config_template(modules=[{
+            "name": "sql",
+            "metricsets": ["query"],
+            "hosts": self.get_hosts(),
+            "period": "5s",
+            "additional_content": """
+  driver: mysql
+  sql_query: 'select table_schema, table_name, engine, table_rows from information_schema.tables where table_rows > 0'"""
+        }])
+        proc = self.start_beat(home=self.beat_path)
+        self.wait_until(lambda: self.output_lines() > 0)
+        proc.check_kill_and_wait()
+        self.assert_no_logged_warnings()
+
+        output = self.read_output_json()
+        self.assertGreater(len(output), 0)
+
+        for evt in output:
+            self.assert_fields_are_documented(evt)
+            self.assertIn("sql", evt.keys(), evt)
+            self.assertIn("query", evt["sql"].keys(), evt)
+
+    def get_hosts(self):
+        return ['root:test@tcp({})/'.format(self.compose_host())]