enable individual labels from aggregate (#269)

* enable individual labels from aggregate

* added to unit test

pkg/pipeline/extract/aggregate/aggregate.go

@@ -53,6 +53,7 @@ type Aggregate struct {
 
 type GroupState struct {
 	normalizedValues NormalizedValues
+	labels           Labels
 	recentRawValues  []float64
 	recentOpValue    float64
 	recentCount      int
@@ -95,14 +96,14 @@ func (labels Labels) getNormalizedValues() NormalizedValues {
 	return NormalizedValues(normalizedAsString)
 }
 
-func (aggregate Aggregate) FilterEntry(entry config.GenericMap) (error, NormalizedValues) {
+func (aggregate Aggregate) FilterEntry(entry config.GenericMap) (error, NormalizedValues, Labels) {
 	labels, allLabelsFound := aggregate.LabelsFromEntry(entry)
 	if !allLabelsFound {
-		return fmt.Errorf("missing keys in entry"), ""
+		return fmt.Errorf("missing keys in entry"), "", nil
 	}
 
 	normalizedValues := labels.getNormalizedValues()
-	return nil, normalizedValues
+	return nil, normalizedValues, labels
 }
 
 func getInitValue(operation string) float64 {
@@ -120,15 +121,15 @@ func getInitValue(operation string) float64 {
 	}
 }
 
-func (aggregate Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValues NormalizedValues) error {
+func (aggregate Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValues NormalizedValues, labels Labels) error {
 
 	aggregate.mutex.Lock()
 	defer aggregate.mutex.Unlock()
 
 	var groupState *GroupState
 	oldEntry, ok := aggregate.cache.GetCacheEntry(string(normalizedValues))
 	if !ok {
-		groupState = &GroupState{normalizedValues: normalizedValues}
+		groupState = &GroupState{normalizedValues: normalizedValues, labels: labels}
 		initVal := getInitValue(string(aggregate.Definition.Operation))
 		groupState.totalValue = initVal
 		groupState.recentOpValue = initVal
@@ -183,13 +184,13 @@ func (aggregate Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValu
 func (aggregate Aggregate) Evaluate(entries []config.GenericMap) error {
 	for _, entry := range entries {
 		// filter entries matching labels with aggregates
-		err, normalizedValues := aggregate.FilterEntry(entry)
+		err, normalizedValues, labels := aggregate.FilterEntry(entry)
 		if err != nil {
 			continue
 		}
 
 		// update aggregate group by entry
-		err = aggregate.UpdateByEntry(entry, normalizedValues)
+		err = aggregate.UpdateByEntry(entry, normalizedValues, labels)
 		if err != nil {
 			log.Debugf("UpdateByEntry error %v", err)
 			continue
@@ -208,7 +209,7 @@ func (aggregate Aggregate) GetMetrics() []config.GenericMap {
 	// iterate over the items in the cache
 	aggregate.cache.Iterate(func(key string, value interface{}) {
 		group := value.(*GroupState)
-		metrics = append(metrics, config.GenericMap{
+		newEntry := config.GenericMap{
 			"name":              aggregate.Definition.Name,
 			"operation":         aggregate.Definition.Operation,
 			"record_key":        aggregate.Definition.RecordKey,
@@ -220,7 +221,12 @@ func (aggregate Aggregate) GetMetrics() []config.GenericMap {
 			"recent_op_value":   group.recentOpValue,
 			"recent_count":      group.recentCount,
 			strings.Join(aggregate.Definition.By, "_"): string(group.normalizedValues),
-		})
+		}
+		// add the items in aggregate.Definition.By individually to the entry
+		for _, key := range aggregate.Definition.By {
+			newEntry[key] = group.labels[key]
+		}
+		metrics = append(metrics, newEntry)
 		// Once reported, we reset the recentXXX fields
 		if aggregate.Definition.Operation == OperationRawValues {
 			group.recentRawValues = make([]float64, 0)

pkg/pipeline/extract/aggregate/aggregate_test.go

@@ -96,12 +96,17 @@ func Test_FilterEntry(t *testing.T) {
 	aggregate := GetMockAggregate()
 	entry := test.GetIngestMockEntry(false)
 
-	err, _ := aggregate.FilterEntry(entry)
+	err, _, _ := aggregate.FilterEntry(entry)
+	require.Equal(t, err, nil)
 
+	err, normalizedLabels, labels := aggregate.FilterEntry(entry)
 	require.Equal(t, err, nil)
+	require.Equal(t, Labels{"srcIP": "10.0.0.1", "dstIP": "20.0.0.2"}, labels)
+	require.Equal(t, NormalizedValues("20.0.0.2,10.0.0.1"), normalizedLabels)
+
 	entry = test.GetIngestMockEntry(true)
 
-	err, _ = aggregate.FilterEntry(entry)
+	err, _, _ = aggregate.FilterEntry(entry)
 
 	require.EqualError(t, err, "missing keys in entry")
 }