apply review

metricbeat/docs/fields.asciidoc

@@ -3672,6 +3672,16 @@ 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/module/elasticsearch/ml_job/_meta/docs.asciidoc

@@ -1,3 +1,3 @@
-=== elasticsearch ml MetricSet
+=== 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

@@ -15,3 +15,9 @@
       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.
+
+

metricbeat/module/elasticsearch/ml_job/data.go

@@ -20,6 +20,8 @@ package ml_job
 import (
 	"encoding/json"
 
+	"github.com/joeshaw/multierror"
+
 	"github.com/elastic/beats/libbeat/common"
 	s "github.com/elastic/beats/libbeat/common/schema"
 	c "github.com/elastic/beats/libbeat/common/schema/mapstriface"
@@ -32,6 +34,7 @@ var (
 		"state": c.Str("state"),
 		"data_counts": c.Dict("data_counts", s.Schema{
 			"processed_record_count": c.Int("processed_record_count"),
+			"invalid_date_count":     c.Int("invalid_date_count"),
 		}),
 	}
 )
@@ -40,16 +43,15 @@ type jobsStruct struct {
 	Jobs []map[string]interface{} `json:"jobs"`
 }
 
-func eventsMapping(r mb.ReporterV2, content []byte) []error {
+func eventsMapping(r mb.ReporterV2, content []byte) error {
 
 	jobsData := &jobsStruct{}
 	err := json.Unmarshal(content, jobsData)
 	if err != nil {
-		r.Error(err)
-		return []error{err}
+		return err
 	}
 
-	var errs []error
+	var errs multierror.Errors
 	for _, job := range jobsData.Jobs {
 
 		event := mb.Event{}
@@ -63,5 +65,5 @@ func eventsMapping(r mb.ReporterV2, content []byte) []error {
 		event.RootFields.Put("service.name", "elasticsearch")
 		r.Event(event)
 	}
-	return errs
+	return errs.Err()
 }

metricbeat/module/elasticsearch/ml_job/ml.go → metricbeat/module/elasticsearch/ml_job/ml_job.go

@@ -64,7 +64,7 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 
 	// Not master, no event sent
 	if !isMaster {
-		logp.Debug("elasticsearch", "Trying to fetch index summary stats from a none master node.")
+		logp.Debug("elasticsearch", "Trying to fetch machine learning job stats from a non-master node.")
 		return
 	}
 
@@ -74,5 +74,9 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 		return
 	}
 
-	eventsMapping(r, content)
+	err = eventsMapping(r, content)
+	if err != nil {
+		r.Error(err)
+		return
+	}
 }

metricbeat/module/elasticsearch/test_elasticsearch.py

@@ -47,7 +47,8 @@ def create_ml_job(self):
         try:
             es.transport.perform_request('POST', "/_xpack/license/start_trial?acknowledge=true")
         except:
-            print "Trial already enabled"
+            e = sys.exc_info()[0]
+            print "Trial already enabled. Error: {}".format(e)
 
         # Check if an ml job already exists
         response = es.transport.perform_request('GET', "/_xpack/ml/anomaly_detectors/_all/")