Fix mapping and race condition in sql module

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())]