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

Add `count` aggregation function for MAL (#11869) 

Co-authored-by: Wan Kai <wankai123@foxmail.com>
This commit is contained in:
peachisai 2024-02-06 10:48:51 +08:00 committed by GitHub
parent 78d9e33f10
commit a688500fb4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 110 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -40,6 +40,7 @@
- VIRTUAL_DATABASE -> POSTGRESQL
* Add Golang as a supported language for AMQP.
* Support available layers of service in the topology.
* Add `count` aggregation function for MAL
* Add Service Hierarchy auto matching layer relationships (upper -> lower) as following:
- NGINX -> K8S_SERVICE
- APISIX -> K8S_SERVICE

1
docs/en/concepts-and-designs/mal.md
View File

@ -154,6 +154,7 @@ resulting in a new sample family having fewer samples (sometimes having just a s
- min (select minimum over dimensions)
- max (select maximum over dimensions)
- avg (calculate the average over dimensions)
- count (calculate the count over dimensions, the last tag will be counted)
These operations can be used to aggregate overall label dimensions or preserve distinct dimensions by inputting `by` parameter( the keyword `by` could be omitted)

63
oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java
View File

@ -47,6 +47,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.DoubleBinaryOperator;
import java.util.function.Function;
@ -242,6 +243,59 @@ public class SampleFamily {
);
}
public SampleFamily count(List<String> by) {
ExpressionParsingContext.get().ifPresent(ctx -> ctx.aggregationLabels.addAll(by));
if (this == EMPTY) {
return EMPTY;
}
if (by == null) {
long result = Arrays.stream(samples).count();
return SampleFamily.build(
this.context, InternalOps.newSample(samples[0].name, ImmutableMap.of(), samples[0].timestamp, result));
}
if (by.size() == 1) {
Set<String> set = Arrays
.stream(samples)
.map(sample -> sample.labels.get(by.get(0)))
.filter(StringUtils::isNotBlank)
.collect(Collectors.toSet());
return SampleFamily.build(
this.context, InternalOps.newSample(samples[0].name, ImmutableMap.of(), samples[0].timestamp, set.size()));
}
Stream<Map.Entry<ImmutableMap<String, String>, List<Sample>>> stream = Arrays
.stream(samples)
.filter(sample -> sample.labels.keySet().containsAll(by))
.collect(groupingBy(it -> InternalOps.getLabels(by, it)))
.entrySet()
.stream()
.map(entry -> InternalOps.newSample(
entry.getValue().get(0).getName(),
entry.getKey(),
entry.getValue().get(0).getTimestamp(),
entry.getValue().size()))
.collect(groupingBy(it -> InternalOps.groupByExcludedLabel(by.get(by.size() - 1), it), mapping(identity(), toList())))
.entrySet()
.stream();
Sample[] array = stream
.map(entry -> InternalOps.newSample(
entry.getValue().get(0).getName(),
entry.getKey(),
entry.getValue().get(0).getTimestamp(),
entry.getValue().size()
))
.toArray(Sample[]::new);
SampleFamily sampleFamily = SampleFamily.build(
this.context,
array
);
return sampleFamily;
}
protected SampleFamily aggregate(List<String> by, DoubleBinaryOperator aggregator) {
ExpressionParsingContext.get().ifPresent(ctx -> ctx.aggregationLabels.addAll(by));
if (this == EMPTY) {
@ -801,6 +855,15 @@ public class SampleFamily {
labelKey -> sample.labels.getOrDefault(labelKey, "")
));
}
private static ImmutableMap<String, String> groupByExcludedLabel(final String excludedLabelKey, final Sample sample) {
return sample
.labels
.entrySet()
.stream()
.filter(v -> !v.getKey().equals(excludedLabelKey))
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
}
}
private enum CompType {

45
oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java
View File

@ -147,6 +147,51 @@ public class AggregationTest {
).build()),
false,
},
{
"count",
of("http_success_request", SampleFamilyBuilder.newBuilder(
Sample.builder().labels(of("idc", "t3")).value(100).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t2")).value(3).name("http_success_request").build()
).build()),
"http_success_request.count()",
Result.success(SampleFamilyBuilder.newBuilder(Sample.builder().labels(ImmutableMap.of()).value(3).name("http_success_request").build()).build()),
false,
},
{
"count-by-one",
of("http_success_request", SampleFamilyBuilder.newBuilder(
Sample.builder().labels(of("idc", "t1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "cn", "instance", "10.0.0.1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t2", "region", "us", "instance", "10.0.0.2")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.3")).value(100).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t2", "region", "cn", "instance", "10.0.0.3")).value(3).name("http_success_request").build()
).build()),
"http_success_request.count(by = ['instance'])",
Result.success(SampleFamilyBuilder.newBuilder(
Sample.builder().labels(ImmutableMap.of()).value(3).name("http_success_request").build()
).build()),
false,
},
{
"count-by-multi",
of("http_success_request", SampleFamilyBuilder.newBuilder(
Sample.builder().labels(of("idc", "t1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "cn", "instance", "10.0.0.1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "cn", "instance", "10.0.0.1")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t2", "region", "us", "instance", "10.0.0.2")).value(50).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.3")).value(100).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t1", "region", "us", "instance", "10.0.0.4")).value(100).name("http_success_request").build(),
Sample.builder().labels(of("idc", "t2", "region", "cn", "instance", "10.0.0.5")).value(3).name("http_success_request").build()
).build()),
"http_success_request.count(by = ['region','instance'])",
Result.success(SampleFamilyBuilder.newBuilder(
Sample.builder().labels(of("region", "us")).value(3).name("http_success_request").build(),
Sample.builder().labels(of("region", "cn")).value(2).name("http_success_request").build()
).build()),
false,
},
});
}