Add self obs metrics for L1/L2 queue and persistent cache. (apache#13405)

* 1.[Break Change] MQE function `sort_values` sorts according to the aggregation result and labels rather than the simple time series values.
2. Fix `MetricsPersistentWorker`, remove DataCarrier queue from `Hour/Day` dimensions metrics persistent process.
3. Self Observability: add `metrics_aggregation_queue_used_percentage` and `metrics_persistent_collection_cached_size` metrics for the OAP server.

Author: wankai123

docs/en/api/metrics-query-expression.md

@@ -440,19 +440,42 @@ TIME_SERIES_VALUES.
 
 ## Sort Operation
 ### SortValues Operation
-SortValues Operation takes an expression and sorts the values of the input expression result.
-
+SortValues Operation takes an expression used to sort and pick the top N label value groups, which according to
+the values of a given ExpressionResult and based on the specified order, limit and aggregation type.
+If the input expression is not a labeled result, it will retrurn the original expression result.
 Expression:
 ```text
-sort_values(Expression, <limit>, <order>)
+sort_values(Expression, <limit>, <order>, <aggregation_type>)
 ```
-- `limit` is the number of the sort results, should be a positive integer, if not specified, will return all results. Optional.
+- `limit` is the number of the sort results, should be a positive integer.
 - `order` is the order of the sort results. The value of `order` can be `asc` or `des`.
+-  `aggregation_type` is the type of the aggregation operation. The type can be `avg`, `sum`, `max`, `min`.
 
-For example:
-If we want to sort the `service_resp_time` metric values in descending order and get the top 10 values, we can use the following expression:
+For example, the following metrics in time series T1 and T2:
+```text
+T1:
+http_requests_total{service="api"}   160
+http_requests_total{service="web"}   120
+http_requests_total{service="auth"}  80
+
+T2:
+http_requests_total{service="api"}   100
+http_requests_total{service="web"}   180
+http_requests_total{service="auth"}  10
+```
+We can use SortValuesOp to pick the top 2 services with the most avg requests in descending order:
+```text
+sort_values(http_requests_total, 2, desc, avg)
+```
+The result will be:
 ```text
-sort_values(service_resp_time, 10, des)
+T1:
+http_requests_total{service="web"}   120
+http_requests_total{service="api"}   160
+
+T2:
+http_requests_total{service="web"}   180
+http_requests_total{service="api"}   100
 ```
 
 #### Result Type

docs/en/changes/changes.md

@@ -4,6 +4,8 @@
 
 * Bump up BanyanDB dependency version(server and java-client) to 0.9.0.
 * Fix CVE-2025-54057, restrict and validate url for widgets.
+* Fix `MetricsPersistentWorker`, remove DataCarrier queue from `Hour/Day` dimensions metrics persistent process.
+  This is important to reduce memory cost and `Hour/Day` dimensions metrics persistent latency.
 
 #### OAP Server
 
@@ -46,6 +48,8 @@
 * Add UI dashboard for Ruby runtime metrics.
 * Tracing Query Execution HTTP APIs: make the argument `service layer` optional.
 * GraphQL API: metadata, topology, log and trace support query by name.
+* [Break Change] MQE function `sort_values` sorts according to the aggregation result and labels rather than the simple time series values.
+* Self Observability: add `metrics_aggregation_queue_used_percentage` and `metrics_persistent_collection_cached_size` metrics for the OAP server.
 
 #### UI
 
@@ -63,6 +67,7 @@
 * Fix the snapshot charts unable to display.
 * Bump vue-i18n from 9.14.3 to 9.14.5.
 * Fix split queries for topology to avoid page crash.
+* Self Observability ui-template: Add new panels for monitor `metrics aggregation queue used percentage` and `metrics persistent collection cached size`.
 
 #### Documentation
 

oap-server/mqe-grammar/src/main/antlr4/org/apache/skywalking/mqe/rt/grammar/MQEParser.g4

@@ -38,7 +38,7 @@ expression
     | topNOf L_PAREN topN (COMMA topN)* COMMA INTEGER COMMA order R_PAREN           #topNOfOP
     | relabels L_PAREN expression COMMA label COMMA replaceLabel R_PAREN #relablesOP
     | aggregateLabels L_PAREN expression COMMA aggregateLabelsFunc R_PAREN #aggregateLabelsOp
-    | sort_values L_PAREN expression (COMMA INTEGER)? COMMA order R_PAREN #sortValuesOP
+    | sort_values L_PAREN expression COMMA INTEGER COMMA order COMMA aggregation R_PAREN #sortValuesOP
     | sort_label_values L_PAREN expression COMMA order COMMA labelNameList R_PAREN #sortLabelValuesOP
     | baseline L_PAREN metric COMMA baseline_type R_PAREN #baselineOP
     ;

oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/MQEVisitorBase.java

@@ -23,7 +23,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.function.BiFunction;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -437,12 +436,9 @@ public ExpressionResult visitSortValuesOP(MQEParser.SortValuesOPContext ctx) {
         try {
             ExpressionResult result = visit(ctx.expression());
             int order = ctx.order().getStart().getType();
-            Optional<Integer> limit = Optional.empty();
-            if (ctx.INTEGER() != null) {
-                limit = Optional.of(Integer.valueOf(ctx.INTEGER().getText()));
-            }
+            int limit = Integer.parseInt(ctx.INTEGER().getText());
             try {
-                return SortValuesOp.doSortValuesOp(result, limit, order);
+                return SortValuesOp.doSortValuesOp(result, limit, order, MQEParser.AVG);
             } catch (IllegalExpressionException e) {
                 return getErrorResult(e.getMessage());
             }

oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/operation/SortValuesOp.java

@@ -18,40 +18,98 @@
 
 package org.apache.skywalking.mqe.rt.operation;
 
+import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
-import java.util.Optional;
+import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.skywalking.mqe.rt.exception.IllegalExpressionException;
 import org.apache.skywalking.mqe.rt.grammar.MQEParser;
 import org.apache.skywalking.oap.server.core.query.mqe.ExpressionResult;
-import org.apache.skywalking.oap.server.core.query.mqe.MQEValue;
+import org.apache.skywalking.oap.server.core.query.mqe.ExpressionResultType;
+import org.apache.skywalking.oap.server.core.query.mqe.MQEValues;
+import org.apache.skywalking.oap.server.core.query.mqe.Metadata;
 
+/**
+ * When a result has multiple label value groups, it is often required to sort the values of these groups.
+ * SortValuesOp is used to sort and pick the top N label value groups, which according to
+ * the values of a given ExpressionResult and based on the specified order, limit and aggregation type.
+ *
+ * It first performs aggregation on the results, then sorts them according to the specified order (ascending or descending),
+ * and finally limits the number of results if a limit is provided.
+ *
+ * for example, the following metrics in time series T1 and T2:
+ * T1:
+ * http_requests_total{service="api"}   160
+ * http_requests_total{service="web"}   120
+ * http_requests_total{service="auth"}  80
+ *
+ * T2:
+ * http_requests_total{service="api"}   100
+ * http_requests_total{service="web"}   180
+ * http_requests_total{service="auth"}  10
+ *
+ * We can use SortValuesOp to pick the top 2 services with the most avg requests in descending order:
+ * `sort_values(http_requests_total, 2, desc, avg)`
+ * The result will be:
+ * T1:
+ * http_requests_total{service="web"}   120
+ * http_requests_total{service="api"}   160
+ *
+ * T2:
+ * http_requests_total{service="web"}   180
+ * http_requests_total{service="api"}   100
+ */
 public class SortValuesOp {
     public static ExpressionResult doSortValuesOp(ExpressionResult expResult,
-                                                 Optional<Integer> limit,
-                                                 int order) throws IllegalExpressionException {
-        if (MQEParser.ASC == order || MQEParser.DES == order) {
-            expResult.getResults().forEach(mqeValues -> {
-                List<MQEValue> values = mqeValues.getValues()
-                                                 .stream()
-                                                 // Filter out empty values
-                                                 .filter(mqeValue -> !mqeValue.isEmptyValue())
-                                                 .sorted(MQEParser.ASC == order ? Comparator.comparingDouble(
-                                                     MQEValue::getDoubleValue) :
-                                                             Comparator.comparingDouble(MQEValue::getDoubleValue)
-                                                                       .reversed())
-                                                 .collect(
-                                                     Collectors.toList());
-                if (limit.isPresent() && limit.get() < values.size()) {
-                    mqeValues.setValues(values.subList(0, limit.get()));
-                } else {
-                    mqeValues.setValues(values);
-                }
-            });
-        } else {
-            throw new IllegalExpressionException("Unsupported sort order.");
+                                                  int limit,
+                                                  int order,
+                                                  int aggregationType) throws IllegalExpressionException {
+        // no label result, no need to sort
+        if (!expResult.isLabeledResult()) {
+            return expResult;
         }
+        // store the original results in a map to avoid losing data during aggregation
+        Map<Metadata, MQEValues> resultMap = expResult.getResults()
+                                                      .stream()
+                                                      .collect(Collectors.toMap(
+                                                          MQEValues::getMetric, v -> {
+                                                              MQEValues newValues = new MQEValues();
+                                                              newValues.setMetric(v.getMetric());
+                                                              newValues.setValues(v.getValues());
+                                                              return newValues;
+                                                          }
+                                                      ));
+        // do aggregation first
+        ExpressionResult aggResult = AggregationOp.doAggregationOp(expResult, aggregationType);
+
+        List<MQEValues> sorted =
+            aggResult.getResults().stream()
+                     .sorted(getComparator(order))
+                     .collect(Collectors.toList());
+        if (limit < sorted.size()) {
+            sorted = sorted.subList(0, limit);
+        }
+        List<MQEValues> results = new ArrayList<>();
+        sorted.forEach(v -> {
+                           MQEValues mqeValues = resultMap.get(v.getMetric());
+                           if (mqeValues != null) {
+                               results.add(mqeValues);
+                           }
+                       }
+        );
+
+        expResult.setResults(results);
+        expResult.setType(ExpressionResultType.TIME_SERIES_VALUES);
         return expResult;
     }
+
+    private static Comparator<MQEValues> getComparator(int order) {
+        Comparator<MQEValues> comparator = Comparator.comparingDouble(
+            mqeValues -> mqeValues.getValues().isEmpty()
+                ? Double.NaN
+                : mqeValues.getValues().get(0).getDoubleValue()
+        );
+        return order == MQEParser.ASC ? comparator : comparator.reversed();
+    }
 }

oap-server/mqe-rt/src/test/java/org/apache/skywalking/mqe/rt/SortValuesOpTest.java

@@ -18,7 +18,6 @@
 
 package org.apache.skywalking.mqe.rt;
 
-import java.util.Optional;
 import org.apache.skywalking.mqe.rt.exception.IllegalExpressionException;
 import org.apache.skywalking.mqe.rt.grammar.MQEParser;
 import org.apache.skywalking.mqe.rt.operation.SortValuesOp;
@@ -32,35 +31,64 @@ public class SortValuesOpTest {
     @Test
     public void sortValueTest() throws IllegalExpressionException {
         //no label
-        ExpressionResult des = SortValuesOp.doSortValuesOp(mockData.newSeriesNoLabeledResult(), Optional.of(3),
-                                                          MQEParser.DES);
-        Assertions.assertEquals(300, des.getResults().get(0).getValues().get(0).getDoubleValue());
-        Assertions.assertEquals(100, des.getResults().get(0).getValues().get(1).getDoubleValue());
-        ExpressionResult asc = SortValuesOp.doSortValuesOp(mockData.newSeriesNoLabeledResult(), Optional.of(3),
-                                                          MQEParser.ASC);
+        ExpressionResult des = SortValuesOp.doSortValuesOp(
+            mockData.newSeriesNoLabeledResult(), 3,
+            MQEParser.DES, MQEParser.AVG
+        );
+        Assertions.assertEquals(100, des.getResults().get(0).getValues().get(0).getDoubleValue());
+        Assertions.assertEquals(300, des.getResults().get(0).getValues().get(1).getDoubleValue());
+        ExpressionResult asc = SortValuesOp.doSortValuesOp(
+            mockData.newSeriesNoLabeledResult(), 3,
+            MQEParser.ASC, MQEParser.AVG
+        );
         Assertions.assertEquals(100, asc.getResults().get(0).getValues().get(0).getDoubleValue());
         Assertions.assertEquals(300, asc.getResults().get(0).getValues().get(1).getDoubleValue());
 
         //labeled
-        ExpressionResult desLabeled = SortValuesOp.doSortValuesOp(mockData.newSeriesLabeledResult(), Optional.of(3),
-                                                                 MQEParser.DES);
-        Assertions.assertEquals(300, desLabeled.getResults().get(0).getValues().get(0).getDoubleValue());
-        Assertions.assertEquals(100, desLabeled.getResults().get(0).getValues().get(1).getDoubleValue());
-        Assertions.assertEquals(301, desLabeled.getResults().get(1).getValues().get(0).getDoubleValue());
-        Assertions.assertEquals(101, desLabeled.getResults().get(1).getValues().get(1).getDoubleValue());
-        ExpressionResult ascLabeled = SortValuesOp.doSortValuesOp(mockData.newSeriesLabeledResult(), Optional.of(2),
-                                                                 MQEParser.ASC);
+        ExpressionResult desLabeled = SortValuesOp.doSortValuesOp(
+            mockData.newSeriesLabeledResult(), 3,
+            MQEParser.DES, MQEParser.AVG
+        );
+        Assertions.assertEquals(101, desLabeled.getResults().get(0).getValues().get(0).getDoubleValue());
+        Assertions.assertEquals(301, desLabeled.getResults().get(0).getValues().get(1).getDoubleValue());
+        Assertions.assertEquals("label", desLabeled.getResults().get(0).getMetric().getLabels().get(0).getKey());
+        Assertions.assertEquals("2", desLabeled.getResults().get(0).getMetric().getLabels().get(0).getValue());
+        Assertions.assertEquals("label2", desLabeled.getResults().get(0).getMetric().getLabels().get(1).getKey());
+        Assertions.assertEquals("21", desLabeled.getResults().get(0).getMetric().getLabels().get(1).getValue());
+        Assertions.assertEquals(100, desLabeled.getResults().get(1).getValues().get(0).getDoubleValue());
+        Assertions.assertEquals(300, desLabeled.getResults().get(1).getValues().get(1).getDoubleValue());
+        Assertions.assertEquals("label", desLabeled.getResults().get(1).getMetric().getLabels().get(0).getKey());
+        Assertions.assertEquals("1", desLabeled.getResults().get(1).getMetric().getLabels().get(0).getValue());
+        Assertions.assertEquals("label2", desLabeled.getResults().get(1).getMetric().getLabels().get(1).getKey());
+        Assertions.assertEquals("21", desLabeled.getResults().get(1).getMetric().getLabels().get(1).getValue());
+
+        ExpressionResult ascLabeled = SortValuesOp.doSortValuesOp(
+            mockData.newSeriesLabeledResult(), 3,
+            MQEParser.ASC, MQEParser.AVG
+        );
         Assertions.assertEquals(100, ascLabeled.getResults().get(0).getValues().get(0).getDoubleValue());
         Assertions.assertEquals(300, ascLabeled.getResults().get(0).getValues().get(1).getDoubleValue());
+        Assertions.assertEquals("label", ascLabeled.getResults().get(0).getMetric().getLabels().get(0).getKey());
+        Assertions.assertEquals("1", ascLabeled.getResults().get(0).getMetric().getLabels().get(0).getValue());
+        Assertions.assertEquals("label2", ascLabeled.getResults().get(0).getMetric().getLabels().get(1).getKey());
+        Assertions.assertEquals("21", ascLabeled.getResults().get(0).getMetric().getLabels().get(1).getValue());
         Assertions.assertEquals(101, ascLabeled.getResults().get(1).getValues().get(0).getDoubleValue());
         Assertions.assertEquals(301, ascLabeled.getResults().get(1).getValues().get(1).getDoubleValue());
+        Assertions.assertEquals("label", ascLabeled.getResults().get(1).getMetric().getLabels().get(0).getKey());
+        Assertions.assertEquals("2", ascLabeled.getResults().get(1).getMetric().getLabels().get(0).getValue());
+        Assertions.assertEquals("label2", ascLabeled.getResults().get(1).getMetric().getLabels().get(1).getKey());
+        Assertions.assertEquals("21", ascLabeled.getResults().get(1).getMetric().getLabels().get(1).getValue());
 
         //limit
-        ExpressionResult desLabeledLimit = SortValuesOp.doSortValuesOp(mockData.newSeriesLabeledResult(), Optional.of(1),
-                                                                      MQEParser.DES);
-        Assertions.assertEquals(1, desLabeledLimit.getResults().get(0).getValues().size());
-        Assertions.assertEquals(1, desLabeledLimit.getResults().get(1).getValues().size());
-        Assertions.assertEquals(300, desLabeledLimit.getResults().get(0).getValues().get(0).getDoubleValue());
-        Assertions.assertEquals(301, desLabeledLimit.getResults().get(1).getValues().get(0).getDoubleValue());
+        ExpressionResult desLabeledLimit = SortValuesOp.doSortValuesOp(
+            mockData.newSeriesLabeledResult(), 1,
+            MQEParser.DES, MQEParser.AVG
+        );
+        Assertions.assertEquals(101, desLabeledLimit.getResults().get(0).getValues().get(0).getDoubleValue());
+        Assertions.assertEquals(301, desLabeledLimit.getResults().get(0).getValues().get(1).getDoubleValue());
+        Assertions.assertEquals("label", desLabeledLimit.getResults().get(0).getMetric().getLabels().get(0).getKey());
+        Assertions.assertEquals("2", desLabeledLimit.getResults().get(0).getMetric().getLabels().get(0).getValue());
+        Assertions.assertEquals("label2", desLabeledLimit.getResults().get(0).getMetric().getLabels().get(1).getKey());
+        Assertions.assertEquals("21", desLabeledLimit.getResults().get(0).getMetric().getLabels().get(1).getValue());
     }
 }