allow forcing dynamic object type mapping (#6691)

Currently dynamic mappings are tolerant of unexpected data types.  For example, this definition:
```
- name: test
    type: object
    object_type: long
    dynamic: true
```

produces this dynamic template:
```
"test": {
  "path_match": "test.*",
  "match_mapping_type": "long",
  "mapping": {
    "type": "long"
  }
}
```

If the first value seen is a string, say indexing this document:
```
"test": {
      "somelong": 16,
      "somestring": "foo"
}
```

the resulting mapping is actually:
```
"test": {
  "dynamic": "true",
  "properties": {
    "test": {
      "properties": {
        "somelong": {
          "type": "long"
        },
        "somestring": {
          "type": "keyword",
          "ignore_above": 1024
        }
      }
    }
  }
}
```

In apm-server, data is already validated before being sent along to elasticsearch, so barring a bug, a string will not be sent where a long is required.  This change allows forcing the data for a dynamic field to be a specific type via field config, eg for:

```
- name: test
    type: object
    object_type: long
    object_type_mapping_type: "*"
    dynamic: true
```

The resulting dynamic template would be:

```
"test": {
  "path_match": "test.*",
  "match_mapping_type": "*",
  "mapping": {
    "type": "long"
  }
}
```

Any value that can't be mapped to `long` will result in an indexing error.  Continuing this example:
```
"error": {
  "root_cause": [
    {
      "type": "mapper_parsing_exception",
       "reason": "failed to parse [test.somestring]"
    }
  ]
}
```

CHANGELOG.asciidoc

@@ -25,6 +25,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Add logging when monitoring cannot connect to Elasticsearch. {pull}6365[6365]
 - Rename beat.cpu.*.time metrics to beat.cpu.*.time.ms. {pull}6449[6449]
 - Mark `system.syslog.message` and `system.auth.message` as `text` instead of `keyword`. {pull}6589[6589]
+- Allow override of dynamic template `match_mapping_type` for fields with object_type. {pull}6691[6691]
 
 *Auditbeat*
 

libbeat/common/field.go

@@ -17,22 +17,23 @@ import (
 type Fields []Field
 
 type Field struct {
-	Name           string      `config:"name"`
-	Type           string      `config:"type"`
-	Description    string      `config:"description"`
-	Format         string      `config:"format"`
-	ScalingFactor  int         `config:"scaling_factor"`
-	Fields         Fields      `config:"fields"`
-	MultiFields    Fields      `config:"multi_fields"`
-	ObjectType     string      `config:"object_type"`
-	Enabled        *bool       `config:"enabled"`
-	Analyzer       string      `config:"analyzer"`
-	SearchAnalyzer string      `config:"search_analyzer"`
-	Norms          bool        `config:"norms"`
-	Dynamic        DynamicType `config:"dynamic"`
-	Index          *bool       `config:"index"`
-	DocValues      *bool       `config:"doc_values"`
-	CopyTo         string      `config:"copy_to"`
+	Name                  string      `config:"name"`
+	Type                  string      `config:"type"`
+	Description           string      `config:"description"`
+	Format                string      `config:"format"`
+	ScalingFactor         int         `config:"scaling_factor"`
+	Fields                Fields      `config:"fields"`
+	MultiFields           Fields      `config:"multi_fields"`
+	ObjectType            string      `config:"object_type"`
+	ObjectTypeMappingType string      `config:"object_type_mapping_type"`
+	Enabled               *bool       `config:"enabled"`
+	Analyzer              string      `config:"analyzer"`
+	SearchAnalyzer        string      `config:"search_analyzer"`
+	Norms                 bool        `config:"norms"`
+	Dynamic               DynamicType `config:"dynamic"`
+	Index                 *bool       `config:"index"`
+	DocValues             *bool       `config:"doc_values"`
+	CopyTo                string      `config:"copy_to"`
 
 	// Kibana specific
 	Analyzed     *bool  `config:"analyzed"`

libbeat/template/processor.go

@@ -200,6 +200,13 @@ func (p *Processor) array(f *common.Field) common.MapStr {
 func (p *Processor) object(f *common.Field) common.MapStr {
 	dynProperties := getDefaultProperties(f)
 
+	matchType := func(onlyType string) string {
+		if f.ObjectTypeMappingType != "" {
+			return f.ObjectTypeMappingType
+		}
+		return onlyType
+	}
+
 	switch f.ObjectType {
 	case "text":
 		dynProperties["type"] = "text"
@@ -208,13 +215,13 @@ func (p *Processor) object(f *common.Field) common.MapStr {
 			dynProperties["type"] = "string"
 			dynProperties["index"] = "analyzed"
 		}
-		addDynamicTemplate(f, dynProperties, "string")
+		addDynamicTemplate(f, dynProperties, matchType("string"))
 	case "long":
 		dynProperties["type"] = f.ObjectType
-		addDynamicTemplate(f, dynProperties, "long")
+		addDynamicTemplate(f, dynProperties, matchType("long"))
 	case "keyword":
 		dynProperties["type"] = f.ObjectType
-		addDynamicTemplate(f, dynProperties, "string")
+		addDynamicTemplate(f, dynProperties, matchType("string"))
 	}
 
 	properties := getDefaultProperties(f)

libbeat/template/processor_test.go

@@ -205,6 +205,32 @@ func TestDynamicTemplate(t *testing.T) {
 				},
 			},
 		},
+		{
+			field: common.Field{
+				Type: "object", ObjectType: "long", ObjectTypeMappingType: "futuretype",
+				Path: "language", Name: "english",
+			},
+			expected: common.MapStr{
+				"language.english": common.MapStr{
+					"mapping":            common.MapStr{"type": "long"},
+					"match_mapping_type": "futuretype",
+					"path_match":         "language.english.*",
+				},
+			},
+		},
+		{
+			field: common.Field{
+				Type: "object", ObjectType: "long", ObjectTypeMappingType: "*",
+				Path: "language", Name: "english",
+			},
+			expected: common.MapStr{
+				"language.english": common.MapStr{
+					"mapping":            common.MapStr{"type": "long"},
+					"match_mapping_type": "*",
+					"path_match":         "language.english.*",
+				},
+			},
+		},
 		{
 			field: common.Field{
 				Type: "object", ObjectType: "long",