Cherry-pick #8504 to 6.x: Add kafka fields needed for dashboards

CHANGELOG.asciidoc

@@ -135,6 +135,7 @@ https://github.com/elastic/beats/compare/v6.4.0...6.x[Check the HEAD diff]
 - Support for Kafka 2.0.0 {pull}8399[8399]
 - Added support for query params in configuration {issue}8286[8286] {pull}8292[8292]
 - Add container image for docker metricsets. {issue}8214[8214] {pull}8438[8438]
+- Precalculate composed id fields for kafka dashboards. {pull}8504[8504]
 - Add support for `full` status page output for php-fpm module as a separate metricset called `process`. {pull}8394[8394]
 
 *Packetbeat*

metricbeat/docs/fields.asciidoc

@@ -7108,6 +7108,24 @@ type: long
 Partition id.
 
 
+--
+
+*`kafka.partition.topic_id`*::
++
+--
+type: keyword
+
+Unique id of the partition in the topic.
+
+--
+
+*`kafka.partition.topic_broker_id`*::
++
+--
+type: keyword
+
+Unique id of the partition in the topic and the broker.
+
 --
 
 [float]

metricbeat/module/kafka/_meta/fields.yml

@@ -39,3 +39,13 @@
           type: long
           description: >
             Partition id.
+
+        - name: partition.topic_id
+          type: keyword
+          description:
+            Unique id of the partition in the topic.
+
+        - name: partition.topic_broker_id
+          type: keyword
+          description:
+            Unique id of the partition in the topic and the broker.

metricbeat/module/kafka/consumergroup/_meta/data.json

@@ -17,7 +17,7 @@
             "client": {
                 "host": "172.18.0.1",
                 "id": "sarama",
-                "member_id": "sarama-714cfb8b-39e5-4128-9109-e5056c5e8f56"
+                "member_id": "sarama-fcb5a5db-0474-4f3a-a5af-29e2f14549c5"
             },
             "error": {
                 "code": 0
@@ -29,7 +29,8 @@
             "topic": "metricbeat-test"
         },
         "partition": {
-            "id": 0
+            "id": 0,
+            "topic_id": "0-metricbeat-test"
         },
         "topic": {
             "name": "metricbeat-test"
@@ -41,4 +42,4 @@
         "name": "consumergroup",
         "rtt": 115
     }
-}
+}

metricbeat/module/kafka/consumergroup/consumergroup.go

@@ -19,6 +19,7 @@ package consumergroup
 
 import (
 	"crypto/tls"
+	"fmt"
 
 	"github.com/pkg/errors"
 
@@ -111,6 +112,9 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 	}
 
 	emitEvent := func(event common.MapStr) {
+		// Helpful IDs to avoid scripts on queries
+		partitionTopicID := fmt.Sprintf("%d-%s", event["partition"], event["topic"])
+
 		// TODO (deprecation): Remove fields from MetricSetFields moved to ModuleFields
 		event["broker"] = brokerInfo
 		r.Event(mb.Event{
@@ -120,7 +124,8 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 					"name": event["topic"],
 				},
 				"partition": common.MapStr{
-					"id": event["partition"],
+					"id":       event["partition"],
+					"topic_id": partitionTopicID,
 				},
 			},
 			MetricSetFields: event,

metricbeat/module/kafka/partition/_meta/data.json

@@ -14,6 +14,7 @@
                 "address": "172.18.0.2:9092",
                 "id": 0
             },
+            "id": 0,
             "offset": {
                 "newest": 0,
                 "oldest": 0
@@ -26,11 +27,13 @@
                 "replica": 0
             },
             "topic": {
-                "name": "foo-1532945633-185372965"
-            }
+                "name": "foo-1538389014-739473801"
+            },
+            "topic_broker_id": "0-foo-1538389014-739473801-0",
+            "topic_id": "0-foo-1538389014-739473801"
         },
         "topic": {
-            "name": "foo-1532945633-185372965"
+            "name": "foo-1538389014-739473801"
         }
     },
     "metricset": {

metricbeat/module/kafka/partition/partition.go

@@ -20,6 +20,7 @@ package partition
 import (
 	"crypto/tls"
 	"errors"
+	"fmt"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
@@ -167,8 +168,17 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 					}
 				}
 
+				// Helpful IDs to avoid scripts on queries
+				partitionTopicID := fmt.Sprintf("%d-%s", partition.ID, topic.Name)
+				partitionTopicBrokerID := fmt.Sprintf("%s-%d", partitionTopicID, id)
+
 				// create event
 				event := common.MapStr{
+					// Common `kafka.partition` fields
+					"id":              partition.ID,
+					"topic_id":        partitionTopicID,
+					"topic_broker_id": partitionTopicBrokerID,
+
 					"topic":     evtTopic,
 					"broker":    evtBroker,
 					"partition": partitionEvent,
@@ -183,9 +193,6 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 					ModuleFields: common.MapStr{
 						"broker": evtBroker,
 						"topic":  evtTopic,
-						"partition": common.MapStr{
-							"id": partition.ID,
-						},
 					},
 					MetricSetFields: event,
 				})