[azure] [app_insights] Group metrics by dimensions (segments) and timestamp

CHANGELOG.next.asciidoc

@@ -233,6 +233,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add GCP CloudSQL metadata {pull}33066[33066]
 - Add GCP Carbon Footprint metricbeat data {pull}34820[34820]
 - Add event loop utilization metric to Kibana module {pull}35020[35020]
+- Add metrics grouping by dimensions and time to Azure app insights {pull}36634[36634]
 
 
 *Osquerybeat*

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

@@ -7,7 +7,9 @@ package app_insights
 import (
 	"fmt"
 	"regexp"
+	"sort"
 	"strings"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/services/preview/appinsights/v1/insights"
 	"github.com/Azure/go-autorest/autorest/date"
@@ -116,6 +118,15 @@ func isSegment(metric string) bool {
 	return false
 }
 
+type metricTimeKey struct {
+	Start time.Time
+	End   time.Time
+}
+
+func newMetricTimeKey(start, end time.Time) metricTimeKey {
+	return metricTimeKey{Start: start, End: end}
+}
+
 func EventsMapping(metricValues insights.ListMetricsResultsItem, applicationId string, namespace string) []mb.Event {
 	var events []mb.Event
 	if metricValues.Value == nil {
@@ -124,12 +135,9 @@ func EventsMapping(metricValues insights.ListMetricsResultsItem, applicationId s
 	groupedAddProp := make(map[string][]MetricValue)
 	mValues := mapMetricValues(metricValues)
 
-	var segValues []MetricValue
 	for _, mv := range mValues {
 		if len(mv.Segments) == 0 {
 			groupedAddProp[mv.Interval] = append(groupedAddProp[mv.Interval], mv)
-		} else {
-			segValues = append(segValues, mv)
 		}
 	}
 
@@ -139,43 +147,156 @@ func EventsMapping(metricValues insights.ListMetricsResultsItem, applicationId s
 			events = append(events, event)
 		}
 	}
-	for _, val := range segValues {
-		for _, seg := range val.Segments {
-			lastSeg := getValue(seg)
-			for _, ls := range lastSeg {
-				events = append(events, createSegEvent(val, ls, applicationId, namespace))
-			}
+
+	groupedByDimensions := groupMetricsByDimension(mValues)
+
+	for _, group := range groupedByDimensions {
+		groupedByTime := groupMetricsByTime(group)
+
+		for ts, group := range groupedByTime {
+			events = append(
+				events,
+				createGroupEvent(group, ts, applicationId, namespace),
+			)
 		}
 	}
 	return events
 }
 
-func getValue(metric MetricValue) []MetricValue {
-	var values []MetricValue
-	if metric.Segments == nil {
-		return []MetricValue{metric}
+// groupMetricsByTime groups metrics by their start and end times truncated to the second.
+func groupMetricsByTime(metrics []MetricValue) map[metricTimeKey][]MetricValue {
+	result := make(map[metricTimeKey][]MetricValue, len(metrics)/2)
+
+	for _, metric := range metrics {
+		// The start and end times are truncated to the nearest second.
+		// This is done to ensure that metrics that fall within the same
+		// second are grouped together, even if their actual time are
+		// slightly different.
+		timeKey := newMetricTimeKey(
+			metric.Start.Time.Truncate(time.Second),
+			metric.End.Time.Truncate(time.Second),
+		)
+		result[timeKey] = append(result[timeKey], metric)
 	}
-	for _, met := range metric.Segments {
-		values = append(values, getValue(met)...)
+
+	return result
+}
+
+// groupMetricsByDimension groups the given metrics by their dimension keys.
+func groupMetricsByDimension(metrics []MetricValue) map[string][]MetricValue {
+	var (
+		keys                 = make(map[string][]MetricValue)
+		firstStart, firstEnd *date.Time
+		helper               func(metrics []MetricValue)
+	)
+
+	helper = func(metrics []MetricValue) {
+		for _, metric := range metrics {
+			dimensionKey := getSortedKeys(metric.SegmentName)
+
+			if metric.Start != nil && !metric.Start.IsZero() {
+				firstStart = metric.Start
+			}
+
+			if metric.End != nil && !metric.End.IsZero() {
+				firstEnd = metric.End
+			}
+
+			if len(metric.Segments) > 0 {
+				for _, segment := range metric.Segments {
+					segmentKey := getSortedKeys(segment.SegmentName)
+					if segmentKey != "" {
+						combinedKey := dimensionKey + segmentKey
+
+						newMetric := MetricValue{
+							SegmentName: segment.SegmentName,
+							Value:       segment.Value,
+							Segments:    segment.Segments,
+							Interval:    segment.Interval,
+							Start:       firstStart,
+							End:         firstEnd,
+						}
+
+						keys[combinedKey] = append(keys[combinedKey], newMetric)
+					}
+				}
+
+				for _, segment := range metric.Segments {
+					helper(segment.Segments)
+				}
+			} else if dimensionKey != "" {
+				m := metric
+				m.Start, m.End = firstStart, firstEnd
+				keys[dimensionKey] = append(keys[dimensionKey], m)
+			}
+		}
 	}
-	return values
+
+	helper(metrics)
+
+	return keys
 }
 
-func createSegEvent(parentMetricValue MetricValue, metricValue MetricValue, applicationId string, namespace string) mb.Event {
+// getSortedKeys returns a string of sorted keys.
+// The keys are sorted in alphabetical order.
+func getSortedKeys(m map[string]string) string {
+	keys := make([]string, 0, len(m))
+	for k, v := range m {
+		keys = append(keys, k+v)
+	}
+	sort.Strings(keys)
+	return strings.Join(keys, "")
+}
+
+func createGroupEvent(metricValue []MetricValue, metricTime metricTimeKey, applicationId, namespace string) mb.Event {
+	if metricTime.Start.IsZero() || metricTime.End.IsZero() {
+		return mb.Event{}
+	}
+
 	metricList := mapstr.M{}
-	for key, metric := range metricValue.Value {
-		metricList.Put(key, metric)
+
+	for _, v := range metricValue {
+		for key, metric := range v.Value {
+			_, _ = metricList.Put(key, metric)
+		}
 	}
+
 	if len(metricList) == 0 {
 		return mb.Event{}
 	}
-	event := createEvent(parentMetricValue.Start, parentMetricValue.End, applicationId, namespace, metricList)
-	if len(parentMetricValue.SegmentName) > 0 {
-		event.ModuleFields.Put("dimensions", parentMetricValue.SegmentName)
+
+	event := mb.Event{
+		ModuleFields: mapstr.M{"application_id": applicationId},
+		MetricSetFields: mapstr.M{
+			"start_date": metricTime.Start,
+			"end_date":   metricTime.End,
+		},
+		Timestamp: metricTime.End,
 	}
-	if len(metricValue.SegmentName) > 0 {
-		event.ModuleFields.Put("dimensions", metricValue.SegmentName)
+
+	event.RootFields = mapstr.M{}
+	_, _ = event.RootFields.Put("cloud.provider", "azure")
+
+	segments := make(map[string]string)
+
+	for _, v := range metricValue {
+		for sn, sv := range v.SegmentName {
+			segments[sn] = sv
+		}
 	}
+
+	if len(segments) > 0 {
+		_, _ = event.ModuleFields.Put("dimensions", segments)
+	}
+
+	if namespace == "" {
+		_, _ = event.ModuleFields.Put("metrics", metricList)
+	} else {
+		for key, metric := range metricList {
+			_, _ = event.MetricSetFields.Put(key, metric)
+		}
+	}
+
 	return event
 }
 
@@ -219,9 +340,9 @@ func createNoSegEvent(values []MetricValue, applicationId string, namespace stri
 func getAdditionalPropMetric(addProp map[string]interface{}) map[string]interface{} {
 	metricNames := make(map[string]interface{})
 	for key, val := range addProp {
-		switch val.(type) {
+		switch v := val.(type) {
 		case map[string]interface{}:
-			for subKey, subVal := range val.(map[string]interface{}) {
+			for subKey, subVal := range v {
 				if subVal != nil {
 					metricNames[cleanMetricNames(fmt.Sprintf("%s.%s", key, subKey))] = subVal
 				}