Add a way to control the aggregation type for the SelectSeries API (#…

…2758) * Add mergeFunction to the SelectSeries API * Rename mergeFunction to aggregation, make it optional * Rename accumulator to aggregator * Use aggregation param in series merger as well * Switch to an enum for the time series aggregation type * Restore removed doc * Add aggregation to frontend state * Fix formatting * Simplify series aggregation: - perform only sum aggregation in series merger - retain duplicate samples in series merger (when sum=false) - remove "first value" aggregator * Fix issues after merge with main * Simplify test
grafana · Dec 4, 2023 · b9966bd · b9966bd
1 parent fd89bda
commit b9966bd
Showing 22 changed files with 837 additions and 452 deletions.
diff --git a/api/gen/proto/go/ingester/v1/ingester.pb.go b/api/gen/proto/go/ingester/v1/ingester.pb.go
diff --git a/api/gen/proto/go/ingester/v1/ingester_vtproto.pb.go b/api/gen/proto/go/ingester/v1/ingester_vtproto.pb.go
diff --git a/api/gen/proto/go/querier/v1/querier.pb.go b/api/gen/proto/go/querier/v1/querier.pb.go
diff --git a/api/gen/proto/go/querier/v1/querier_vtproto.pb.go b/api/gen/proto/go/querier/v1/querier_vtproto.pb.go
diff --git a/api/gen/proto/go/types/v1/types.pb.go b/api/gen/proto/go/types/v1/types.pb.go
diff --git a/api/ingester/v1/ingester.proto b/api/ingester/v1/ingester.proto
@@ -59,6 +59,8 @@ message SelectProfilesRequest {
   int64 end = 4;
   // Optional: Hints for querying
   optional Hints hints = 5;
+  // Optional: Aggregation
+  optional types.v1.TimeSeriesAggregationType aggregation = 6;
 }
 
 message MergeProfilesStacktracesRequest {

diff --git a/api/openapiv2/gen/phlare.swagger.json b/api/openapiv2/gen/phlare.swagger.json
@@ -1001,6 +1001,10 @@
         "hints": {
           "$ref": "#/definitions/v1Hints",
           "title": "Optional: Hints for querying"
+        },
+        "aggregation": {
+          "$ref": "#/definitions/v1TimeSeriesAggregationType",
+          "title": "Optional: Aggregation"
         }
       }
     },
@@ -1100,6 +1104,14 @@
       ],
       "default": "MERGE_FORMAT_UNSPECIFIED"
     },
+    "v1TimeSeriesAggregationType": {
+      "type": "string",
+      "enum": [
+        "TIME_SERIES_AGGREGATION_TYPE_SUM",
+        "TIME_SERIES_AGGREGATION_TYPE_AVERAGE"
+      ],
+      "default": "TIME_SERIES_AGGREGATION_TYPE_SUM"
+    },
     "v1ValueType": {
       "type": "object",
       "properties": {

diff --git a/api/querier/v1/querier.proto b/api/querier/v1/querier.proto
@@ -124,6 +124,7 @@ message SelectSeriesRequest {
   int64 end = 4; // milliseconds since epoch
   repeated string group_by = 5;
   double step = 6; // Query resolution step width in seconds
+  optional types.v1.TimeSeriesAggregationType aggregation = 7;
 }
 
 message SelectSeriesResponse {

diff --git a/api/types/v1/types.proto b/api/types/v1/types.proto
@@ -66,3 +66,8 @@ message BlockCompaction {
   repeated string sources = 2;
   repeated string parents = 3;
 }
+
+enum TimeSeriesAggregationType {
+  TIME_SERIES_AGGREGATION_TYPE_SUM = 0;
+  TIME_SERIES_AGGREGATION_TYPE_AVERAGE = 1;
+}
diff --git a/pkg/frontend/frontend_select_series.go b/pkg/frontend/frontend_select_series.go
@@ -66,6 +66,7 @@ func (f *Frontend) SelectSeries(ctx context.Context,
 				End:           r.End.UnixMilli(),
 				GroupBy:       c.Msg.GroupBy,
 				Step:          c.Msg.Step,
+				Aggregation:   c.Msg.Aggregation,
 			})
 			resp, err := connectgrpc.RoundTripUnary[
 				querierv1.SelectSeriesRequest,