Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

MQE: code cleanup #10488

Merged
merged 4 commits into from
Jan 21, 2025
Merged

MQE: code cleanup #10488

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
915aad2
Reorder comments to make `tools/check-for-disabled-but-supported-mqe-…
charleskorn Jan 21, 2025
2eacf98
Remove unnecessary indirection
charleskorn Jan 21, 2025
5307128
Fix grammar in error messages
charleskorn Jan 21, 2025
ce52a4c
Remove unused constants
charleskorn Jan 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
280 changes: 135 additions & 145 deletions pkg/streamingpromql/functions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,96 +140,92 @@ func scalarToInstantVectorOperatorFactory(args []types.Operator, _ *limiting.Mem
return scalars.NewScalarToInstantVector(inner, expressionPosition), nil
}

func LabelReplaceFunctionOperatorFactory() InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 5 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 5 argument for label_replace, got %v", len(args))
}
func LabelReplaceFunctionOperatorFactory(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 5 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 5 arguments for label_replace, got %v", len(args))
}

inner, ok := args[0].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for label_replace, got %T", args[0])
}
inner, ok := args[0].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for label_replace, got %T", args[0])
}

dstLabel, ok := args[1].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 2nd argument for label_replace, got %T", args[1])
}
dstLabel, ok := args[1].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 2nd argument for label_replace, got %T", args[1])
}

replacement, ok := args[2].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 3rd argument for label_replace, got %T", args[2])
}
replacement, ok := args[2].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 3rd argument for label_replace, got %T", args[2])
}

srcLabel, ok := args[3].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 4th argument for label_replace, got %T", args[3])
}
srcLabel, ok := args[3].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 4th argument for label_replace, got %T", args[3])
}

regex, ok := args[4].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 5th argument for label_replace, got %T", args[4])
}
regex, ok := args[4].(types.StringOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a string for 5th argument for label_replace, got %T", args[4])
}

f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.PassthroughData,
SeriesMetadataFunction: functions.SeriesMetadataFunctionDefinition{
Func: functions.LabelReplaceFactory(dstLabel, replacement, srcLabel, regex),
NeedsSeriesDeduplication: true,
},
}
f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.PassthroughData,
SeriesMetadataFunction: functions.SeriesMetadataFunctionDefinition{
Func: functions.LabelReplaceFactory(dstLabel, replacement, srcLabel, regex),
NeedsSeriesDeduplication: true,
},
}

o := functions.NewFunctionOverInstantVector(inner, nil, memoryConsumptionTracker, f, expressionPosition, timeRange)
o := functions.NewFunctionOverInstantVector(inner, nil, memoryConsumptionTracker, f, expressionPosition, timeRange)

return operators.NewDeduplicateAndMerge(o, memoryConsumptionTracker), nil
}
return operators.NewDeduplicateAndMerge(o, memoryConsumptionTracker), nil
}

func ClampFunctionOperatorFactory() InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 3 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 3 argument for clamp, got %v", len(args))
}

inner, ok := args[0].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for clamp, got %T", args[0])
}
func ClampFunctionOperatorFactory(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 3 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 3 arguments for clamp, got %v", len(args))
}

min, ok := args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 2nd argument for clamp, got %T", args[1])
}
inner, ok := args[0].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for clamp, got %T", args[0])
}

max, ok := args[2].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 3rd argument for clamp, got %T", args[2])
}
min, ok := args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 2nd argument for clamp, got %T", args[1])
}

f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.Clamp,
SeriesMetadataFunction: functions.DropSeriesName,
}
max, ok := args[2].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 3rd argument for clamp, got %T", args[2])
}

return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{min, max}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.Clamp,
SeriesMetadataFunction: functions.DropSeriesName,
}

return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{min, max}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
}

func ClampMinMaxFunctionOperatorFactory(functionName string, isMin bool) InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 2 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 2 argument for %s, got %v", functionName, len(args))
return nil, fmt.Errorf("expected exactly 2 arguments for %s, got %v", functionName, len(args))
}

inner, ok := args[0].(types.InstantVectorOperator)
Expand All @@ -253,95 +249,89 @@ func ClampMinMaxFunctionOperatorFactory(functionName string, isMin bool) Instant
}
}

func RoundFunctionOperatorFactory() InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 1 && len(args) != 2 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 1 or 2 argument for round, got %v", len(args))
}
func RoundFunctionOperatorFactory(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 1 && len(args) != 2 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected 1 or 2 arguments for round, got %v", len(args))
}

inner, ok := args[0].(types.InstantVectorOperator)
inner, ok := args[0].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for round, got %T", args[0])
}

var toNearest types.ScalarOperator
if len(args) == 2 {
toNearest, ok = args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 1st argument for round, got %T", args[0])
}

var toNearest types.ScalarOperator
if len(args) == 2 {
toNearest, ok = args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 2nd argument for round, got %T", args[1])
}
} else {
toNearest = scalars.NewScalarConstant(float64(1), timeRange, memoryConsumptionTracker, expressionPosition)
}

f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.Round,
SeriesMetadataFunction: functions.DropSeriesName,
return nil, fmt.Errorf("expected a scalar for 2nd argument for round, got %T", args[1])
}
} else {
toNearest = scalars.NewScalarConstant(float64(1), timeRange, memoryConsumptionTracker, expressionPosition)
}

return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{toNearest}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.Round,
SeriesMetadataFunction: functions.DropSeriesName,
}
}

func HistogramQuantileFunctionOperatorFactory() InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, annotations *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 2 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 2 argument for histogram_quantile, got %v", len(args))
}
return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{toNearest}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
}

ph, ok := args[0].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 1st argument for histogram_quantile, got %T", args[0])
}
func HistogramQuantileFunctionOperatorFactory(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, annotations *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 2 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 2 arguments for histogram_quantile, got %v", len(args))
}

inner, ok := args[1].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 2nd argument for histogram_quantile, got %T", args[1])
}
ph, ok := args[0].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 1st argument for histogram_quantile, got %T", args[0])
}

o := functions.NewHistogramQuantileFunction(ph, inner, memoryConsumptionTracker, annotations, expressionPosition, timeRange)
return operators.NewDeduplicateAndMerge(o, memoryConsumptionTracker), nil
inner, ok := args[1].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 2nd argument for histogram_quantile, got %T", args[1])
}
}

func HistogramFractionFunctionOperatorFactory() InstantVectorFunctionOperatorFactory {
return func(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 3 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 3 argument for histogram_fraction, got %v", len(args))
}
o := functions.NewHistogramQuantileFunction(ph, inner, memoryConsumptionTracker, annotations, expressionPosition, timeRange)
return operators.NewDeduplicateAndMerge(o, memoryConsumptionTracker), nil
}

lower, ok := args[0].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 1st argument for histogram_fraction, got %T", args[0])
}
func HistogramFractionFunctionOperatorFactory(args []types.Operator, memoryConsumptionTracker *limiting.MemoryConsumptionTracker, _ *annotations.Annotations, expressionPosition posrange.PositionRange, timeRange types.QueryTimeRange) (types.InstantVectorOperator, error) {
if len(args) != 3 {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected exactly 3 arguments for histogram_fraction, got %v", len(args))
}

upper, ok := args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 2nd argument for histogram_fraction, got %T", args[1])
}
lower, ok := args[0].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 1st argument for histogram_fraction, got %T", args[0])
}

inner, ok := args[2].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 3rd argument for histogram_fraction, got %T", args[2])
}
upper, ok := args[1].(types.ScalarOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected a scalar for 2nd argument for histogram_fraction, got %T", args[1])
}

f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.HistogramFraction,
SeriesMetadataFunction: functions.DropSeriesName,
}
inner, ok := args[2].(types.InstantVectorOperator)
if !ok {
// Should be caught by the PromQL parser, but we check here for safety.
return nil, fmt.Errorf("expected an instant vector for 3rd argument for histogram_fraction, got %T", args[2])
}

return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{lower, upper}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
f := functions.FunctionOverInstantVectorDefinition{
SeriesDataFunc: functions.HistogramFraction,
SeriesMetadataFunction: functions.DropSeriesName,
}

return functions.NewFunctionOverInstantVector(inner, []types.ScalarOperator{lower, upper}, memoryConsumptionTracker, f, expressionPosition, timeRange), nil
}

// These functions return an instant-vector.
Expand All @@ -357,7 +347,7 @@ var instantVectorFunctionOperatorFactories = map[string]InstantVectorFunctionOpe
"avg_over_time": FunctionOverRangeVectorOperatorFactory("avg_over_time", functions.AvgOverTime),
"ceil": InstantVectorTransformationFunctionOperatorFactory("ceil", functions.Ceil),
"changes": FunctionOverRangeVectorOperatorFactory("changes", functions.Changes),
"clamp": ClampFunctionOperatorFactory(),
"clamp": ClampFunctionOperatorFactory,
"clamp_max": ClampMinMaxFunctionOperatorFactory("clamp_max", false),
"clamp_min": ClampMinMaxFunctionOperatorFactory("clamp_min", true),
"cos": InstantVectorTransformationFunctionOperatorFactory("cos", functions.Cos),
Expand All @@ -370,15 +360,15 @@ var instantVectorFunctionOperatorFactories = map[string]InstantVectorFunctionOpe
"floor": InstantVectorTransformationFunctionOperatorFactory("floor", functions.Floor),
"histogram_avg": InstantVectorTransformationFunctionOperatorFactory("histogram_avg", functions.HistogramAvg),
"histogram_count": InstantVectorTransformationFunctionOperatorFactory("histogram_count", functions.HistogramCount),
"histogram_fraction": HistogramFractionFunctionOperatorFactory(),
"histogram_quantile": HistogramQuantileFunctionOperatorFactory(),
"histogram_fraction": HistogramFractionFunctionOperatorFactory,
"histogram_quantile": HistogramQuantileFunctionOperatorFactory,
"histogram_stddev": InstantVectorTransformationFunctionOperatorFactory("histogram_stddev", functions.HistogramStdDevStdVar(true)),
"histogram_stdvar": InstantVectorTransformationFunctionOperatorFactory("histogram_stdvar", functions.HistogramStdDevStdVar(false)),
"histogram_sum": InstantVectorTransformationFunctionOperatorFactory("histogram_sum", functions.HistogramSum),
"idelta": FunctionOverRangeVectorOperatorFactory("idelta", functions.Idelta),
"increase": FunctionOverRangeVectorOperatorFactory("increase", functions.Increase),
"irate": FunctionOverRangeVectorOperatorFactory("irate", functions.Irate),
"label_replace": LabelReplaceFunctionOperatorFactory(),
"label_replace": LabelReplaceFunctionOperatorFactory,
"last_over_time": FunctionOverRangeVectorOperatorFactory("last_over_time", functions.LastOverTime),
"ln": InstantVectorTransformationFunctionOperatorFactory("ln", functions.Ln),
"log10": InstantVectorTransformationFunctionOperatorFactory("log10", functions.Log10),
Expand All @@ -389,7 +379,7 @@ var instantVectorFunctionOperatorFactories = map[string]InstantVectorFunctionOpe
"rad": InstantVectorTransformationFunctionOperatorFactory("rad", functions.Rad),
"rate": FunctionOverRangeVectorOperatorFactory("rate", functions.Rate),
"resets": FunctionOverRangeVectorOperatorFactory("resets", functions.Resets),
"round": RoundFunctionOperatorFactory(),
"round": RoundFunctionOperatorFactory,
"sgn": InstantVectorTransformationFunctionOperatorFactory("sgn", functions.Sgn),
"sin": InstantVectorTransformationFunctionOperatorFactory("sin", functions.Sin),
"sinh": InstantVectorTransformationFunctionOperatorFactory("sinh", functions.Sinh),
Expand Down
Loading
Loading