Add support for histograms to metrics intake (#5360)

* model/modeldecoder: add metric type and unit

* systemtest: test histogram metrics

* Update changelog

* systemtest: fix min docs expectation in test

Author: axw

beater/test_approved_es_documents/TestPublishIntegrationMetricsets.approved.json

@@ -298,6 +298,72 @@
                 "id": "axb123hg",
                 "name": "logged-in-user"
             }
+        },
+        {
+            "@timestamp": "2017-05-30T18:53:41.366Z",
+            "_doc_count": 6,
+            "_metric_descriptions": {
+                "latency_distribution": {
+                    "type": "histogram",
+                    "unit": "s"
+                }
+            },
+            "agent": {
+                "name": "elastic-node",
+                "version": "3.14.0"
+            },
+            "ecs": {
+                "version": "1.8.0"
+            },
+            "host": {
+                "ip": "127.0.0.1"
+            },
+            "labels": {
+                "tag1": "one",
+                "tag2": 2
+            },
+            "latency_distribution": {
+                "counts": [
+                    1,
+                    2,
+                    3
+                ],
+                "values": [
+                    1.1,
+                    2.2,
+                    3.3
+                ]
+            },
+            "metricset.name": "app",
+            "observer": {
+                "ephemeral_id": "00000000-0000-0000-0000-000000000000",
+                "hostname": "",
+                "id": "fbba762a-14dd-412c-b7e9-b79f903eb492",
+                "type": "test-apm-server",
+                "version": "1.2.3",
+                "version_major": 1
+            },
+            "process": {
+                "pid": 1234
+            },
+            "processor": {
+                "event": "metric",
+                "name": "metric"
+            },
+            "service": {
+                "language": {
+                    "name": "ecmascript"
+                },
+                "name": "1234_service-12a3",
+                "node": {
+                    "name": "node-1"
+                }
+            },
+            "user": {
+                "email": "user@mail.com",
+                "id": "axb123hg",
+                "name": "logged-in-user"
+            }
         }
     ]
 }

changelogs/head.asciidoc

@@ -26,6 +26,7 @@ https://github.com/elastic/apm-server/compare/7.13\...master[View commits]
 * Translate otel messaging.* semantic conventions to ECS {pull}5334[5334]
 * Add support for dynamic histogram metrics {pull}5239[5239]
 * Tail-sampling processor now resumes subscription from previous position after restart {pull}5350[5350]
+* Add support for histograms to metrics intake {pull}5360[5360]
 
 [float]
 ==== Deprecated

docs/spec/v2/metricset.json

@@ -13,13 +13,118 @@
             "object"
           ],
           "properties": {
+            "counts": {
+              "description": "Counts holds the bucket counts for histogram metrics.  These numbers must be positive or zero.  If Counts is specified, then Values is expected to be specified with the same number of elements, and with the same order.",
+              "type": [
+                "null",
+                "array"
+              ],
+              "items": {
+                "type": "integer",
+                "minimum": 0
+              },
+              "minItems": 0
+            },
+            "type": {
+              "description": "Type holds an optional metric type: gauge, counter, or histogram.  If Type is unknown, it will be ignored.",
+              "type": [
+                "null",
+                "string"
+              ]
+            },
+            "unit": {
+              "description": "Unit holds an optional unit for the metric.  - \"percent\" (value is in the range [0,1]) - \"byte\" - a time unit: \"nanos\", \"micros\", \"ms\", \"s\", \"m\", \"h\", \"d\"  If Unit is unknown, it will be ignored.",
+              "type": [
+                "null",
+                "string"
+              ]
+            },
             "value": {
               "description": "Value holds the value of a single metric sample.",
-              "type": "number"
+              "type": [
+                "null",
+                "number"
+              ]
+            },
+            "values": {
+              "description": "Values holds the bucket values for histogram metrics.  Values must be provided in ascending order; failure to do so will result in the metric being discarded.",
+              "type": [
+                "null",
+                "array"
+              ],
+              "items": {
+                "type": "number"
+              },
+              "minItems": 0
             }
           },
-          "required": [
-            "value"
+          "allOf": [
+            {
+              "if": {
+                "properties": {
+                  "counts": {
+                    "type": "array"
+                  }
+                },
+                "required": [
+                  "counts"
+                ]
+              },
+              "then": {
+                "properties": {
+                  "values": {
+                    "type": "array"
+                  }
+                },
+                "required": [
+                  "values"
+                ]
+              }
+            },
+            {
+              "if": {
+                "properties": {
+                  "values": {
+                    "type": "array"
+                  }
+                },
+                "required": [
+                  "values"
+                ]
+              },
+              "then": {
+                "properties": {
+                  "counts": {
+                    "type": "array"
+                  }
+                },
+                "required": [
+                  "counts"
+                ]
+              }
+            }
+          ],
+          "anyOf": [
+            {
+              "properties": {
+                "value": {
+                  "type": "number"
+                }
+              },
+              "required": [
+                "value"
+              ]
+            },
+            {
+              "properties": {
+                "values": {
+                  "type": "array"
+                }
+              },
+              "required": [
+                "values"
+              ]
+            }
           ]
         }
       }

model/modeldecoder/generator/code.go

@@ -161,28 +161,37 @@ func (val *%s) IsSet() bool {
 	if key != "" {
 		key += "."
 	}
-	prefix := ``
+	prefix := ` `
 	for i := 0; i < len(structTyp.fields); i++ {
 		f := structTyp.fields[i]
 		if !f.Exported() {
 			continue
 		}
-		switch t := f.Type().Underlying().(type) {
-		case *types.Slice, *types.Map:
-			fmt.Fprintf(&g.buf, `%s len(val.%s) > 0`, prefix, f.Name())
-		case *types.Struct:
-			fmt.Fprintf(&g.buf, `%s val.%s.IsSet()`, prefix, f.Name())
-		default:
-			return fmt.Errorf("unhandled type %T for IsSet() for '%s%s'", t, key, jsonName(f))
+		g.buf.WriteString(prefix)
+		if err := generateIsSet(&g.buf, f, "val."); err != nil {
+			return errors.Wrapf(err, "error generating IsSet() for '%s%s'", key, jsonName(f))
 		}
-		prefix = ` ||`
+		prefix = ` || `
 	}
 	fmt.Fprint(&g.buf, `
 }
 `)
 	return nil
 }
 
+func generateIsSet(w io.Writer, field structField, fieldSelectorPrefix string) error {
+	switch typ := field.Type().Underlying(); typ.(type) {
+	case *types.Slice, *types.Map:
+		fmt.Fprintf(w, "(len(%s%s) > 0)", fieldSelectorPrefix, field.Name())
+		return nil
+	case *types.Struct:
+		fmt.Fprintf(w, "%s%s.IsSet()", fieldSelectorPrefix, field.Name())
+		return nil
+	default:
+		return fmt.Errorf("unhandled type %T generating IsSet() for '%s'", typ, jsonName(field))
+	}
+}
+
 // generateReset creates `Reset` methods for struct fields setting them to
 // their zero values or calling their `Reset` methods
 // it only considers exported fields

model/modeldecoder/generator/jsonnumber.go

@@ -17,8 +17,24 @@
 
 package generator
 
+import "encoding/json"
+
 func generateJSONPropertyJSONNumber(info *fieldInfo, parent *property, child *property) error {
 	child.Type.add(TypeNameNumber)
 	parent.Properties[jsonSchemaName(info.field)] = child
 	return setPropertyRulesInteger(info, child)
 }
+
+func setPropertyRulesNumber(info *fieldInfo, p *property) error {
+	for tagName, tagValue := range info.tags {
+		switch tagName {
+		case tagMax:
+			p.Max = json.Number(tagValue)
+			delete(info.tags, tagName)
+		case tagMin:
+			p.Min = json.Number(tagValue)
+			delete(info.tags, tagName)
+		}
+	}
+	return nil
+}

model/modeldecoder/generator/jsonschema.go

@@ -215,6 +215,7 @@ var (
 		"float64":                  TypeNameNumber,
 		nullableTypeInt:            TypeNameInteger,
 		"int":                      TypeNameInteger,
+		"int64":                    TypeNameInteger,
 		nullableTypeTimeMicrosUnix: TypeNameInteger,
 		nullableTypeString:         TypeNameString,
 		"string":                   TypeNameString,

model/modeldecoder/generator/nstring.go

@@ -40,7 +40,7 @@ func generateNullableStringValidation(w io.Writer, fields []structField, f struc
 		case tagRequired:
 			ruleNullableRequired(w, f)
 		case tagRequiredIfAny:
-			if err = ruleRequiredIfAny(w, fields, f, rule.value); err != nil {
+			if err := ruleRequiredIfAny(w, fields, f, rule.value); err != nil {
 				return errors.Wrap(err, "nullableString")
 			}
 		default:

model/modeldecoder/generator/slice.go

@@ -18,6 +18,7 @@
 package generator
 
 import (
+	"encoding/json"
 	"fmt"
 	"go/types"
 	"io"
@@ -45,10 +46,14 @@ for _, elem := range val.%s{
 		switch rule.name {
 		case tagMinLength, tagMaxLength:
 			err = sliceRuleMinMaxLength(w, f, rule)
+		case tagMinVals:
+			err = sliceRuleMinVals(w, f, rule)
 		case tagRequired:
 			sliceRuleRequired(w, f, rule)
 		case tagRequiredAnyOf:
 			err = ruleRequiredOneOf(w, fields, rule.value)
+		case tagRequiredIfAny:
+			err = ruleRequiredIfAny(w, fields, f, rule.value)
 		default:
 			return errors.Wrap(errUnhandledTagRule(rule), "slice")
 		}
@@ -79,6 +84,17 @@ for _, elem := range val.%s{
 	return fmt.Errorf("unhandled tag rule max for type %s", f.Type().Underlying())
 }
 
+func sliceRuleMinVals(w io.Writer, f structField, rule validationRule) error {
+	fmt.Fprintf(w, `
+for _, elem := range val.%s{
+	if elem %s %s{
+		return fmt.Errorf("'%s': validation rule '%s(%s)' violated")
+	}
+}
+`[1:], f.Name(), ruleMinMaxOperator(rule.name), rule.value, jsonName(f), rule.name, rule.value)
+	return nil
+}
+
 func sliceRuleRequired(w io.Writer, f structField, rule validationRule) {
 	fmt.Fprintf(w, `
 if len(val.%s) == 0{
@@ -110,11 +126,19 @@ func generateJSONPropertySlice(info *fieldInfo, parent *property, child *propert
 	// NOTE(simi): set required=true to be aligned with previous JSON schema definitions
 	items := property{Type: &propertyType{names: []propertyTypeName{itemsType}, required: true}}
 	switch itemsType {
+	case TypeNameInteger:
+		setPropertyRulesInteger(info, &items)
+	case TypeNameNumber:
+		setPropertyRulesNumber(info, &items)
 	case TypeNameString:
 		setPropertyRulesString(info, &items)
 	default:
 		return fmt.Errorf("unhandled slice item type %s", itemsType)
 	}
+	if minVals, ok := info.tags[tagMinVals]; ok {
+		items.Min = json.Number(minVals)
+		delete(info.tags, tagMinVals)
+	}
 	child.Items = &items
 	return nil
 }