Add ml_job metricset to Elasticsearch module

CHANGELOG.asciidoc

@@ -298,6 +298,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Collect accumulated docker network metrics and mark old ones as deprecated. {pull}7253[7253]
 - Add TLS support to MongoDB module. {pull}7401[7401]
 - Added Traefik module with health metricset. {pull}7413[7413]
+- Add Elasticsearch ml_job metricsets. {pull}7196[7196]
 
 *Packetbeat*
 

metricbeat/docs/fields.asciidoc

@@ -3635,6 +3635,53 @@ format: bytes
 Total number of memory used by the segments in bytes.
 
 
+--
+
+[float]
+== ml.job fields
+
+ml
+
+
+
+*`elasticsearch.ml.job.id`*::
++
+--
+type: keyword
+
+Unique ml job id.
+
+
+--
+
+*`elasticsearch.ml.job.state`*::
++
+--
+type: keyword
+
+Job state.
+
+
+--
+
+*`elasticsearch.ml.job.data_counts.processed_record_count`*::
++
+--
+type: long
+
+Processed data events.
+
+
+--
+
+*`elasticsearch.ml.job.data_counts.invalid_date_count`*::
++
+--
+type: long
+
+The number of records with either a missing date field or a date that could not be parsed.
+
+
 --
 
 [float]

metricbeat/docs/modules/elasticsearch.asciidoc

@@ -28,6 +28,7 @@ metricbeat.modules:
     #- index
     #- index_summary
     #- shard
+    #- ml_job
   period: 10s
   hosts: ["localhost:9200"]
 ----
@@ -43,6 +44,8 @@ The following metricsets are available:
 
 * <<metricbeat-metricset-elasticsearch-index_summary,index_summary>>
 
+* <<metricbeat-metricset-elasticsearch-ml_job,ml_job>>
+
 * <<metricbeat-metricset-elasticsearch-node,node>>
 
 * <<metricbeat-metricset-elasticsearch-node_stats,node_stats>>
@@ -55,6 +58,8 @@ include::elasticsearch/index.asciidoc[]
 
 include::elasticsearch/index_summary.asciidoc[]
 
+include::elasticsearch/ml_job.asciidoc[]
+
 include::elasticsearch/node.asciidoc[]
 
 include::elasticsearch/node_stats.asciidoc[]

metricbeat/docs/modules/elasticsearch/ml_job.asciidoc

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

metricbeat/docs/modules_list.asciidoc

@@ -33,8 +33,9 @@ This file is generated! See scripts/docs_collector.py
 |<<metricbeat-module-dropwizard,Dropwizard>>  beta[]   |image:./images/icon-no.png[No prebuilt dashboards]    |  
 .1+| .1+|  |<<metricbeat-metricset-dropwizard-collector,collector>> beta[]  
 |<<metricbeat-module-elasticsearch,Elasticsearch>>  beta[]   |image:./images/icon-no.png[No prebuilt dashboards]    |  
-.6+| .6+|  |<<metricbeat-metricset-elasticsearch-index,index>> experimental[]  
+.7+| .7+|  |<<metricbeat-metricset-elasticsearch-index,index>> experimental[]  
 |<<metricbeat-metricset-elasticsearch-index_summary,index_summary>> experimental[]  
+|<<metricbeat-metricset-elasticsearch-ml_job,ml_job>> experimental[]  
 |<<metricbeat-metricset-elasticsearch-node,node>> beta[]  
 |<<metricbeat-metricset-elasticsearch-node_stats,node_stats>> beta[]  
 |<<metricbeat-metricset-elasticsearch-pending_tasks,pending_tasks>> experimental[]  

metricbeat/metricbeat.reference.yml

@@ -201,6 +201,7 @@ metricbeat.modules:
     #- index
     #- index_summary
     #- shard
+    #- ml_job
   period: 10s
   hosts: ["localhost:9200"]
 

metricbeat/module/elasticsearch/_meta/config.reference.yml

@@ -5,5 +5,6 @@
     #- index
     #- index_summary
     #- shard
+    #- ml_job
   period: 10s
   hosts: ["localhost:9200"]

metricbeat/module/elasticsearch/elasticsearch_integration_test.go

@@ -21,6 +21,8 @@ package elasticsearch_test
 
 import (
 	"fmt"
+	"io/ioutil"
+	"net"
 	"net/http"
 	"os"
 	"testing"
@@ -30,8 +32,11 @@ import (
 	"github.com/elastic/beats/libbeat/tests/compose"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 
+	"bytes"
+
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/index"
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/index_summary"
+	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/ml_job"
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/node"
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/node_stats"
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/shard"
@@ -40,6 +45,7 @@ import (
 var metricSets = []string{
 	"index",
 	"index_summary",
+	"ml_job",
 	"node",
 	"node_stats",
 	"shard",
@@ -48,7 +54,14 @@ var metricSets = []string{
 func TestFetch(t *testing.T) {
 	compose.EnsureUp(t, "elasticsearch")
 
-	err := createIndex(getEnvHost() + ":" + getEnvPort())
+	host := net.JoinHostPort(getEnvHost(), getEnvPort())
+	err := createIndex(host)
+	assert.NoError(t, err)
+
+	err = enableTrialLicense(host)
+	assert.NoError(t, err)
+
+	err = createMLJob(host)
 	assert.NoError(t, err)
 
 	for _, metricSet := range metricSets {
@@ -110,31 +123,97 @@ func getConfig(metricset string) map[string]interface{} {
 func createIndex(host string) error {
 	client := &http.Client{}
 
-	resp, err := http.Get("http://" + host + "/testindex")
+	if checkExists("http://" + host + "/testindex") {
+		return nil
+	}
+
+	req, err := http.NewRequest("PUT", "http://"+host+"/testindex", nil)
 	if err != nil {
 		return err
 	}
-	resp.Body.Close()
 
-	// This means index already exists
-	if resp.StatusCode == 200 {
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return fmt.Errorf("HTTP error %d: %s", resp.StatusCode, resp.Status)
+	}
+
+	return nil
+}
+
+// createIndex creates and elasticsearch index in case it does not exit yet
+func enableTrialLicense(host string) error {
+	client := &http.Client{}
+
+	enableXPackURL := "/_xpack/license/start_trial?acknowledge=true"
+
+	req, err := http.NewRequest("POST", "http://"+host+enableXPackURL, nil)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	return nil
+}
+
+func createMLJob(host string) error {
+
+	mlJob, err := ioutil.ReadFile("ml_job/_meta/test/test_job.json")
+	if err != nil {
+		return err
+	}
+
+	client := &http.Client{}
+
+	jobURL := "/_xpack/ml/anomaly_detectors/total-requests"
+
+	if checkExists("http://" + host + jobURL) {
 		return nil
 	}
 
-	req, err := http.NewRequest("PUT", "http://"+host+"/testindex", nil)
+	req, err := http.NewRequest("PUT", "http://"+host+jobURL, bytes.NewReader(mlJob))
 	if err != nil {
 		return err
 	}
+	req.Header.Add("Content-Type", "application/json")
 
-	resp, err = client.Do(req)
+	resp, err := client.Do(req)
 	if err != nil {
 		return err
 	}
 	defer resp.Body.Close()
 
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
 	if resp.StatusCode != 200 {
-		return fmt.Errorf("HTTP error %d: %s", resp.StatusCode, resp.Status)
+		return fmt.Errorf("HTTP error loading ml job %d: %s, %s", resp.StatusCode, resp.Status, body)
 	}
 
 	return nil
 }
+
+func checkExists(url string) bool {
+	resp, err := http.Get(url)
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+
+	// Entry exists
+	if resp.StatusCode == 200 {
+		return true
+	}
+	return false
+}

metricbeat/module/elasticsearch/ml_job/_meta/data.json

@@ -0,0 +1,28 @@
+{
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "beat": {
+        "hostname": "host.example.com",
+        "name": "host.example.com"
+    },
+    "elasticsearch": {
+        "ml": {
+            "job": {
+                "data_counts": {
+                    "processed_record_count": 0
+                },
+                "id": "filebeat-apache2-access-low_request_rate",
+                "state": "closed"
+            }
+        }
+    },
+    "metricset": {
+        "host": "127.0.0.1:9200",
+        "module": "elasticsearch",
+        "name": "ml_job",
+        "namespace": "elasticsearch.ml.job",
+        "rtt": 115
+    },
+    "service": {
+        "name": "elasticsearch"
+    }
+}

metricbeat/module/elasticsearch/ml_job/_meta/docs.asciidoc

@@ -0,0 +1,3 @@
+=== elasticsearch ml_job MetricSet
+
+This is the ml_job metricset of the module Elasticsearch. This metricset requires https://www.elastic.co/products/x-pack/machine-learning[Machine Learning] to be enabled.

metricbeat/module/elasticsearch/ml_job/_meta/fields.yml

@@ -0,0 +1,23 @@
+- name: ml.job
+  type: group
+  description: >
+    ml
+  fields:
+    - name: id
+      type: keyword
+      description: >
+        Unique ml job id.
+    - name: state
+      type: keyword
+      description: >
+        Job state.
+    - name: data_counts.processed_record_count
+      type: long
+      description: >
+        Processed data events.
+    - name: data_counts.invalid_date_count
+      type: long
+      description: >
+        The number of records with either a missing date field or a date that could not be parsed.
+
+