Group same timestamp metrics values in app_insights metricset

CHANGELOG.next.asciidoc

@@ -302,6 +302,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix k8s scheduler compatibility issue. {pull}19699[19699]
 - Fix SQL module mapping NULL values as string {pull}18955[18955] {issue}18898[18898
 - Modify doc for app_insights metricset to contain example of config. {pull}20185[20185]
+- Groups same timestamp metric values to one event in the app_insights metricset. {pull}20403[20403]
 
 *Packetbeat*
 

x-pack/metricbeat/module/azure/app_insights/data.go

@@ -8,6 +8,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/Azure/go-autorest/autorest/date"
+
 	"github.com/Azure/azure-sdk-for-go/services/preview/appinsights/v1/insights"
 
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -19,45 +21,80 @@ func EventsMapping(metricValues insights.ListMetricsResultsItem, applicationId s
 	if metricValues.Value == nil {
 		return events
 	}
+	groupedAddProp := make(map[string][]insights.MetricsResultInfo)
 	for _, item := range *metricValues.Value {
 		if item.Body != nil && item.Body.Value != nil {
 			if item.Body.Value.AdditionalProperties != nil {
-				events = append(events, createEvent(*item.Body.Value, insights.MetricsSegmentInfo{}, applicationId))
+				groupedAddProp[fmt.Sprintf("%sTO%s", item.Body.Value.Start, item.Body.Value.End)] =
+					append(groupedAddProp[fmt.Sprintf("%sTO%s", item.Body.Value.Start, item.Body.Value.End)], *item.Body.Value)
 			} else if item.Body.Value.Segments != nil {
 				for _, segment := range *item.Body.Value.Segments {
-					events = append(events, createEvent(*item.Body.Value, segment, applicationId))
+					event, ok := createSegmentEvent(*item.Body.Value.Start, *item.Body.Value.End, segment, applicationId)
+					if ok {
+						events = append(events, event)
+					}
 				}
 			}
 		}
 	}
+	if len(groupedAddProp) > 0 {
+		for _, val := range groupedAddProp {
+			event, ok := createEvent(val, applicationId)
+			if ok {
+				events = append(events, event)
+			}
+		}
+	}
 	return events
 }
 
-func createEvent(value insights.MetricsResultInfo, segment insights.MetricsSegmentInfo, applicationId string) mb.Event {
+func createSegmentEvent(start date.Time, end date.Time, segment insights.MetricsSegmentInfo, applicationId string) (mb.Event, bool) {
 	metricList := common.MapStr{}
-	if value.AdditionalProperties != nil {
+	metrics := getMetric(segment.AdditionalProperties)
+	if len(metrics) == 0 {
+		return mb.Event{}, false
+	}
+	for key, metric := range metrics {
+		metricList.Put(key, metric)
+	}
+	event := mb.Event{
+		MetricSetFields: common.MapStr{
+			"start_date":     start,
+			"end_date":       end,
+			"application_id": applicationId,
+		},
+		Timestamp: end.Time,
+	}
+	event.RootFields = common.MapStr{}
+	event.RootFields.Put("cloud.provider", "azure")
+	event.MetricSetFields.Put("metrics", metricList)
+	return event, true
+}
+
+func createEvent(values []insights.MetricsResultInfo, applicationId string) (mb.Event, bool) {
+	metricList := common.MapStr{}
+	for _, value := range values {
 		metrics := getMetric(value.AdditionalProperties)
 		for key, metric := range metrics {
 			metricList.Put(key, metric)
 		}
-	} else {
-		metrics := getMetric(segment.AdditionalProperties)
-		for key, metric := range metrics {
-			metricList.Put(key, metric)
-		}
 	}
+	if len(metricList) == 0 {
+		return mb.Event{}, false
+	}
+
 	event := mb.Event{
 		MetricSetFields: common.MapStr{
-			"start_date":     value.Start,
-			"end_date":       value.End,
+			"start_date":     values[0].Start,
+			"end_date":       values[0].End,
 			"application_id": applicationId,
 		},
-		Timestamp: value.End.Time,
+		Timestamp: values[0].End.Time,
 	}
 	event.RootFields = common.MapStr{}
 	event.RootFields.Put("cloud.provider", "azure")
 	event.MetricSetFields.Put("metrics", metricList)
-	return event
+	return event, true
 }
 
 func getMetric(addProp map[string]interface{}) map[string]interface{} {
@@ -66,7 +103,9 @@ func getMetric(addProp map[string]interface{}) map[string]interface{} {
 		switch val.(type) {
 		case map[string]interface{}:
 			for subKey, subVal := range val.(map[string]interface{}) {
-				metricNames[cleanMetricNames(fmt.Sprintf("%s.%s", key, subKey))] = subVal
+				if subVal != nil {
+					metricNames[cleanMetricNames(fmt.Sprintf("%s.%s", key, subKey))] = subVal
+				}
 			}
 		default:
 			metricNames[cleanMetricNames(key)] = val