diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 371964a6bc..ae292bbd2d 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -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 diff --git a/docs/en/concepts-and-designs/mal.md b/docs/en/concepts-and-designs/mal.md index 05da076fb8..50ff6decc6 100644 --- a/docs/en/concepts-and-designs/mal.md +++ b/docs/en/concepts-and-designs/mal.md @@ -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) diff --git a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java index 3a0f825ee0..e91a98f2ac 100644 --- a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java +++ b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/dsl/SampleFamily.java @@ -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 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 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, List>> 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 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 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 { diff --git a/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java index 4f48fdc267..2afeeb57b2 100644 --- a/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java +++ b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/AggregationTest.java @@ -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, + }, }); }