Add functions in MAL to filter metrics according to the metric value (#6587)

CHANGES.md

@@ -53,6 +53,7 @@ Release Notes.
 * Fix receiver don't need to get itself when healthCheck
 * Remove group concept from AvgHistogramFunction. Heatmap(function result) doesn't support labels.
 * Support metrics grouped by scope labelValue in MAL, no need global same labelValue as before.
+* Add functions in MAL to filter metrics according to the metric value.
 * Optimize the self monitoring grafana dashboard.
 
 #### UI

docs/en/concepts-and-designs/mal.md

@@ -41,6 +41,22 @@ For example, this filters all instance_trace_count samples for us-west and asia-
 ```
 instance_trace_count.tagMatch("region", "us-west|asia-north").tagEqual("az", "az-1")
 ```
+### Value filter
+
+MAL support six type operations to filter samples in a sample family by value:
+
+- valueEqual: Filter values that are exactly equal to the provided value.
+- valueNotEqual: Filter values that are not equal to the provided value.
+- valueGreater: Filter values that greater than the provided value.
+- valueGreaterEqual: Filter values that greater or equal the provided value.
+- valueLess: Filter values that less than the provided value.
+- valueLessEqual: Filter values that less or equal the provided value.
+
+For example, this filters all instance_trace_count samples for values >= 33:
+
+```
+instance_trace_count.valueGreaterEqual(33)
+```
 
 ### Binary operators
 

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

@@ -26,6 +26,7 @@ import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.AtomicDouble;
 import groovy.lang.Closure;
 import io.vavr.Function2;
+import io.vavr.Function3;
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
@@ -101,6 +102,31 @@ public class SampleFamily {
         return match(labels, (sv, lv) -> !sv.matches(lv));
     }
 
+    /* value filter operations*/
+    public SampleFamily valueEqual(double compValue) {
+        return valueMatch(CompType.EQUAL, compValue, InternalOps::doubleComp);
+    }
+
+    public SampleFamily valueNotEqual(double compValue) {
+        return valueMatch(CompType.NOT_EQUAL, compValue, InternalOps::doubleComp);
+    }
+
+    public SampleFamily valueGreater(double compValue) {
+        return valueMatch(CompType.GREATER, compValue, InternalOps::doubleComp);
+    }
+
+    public SampleFamily valueGreaterEqual(double compValue) {
+        return valueMatch(CompType.GREATER_EQUAL, compValue, InternalOps::doubleComp);
+    }
+
+    public SampleFamily valueLess(double compValue) {
+        return valueMatch(CompType.LESS, compValue, InternalOps::doubleComp);
+    }
+
+    public SampleFamily valueLessEqual(double compValue) {
+        return valueMatch(CompType.LESS_EQUAL, compValue, InternalOps::doubleComp);
+    }
+
     /* Binary operator overloading*/
     public SampleFamily plus(Number number) {
         return newValue(v -> v + number.doubleValue());
@@ -401,6 +427,14 @@ public class SampleFamily {
         return ss.length > 0 ? SampleFamily.build(this.context, ss) : EMPTY;
     }
 
+    private SampleFamily valueMatch(CompType compType,
+                                    double compValue,
+                                    Function3<CompType, Double, Double, Boolean> op) {
+        Sample[] ss = Arrays.stream(samples)
+                            .filter(sample -> op.apply(compType, sample.value, compValue)).toArray(Sample[]::new);
+        return ss.length > 0 ? SampleFamily.build(this.context, ss) : EMPTY;
+    }
+
     SampleFamily newValue(Function<Double, Double> transform) {
         if (this == EMPTY) {
             return EMPTY;
@@ -509,6 +543,26 @@ public class SampleFamily {
             return a.equals(b);
         }
 
+        private static boolean doubleComp(CompType compType, double a, double b) {
+            int result = Double.compare(a, b);
+            switch (compType) {
+                case EQUAL:
+                    return result == 0;
+                case NOT_EQUAL:
+                    return result != 0;
+                case GREATER:
+                    return result == 1;
+                case GREATER_EQUAL:
+                    return result == 0 || result == 1;
+                case LESS:
+                    return result == -1;
+                case LESS_EQUAL:
+                    return result == 0 || result == -1;
+            }
+
+            return false;
+        }
+
         private static ImmutableMap<String, String> getLabels(final List<String> labelKeys, final Sample sample) {
             return labelKeys.stream()
                             .collect(toImmutableMap(
@@ -517,4 +571,8 @@ public class SampleFamily {
                             ));
         }
     }
+
+    private enum CompType {
+        EQUAL, NOT_EQUAL, LESS, LESS_EQUAL, GREATER, GREATER_EQUAL
+    }
 }

oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/ValueFilterTest.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.oap.meter.analyzer.dsl;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Arrays;
+import java.util.Collection;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static com.google.common.collect.ImmutableMap.of;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+@Slf4j
+@RunWith(Parameterized.class)
+public class ValueFilterTest {
+
+    @Parameterized.Parameter
+    public String name;
+
+    @Parameterized.Parameter(1)
+    public ImmutableMap<String, SampleFamily> input;
+
+    @Parameterized.Parameter(2)
+    public String expression;
+
+    @Parameterized.Parameter(3)
+    public Result want;
+
+    @Parameterized.Parameter(4)
+    public boolean isThrow;
+
+    @Parameterized.Parameters(name = "{index}: {0}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][] {
+            {
+                "valueEqual",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueEqual(1)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                false,
+                },
+            {
+                "valueNotEqual",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueNotEqual(1)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build()
+                ).build()),
+                false,
+                },
+            {
+                "valueGreater",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueGreater(1)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build()
+                ).build()),
+                false,
+                },
+            {
+                "valueGreaterEqual",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueGreaterEqual(1)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                false,
+                },
+            {
+                "valueLess",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueLess(2)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                false,
+                },
+            {
+                "valueLessEqual",
+                of("http_success_request", SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                "http_success_request.valueLessEqual(2)",
+                Result.success(SampleFamilyBuilder.newBuilder(
+                    Sample.builder().labels(of("idc", "t1")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t2")).value(2).build(),
+                    Sample.builder().labels(of("idc", "t3")).value(1).build()
+                ).build()),
+                false,
+                },
+            });
+    }
+
+    @Test
+    public void test() {
+        Expression e = DSL.parse(expression);
+        Result r = null;
+        try {
+            r = e.run(input);
+        } catch (Throwable t) {
+            if (isThrow) {
+                return;
+            }
+            log.error("Test failed", t);
+            fail("Should not throw anything");
+        }
+        if (isThrow) {
+            fail("Should throw something");
+        }
+        assertThat(r, is(want));
+    }
+}
\ No newline at end of file