Proposal: Dynamic Metricsets (MySQL)

metricbeat/beater/event.go

@@ -56,16 +56,23 @@ func (b EventBuilder) Build() (common.MapStr, error) {
 		delete(event, mb.ModuleData)
 	}
 
+	metricsetName := b.MetricSetName
+	mName, metricsetNameExists := event["_metricsetName"]
+	if metricsetNameExists {
+		metricsetName = mName.(string)
+		delete(event, "_metricsetName")
+	}
+
 	event = common.MapStr{
 		"@timestamp":            timestamp,
 		"type":                  typeName,
 		common.EventMetadataKey: b.metadata,
 		b.ModuleName: common.MapStr{
-			b.MetricSetName: event,
+			metricsetName: event,
 		},
 		"metricset": common.MapStr{
 			"module": b.ModuleName,
-			"name":   b.MetricSetName,
+			"name":   metricsetName,
 			"rtt":    b.FetchDuration.Nanoseconds() / int64(time.Microsecond),
 		},
 	}
@@ -77,6 +84,13 @@ func (b EventBuilder) Build() (common.MapStr, error) {
 		}
 	}
 
+	//
+	if metricsetNameExists {
+		if _, ok := event["metricset"].(common.MapStr); ok {
+			event["metricset"].(common.MapStr)["dynamic"] = b.MetricSetName
+		}
+	}
+
 	// Overwrite default index if set.
 	if indexName != "" {
 		event["beat"] = common.MapStr{

metricbeat/docs/fields.asciidoc

@@ -2765,6 +2765,21 @@ MySQL server status metrics collected from MySQL.
 
 
 
+[float]
+== query Fields
+
+query
+
+
+
+[float]
+=== mysql.query.example
+
+type: keyword
+
+Example field
+
+
 [float]
 == status Fields
 

metricbeat/docs/modules/mysql.asciidoc

@@ -55,14 +55,25 @@ metricbeat.modules:
   # The username and password can either be set in the DSN or using the username
   # and password config options. Those specified in the DSN take precedence.
   hosts: ["root:secret@tcp(127.0.0.1:3306)/"]
+
+
+- module: mysql
+  metricsets: ["query"]
+  namespace: "test"
+  query: "SELECT * FROM example"
+  hosts: ["root:test@tcp(127.0.0.1:3306)/exampledatabase"]
 ----
 
 [float]
 === Metricsets
 
 The following metricsets are available:
 
+* <<metricbeat-metricset-mysql-query,query>>
+
 * <<metricbeat-metricset-mysql-status,status>>
 
+include::mysql/query.asciidoc[]
+
 include::mysql/status.asciidoc[]
 

metricbeat/docs/modules/mysql/query.asciidoc

@@ -0,0 +1,19 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[metricbeat-metricset-mysql-query]]
+include::../../../module/mysql/query/_meta/docs.asciidoc[]
+
+
+==== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-mysql,exported fields>> section.
+
+Here is an example document generated by this metricset:
+
+[source,json]
+----
+include::../../../module/mysql/query/_meta/data.json[]
+----

metricbeat/include/list.go

@@ -25,6 +25,7 @@ import (
 	_ "github.com/elastic/beats/metricbeat/module/mongodb"
 	_ "github.com/elastic/beats/metricbeat/module/mongodb/status"
 	_ "github.com/elastic/beats/metricbeat/module/mysql"
+	_ "github.com/elastic/beats/metricbeat/module/mysql/query"
 	_ "github.com/elastic/beats/metricbeat/module/mysql/status"
 	_ "github.com/elastic/beats/metricbeat/module/nginx"
 	_ "github.com/elastic/beats/metricbeat/module/nginx/stubstatus"

metricbeat/metricbeat.template-es2x.json

@@ -1458,6 +1458,15 @@
         },
         "mysql": {
           "properties": {
+            "query": {
+              "properties": {
+                "example": {
+                  "ignore_above": 1024,
+                  "index": "not_analyzed",
+                  "type": "string"
+                }
+              }
+            },
             "status": {
               "properties": {
                 "aborted": {

metricbeat/metricbeat.template.json

@@ -1453,6 +1453,14 @@
         },
         "mysql": {
           "properties": {
+            "query": {
+              "properties": {
+                "example": {
+                  "ignore_above": 1024,
+                  "type": "keyword"
+                }
+              }
+            },
             "status": {
               "properties": {
                 "aborted": {

metricbeat/module/mysql/_meta/config.yml

@@ -7,3 +7,10 @@
   # The username and password can either be set in the DSN or using the username
   # and password config options. Those specified in the DSN take precedence.
   hosts: ["root:secret@tcp(127.0.0.1:3306)/"]
+
+
+- module: mysql
+  metricsets: ["query"]
+  namespace: "test"
+  query: "SELECT * FROM example"
+  hosts: ["root:test@tcp(127.0.0.1:3306)/exampledatabase"]

metricbeat/module/mysql/query/_meta/data.json

@@ -0,0 +1,19 @@
+{
+    "@timestamp":"2016-05-23T08:05:34.853Z",
+    "beat":{
+        "hostname":"beathost",
+        "name":"beathost"
+    },
+    "metricset":{
+        "host":"localhost",
+        "module":"mysql",
+        "name":"status",
+        "rtt":44269
+    },
+    "mysql":{
+        "query":{
+            "example": "query"
+        }
+    },
+    "type":"metricsets"
+}

metricbeat/module/mysql/query/_meta/docs.asciidoc

@@ -0,0 +1,3 @@
+=== mysql query MetricSet
+
+This is the query metricset of the module mysql.

metricbeat/module/mysql/query/_meta/fields.yml

@@ -0,0 +1,9 @@
+- name: query
+  type: group
+  description: >
+    query
+  fields:
+    - name: example
+      type: keyword
+      description: >
+        Example field

metricbeat/module/mysql/query/data.go

@@ -0,0 +1 @@
+package query

metricbeat/module/mysql/query/query.go

@@ -0,0 +1,96 @@
+package query
+
+import (
+	"database/sql"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/metricbeat/mb"
+	"github.com/elastic/beats/metricbeat/module/mysql"
+
+	"github.com/pkg/errors"
+)
+
+var (
+	debugf = logp.MakeDebug("mysql-query")
+)
+
+func init() {
+	if err := mb.Registry.AddMetricSet("mysql", "query", New, mysql.ParseDSN); err != nil {
+		panic(err)
+	}
+}
+
+// MetricSet for fetching MySQL server query.
+type MetricSet struct {
+	mb.BaseMetricSet
+	db *sql.DB
+	query string
+	namespace string
+}
+
+// New creates and returns a new MetricSet instance.
+func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
+
+	// Unpack additional configuration options.
+	config := struct {
+		Query string `config:"query"`
+		Namespace string `config:"namespace"`
+	}{}
+	err := base.Module().UnpackConfig(&config)
+	if err != nil {
+		return nil, err
+	}
+
+	return &MetricSet{
+		BaseMetricSet: base,
+		query: config.Query,
+		namespace: config.Namespace,
+	}, nil
+}
+
+// Fetch fetches query messages from a mysql host.
+func (m *MetricSet) Fetch() (common.MapStr, error) {
+	if m.db == nil {
+		var err error
+		m.db, err = mysql.NewDB(m.HostData().URI)
+		if err != nil {
+			return nil, errors.Wrap(err, "mysql-query fetch failed")
+		}
+	}
+
+	event, err := m.loadQuery(m.db, m.query)
+	event["_metricsetName"] = m.namespace
+
+	if err != nil {
+		return event, err
+	}
+
+	return event, nil
+}
+
+// loadStatus loads all status entries from the given database into an array.
+func (m *MetricSet) loadQuery(db *sql.DB, query string) (common.MapStr, error) {
+	// Returns all rows for the given query
+	rows, err := db.Query(query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	data := common.MapStr{}
+
+	for rows.Next() {
+		var name string
+		var value string
+
+		err = rows.Scan(&name, &value)
+		if err != nil {
+			return nil, err
+		}
+
+		data[name] = value
+	}
+
+	return data, nil
+}