[Metricbeat] Convert millis-since-epoch timestamps in elasticsearch/ml_job metricset to ints

CHANGELOG.next.asciidoc

@@ -236,6 +236,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Change kubernetes.event.message to text. {pull}13964[13964]
 - Fix performance counter values for windows/perfmon metricset. {issue}14036[14036] {pull}14039[14039]
 - Add FailOnRequired when applying schema and fix metric names in mongodb metrics metricset. {pull}14143[14143]
+- Convert indexed ms-since-epoch timestamp fields in `elasticsearch/ml_job` metricset to ints from float64s. {issue}14220[14220] {pull}14222[14222]
 
 *Packetbeat*
 

metricbeat/helper/elastic/elastic.go

@@ -106,3 +106,24 @@ func ReportAndLogError(err error, r mb.ReporterV2, l *logp.Logger) {
 	r.Error(err)
 	l.Error(err)
 }
+
+// FixTimestampField converts the given timestamp field in the given map from a float64 to an
+// int, so that it is not serialized in scientific notation in the event. This is because
+// Elasticsearch cannot accepts scientific notation to represent millis-since-epoch values
+// for it's date fields: https://github.com/elastic/elasticsearch/pull/36691
+func FixTimestampField(m common.MapStr, field string) error {
+	v, err := m.GetValue(field)
+	if err == common.ErrKeyNotFound {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	switch vv := v.(type) {
+	case float64:
+		_, err := m.Put(field, int(vv))
+		return err
+	}
+	return nil
+}

metricbeat/helper/elastic/elastic_test.go

@@ -87,3 +87,56 @@ func TestReportErrorForMissingField(t *testing.T) {
 	assert.Equal(t, expectedError, err)
 	assert.Equal(t, expectedError, currentErr)
 }
+
+func TestFixTimestampField(t *testing.T) {
+	tests := []struct {
+		Name          string
+		OriginalValue map[string]interface{}
+		ExpectedValue map[string]interface{}
+	}{
+		{
+			"converts float64s in scientific notation to ints",
+			map[string]interface{}{
+				"foo": 1.571284349E12,
+			},
+			map[string]interface{}{
+				"foo": 1571284349000,
+			},
+		},
+		{
+			"converts regular notation float64s to ints",
+			map[string]interface{}{
+				"foo": float64(1234),
+			},
+			map[string]interface{}{
+				"foo": 1234,
+			},
+		},
+		{
+			"ignores missing fields",
+			map[string]interface{}{
+				"bar": 12345,
+			},
+			map[string]interface{}{
+				"bar": 12345,
+			},
+		},
+		{
+			"leaves strings untouched",
+			map[string]interface{}{
+				"foo": "bar",
+			},
+			map[string]interface{}{
+				"foo": "bar",
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.Name, func(t *testing.T) {
+			err := FixTimestampField(test.OriginalValue, "foo")
+			assert.NoError(t, err)
+			assert.Equal(t, test.ExpectedValue, test.OriginalValue)
+		})
+	}
+}

metricbeat/module/elasticsearch/ml_job/data_xpack.go

@@ -49,13 +49,22 @@ func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info,
 	}
 
 	var errs multierror.Errors
-	for _, job := range jobsArr {
-		job, ok = job.(map[string]interface{})
+	for _, j := range jobsArr {
+		job, ok := j.(map[string]interface{})
 		if !ok {
 			errs = append(errs, fmt.Errorf("job is not a map"))
 			continue
 		}
 
+		if err := elastic.FixTimestampField(job, "data_counts.earliest_record_timestamp"); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		if err := elastic.FixTimestampField(job, "data_counts.latest_record_timestamp"); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+
 		event := mb.Event{}
 		event.RootFields = common.MapStr{
 			"cluster_uuid": info.ClusterID,