-
Notifications
You must be signed in to change notification settings - Fork 544
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
MQE: Add support for more aggregation functions (#9008)
* MQE: Add feature flag for disabling aggregation operations * Move sum aggregation into AggregationGroup struct * Use AggregationFunctionFactories in prep to expand * Add MAX aggregation * Update comment from promql issue * Rename aggregation group * Split out accumulatePoint * Move error declaration * Use emitAnnotationFunc optimisation * Remove old comment * Remove unncessary resetting of values * Finish renaming AggregationGroup * Add support for min aggregation * Optimisation: Check for type conflicts as we go * Move invalidCombinationOfHistograms to common in prep for avg * Compact resulting histogram (consistent with promql) * Remove avg that was not ready to be committed * Do not compact for now * Rename aggregation * Construct min/max accumulation point once * Rename compensation values * Add extra test for conflicting series * Revert "Optimisation: Check for type conflicts as we go" This reverts commit 6605803. * Fix lint * Update CHANGELOG * Update docs * Address review feedback
- Loading branch information
Showing
14 changed files
with
460 additions
and
233 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package aggregations | ||
|
|
||
| import ( | ||
| "github.com/prometheus/prometheus/model/histogram" | ||
| "github.com/prometheus/prometheus/promql/parser" | ||
|
|
||
| "github.com/grafana/mimir/pkg/streamingpromql/functions" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/limiting" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/types" | ||
| ) | ||
|
|
||
| // AggregationGroup accumulates series that have been grouped together and computes the output series data. | ||
| type AggregationGroup interface { | ||
| // AccumulateSeries takes in a series as part of the group | ||
| AccumulateSeries(data types.InstantVectorSeriesData, steps int, start int64, interval int64, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, emitAnnotationFunc functions.EmitAnnotationFunc) error | ||
| // ComputeOutputSeries does any final calculations and returns the grouped series data | ||
| ComputeOutputSeries(start int64, interval int64, memoryConsumptionTracker *limiting.MemoryConsumptionTracker) (types.InstantVectorSeriesData, bool, error) | ||
| } | ||
|
|
||
| type AggregationGroupFactory func() AggregationGroup | ||
|
|
||
| var AggregationGroupFactories = map[parser.ItemType]AggregationGroupFactory{ | ||
| parser.MAX: func() AggregationGroup { return NewMinMaxAggregationGroup(true) }, | ||
| parser.MIN: func() AggregationGroup { return NewMinMaxAggregationGroup(false) }, | ||
| parser.SUM: func() AggregationGroup { return &SumAggregationGroup{} }, | ||
| } | ||
|
|
||
| // Sentinel value used to indicate a sample has seen an invalid combination of histograms and should be ignored. | ||
| // | ||
| // Invalid combinations include exponential and custom buckets, and histograms with incompatible custom buckets. | ||
| var invalidCombinationOfHistograms = &histogram.FloatHistogram{} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
| // Provenance-includes-location: https://github.com/prometheus/prometheus/blob/main/promql/engine.go | ||
| // Provenance-includes-license: Apache-2.0 | ||
| // Provenance-includes-copyright: The Prometheus Authors | ||
|
|
||
| package aggregations | ||
|
|
||
| import ( | ||
| "math" | ||
|
|
||
| "github.com/prometheus/prometheus/promql" | ||
|
|
||
| "github.com/grafana/mimir/pkg/streamingpromql/functions" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/limiting" | ||
| "github.com/grafana/mimir/pkg/streamingpromql/types" | ||
| ) | ||
|
|
||
| type MinMaxAggregationGroup struct { | ||
| floatValues []float64 | ||
| floatPresent []bool | ||
|
|
||
| accumulatePoint func(idx int64, f float64) | ||
| } | ||
|
|
||
| // max represents whether this aggregation is `max` (true), or `min` (false) | ||
| func NewMinMaxAggregationGroup(max bool) *MinMaxAggregationGroup { | ||
| g := &MinMaxAggregationGroup{} | ||
| if max { | ||
| g.accumulatePoint = g.maxAccumulatePoint | ||
| } else { | ||
| g.accumulatePoint = g.minAccumulatePoint | ||
| } | ||
| return g | ||
| } | ||
|
|
||
| func (g *MinMaxAggregationGroup) maxAccumulatePoint(idx int64, f float64) { | ||
| if !g.floatPresent[idx] || g.floatPresent[idx] && f > g.floatValues[idx] { | ||
| g.floatValues[idx] = f | ||
| g.floatPresent[idx] = true | ||
| } | ||
| } | ||
|
|
||
| func (g *MinMaxAggregationGroup) minAccumulatePoint(idx int64, f float64) { | ||
| if !g.floatPresent[idx] || g.floatPresent[idx] && f < g.floatValues[idx] { | ||
| g.floatValues[idx] = f | ||
| g.floatPresent[idx] = true | ||
| } | ||
| } | ||
|
|
||
| func (g *MinMaxAggregationGroup) AccumulateSeries(data types.InstantVectorSeriesData, steps int, start int64, interval int64, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ functions.EmitAnnotationFunc) error { | ||
| if len(data.Floats) > 0 && g.floatValues == nil { | ||
| var err error | ||
| // First series with float values for this group, populate it. | ||
| g.floatValues, err = types.Float64SlicePool.Get(steps, memoryConsumptionTracker) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| g.floatPresent, err = types.BoolSlicePool.Get(steps, memoryConsumptionTracker) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| g.floatValues = g.floatValues[:steps] | ||
| g.floatPresent = g.floatPresent[:steps] | ||
| } | ||
|
|
||
| for _, p := range data.Floats { | ||
| if math.IsNaN(p.F) { | ||
| continue | ||
| } | ||
| idx := (p.T - start) / interval | ||
| g.accumulatePoint(idx, p.F) | ||
| } | ||
|
|
||
| // If a histogram exists max treats it as 0. We have to detect this here so that we return a 0 value instead of nothing. | ||
| // This is consistent with prometheus but may not be desired value: https://github.com/prometheus/prometheus/issues/14711 | ||
| for _, p := range data.Histograms { | ||
| idx := (p.T - start) / interval | ||
| g.accumulatePoint(idx, 0) | ||
| } | ||
|
|
||
| types.PutInstantVectorSeriesData(data, memoryConsumptionTracker) | ||
| return nil | ||
| } | ||
|
|
||
| func (g *MinMaxAggregationGroup) ComputeOutputSeries(start int64, interval int64, memoryConsumptionTracker *limiting.MemoryConsumptionTracker) (types.InstantVectorSeriesData, bool, error) { | ||
| floatPointCount := 0 | ||
| for _, p := range g.floatPresent { | ||
| if p { | ||
| floatPointCount++ | ||
| } | ||
| } | ||
| var floatPoints []promql.FPoint | ||
| var err error | ||
| if floatPointCount > 0 { | ||
| floatPoints, err = types.FPointSlicePool.Get(floatPointCount, memoryConsumptionTracker) | ||
| if err != nil { | ||
| return types.InstantVectorSeriesData{}, false, err | ||
| } | ||
|
|
||
| for i, havePoint := range g.floatPresent { | ||
| if havePoint { | ||
| t := start + int64(i)*interval | ||
| f := g.floatValues[i] | ||
| floatPoints = append(floatPoints, promql.FPoint{T: t, F: f}) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| types.Float64SlicePool.Put(g.floatValues, memoryConsumptionTracker) | ||
| types.BoolSlicePool.Put(g.floatPresent, memoryConsumptionTracker) | ||
|
|
||
| return types.InstantVectorSeriesData{Floats: floatPoints}, false, nil | ||
| } |
Oops, something went wrong.