Support labelAvg function in the OAL engine

mrproliu · mrproliu · commit a1ef35641f96 · 2025-01-08T18:16:10.000+08:00
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
@@ -13,6 +13,8 @@
     source tar from the website and publish them to your private maven repository.
 * [Breaking Change] Remove H2 as storage option permanently. BanyanDB 0.8(OAP 10.2 required) is easy, stable and 
   production-ready. Don't need H2 as default storage anymore.
+* Support `labelCount` function in the OAL engine.
+* Added `maxLabelCount` parameter in the `labelCount` function of OAL to limit the number of labels can be counted.
 
 #### OAP Server
 
diff --git a/docs/en/concepts-and-designs/oal.md b/docs/en/concepts-and-designs/oal.md
@@ -103,9 +103,14 @@ In this case, see `p99`, `p95`, `p90`, `p75`, and `p50` of all incoming requests
 In this case, the p99 value of all incoming requests. The parameter is precise to a latency at p99, such as in the above case, and 120ms and 124ms are considered to produce the same response time.
 
 - `labelCount`. The count of the label value.
-> drop_reason_count = from(CiliumService.*).filter(verdict == "dropped").labelCount(dropReason);
+> drop_reason_count = from(CiliumService.*).filter(verdict == "dropped").labelCount(dropReason, 100);
 
-In this case, the count of the drop reason of each Cilium service. 
+In this case, the count of the drop reason of each Cilium service, max support calculate `100` reasons(optional configuration). 
+
+- `labelAvg`. The avg of the label value.
+> drop_reason_avg = from(BrowserResourcePerf.*).labelAvg(name, duration, 100);
+
+In this case, the avg of the duration of each browser resource file, max support calculate `100` resource file(optional configuration).
 
 ## Metrics name
 The metrics name for storage implementor, alarm and query modules. The type inference is supported by core.
diff --git a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AggregationFuncStmt.java b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/AggregationFuncStmt.java
@@ -67,4 +67,8 @@ public Argument getLastArgument() {
     public Argument getNextFuncArg() {
         return funcArgs.get(argGetIdx++);
     }
+
+    public boolean hasNextArg() {
+        return argGetIdx < funcArgs.size();
+    }
 }
diff --git a/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/DeepAnalysis.java b/oap-server/oal-rt/src/main/java/org/apache/skywalking/oal/rt/parser/DeepAnalysis.java
@@ -23,6 +23,7 @@
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Arg;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.ConstOne;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.DefaultValue;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Entrance;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.SourceFrom;
 import org.apache.skywalking.oap.server.core.storage.annotation.Column;
@@ -123,6 +124,12 @@ public AnalysisResult analysis(AnalysisResult result) {
                 }
             } else if (annotation instanceof Arg) {
                 entryMethod.addArg(parameterType, result.getAggregationFuncStmt().getNextFuncArg());
+            } else if (annotation instanceof DefaultValue) {
+                if (result.getAggregationFuncStmt().hasNextArg()) {
+                    entryMethod.addArg(parameterType, result.getAggregationFuncStmt().getNextFuncArg());
+                } else {
+                    entryMethod.addArg(parameterType, ((DefaultValue) annotation).value());
+                }
             } else {
                 throw new IllegalArgumentException(
                     "Entrance method:" + entranceMethod + " doesn't the expected annotation.");
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/DataTable.java
@@ -70,8 +70,18 @@ public void put(DataLabel labels, Long value) {
      * Accumulate the value with existing value in the same given key.
      */
     public void valueAccumulation(String key, Long value) {
+        this.valueAccumulation(key, value, 0);
+    }
+
+    /**
+     * Accumulate the value with existing value in the same given key, and limit the data size.
+     */
+    public void valueAccumulation(String key, Long value, int maxDataSize) {
         Long element = data.get(key);
         if (element == null) {
+            if (maxDataSize > 0 && data.size() >= maxDataSize) {
+                return;
+            }
             element = value;
         } else {
             element += value;
@@ -155,9 +165,16 @@ public void copyFrom(final DataTable source) {
     }
 
     public DataTable append(DataTable dataTable) {
+        return this.append(dataTable, 0);
+    }
+
+    public DataTable append(DataTable dataTable, int maxDataSize) {
         dataTable.data.forEach((key, value) -> {
             Long current = this.data.get(key);
             if (current == null) {
+                if (maxDataSize > 0 && data.size() >= maxDataSize) {
+                    return;
+                }
                 current = value;
             } else {
                 current += value;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/LabelAvgMetrics.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/LabelAvgMetrics.java
@@ -0,0 +1,117 @@
+/*
+ * 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.server.core.analysis.metrics;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Arg;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.DefaultValue;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Entrance;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.MetricsFunction;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ElasticSearch;
+
+import java.util.Objects;
+import java.util.Set;
+
+@MetricsFunction(functionName = "labelAvg")
+public abstract class LabelAvgMetrics extends Metrics implements LabeledValueHolder {
+    protected static final String SUMMATION = "datatable_summation";
+    protected static final String COUNT = "datatable_count";
+    protected static final String VALUE = "datatable_value";
+
+    protected static final String LABEL_NAME = "n";
+
+    @Getter
+    @Setter
+    @Column(name = SUMMATION, storageOnly = true)
+    @ElasticSearch.Column(legacyName = "summation")
+    @BanyanDB.MeasureField
+    protected DataTable summation;
+    @Getter
+    @Setter
+    @Column(name = COUNT, storageOnly = true)
+    @ElasticSearch.Column(legacyName = "count")
+    @BanyanDB.MeasureField
+    protected DataTable count;
+    @Getter
+    @Setter
+    @Column(name = VALUE, dataType = Column.ValueDataType.LABELED_VALUE, storageOnly = true)
+    @ElasticSearch.Column(legacyName = "value")
+    @BanyanDB.MeasureField
+    private DataTable value;
+
+    private boolean isCalculated;
+    private int maxLabelCount;
+
+    public LabelAvgMetrics() {
+        this.summation = new DataTable(30);
+        this.count = new DataTable(30);
+        this.value = new DataTable(30);
+    }
+
+    @Entrance
+    public final void combine(@Arg String label, @Arg long count, @DefaultValue("1024") int maxLabelCount) {
+        this.isCalculated = false;
+        this.maxLabelCount = maxLabelCount;
+        this.summation.valueAccumulation(label, count, maxLabelCount);
+        this.count.valueAccumulation(label, 1L, maxLabelCount);
+    }
+
+    @Override
+    public boolean combine(Metrics metrics) {
+        this.isCalculated = false;
+        final LabelAvgMetrics labelCountMetrics = (LabelAvgMetrics) metrics;
+        this.summation.append(labelCountMetrics.summation, labelCountMetrics.maxLabelCount);
+        this.count.append(labelCountMetrics.count, labelCountMetrics.maxLabelCount);
+        return true;
+    }
+
+    @Override
+    public void calculate() {
+        if (isCalculated) {
+            return;
+        }
+
+        Set<String> keys = count.keys();
+        for (String key : keys) {
+            Long s = summation.get(key);
+            if (Objects.isNull(s)) {
+                continue;
+            }
+            Long c = count.get(key);
+            if (Objects.isNull(c)) {
+                continue;
+            }
+            long result = s / c;
+            if (result == 0 && s > 0) {
+                result = 1;
+            }
+            final DataLabel label = new DataLabel();
+            label.put(LABEL_NAME, key);
+            value.put(label, result);
+        }
+    }
+
+    @Override
+    public DataTable getValue() {
+        return this.value;
+    }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/LabelCountMetrics.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/LabelCountMetrics.java
@@ -22,6 +22,7 @@
 import lombok.Setter;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Arg;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.ConstOne;
+import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.DefaultValue;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.Entrance;
 import org.apache.skywalking.oap.server.core.analysis.metrics.annotation.MetricsFunction;
 import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
@@ -49,23 +50,25 @@ public abstract class LabelCountMetrics extends Metrics implements LabeledValueH
     private DataTable value;
 
     private boolean isCalculated;
+    private int maxLabelCount;
 
     public LabelCountMetrics() {
         this.dataset = new DataTable(30);
         this.value = new DataTable(30);
     }
 
     @Entrance
-    public final void combine(@Arg String label, @ConstOne long count) {
+    public final void combine(@Arg String label, @ConstOne long count, @DefaultValue("1024") int maxLabelCount) {
         this.isCalculated = false;
-       this.dataset.valueAccumulation(label, count);
+        this.maxLabelCount = maxLabelCount;
+        this.dataset.valueAccumulation(label, count, maxLabelCount);
     }
 
     @Override
     public boolean combine(Metrics metrics) {
         this.isCalculated = false;
         final LabelCountMetrics labelCountMetrics = (LabelCountMetrics) metrics;
-        this.dataset.append(labelCountMetrics.dataset);
+        this.dataset.append(labelCountMetrics.dataset, labelCountMetrics.maxLabelCount);
         return true;
     }
 
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/annotation/DefaultValue.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/annotation/DefaultValue.java
@@ -0,0 +1,30 @@
+/*
+ * 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.server.core.analysis.metrics.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.PARAMETER)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface DefaultValue {
+    String value();
+}

Original file line number	Diff line number	Diff line change
`@@ -67,4 +67,8 @@ public Argument getLastArgument() {`
`67`	`67`	`public Argument getNextFuncArg() {`
`68`	`68`	`return funcArgs.get(argGetIdx++);`
`69`	`69`	`}`
	`70`	`+`
	`71`	`+ public boolean hasNextArg() {`
	`72`	`+ return argGetIdx < funcArgs.size();`
	`73`	`+ }`
`70`	`74`	`}`