Baseline: Support query baseline with MQE and use in the Alarm Kernel. (#13024)

Author: wankai123

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

@@ -432,7 +432,7 @@ then the expression result is:
 V(T1)-V(T1-2), V(T2)-V(T1-1), V(T3)-V(T1)
 ```
 
-**Note**:
+**Notice**
 * If the calculated metric value is empty, the result will be empty. Assume in the T3 point, the increase value = V(T3)-V(T1), If the metric V(T3) or V(T1) is empty, the result value in T3 will be empty.
 
 ### Result Type
@@ -494,6 +494,34 @@ metric{label1='a', label2='2c'}
 metric{label1='a', label2='2a'}
 ```
 
+### Baseline Operation
+Baseline Operation takes an expression and gets the baseline predicted values of the input metric.
+
+Expression:
+```text
+baseline(Expression, <baseline_type>)
+```
+
+- `baseline_type` is the type of the baseline predicted value. The type can be `value`, `upper`, `lower`.
+
+for example:
+If we want to get the baseline predicted `upper` values of the `service_resp_time` metric, we can use the following expression:
+```text
+baseline(service_resp_time, upper)
+```
+
+**Notice**:
+- This feature is required to enable the `baseline module` and deploy a baseline service. And the baseline service should implement the protocol of the [baseline.proto](../../../oap-server/metrics-baseline/src/main/proto/baseline.proto).
+Otherwise, the result will be empty.
+- The baseline operation requires the relative metrics declared through baseline service.
+Otherwise, the result will be empty, which means there is no baseline or predicated value.
+- For now, the predictions aim to every hour.
+And the predicated values provided within this baseline are at a minute-level granularity. 
+As a result, for CPM(calls per minute), when the query step is `MINUTE` and duration is in a full hour, the returned values are same in every minute of this whole hour.
+
+### Result Type
+TIME_SERIES_VALUES.
+
 ## Expression Query Example
 ### Labeled Value Metrics
 ```text

docs/en/changes/changes.md

@@ -68,6 +68,7 @@
 * BaseLine: Support query baseline metrics names.
 * Add `Get Node List in the Cluster` API.
 * Add type descriptor when converting Envoy logs to JSON for persistence, to avoid conversion error.
+* Bseline: Support query baseline with MQE and use in the Alarm Rule.
 
 #### UI
 

docs/en/setup/backend/backend-alarm.md

@@ -125,6 +125,21 @@ Currently, metrics from the **Service**, **Service Instance**, **Endpoint**, **S
 
 Submit an issue or a pull request if you want to support any other scopes in Alarm.
 
+### Use the Baseline Predicted Value to trigger the Alarm
+Since 10.2.0, SkyWalking supports using the baseline predicted value in the alarm rule expression. 
+The MQE expression can refer to [Baseline Operation](../../api/metrics-query-expression.md#baseline-operation).
+
+For example, the following rule will compare the service response time with the baseline predicted value in each time bucket, and
+when the service response time is higher than the baseline predicted value in 3 minutes of the last 10 minutes, the alarm will be triggered.
+
+```yaml
+rules:
+  service_resp_time_rule:
+    expression: sum(service_resp_time > baseline(service_resp_time, upper)) > 3
+    period: 10
+    message: Service {name} response time is higher than the baseline predicted value in 3 minutes of last 10 minutes.
+```
+
 ## Hooks
 Hooks are a way to send alarm messages to the outside world. SkyWalking supports multiple hooks of the same type, each hook can support different configurations. 
 For example, you can configure two Slack hooks, one named `default` and set `is-default: true` means this hook will apply on all `Alarm Rules` **without config** `hooks`.

oap-server/metrics-baseline/src/main/java/org/apache/skywalking/oap/server/baseline/service/BaselineQueryService.java

@@ -18,19 +18,28 @@
 
 package org.apache.skywalking.oap.server.baseline.service;
 
+import java.util.Map;
 import org.apache.skywalking.oap.server.library.module.Service;
 
 import java.util.List;
 
 public interface BaselineQueryService extends Service {
     /**
      * query supported query baseline metrics names
+     *
      * @return
      */
     List<String> querySupportedMetrics();
 
     /**
-     * query predict metrics
+     * query predicted metrics
      */
-    List<PredictServiceMetrics> queryPredictMetrics(List<ServiceMetrics> serviceMetrics, long startTimeBucket, long endTimeBucket);
+    List<PredictServiceMetrics> queryPredictMetrics(List<ServiceMetrics> serviceMetrics,
+                                                    long startTimeBucket,
+                                                    long endTimeBucket);
+
+    /**
+     * query predicted metrics from cache, return all predicted metrics for the given service name and time bucket hour
+     */
+    Map<String, PredictServiceMetrics.PredictMetricsValue> queryPredictMetricsFromCache(String serviceName, String timeBucketHour);
 }

oap-server/metrics-baseline/src/main/java/org/apache/skywalking/oap/server/baseline/service/BaselineQueryServiceImpl.java

@@ -18,8 +18,13 @@
 
 package org.apache.skywalking.oap.server.baseline.service;
 
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import com.google.protobuf.Empty;
 import io.grpc.ManagedChannel;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.baseline.v3.AlarmBaselineMetricPrediction;
 import org.apache.skywalking.apm.baseline.v3.AlarmBaselineMetricsNames;
@@ -29,7 +34,11 @@
 import org.apache.skywalking.apm.baseline.v3.AlarmBaselineServiceMetricName;
 import org.apache.skywalking.apm.baseline.v3.KeyStringValuePair;
 import org.apache.skywalking.apm.baseline.v3.TimeBucketStep;
+import org.apache.skywalking.oap.server.core.Const;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
 import org.apache.skywalking.oap.server.library.client.grpc.GRPCClient;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 
 import java.util.ArrayList;
@@ -41,8 +50,12 @@
 @Slf4j
 public class BaselineQueryServiceImpl implements BaselineQueryService {
     private AlarmBaselineServiceGrpc.AlarmBaselineServiceBlockingStub stub;
+    private final Cache<String/*timeBucket,serviceName*/, Map<String/*metricName*/, PredictServiceMetrics.PredictMetricsValue>> baselineCache;
 
     public BaselineQueryServiceImpl(String addr, int port) {
+        this.baselineCache = CacheBuilder.newBuilder()
+                                         .expireAfterAccess(1, TimeUnit.HOURS)
+                                         .build();
         if (StringUtil.isEmpty(addr) || port <= 0) {
             return;
         }
@@ -55,6 +68,7 @@ public BaselineQueryServiceImpl(String addr, int port) {
     @Override
     public List<String> querySupportedMetrics() {
         if (stub == null) {
+            log.warn("Baseline service is not set up, return empty list.");
             return Collections.emptyList();
         }
 
@@ -66,6 +80,7 @@ public List<String> querySupportedMetrics() {
 
     public List<PredictServiceMetrics> queryPredictMetrics(List<ServiceMetrics> serviceMetrics, long startTimeBucket, long endTimeBucket) {
         if (stub == null) {
+            log.warn("Baseline service is not set up, return empty baseline values.");
             return Collections.emptyList();
         }
 
@@ -77,6 +92,58 @@ public List<PredictServiceMetrics> queryPredictMetrics(List<ServiceMetrics> serv
         return Collections.emptyList();
     }
 
+    public Map<String, PredictServiceMetrics.PredictMetricsValue> queryPredictMetricsFromCache(String serviceName,
+                                                                                               String timeBucketHour) {
+        if (stub == null) {
+            log.warn("Baseline service is not set up, return empty baseline values.");
+            return Collections.emptyMap();
+        }
+        String key = timeBucketHour + Const.COMMA + serviceName;
+        Map<String, PredictServiceMetrics.PredictMetricsValue> baselineValues = this.baselineCache.asMap().get(key);
+
+        if (CollectionUtils.isNotEmpty(baselineValues)) {
+            return baselineValues;
+        }
+        //reload all metrics and timeBucket baseline values for this service
+        List<String> metrics = querySupportedMetrics();
+        ServiceMetrics serviceMetrics = ServiceMetrics.builder()
+                                                      .serviceName(serviceName)
+                                                      .metricsNames(metrics)
+                                                      .build();
+        //todo: need config?
+        long startTimeBucket = TimeBucket.getTimeBucket(
+            System.currentTimeMillis() - TimeUnit.HOURS.toMillis(24), DownSampling.Hour);
+        long endTimeBucket = TimeBucket.getTimeBucket(
+            System.currentTimeMillis() + TimeUnit.HOURS.toMillis(24), DownSampling.Hour);
+        List<PredictServiceMetrics> predictServiceMetricsList = queryPredictMetrics(
+            Collections.singletonList(serviceMetrics), startTimeBucket, endTimeBucket);
+        if (CollectionUtils.isEmpty(predictServiceMetricsList)) {
+            return Collections.emptyMap();
+        }
+        for (String metricName : metrics) {
+            for (PredictServiceMetrics predictServiceMetrics : predictServiceMetricsList) {
+                List<PredictServiceMetrics.PredictMetricsValue> predictMetricsValues = predictServiceMetrics.getMetricsValues()
+                                                                                                            .get(
+                                                                                                                metricName);
+                if (CollectionUtils.isEmpty(predictMetricsValues)) {
+                    continue;
+                }
+                for (PredictServiceMetrics.PredictMetricsValue predictMetricsValue : predictMetricsValues) {
+                    if (predictMetricsValue == null) {
+                        continue;
+                    }
+                    this.baselineCache.asMap()
+                                      .computeIfAbsent(
+                                          predictMetricsValue.getTimeBucket() + Const.COMMA + serviceName,
+                                          k -> new HashMap<>()
+                                      )
+                                      .put(metricName, predictMetricsValue);
+                }
+            }
+        }
+        return this.baselineCache.asMap().getOrDefault(key, Collections.emptyMap());
+    }
+
     private List<PredictServiceMetrics> queryPredictMetrics0(List<ServiceMetrics> serviceMetrics, long startTimeBucket, long endTimeBucket) {
         // building request
         final AlarmBaselineRequest.Builder request = AlarmBaselineRequest.newBuilder();

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

@@ -98,6 +98,11 @@ ATTR3: 'attr3';
 ATTR4: 'attr4';
 ATTR5: 'attr5';
 
+BASELINE: 'baseline';
+VALUE: 'value';
+UPPER: 'upper';
+LOWER: 'lower';
+
 // Literals
 INTEGER: Digit+;
 DECIMAL: Digit+ DOT Digit+;

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

@@ -40,6 +40,7 @@ expression
     | aggregateLabels L_PAREN expression COMMA aggregateLabelsFunc R_PAREN #aggregateLabelsOp
     | sort_values L_PAREN expression (COMMA INTEGER)? COMMA order 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
     ;
 
 expressionList
@@ -110,3 +111,6 @@ attributeName:
     ATTR0 | ATTR1 | ATTR2 | ATTR3 | ATTR4 | ATTR5;
 attribute: attributeName (EQ | NEQ) VALUE_STRING;
 attributeList: attribute (COMMA attribute)*;
+
+baseline: BASELINE;
+baseline_type: VALUE | UPPER | LOWER;

oap-server/mqe-rt/pom.xml

@@ -44,5 +44,10 @@
             <artifactId>mqe-grammar</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>metrics-baseline</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>