kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support aggregation operators in PromQL query. (#12431)

This commit is contained in:
weixiang1862 2024-07-12 16:29:03 +08:00 committed by GitHub
parent 641163cf1d
commit 61b4674c16
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 415 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
docs/en/api/promql-service.md
View File

@ -106,6 +106,33 @@ For example:
service_cpm{service='service_A', layer='$layer'} > service_cpm{service='service_B', layer='$layer'}
```
### Aggregation operators
[Prometheus Docs Reference](https://prometheus.io/docs/prometheus/latest/querying/operators/#aggregation-operators)
| Operator | Definition | Support |
|----------|---------------------------------------|---------|
| sum | calculate sum over dimensions | yes |
| min | select minimum over dimensions | yes |
| max | select maximum over dimensions | yes |
| avg | calculate the average over dimensions | yes |
For example:
If the metric `http_requests_total` had time series that fan out by `service`, `service_instance_id`, and `group` labels,
we could calculate the total number of seen HTTP requests per service and group over all service instances via:
```
sum by (service, group) (http_requests_total{service='$service', layer='$layer'})
```
Which is equivalent to:
```
sum without (service_instance_id) (http_requests_total{service='$service', layer='$layer'})
```
If we are just interested in the total of HTTP requests we have seen in all services, we could simply write:
```
sum(http_requests_total{service='$service', layer='$layer'})
```
### HTTP API
#### Expression queries

1
docs/en/changes/changes.md
View File

@ -31,6 +31,7 @@
* Fix expression of graph `Current QPS` in MySQL dashboard.
* Support tracing logs query for debugging.
* BanyanDB: fix Tag autocomplete data storage and query.
* Support aggregation operators in PromQL query.
#### UI
* Highlight search log keywords.

9
oap-server/server-query-plugin/promql-plugin/src/main/antlr4/org/apache/skywalking/promql/rt/grammar/PromQLLexer.g4
View File

@ -45,6 +45,15 @@ LT: '<';
GTE: '>=';
GT: '>';
// Aggregation operators
AVG: 'avg';
MAX: 'max';
MIN: 'min';
SUM: 'sum';
BY: 'by';
WITHOUT: 'without';
// Literals
NUMBER: Digit+ (DOT Digit+)?;
DURATION: Digit+ ('ms' | 's' | 'm' | 'h' | 'd' | 'w');

9
oap-server/server-query-plugin/promql-plugin/src/main/antlr4/org/apache/skywalking/promql/rt/grammar/PromQLParser.g4
View File

@ -27,6 +27,8 @@ expression
| expression mulDivMod expression # mulDivModOp
| expression addSub expression # addSubOp
| expression compare expression # compareOp
| aggregationFunc (aggregationClause)? L_PAREN expression R_PAREN # aggregationOp
| aggregationFunc L_PAREN expression R_PAREN (aggregationClause)? # aggregationOp
;
expressionNode: metricInstant| metricRange| numberLiteral| badRange;
@ -35,6 +37,12 @@ addSub: ADD | SUB ;
mulDivMod: MUL | DIV | MOD;
compare: (DEQ | NEQ | LTE | LT | GTE | GT) BOOL?;
aggregationFunc:
AVG | SUM | MAX | MIN;
aggregationClause:
(BY | WITHOUT) L_PAREN labelNameList R_PAREN;
metricName: NAME_STRING;
metricInstant: metricName | metricName L_BRACE labelList? R_BRACE;
metricRange: metricInstant L_BRACKET DURATION R_BRACKET;
@ -43,6 +51,7 @@ labelName: NAME_STRING;
labelValue: VALUE_STRING;
label: labelName EQ labelValue;
labelList: label (COMMA label)*;
labelNameList: labelName (COMMA labelName)*;
numberLiteral: NUMBER;

4
oap-server/server-query-plugin/promql-plugin/src/main/java/org/apache/skywalking/oap/query/promql/entity/TimeValuePair.java
View File

@ -25,8 +25,8 @@ import org.apache.skywalking.oap.query.promql.entity.codec.TimeValuePairSerializ
@Data
@JsonSerialize(using = TimeValuePairSerializer.class)
public class TimeValuePair {
private final long time;
private final String value;
private long time;
private String value;
public TimeValuePair(final long time, String value) {
this.time = time;

78
oap-server/server-query-plugin/promql-plugin/src/main/java/org/apache/skywalking/oap/query/promql/rt/PromOpUtils.java
View File

@ -20,8 +20,18 @@ package org.apache.skywalking.oap.query.promql.rt;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.AggregateLabelsFunc;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.AggregateLabelsFuncFactory;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.AvgAggregateLabelsFunc;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.MaxAggregateLabelsFunc;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.MinAggregateLabelsFunc;
import org.apache.skywalking.mqe.rt.operation.aggregatelabels.SumAggregateLabelsFunc;
import org.apache.skywalking.oap.query.promql.entity.LabelValuePair;
import org.apache.skywalking.oap.query.promql.entity.MetricInfo;
import org.apache.skywalking.oap.query.promql.entity.MetricRangeData;
import org.apache.skywalking.oap.query.promql.entity.TimeValuePair;
import org.apache.skywalking.oap.query.promql.rt.exception.IllegalExpressionException;
@ -33,10 +43,14 @@ import org.apache.skywalking.oap.server.core.query.PointOfTime;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.query.type.KVInt;
import org.apache.skywalking.oap.server.core.query.type.MetricsValues;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.promql.rt.grammar.PromQLParser;
import org.joda.time.format.PeriodFormatter;
import org.joda.time.format.PeriodFormatterBuilder;
import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.toList;
public class PromOpUtils {
static MetricsRangeResult matrixScalarBinaryOp(MetricsRangeResult matrix, ScalarResult scalar, int opType) {
@ -92,6 +106,70 @@ public class PromOpUtils {
return result;
}
static MetricsRangeResult matrixAggregateOp(MetricsRangeResult result, int funcType, List<String> groupingBy) {
List<MetricRangeData> metricDataList = result.getMetricDataList();
Map<List<LabelValuePair>, List<MetricRangeData>> groupedResult = metricDataList
.stream().collect(groupingBy(rangeData -> getLabels(groupingBy, rangeData), LinkedHashMap::new, toList()));
MetricsRangeResult rangeResult = new MetricsRangeResult();
rangeResult.setResultType(ParseResultType.METRICS_RANGE);
AggregateLabelsFuncFactory factory = getAggregateFuncFactory(funcType);
groupedResult.forEach((labels, dataList) -> {
if (dataList.isEmpty()) {
return;
}
List<TimeValuePair> combineTo = dataList.get(0).getValues();
for (int i = 0; i < combineTo.size(); i++) {
AggregateLabelsFunc aggregateLabelsFunc = factory.getAggregateLabelsFunc();
for (MetricRangeData rangeData : dataList) {
TimeValuePair toCombine = rangeData.getValues().get(i);
if (StringUtil.isNotBlank(toCombine.getValue())) {
aggregateLabelsFunc.combine(Double.parseDouble(toCombine.getValue()));
}
}
TimeValuePair timeValuePair = combineTo.get(i);
Double aggResult = aggregateLabelsFunc.getResult();
if (aggResult != null) {
timeValuePair.setValue(aggResult.toString());
}
}
MetricRangeData rangeData = new MetricRangeData();
rangeData.setMetric(new MetricInfo(null));
rangeData.getMetric().setLabels(labels);
rangeData.setValues(combineTo);
rangeResult.getMetricDataList().add(rangeData);
});
return rangeResult;
}
private static AggregateLabelsFuncFactory getAggregateFuncFactory(int funcType) {
switch (funcType) {
case PromQLParser.AVG:
return AvgAggregateLabelsFunc::new;
case PromQLParser.SUM:
return SumAggregateLabelsFunc::new;
case PromQLParser.MAX:
return MaxAggregateLabelsFunc::new;
case PromQLParser.MIN:
return MinAggregateLabelsFunc::new;
default:
throw new IllegalArgumentException("Unsupported aggregate function type: " + funcType);
}
}
private static List<LabelValuePair> getLabels(List<String> groupingBy, MetricRangeData data) {
return groupingBy.stream()
.map(
labelName ->
data.getMetric().getLabels()
.stream().filter(label -> labelName.equals(label.getLabelName()))
.findAny().orElseGet(() -> new LabelValuePair(labelName, ""))
)
.collect(toList());
}
static double scalarBinaryOp(double leftValue, double rightValue, int opType) {
double calculatedResult = 0;
switch (opType) {

48
oap-server/server-query-plugin/promql-plugin/src/main/java/org/apache/skywalking/oap/query/promql/rt/PromQLExprQueryVisitor.java
View File

@ -26,7 +26,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.antlr.v4.runtime.RuleContext;
import org.apache.skywalking.oap.query.promql.entity.ErrorType;
import org.apache.skywalking.oap.query.promql.entity.LabelName;
import org.apache.skywalking.oap.query.promql.entity.LabelValuePair;
@ -58,6 +60,7 @@ import org.apache.skywalking.oap.server.core.query.type.Record;
import org.apache.skywalking.oap.server.core.query.type.SelectedRecord;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.promql.rt.grammar.PromQLParser;
import org.apache.skywalking.promql.rt.grammar.PromQLParserBaseVisitor;
@ -135,6 +138,51 @@ public class PromQLExprQueryVisitor extends PromQLParserBaseVisitor<ParseResult>
return compareOp(left, right, opType, boolModifier);
}
@Override
public ParseResult visitAggregationOp(final PromQLParser.AggregationOpContext ctx) {
ParseResult parseResult = visit(ctx.expression());
if (StringUtil.isNotBlank(parseResult.getErrorInfo())) {
return parseResult;
}
if (!parseResult.getResultType().equals(ParseResultType.METRICS_RANGE)) {
ParseResult result = new ParseResult();
result.setErrorType(ErrorType.BAD_DATA);
result.setErrorInfo("Expected type instant vector in aggregation expression.");
return result;
}
MetricsRangeResult metricsRangeResult = (MetricsRangeResult) parseResult;
if (CollectionUtils.isEmpty(metricsRangeResult.getMetricDataList())) {
return metricsRangeResult;
}
List<String> resultLabelNames = metricsRangeResult.getMetricDataList()
.get(0).getMetric().getLabels()
.stream().map(LabelValuePair::getLabelName)
.collect(Collectors.toList());
List<String> groupingBy = new ArrayList<>();
PromQLParser.AggregationClauseContext clauseContext = ctx.aggregationClause();
if (clauseContext != null) {
List<String> clauseGroupingBy = clauseContext.labelNameList().labelName().stream()
.map(RuleContext::getText)
.filter(resultLabelNames::contains)
.collect(Collectors.toList());
if (clauseContext.getStart().getType() == PromQLParser.WITHOUT) {
groupingBy = resultLabelNames.stream()
.filter(labelName -> !clauseGroupingBy.contains(labelName))
.collect(Collectors.toList());
} else {
groupingBy = clauseGroupingBy;
}
}
return PromOpUtils.matrixAggregateOp(
(MetricsRangeResult) parseResult, ctx.aggregationFunc().getStart().getType(), groupingBy
);
}
private ParseResult compareOp(ParseResult left, ParseResult right, int opType, boolean boolModifier) {
try {
if (left.getResultType() == ParseResultType.SCALAR && right.getResultType() == ParseResultType.SCALAR) {

159
oap-server/server-query-plugin/promql-plugin/src/test/java/org/apache/skywalking/promql/rt/parser/PromQLExprQueryVisitorTest.java
View File

@ -25,6 +25,7 @@ import lombok.SneakyThrows;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.ParseTree;
import org.apache.skywalking.oap.query.promql.entity.LabelValuePair;
import org.apache.skywalking.oap.query.promql.entity.TimeValuePair;
import org.apache.skywalking.oap.query.promql.handler.PromQLApiHandler;
import org.apache.skywalking.oap.query.promql.rt.result.ParseResultType;
@ -121,6 +122,108 @@ public class PromQLExprQueryVisitorTest {
});
}
public static Collection<Object[]> aggregateData() {
// {service_instance_id=a,group=g} 0, 1, 2
// {service_instance_id=b,group=g} 2, 3, 4
return Arrays.asList(new Object[][] {
{
"MetricsAggregationOpSum",
PromQLApiHandler.QueryType.RANGE,
"sum by(group) (http_requests_total{service='serviceA', layer='GENERAL'})",
List.of(
List.of(
new TimeValuePair(TIME_2023022010, "2.0"),
new TimeValuePair(TIME_2023022011, "4.0"),
new TimeValuePair(TIME_2023022012, "6.0")
)
),
List.of(
List.of(
new LabelValuePair("group", "g")
)
)
},
{
"MetricsAggregationOpAvg",
PromQLApiHandler.QueryType.RANGE,
"avg by(group) (http_requests_total{service='serviceA', layer='GENERAL'})",
List.of(
List.of(
new TimeValuePair(TIME_2023022010, "1.0"),
new TimeValuePair(TIME_2023022011, "2.0"),
new TimeValuePair(TIME_2023022012, "3.0")
)
),
List.of(
List.of(
new LabelValuePair("group", "g")
)
)
},
{
"MetricsAggregationOpMax",
PromQLApiHandler.QueryType.RANGE,
"max (http_requests_total{service='serviceA', layer='GENERAL'}) by (group)",
List.of(
List.of(
new TimeValuePair(TIME_2023022010, "2.0"),
new TimeValuePair(TIME_2023022011, "3.0"),
new TimeValuePair(TIME_2023022012, "4.0")
)
),
List.of(
List.of(
new LabelValuePair("group", "g")
)
)
},
{
"MetricsAggregationOpMin",
PromQLApiHandler.QueryType.RANGE,
"min (http_requests_total{service='serviceA', layer='GENERAL'}) by (group)",
List.of(
List.of(
new TimeValuePair(TIME_2023022010, "0.0"),
new TimeValuePair(TIME_2023022011, "1.0"),
new TimeValuePair(TIME_2023022012, "2.0")
)
),
List.of(
List.of(
new LabelValuePair("group", "g")
)
)
},
{
"MetricsAggregationOpMinWithout",
PromQLApiHandler.QueryType.RANGE,
"min (http_requests_total{service='serviceA', layer='GENERAL'}) without (group)",
List.of(
List.of(
new TimeValuePair(TIME_2023022010, "0.0"),
new TimeValuePair(TIME_2023022011, "1.0"),
new TimeValuePair(TIME_2023022012, "2.0")
),
List.of(
new TimeValuePair(TIME_2023022010, "2.0"),
new TimeValuePair(TIME_2023022011, "3.0"),
new TimeValuePair(TIME_2023022012, "4.0")
)
),
List.of(
List.of(
new LabelValuePair("service_instance_id", "a"),
new LabelValuePair("layer", "GENERAL")
),
List.of(
new LabelValuePair("service_instance_id", "b"),
new LabelValuePair("layer", "GENERAL")
)
)
}
});
}
@SneakyThrows
@BeforeEach
public void setup() {
@ -128,6 +231,10 @@ public class PromQLExprQueryVisitorTest {
0,
DefaultScopeDefine.SERVICE
);
ValueColumnMetadata.INSTANCE.putIfAbsent("http_requests_total", "value", Column.ValueDataType.LABELED_VALUE,
0,
DefaultScopeDefine.SERVICE
);
metricsQueryService = mock(MetricsQueryService.class);
recordQueryService = mock(RecordQueryService.class);
aggregationQueryService = mock(AggregationQueryService.class);
@ -138,6 +245,10 @@ public class PromQLExprQueryVisitorTest {
Mockito.doReturn(mockMetricsValues())
.when(metricsQueryService)
.readMetricsValues(any(MetricsCondition.class), any(Duration.class));
Mockito.doReturn(mockLabeledMetricsValues())
.when(metricsQueryService)
.readLabeledMetricsValues(any(MetricsCondition.class), any(), any(Duration.class));
}
private MetricsValues mockMetricsValues() {
@ -152,6 +263,30 @@ public class PromQLExprQueryVisitorTest {
return values;
}
private List<MetricsValues> mockLabeledMetricsValues() {
final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
// {service_instance_id=a,group=g} 0, 1, 2
MetricsValues values1 = new MetricsValues();
values1.setLabel("{service_instance_id=a,group=g}");
for (int i = 0; i < pointOfTimes.size(); i++) {
final KVInt kvInt = new KVInt();
kvInt.setId(String.valueOf(pointOfTimes.get(i).getPoint()));
kvInt.setValue(i);
values1.getValues().addKVInt(kvInt);
}
// {service_instance_id=b,group=g} 2, 3, 4
MetricsValues values2 = new MetricsValues();
values2.setLabel("{service_instance_id=b,group=g}");
for (int i = 0; i < pointOfTimes.size(); i++) {
final KVInt kvInt = new KVInt();
kvInt.setId(String.valueOf(pointOfTimes.get(i).getPoint()));
kvInt.setValue(i + 2);
values2.getValues().addKVInt(kvInt);
}
return List.of(values1, values2);
}
@ParameterizedTest(name = "{0}")
@MethodSource("data")
public void test(String name,
@ -182,4 +317,28 @@ public class PromQLExprQueryVisitorTest {
Assertions.fail();
}
}
@ParameterizedTest(name = "{0}")
@MethodSource("aggregateData")
public void testAggregate(String name,
PromQLApiHandler.QueryType queryType,
String expression,
List<Object> wantResultValues,
List<Object> wantResultLabels) {
PromQLLexer lexer = new PromQLLexer(CharStreams.fromString(expression));
CommonTokenStream tokens = new CommonTokenStream(lexer);
PromQLParser parser = new PromQLParser(tokens);
ParseTree tree = parser.expression();
PromQLExprQueryVisitor visitor = new PromQLExprQueryVisitor(
metricsQueryService, recordQueryService, aggregationQueryService, duration, queryType);
ParseResult parseResult = visitor.visit(tree);
Assertions.assertEquals(ParseResultType.METRICS_RANGE, parseResult.getResultType());
MetricsRangeResult result = (MetricsRangeResult) parseResult;
Assertions.assertEquals(result.getMetricDataList().size(), wantResultValues.size());
for (int i = 0; i < result.getMetricDataList().size(); i++) {
Assertions.assertEquals(result.getMetricDataList().get(i).getValues(), wantResultValues.get(i));
Assertions.assertEquals(result.getMetricDataList().get(i).getMetric().getLabels(), wantResultLabels.get(i));
}
}
}

46
test/e2e-v2/cases/promql/expected/service-metric-labeled-matrix-aggregate-by-p.yml Normal file
View File

@ -0,0 +1,46 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
status: success
data:
resultType: matrix
result:
{{- contains .data.result }}
- metric:
__name__:
p: 50
values:
{{- contains .values }}
- - "{{ index . 0 }}"
- "{{ index . 1 }}"
{{- end}}
- metric:
__name__:
p: 75
values:
{{- contains .values }}
- - "{{ index . 0 }}"
- "{{ index . 1 }}"
{{- end}}
- metric:
__name__:
p: 90
values:
{{- contains .values }}
- - "{{ index . 0 }}"
- "{{ index . 1 }}"
{{- end}}
{{- end}}

32
test/e2e-v2/cases/promql/expected/service-metric-labeled-matrix-aggregate-without-p.yml Normal file
View File

@ -0,0 +1,32 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
status: success
data:
resultType: matrix
result:
{{- contains .data.result }}
- metric:
__name__:
layer: GENERAL
scope: Service
service: e2e-service-consumer
values:
{{- contains .values }}
- - "{{ index . 0 }}"
- "{{ index . 1 }}"
{{- end}}
{{- end}}

4
test/e2e-v2/cases/promql/promql-cases.yaml
View File

@ -74,6 +74,10 @@ cases:
expected: expected/service-metric-matrix.yml
- query: curl -X GET http://${oap_host}:${oap_9090}/api/v1/query_range -d 'query=service_percentile{service="e2e-service-consumer", layer="GENERAL", p="50,75,90"}&start='$(($(date +%s)-1800))'&end='$(date +%s)
expected: expected/service-metric-labeled-matrix.yml
- query: curl -X GET http://${oap_host}:${oap_9090}/api/v1/query_range -d 'query=sum by (p) (service_percentile{service="e2e-service-consumer", layer="GENERAL", p="50,75,90"})&start='$(($(date +%s)-1800))'&end='$(date +%s)
expected: expected/service-metric-labeled-matrix-aggregate-by-p.yml
- query: curl -X GET http://${oap_host}:${oap_9090}/api/v1/query_range -d 'query=sum without (p) (service_percentile{service="e2e-service-consumer", layer="GENERAL", p="50,75,90"})&start='$(($(date +%s)-1800))'&end='$(date +%s)
expected: expected/service-metric-labeled-matrix-aggregate-without-p.yml
- query: curl -X GET http://${oap_host}:${oap_9090}/api/v1/query_range -d 'query=service_cpm{layer="GENERAL",top_n="10", order="DES"}&start='$(($(date +%s)-1800))'&end='$(date +%s)
expected: expected/service-metric-sort-matrix.yml
## instance