keeping timeseries and points in order after aggregation in metrics t…

…ransform processor (#663) * timeseries and points in order after aggregation * use simple sort * remove sliceStable * add testcase for aggregate across labels output ordering
open-telemetry · Aug 7, 2020 · e4ff1b8 · e4ff1b8
1 parent a9087fc
commit e4ff1b8
Show file tree

Hide file tree

Showing 5 changed files with 26 additions and 7 deletions.
diff --git a/processor/metricstransformprocessor/datapoint_aggregation.go b/processor/metricstransformprocessor/datapoint_aggregation.go
@@ -17,6 +17,7 @@ package metricstransformprocessor
 import (
 	"math"
 	"math/rand"
+	"sort"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 	"github.com/golang/protobuf/ptypes/timestamp"
@@ -101,6 +102,9 @@ func (mtp *metricsTransformProcessor) mergePoints(timestampToPoints map[int64][]
 
 		}
 	}
+	sort.Slice(newPoints, func(i, j int) bool {
+		return mtp.compareTimestamps(newPoints[i].Timestamp, newPoints[j].Timestamp)
+	})
 	return newPoints
 }
 

diff --git a/processor/metricstransformprocessor/metrics_transform_processor.go b/processor/metricstransformprocessor/metrics_transform_processor.go
@@ -18,6 +18,7 @@ import (
 	"context"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
+	"github.com/golang/protobuf/ptypes/timestamp"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/pdata"
@@ -164,3 +165,8 @@ func (mtp *metricsTransformProcessor) minInt64(num1, num2 int64) int64 {
 	}
 	return num2
 }
+
+// compareTimestamps returns if t1 is a smaller timestamp than t2
+func (mtp *metricsTransformProcessor) compareTimestamps(t1 *timestamp.Timestamp, t2 *timestamp.Timestamp) bool {
+	return t1.Seconds < t2.Seconds || (t1.Seconds == t2.Seconds && t1.Nanos < t2.Nanos)
+}
diff --git a/processor/metricstransformprocessor/metrics_transform_processor_testcases_test.go b/processor/metricstransformprocessor/metrics_transform_processor_testcases_test.go
@@ -378,13 +378,14 @@ var (
 			in: []*metricspb.Metric{
 				metricBuilder().setName("metric1").setLabels([]string{"label1", "label2"}).setDataType(metricspb.MetricDescriptor_GAUGE_DOUBLE).
 					addTimeseries(1, []string{"label1-value1", "label2-value1"}).addTimeseries(1, []string{"label1-value1", "label2-value2"}).
-					addDoublePoint(0, 1, 2).addDoublePoint(1, 3, 2).
+					addTimeseries(2, []string{"label1-value2", "label2-value2"}).
+					addDoublePoint(0, 1, 2).addDoublePoint(1, 3, 2).addDoublePoint(2, 3, 2).
 					build(),
 			},
 			out: []*metricspb.Metric{
 				metricBuilder().setName("metric1").setLabels([]string{"label1"}).setDataType(metricspb.MetricDescriptor_GAUGE_DOUBLE).
-					addTimeseries(1, []string{"label1-value1"}).
-					addDoublePoint(0, 1, 2).
+					addTimeseries(1, []string{"label1-value1"}).addTimeseries(2, []string{"label1-value2"}).
+					addDoublePoint(0, 1, 2).addDoublePoint(1, 3, 2).
 					build(),
 			},
 		},
@@ -409,15 +410,15 @@ var (
 			},
 			in: []*metricspb.Metric{
 				metricBuilder().setName("metric1").setLabels([]string{"label1", "label2"}).setDataType(metricspb.MetricDescriptor_GAUGE_INT64).
-					addTimeseries(1, []string{"label1-value1", "label2-value1"}).addTimeseries(1, []string{"label1-value1", "label2-value2"}).
+					addTimeseries(3, []string{"label1-value1", "label2-value1"}).addTimeseries(3, []string{"label1-value1", "label2-value2"}).
 					addTimeseries(1, []string{"label1-value1", "label2-value3"}).
-					addInt64Point(0, 3, 2).addInt64Point(1, 1, 2).addInt64Point(2, 1, 2).
+					addInt64Point(0, 3, 2).addInt64Point(1, 1, 2).addInt64Point(1, 2, 3).addInt64Point(2, 1, 2).
 					build(),
 			},
 			out: []*metricspb.Metric{
 				metricBuilder().setName("metric1").setLabels([]string{"label1", "label2"}).setDataType(metricspb.MetricDescriptor_GAUGE_INT64).
-					addTimeseries(1, []string{"label1-value1", "new/label2-value"}).addTimeseries(1, []string{"label1-value1", "label2-value3"}).
-					addInt64Point(0, 4, 2).addInt64Point(1, 1, 2).
+					addTimeseries(1, []string{"label1-value1", "label2-value3"}).addTimeseries(3, []string{"label1-value1", "new/label2-value"}).
+					addInt64Point(0, 1, 2).addInt64Point(1, 4, 2).addInt64Point(1, 2, 3).
 					build(),
 			},
 		},

diff --git a/processor/metricstransformprocessor/operation_aggregate_label_values.go b/processor/metricstransformprocessor/operation_aggregate_label_values.go
@@ -16,6 +16,7 @@ package metricstransformprocessor
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
@@ -34,6 +35,9 @@ func (mtp *metricsTransformProcessor) aggregateLabelValuesOp(metric *metricspb.M
 	groupedTimeseries, unchangedTimeseries := mtp.groupTimeseriesByNewLabelValue(metric, operationLabelIdx, op.NewValue, mtpOp.aggregatedValuesSet)
 	aggregatedTimeseries := mtp.mergeTimeseries(groupedTimeseries, op.AggregationType, metric.MetricDescriptor.Type)
 	aggregatedTimeseries = append(aggregatedTimeseries, unchangedTimeseries...)
+	sort.Slice(aggregatedTimeseries, func(i, j int) bool {
+		return mtp.compareTimestamps(aggregatedTimeseries[i].StartTimestamp, aggregatedTimeseries[j].StartTimestamp)
+	})
 	metric.Timeseries = aggregatedTimeseries
 }
 

diff --git a/processor/metricstransformprocessor/operation_aggregate_labels.go b/processor/metricstransformprocessor/operation_aggregate_labels.go
@@ -16,6 +16,7 @@ package metricstransformprocessor
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
@@ -29,6 +30,9 @@ func (mtp *metricsTransformProcessor) aggregateLabelsOp(metric *metricspb.Metric
 	groupedTimeseries := mtp.groupTimeseriesByLabelSet(metric, labelIdxs)
 
 	aggregatedTimeseries := mtp.mergeTimeseries(groupedTimeseries, op.AggregationType, metric.MetricDescriptor.Type)
+	sort.Slice(aggregatedTimeseries, func(i, j int) bool {
+		return mtp.compareTimestamps(aggregatedTimeseries[i].StartTimestamp, aggregatedTimeseries[j].StartTimestamp)
+	})
 	metric.Timeseries = aggregatedTimeseries
 	metric.MetricDescriptor.LabelKeys = labels
 }