Add support for dynamic histogram metrics (#5239)

* model/metricset: add "histogram" dynamic_template

* ingest/pipeline: handle metric descriptions

* model: add metric type and unit

* processor/otel: translate histogram metrics

* Update changelog

* processor/otel: fix histogram translation

Handle histogram bucket counts correctly, following the
approach taken by Prometheus.

* processor/otel: fix merge

* processor/otel: use appropriate slice cap

apmpackage/apm/0.2.0/data_stream/app_metrics/elasticsearch/ingest_pipeline/apm_metrics_dynamic_template.json

@@ -0,0 +1,11 @@
+{
+  "description": "Set dynamic_templates for application metrics",
+  "processors": [
+    {
+      "script": {
+        "if": "ctx._metric_descriptions != null",
+        "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+      }
+    }
+  ]
+}

apmpackage/apm/0.2.0/data_stream/app_metrics/elasticsearch/ingest_pipeline/default.json

@@ -26,6 +26,12 @@
         "name": "metrics-apm.app-0.2.0-apm_error_grouping_name",
         "if": "ctx.processor?.event == 'error'"
       }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.app-0.2.0-apm_metrics_dynamic_template",
+        "if": "ctx.processor?.event == 'metric'"
+      }
     }
   ]
 }

apmpackage/apm/0.2.0/data_stream/app_metrics/fields/fields.yml

@@ -8,6 +8,8 @@
   type: object
   description: Additional experimental data sent by the agents.
   dynamic: true
+- name: histogram
+  type: histogram
 - name: kubernetes.namespace
   type: keyword
   description: |

apmpackage/apm/0.2.0/data_stream/error_logs/elasticsearch/ingest_pipeline/apm_metrics_dynamic_template.json

@@ -0,0 +1,11 @@
+{
+  "description": "Set dynamic_templates for application metrics",
+  "processors": [
+    {
+      "script": {
+        "if": "ctx._metric_descriptions != null",
+        "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+      }
+    }
+  ]
+}

apmpackage/apm/0.2.0/data_stream/error_logs/elasticsearch/ingest_pipeline/default.json

@@ -26,6 +26,12 @@
         "name": "logs-apm.error-0.2.0-apm_error_grouping_name",
         "if": "ctx.processor?.event == 'error'"
       }
+    },
+    {
+      "pipeline": {
+        "name": "logs-apm.error-0.2.0-apm_metrics_dynamic_template",
+        "if": "ctx.processor?.event == 'metric'"
+      }
     }
   ]
 }

apmpackage/apm/0.2.0/data_stream/internal_metrics/elasticsearch/ingest_pipeline/apm_metrics_dynamic_template.json

@@ -0,0 +1,11 @@
+{
+  "description": "Set dynamic_templates for application metrics",
+  "processors": [
+    {
+      "script": {
+        "if": "ctx._metric_descriptions != null",
+        "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+      }
+    }
+  ]
+}

apmpackage/apm/0.2.0/data_stream/internal_metrics/elasticsearch/ingest_pipeline/default.json

@@ -26,6 +26,12 @@
         "name": "metrics-apm.internal-0.2.0-apm_error_grouping_name",
         "if": "ctx.processor?.event == 'error'"
       }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.internal-0.2.0-apm_metrics_dynamic_template",
+        "if": "ctx.processor?.event == 'metric'"
+      }
     }
   ]
 }

apmpackage/apm/0.2.0/data_stream/internal_metrics/fields/fields.yml

@@ -8,6 +8,8 @@
   type: object
   description: Additional experimental data sent by the agents.
   dynamic: true
+- name: histogram
+  type: histogram
 - name: kubernetes.namespace
   type: keyword
   description: |

apmpackage/apm/0.2.0/data_stream/profile_metrics/elasticsearch/ingest_pipeline/apm_metrics_dynamic_template.json

@@ -0,0 +1,11 @@
+{
+  "description": "Set dynamic_templates for application metrics",
+  "processors": [
+    {
+      "script": {
+        "if": "ctx._metric_descriptions != null",
+        "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+      }
+    }
+  ]
+}

apmpackage/apm/0.2.0/data_stream/profile_metrics/elasticsearch/ingest_pipeline/default.json

@@ -26,6 +26,12 @@
         "name": "metrics-apm.profiling-0.2.0-apm_error_grouping_name",
         "if": "ctx.processor?.event == 'error'"
       }
+    },
+    {
+      "pipeline": {
+        "name": "metrics-apm.profiling-0.2.0-apm_metrics_dynamic_template",
+        "if": "ctx.processor?.event == 'metric'"
+      }
     }
   ]
 }

apmpackage/apm/0.2.0/data_stream/traces/elasticsearch/ingest_pipeline/apm_metrics_dynamic_template.json

@@ -0,0 +1,11 @@
+{
+  "description": "Set dynamic_templates for application metrics",
+  "processors": [
+    {
+      "script": {
+        "if": "ctx._metric_descriptions != null",
+        "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+      }
+    }
+  ]
+}

apmpackage/apm/0.2.0/data_stream/traces/elasticsearch/ingest_pipeline/default.json

@@ -26,6 +26,12 @@
         "name": "traces-apm-0.2.0-apm_error_grouping_name",
         "if": "ctx.processor?.event == 'error'"
       }
+    },
+    {
+      "pipeline": {
+        "name": "traces-apm-0.2.0-apm_metrics_dynamic_template",
+        "if": "ctx.processor?.event == 'metric'"
+      }
     }
   ]
 }

apmpackage/apm/0.2.0/docs/README.md

@@ -433,6 +433,7 @@ Metrics are written to `metrics-apm.app.*`, `metrics-apm.internal.*`, and `metri
 |destination.ip|IP addess of the destination. Can be one of multiple IPv4 or IPv6 addresses.|ip|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-yes.png)  |
 |destination.port|Port of the destination.|long|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-yes.png)  |
 |experimental|Additional experimental data sent by the agents.|object|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-no.png)  |
+|histogram||histogram|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-no.png)  |
 |host.architecture|The architecture of the host the event was recorded on.|keyword|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-yes.png)  |
 |host.hostname|The hostname of the host the event was recorded on.|keyword|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-yes.png)  |
 |host.ip|IP of the host that records the event.|ip|  ![](https://doc-icons.s3.us-east-2.amazonaws.com/icon-yes.png)  |

changelogs/head.asciidoc

@@ -23,6 +23,7 @@ https://github.com/elastic/apm-server/compare/7.13\...master[View commits]
 * Support setting agent configuration from apm-server.yml {pull}5177[5177]
 * Add metric_type and unit to field metadata of system metrics {pull}5230[5230]
 * Upgrade Go to 1.15.12 {pull}[]
+* Add support for dynamic histogram metrics {pull}5239[5239]
 
 [float]
 ==== Deprecated

docs/fields.asciidoc

@@ -12,6 +12,7 @@ This file is generated! See _meta/fields.yml and scripts/generate_fields_docs.py
 This document describes the fields that are exported by Apm-Server. They are
 grouped in the following categories:
 
+* <<exported-fields-apm-application-metrics>>
 * <<exported-fields-apm-error>>
 * <<exported-fields-apm-profile>>
 * <<exported-fields-apm-sourcemap>>
@@ -30,6 +31,19 @@ grouped in the following categories:
 * <<exported-fields-system>>
 
 --
+[[exported-fields-apm-application-metrics]]
+== APM Application Metrics fields
+
+APM application metrics.
+
+
+*`histogram`*::
++
+--
+type: histogram
+
+--
+
 [[exported-fields-apm-error]]
 == APM Error fields
 

ingest/pipeline/definition.json

@@ -29,6 +29,12 @@
             "name": "apm_error_grouping_name",
             "if": "ctx.processor?.event == 'error'"
           }
+        },
+        {
+          "pipeline": {
+            "name": "apm_metrics_dynamic_template",
+            "if": "ctx.processor?.event == 'metric'"
+          }
         }
       ]
     }
@@ -137,5 +143,19 @@
         }
       ]
     }
+  },
+  {
+    "id": "apm_metrics_dynamic_template",
+    "body": {
+      "description": "Set dynamic_templates for application metrics",
+      "processors": [
+        {
+          "script": {
+            "if": "ctx._metric_descriptions != null",
+            "source": "Map dynamic_templates = new HashMap();\nfor (entry in ctx._metric_descriptions.entrySet()) {\n  String name = entry.getKey();\n  Map description = entry.getValue();\n  String metric_type = description.type;\n  if (metric_type == \"histogram\") {\n    dynamic_templates[name] = \"histogram\";\n  }\n}\nctx._dynamic_templates = dynamic_templates;\nctx.remove(\"_metric_descriptions\");\n"
+          }
+        }
+      ]
+    }
   }
 ]

ingest/pipeline/definition.yml

@@ -15,6 +15,9 @@ apm:
   - pipeline:
       name: apm_error_grouping_name
       if: ctx.processor?.event == 'error'
+  - pipeline:
+      name: apm_metrics_dynamic_template
+      if: ctx.processor?.event == 'metric'
 
 apm_user_agent:
   description: Add user agent information for APM events
@@ -77,3 +80,23 @@ apm_error_grouping_name:
       field: error.grouping_name
       copy_from: error.log.message
       if: ctx.error?.log?.message != null
+
+# TODO(axw) handle unit in metric descriptions.
+# See https://github.com/elastic/elasticsearch/issues/72536
+apm_metrics_dynamic_template:
+  description: Set dynamic_templates for application metrics
+  processors:
+  - script:
+      if: ctx._metric_descriptions != null
+      source: |
+        Map dynamic_templates = new HashMap();
+        for (entry in ctx._metric_descriptions.entrySet()) {
+          String name = entry.getKey();
+          Map description = entry.getValue();
+          String metric_type = description.type;
+          if (metric_type == "histogram") {
+            dynamic_templates[name] = "histogram";
+          }
+        }
+        ctx._dynamic_templates = dynamic_templates;
+        ctx.remove("_metric_descriptions");

model/metricset.go

@@ -47,6 +47,16 @@ var (
 	metricsetProcessorEntry  = common.MapStr{"name": metricsetProcessorName, "event": metricsetDocType}
 )
 
+// MetricType describes the type of a metric: gauge, counter, or histogram.
+type MetricType string
+
+// Valid MetricType values.
+const (
+	MetricTypeGauge     MetricType = "gauge"
+	MetricTypeCounter   MetricType = "counter"
+	MetricTypeHistogram MetricType = "histogram"
+)
+
 // Metricset describes a set of metrics and associated metadata.
 type Metricset struct {
 	// Timestamp holds the time at which the metrics were published.
@@ -96,6 +106,20 @@ type Sample struct {
 	// Name holds the metric name.
 	Name string
 
+	// Type holds an optional metric type.
+	//
+	// If Type is unspecified or invalid, it will be ignored.
+	Type MetricType
+
+	// Unit holds an optional unit:
+	//
+	// - "percent" (value is in the range [0,1])
+	// - "byte"
+	// - a time unit: "nanos", "micros", "ms", "s", "m", "h", "d"
+	//
+	// If Unit is unspecified or invalid, it will be ignored.
+	Unit string
+
 	// Value holds the metric value for single-value metrics.
 	//
 	// If Counts and Values are specified, then Value will be ignored.
@@ -204,6 +228,16 @@ func (me *Metricset) appendBeatEvents(cfg *transform.Config, events []beat.Event
 
 	fields["processor"] = metricsetProcessorEntry
 
+	// Set a _metric_descriptions field, which holds optional metric types and units.
+	var metricDescriptions mapStr
+	for _, sample := range me.Samples {
+		var m mapStr
+		m.maybeSetString("type", string(sample.Type))
+		m.maybeSetString("unit", sample.Unit)
+		metricDescriptions.maybeSetMapStr(sample.Name, common.MapStr(m))
+	}
+	fields.maybeSetMapStr("_metric_descriptions", common.MapStr(metricDescriptions))
+
 	if cfg.DataStreams {
 		dataset := AppMetricsDataset
 		// Metrics are stored in "metrics" data streams.

model/metricset/_meta/fields.yml

@@ -1,3 +1,12 @@
+- key: apm-application-metrics
+  title: "APM Application Metrics"
+  description: APM application metrics.
+  short_config: true
+  fields:
+    - name: histogram
+      type: histogram
+      dynamic_template: true
+
 - key: apm-transaction-metrics
   title: "APM Transaction Metrics"
   description: >

model/metricset_test.go

@@ -226,6 +226,54 @@ func TestTransform(t *testing.T) {
 			},
 			Msg: "Payload with destination service.",
 		},
+		{
+			Metricset: &Metricset{
+				Timestamp: timestamp,
+				Metadata:  metadata,
+				Samples: []Sample{{
+					Name:   "latency_histogram",
+					Type:   "histogram",
+					Unit:   "s",
+					Counts: []int64{1, 2, 3},
+					Values: []float64{1.1, 2.2, 3.3},
+				}, {
+					Name:  "just_type",
+					Type:  "counter",
+					Value: 123,
+				}, {
+					Name:  "just_unit",
+					Unit:  "percent",
+					Value: 0.99,
+				}},
+			},
+			Output: []common.MapStr{
+				{
+					"data_stream.type":    "metrics",
+					"data_stream.dataset": "apm.app.myservice",
+					"processor":           common.MapStr{"event": "metric", "name": "metric"},
+					"service":             common.MapStr{"name": "myservice"},
+					"latency_histogram": common.MapStr{
+						"counts": []int64{1, 2, 3},
+						"values": []float64{1.1, 2.2, 3.3},
+					},
+					"just_type": 123.0,
+					"just_unit": 0.99,
+					"_metric_descriptions": common.MapStr{
+						"latency_histogram": common.MapStr{
+							"type": "histogram",
+							"unit": "s",
+						},
+						"just_type": common.MapStr{
+							"type": "counter",
+						},
+						"just_unit": common.MapStr{
+							"unit": "percent",
+						},
+					},
+				},
+			},
+			Msg: "Payload with metric type and unit.",
+		},
 	}
 
 	for idx, test := range tests {