Repository: apache/skywalking (language: python)

Issue:
[Breaking Change] [Feature] Accept MQE to replace the old hard-coded alerting rules.
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.


### Description

Alerting core had been there for several years without a significant update, ref https://skywalking.apache.org/docs/main/v9.5.0/en/setup/backend/backend-alarm/. It is based on a shifted time window for specific metrics of given entities. Last year, @kezhenxu94 added composite rules to support alarms triggered by multiple metrics of the same entity.

This year, we have [MQE](https://skywalking.apache.org/docs/main/v9.5.0/en/api/metrics-query-expression/) with [v3 metrics query APIs](https://skywalking.apache.org/docs/main/v9.5.0/en/api/query-protocol/#v3-apis). 

So, @wankai123 and I were trying to immigrate the same query grammar to define query, which would provide a more flexible rule definition. Also, as the calculation capabilities of MQE, the multiple metrics-based alerting would be supported automatically as well.

BTW, as [new hook mechanism](https://skywalking.apache.org/docs/main/next/en/setup/backend/backend-alarm/#hooks) is there since 9.6.0(next release), the whole alerting kernel would be rewritten.

### Use case

We have been working on this direction for weeks. @wankai123 is closing to finish all features and the replacement.

### Related issues

_No response_

### Are you willing to submit a pull request to implement this on your own?

- [ ] Yes I am willing to submit a pull request on my own!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)



Discussion / hints:

Fix patch:
diff --git a/dist-material/alarm-settings.yml b/dist-material/alarm-settings.yml
index d955a6f08153..ce55ab88e826 100644
--- a/dist-material/alarm-settings.yml
+++ b/dist-material/alarm-settings.yml
@@ -18,65 +18,48 @@
 rules:
   # Rule unique name, must be ended with `_rule`.
   service_resp_time_rule:
-    metrics-name: service_resp_time
-    op: ">"
-    threshold: 1000
+    # A MQE expression, the result type must be `SINGLE_VALUE` and the root operation of the expression must be a Compare Operation
+    # which provides `1`(true) or `0`(false) result. When the result is `1`(true), the alarm will be triggered.
+    expression: sum(service_resp_time > 1000) >= 3
     period: 10
-    count: 3
     silence-period: 5
     message: Response time of service {name} is more than 1000ms in 3 minutes of last 10 minutes.
+#  service_resp_time_rule:
+#    expression: avg(service_resp_time) > 1000
+#    period: 10
+#    silence-period: 5
+#    message: Avg response time of service {name} is more than 1000ms in last 10 minutes.
   service_sla_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: service_sla
-    op: "<"
-    threshold: 8000
+    expression: sum(service_sla < 8000) >= 2
     # The length of time to evaluate the metrics
     period: 10
-    # How many times after the metrics match the condition, will trigger alarm
-    count: 2
     # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
     silence-period: 3
     message: Successful rate of service {name} is lower than 80% in 2 minutes of last 10 minutes
   service_resp_time_percentile_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: service_percentile
-    op: ">"
-    threshold: 1000,1000,1000,1000,1000
+    expression: sum(service_percentile{_='0,1,2,3,4'} > 1000) >= 3
     period: 10
-    count: 3
     silence-period: 5
     message: Percentile response time of service {name} alarm in 3 minutes of last 10 minutes, due to more than one condition of p50 > 1000, p75 > 1000, p90 > 1000, p95 > 1000, p99 > 1000
   service_instance_resp_time_rule:
-    metrics-name: service_instance_resp_time
-    op: ">"
-    threshold: 1000
+    expression: sum(service_instance_resp_time > 1000) >= 2
     period: 10
-    count: 2
     silence-period: 5
     message: Response time of service instance {name} is more than 1000ms in 2 minutes of last 10 minutes
   database_access_resp_time_rule:
-    metrics-name: database_access_resp_time
-    threshold: 1000
-    op: ">"
+    expression: sum(database_access_resp_time > 1000) >= 2
     period: 10
-    count: 2
     message: Response time of database access {name} is more than 1000ms in 2 minutes of last 10 minutes
   endpoint_relation_resp_time_rule:
-    metrics-name: endpoint_relation_resp_time
-    threshold: 1000
-    op: ">"
+    expression: sum(endpoint_relation_resp_time > 1000) >= 2
     period: 10
-    count: 2
     message: Response time of endpoint relation {name} is more than 1000ms in 2 minutes of last 10 minutes
 #  Active endpoint related metrics alarm will cost more memory than service and service instance metrics alarm.
 #  Because the number of endpoint is much more than service and instance.
 #
 #  endpoint_resp_time_rule:
-#    metrics-name: endpoint_resp_time
-#    op: ">"
-#    threshold: 1000
+#    expression: sum(endpoint_resp_time > 1000) >= 2
 #    period: 10
-#    count: 2
 #    silence-period: 5
 #    message: Response time of endpoint {name} is more than 1000ms in 2 minutes of last 10 minutes
 
diff --git a/dist-material/config-examples/alarm-settings.yml b/dist-material/config-examples/alarm-settings.yml
index c8bbab35ef46..aa7d7bfa823a 100644
--- a/dist-material/config-examples/alarm-settings.yml
+++ b/dist-material/config-examples/alarm-settings.yml
@@ -18,39 +18,28 @@
 rules:
   # Rule unique name, must be ended with `_rule`.
   endpoint_percent_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: endpoint_percent
-    threshold: 75
-    op: <
+    expression: sum((endpoint_sla / 100) < 75) >= 3
     # The length of time to evaluate the metrics
     period: 10
-    # How many times after the metrics match the condition, will trigger alarm
-    count: 3
     # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
     silence-period: 10
     message: Successful rate of endpoint {name} is lower than 75%
     tags:
       level: WARNING
   service_resp_time_rule:
-    metrics-name: service_resp_time
+    expression: sum(service_resp_time > 1000) >= 1
     # [Optional] Default, match all services in this metrics
     include-names:
       - dubbox-provider
       - dubbox-consumer
-    threshold: 1000
-    op: ">"
     period: 10
-    count: 1
     tags:
       level: CRITICAL
   service_instance_resp_time_rule:
-    metrics-name: service_instance_resp_time
-    op: ">"
-    threshold: 1000
+    expression: sum(service_instance_resp_time > 1000) >= 2
     # [Optional] Default, match all services in this metrics
     include-names-regex: instance\_\d+
     period: 10
-    count: 2
     silence-period: 5
     message: Response time of service instance {name} is more than 1000ms in 2 minutes of last 10 minutes
 
diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE
index 55b78a171165..ef6b5329ee3d 100644
--- a/dist-material/release-docs/LICENSE
+++ b/dist-material/release-docs/LICENSE
@@ -364,7 +364,6 @@ The text of each license is the standard Apache 2.0 license.
     https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-reactive/1.6.4 Apache-2.0
     https://mvnrepository.com/artifact/org.jetbrains/annotations/13.0 Apache-2.0
     https://mvnrepository.com/artifact/org.lz4/lz4-java/1.7.1 Apache-2.0
-    https://mvnrepository.com/artifact/org.mvel/mvel2/2.4.8.Final Apache-2.0
     https://mvnrepository.com/artifact/org.slf4j/jcl-over-slf4j/1.7.30 Apache-2.0
     https://mvnrepository.com/artifact/org.slf4j/log4j-over-slf4j/1.7.30 Apache-2.0
     https://mvnrepository.com/artifact/org.slf4j/slf4j-api/1.7.30 Apache-2.0
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index e69a17bc9ae4..ed3be1e5ca3a 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -74,6 +74,8 @@
 * Fix the Kubernetes resource cache not refreshed.
 * Fix wrong classpath that might cause OOM in startup.
 * Enhance the `serviceRelation` in MAL by adding settings for the `delimiter` and `component` fields. 
+* [Breaking change] Support MQE in the Alerting. The Alarm Rules configuration(alarm-settings.yml), 
+  add `expression` field and remove `metrics-name/count/threshold/op/only-as-condition` fields and remove `composite-rules` configuration.
 
 #### UI
 
diff --git a/docs/en/setup/backend/backend-alarm.md b/docs/en/setup/backend/backend-alarm.md
index 6e81eeba2417..e42b254181c4 100644
--- a/docs/en/setup/backend/backend-alarm.md
+++ b/docs/en/setup/backend/backend-alarm.md
@@ -18,34 +18,22 @@ Defines the relation between scope and entity name.
 - **Endpoint Relation**: {Source endpoint name} in {Source Service name} to {Dest endpoint name} in {Dest service name}
 
 ## Rules
-**There are two types of rules: individual rules and composite rules. A composite rule is a combination of individual rules.**
-### Individual rules
 An alerting rule is made up of the following elements:
 - **Rule name**. A unique name shown in the alarm message. It must end with `_rule`.
-- **Metrics name**. This is also the metrics name in the OAL script. Only long, double, int types are supported. See the
-[list of all potential metrics name](#list-of-all-potential-metrics-name). Events can also be configured as the source
-of Alarm. Please refer to [the event doc](../../concepts-and-designs/event.md) for more details.
+- **Expression**. A [MQE](../../api/metrics-query-expression.md) expression that defines the conditions of the rule.
+The result type must be `SINGLE_VALUE` and the root operation of the expression must be a [Compare Operation](../../api/metrics-query-expression.md#compare-operation) which provides `1`(true) or `0`(false) result.
+When the result is `1`(true), the alarm will be triggered.
+For example, `avg(service_resp_time / 1000) > 1` is a valid expression to indicate the request latency is slower than 1s. The typical illegal expressions are
+    - `avg(service_resp_time > 1000) + 1` expression doesn't use `compare operation`
+    - `service_resp_time > 1000` expression return a `TIME_SERIES_VALUES` type of values rather than a `SINGLE_VALUE` value.
+
+The metrics names in the expression could be found in the [list of all potential metrics name](#list-of-all-potential-metrics-name) doc.
 - **Include names**. Entity names that are included in this rule. Please follow the [entity name definitions](#entity-name).
 - **Exclude names**. Entity names that are excluded from this rule. Please follow the [entity name definitions](#entity-name).
 - **Include names regex**. A regex that includes entity names. If both include-name list and include-name regex are set, both rules will take effect.
 - **Exclude names regex**. A regex that excludes entity names. Both rules will take effect if both include-label list and include-label regex are set.
-- **Include labels**. Metric labels that are included in this rule.
-- **Exclude labels**. Metric labels that are excluded from this rule.
-- **Include labels regex**. A regex that includes labels. If both include-label list and include-label regex are set, both rules will take effect.
-- **Exclude labels regex**. A regex that excludes labels. Both rules will take effect if both exclude-label list and exclude-label regex are set.
 - **Tags**. Tags are key/value pairs that are attached to alarms. Tags are used to specify distinguishing attributes of alarms that are meaningful and relevant to users. If you want to make these tags searchable on the SkyWalking UI, you may set the tag keys in `core/default/searchableAlarmTags` or through the system environment variable `SW_SEARCHABLE_ALARM_TAG_KEYS`. The key `level` is supported by default.
-
-*Label settings are required by the meter system. They are used to store metrics from the label-system platform, such as Prometheus, Micrometer, etc.
-The four label settings mentioned above must implement `LabeledValueHolder`.*
-
-- **Threshold**. The target value. 
-For multiple-value metrics, such as **percentile**, the threshold is an array. It is described as:  `value1, value2, value3, value4, value5`.
-Each value may serve as the threshold for each value of the metrics. Set the value to `-` if you do not wish to trigger the Alarm by one or more of the values.  
-For example, in **percentile**, `value1` is the threshold of P50, and `-, -, value3, value4, value5` means that there is no threshold for P50 and P75 in the percentile alarm rule.
-- **OP**. The operator. It supports `>`, `>=`, `<`, `<=`, `==`, `!=`. We welcome contributions of all OPs.
 - **Period**. The size of metrics cache in minutes for checking the alarm conditions. This is a time window that corresponds to the backend deployment env time.
-- **Count**. Within a period window, if the number of times which **value** goes over the threshold (based on OP) reaches `count`, then an alarm will be sent.
-- **Only as condition**. Indicates if the rule can send notifications or if it simply serves as a condition of the composite rule.
 - **Hooks**. Binding the specific names of the hooks when the alarm is triggered.
   The name format is `{hookType}.{hookName}` (slack.custom1 e.g.) and must be defined in the `hooks` section of the `alarm-settings.yml` file.
   If the hook name is not specified, the global hook will be used.
@@ -60,83 +48,54 @@ Such as for a metric, there is a shifting window as following at T7.
 
 * `Period`(Time point T1 ~ T7) are continuous data points for minutes. Notice, alerts are not supported above minute-by-minute periods as they would not be efficient.
 * Values(Value1 ~ Value7) are the values or labeled values for every time point.
-* `Count`'s value(N) represents there are N values in the window matched the operator and threshold.
-* In every minute, the window would shift automatically. At T8, Value8 would be cached, and T1/Value1 would be removed from the window. 
+* `Expression` is calculated based on the metric values(Value1 ~ Value7). 
+For example, expression `avg(service_resp_time) > 1000`, if the value are `1001, 1001, 1001, 1001, 1001, 1001, 1001`, 
+the calculation is `((1001 + 10001 + ... + 1001) / 7) > 1000` and the result would be `1`(true). Then the alarm would be triggered.
+* In every minute, the window would shift automatically. At T8, Value8 would be cached, and T1/Value1 would be removed from the window.
 
-### Composite rules
-**NOTE**: Composite rules are only applicable to alerting rules targeting the same entity level, such as service-level alarm rules (`service_percent_rule && service_resp_time_percentile_rule`). Do not compose alarm rules of different entity levels, such as an alarm rule of the service metrics with another rule of the endpoint metrics.
-
-A composite rule is made up of the following elements:
-- **Rule name**. A unique name shown in the alarm message. Must end with `_rule`.
-- **Expression**. Specifies how to compose rules, and supports `&&`, `||`, and `()`.
-- **Message**. The notification message to be sent out when the rule is triggered.
-- **Tags**. Tags are key/value pairs that are attached to alarms. Tags are used to specify distinguishing attributes of alarms that are meaningful and relevant to users.
-- **Hooks**. Binding the specific names of the hooks when the alarm is triggered.
-  The name format is `{hookType}.{hookName}` (slack.custom1 e.g.) and must be defined in the `hooks` section of the `alarm-settings.yml` file.
-  If the hook name is not specified, the global hooks will be used.
+Notice: If the expression include labeled metrics and result has multiple labeled value(e.g. `sum(service_percentile{_='0,1'} > 1000) >= 3`), the alarm will be triggered if any of the labeled value result matches 3 times of the condition(P50 > 1000 or P75 > 1000).
 
 ```yaml
 rules:
   # Rule unique name, must be ended with `_rule`.
   endpoint_percent_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: endpoint_percent
-    threshold: 75
-    op: <
+    # A MQE expression and the root operation of the expression must be a Compare Operation.
+    expression: sum((endpoint_sla / 100) < 75) >= 3
     # The length of time to evaluate the metrics
     period: 10
-    # How many times after the metrics match the condition, will trigger alarm
-    count: 3
     # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
     silence-period: 10
-    # Specify if the rule can send notification or just as an condition of composite rule
-    only-as-condition: false
+    message: Successful rate of endpoint {name} is lower than 75%
     tags:
       level: WARNING
   service_percent_rule:
-    metrics-name: service_percent
+    expression: sum((service_sla / 100) < 85) >= 4
     # [Optional] Default, match all services in this metrics
     include-names:
       - service_a
       - service_b
     exclude-names:
       - service_c
-    # Single value metrics threshold.
-    threshold: 85
-    op: <
     period: 10
-    count: 4
-    only-as-condition: false
+    message: Service {name} successful rate is less than 85%
   service_resp_time_percentile_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: service_percentile
-    op: ">"
-    # Multiple value metrics threshold. Thresholds for P50, P75, P90, P95, P99.
-    threshold: 1000,1000,1000,1000,1000
+    expression: sum(service_percentile{_='0,1,2,3,4'} > 1000) >= 3
     period: 10
-    count: 3
     silence-period: 5
     message: Percentile response time of service {name} alarm in 3 minutes of last 10 minutes, due to more than one condition of p50 > 1000, p75 > 1000, p90 > 1000, p95 > 1000, p99 > 1000
-    only-as-condition: false
   meter_service_status_code_rule:
-    metrics-name: meter_status_code
-    exclude-labels:
-      - "200"
-    op: ">"
-    threshold: 10
+    expression: sum(aggregate_labels(meter_status_code{_='4xx,5xx'},sum) > 10) > 3
     period: 10
     count: 3
     silence-period: 5
-    message: The request number of entity {name} non-200 status is more than expected.
-    only-as-condition: false
+    message: The request number of entity {name} 4xx and 5xx status is more than expected.
     hooks:
       - "slack.custom1"
       - "pagerduty.custom1"
-composite-rules:
   comp_rule:
-    # Must satisfied percent rule and resp time rule 
-    expression: service_percent_rule && service_resp_time_percentile_rule
-    message: Service {name} successful rate is less than 80% and P50 of response time is over 1000ms
+    expression: (avg(service_sla / 100) > 80) * (avg(service_percentile{_='0'}) > 1000) == 1
+    period: 10
+    message: Service {name} avg successful rate is less than 80% and P50 of avg response time is over 1000ms in last 10 minutes.
     tags:
       level: CRITICAL
     hooks:
@@ -157,9 +116,7 @@ For convenience's sake, we have provided a default `alarm-setting.yml` in our re
 1. Endpoint relation average response time over 1s in the last 2 minutes.
 
 ### List of all potential metrics name
-The metrics names are defined in the official [OAL scripts](../../guides/backend-oal-scripts.md) and
-[MAL scripts](../../concepts-and-designs/mal.md), the [Event](../../concepts-and-designs/event.md) names can also serve
-as the metrics names, all possible event names can be also found in [the Event doc](../../concepts-and-designs/event.md).
+The metrics names are defined in the official [OAL scripts](../../guides/backend-oal-scripts.md) and [MAL scripts](../../concepts-and-designs/mal.md).
 
 Currently, metrics from the **Service**, **Service Instance**, **Endpoint**, **Service Relation**, **Service Instance Relation**, **Endpoint Relation** scopes could be used in Alarm, and the **Database access** scope is the same as **Service**.
 
@@ -410,27 +367,20 @@ Since 6.5.0, the alerting settings can be updated dynamically at runtime by [Dyn
 which will override the settings in `alarm-settings.yml`.
 
 In order to determine whether an alerting rule is triggered or not, SkyWalking needs to cache the metrics of a time window for
-each alerting rule. If any attribute (`metrics-name`, `op`, `threshold`, `period`, `count`, etc.) of a rule is changed,
+each alerting rule. If any attribute (`expression`, `period`, etc.) of a rule is changed,
 the sliding window will be destroyed and re-created, causing the Alarm of this specific rule to restart again.
 
 ### Keys with data types of alerting rule configuration file
 
 | Alerting element     | Configuration property key | Type           | Description        |
 |----------------------|----------------------------|----------------|--------------------|
+| Expression           | expression                 | string         | MQE expression     |
 | Include names        | include-names              | string array   |                    | 
 | Exclude names        | exclude-names              | string array   |                    | 
 | Include names regex  | include-names-regex        | string         | Java regex Pattern |
 | Exclude names regex  | exclude-names-regex        | string         | Java regex Pattern |
-| Include labels       | include-labels             | string array   |                    |
-| Exclude labels       | exclude-labels             | string array   |                    |
-| Include labels regex | include-labels-regex       | string         | Java regex Pattern |
-| Exclude labels regex | exclude-labels-regex       | string         | Java regex Pattern |
 | Tags                 | tags                       | key-value pair |                    |
-| Threshold            | threshold                  | number         |                    |
-| OP                   | op                         | operator       | example: `>`, `>=` |
 | Period               | Period                     | int            |                    |
-| Count                | count                      | int            |                    |
-| Only as condition    | only-as-condition          | boolean        |                    |
 | Silence period       | silence-period             | int            |                    |
 | Message              | message                    | string         |                    |
 | Hooks                | hooks                      | string array   |                    |
diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml
index b2212ca80169..65932d8ebd3d 100644
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@ -491,12 +491,6 @@
                 <artifactId>groovy</artifactId>
                 <version>${groovy.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.mvel</groupId>
-                <artifactId>mvel2</artifactId>
-                <version>${mvel.version}</version>
-            </dependency>
-
             <dependency>
                 <groupId>commons-beanutils</groupId>
                 <artifactId>commons-beanutils</artifactId>
diff --git a/oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/MQEVisitorBase.java b/oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/MQEVisitorBase.java
index 817d29d757da..2f272dc6c015 100644
--- a/oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/MQEVisitorBase.java
+++ b/oap-server/mqe-rt/src/main/java/org/apache/skywalking/mqe/rt/MQEVisitorBase.java
@@ -40,6 +40,8 @@
 
 @Slf4j
 public abstract class MQEVisitorBase extends MQEParserBaseVisitor<ExpressionResult> {
+    public final static String GENERAL_LABEL_NAME = "_";
+
     @Override
     public ExpressionResult visitParensOp(MQEParser.ParensOpContext ctx) {
         return visit(ctx.expression());
diff --git a/oap-server/server-alarm-plugin/pom.xml b/oap-server/server-alarm-plugin/pom.xml
index e9b4a1d1cf88..355da1030886 100644
--- a/oap-server/server-alarm-plugin/pom.xml
+++ b/oap-server/server-alarm-plugin/pom.xml
@@ -43,15 +43,16 @@
             <artifactId>event-analyzer</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>mqe-rt</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-testing</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.mvel</groupId>
-            <artifactId>mvel2</artifactId>
-        </dependency>
         <dependency>
             <groupId>com.linecorp.armeria</groupId>
             <artifactId>armeria-junit5</artifactId>
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java
index 5c5bc3da81f8..5bd2661abafe 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmCore.java
@@ -18,6 +18,8 @@
 
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
+import java.util.Map;
+import java.util.Set;
 import org.apache.skywalking.oap.server.core.alarm.AlarmCallback;
 import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
 import org.joda.time.LocalDateTime;
@@ -29,7 +31,6 @@
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 /**
  * Alarm core includes metrics values in certain time windows based on alarm settings. By using its internal timer
@@ -45,8 +46,23 @@ public class AlarmCore {
         this.alarmRulesWatcher = alarmRulesWatcher;
     }
 
+    /**
+     * Find the running rules by the metrics name.
+     *
+     * @param metricsName to be found
+     * @return the matched running rule list, or null if not found.
+     */
     public List<RunningRule> findRunningRule(String metricsName) {
-        return alarmRulesWatcher.getRunningContext().get(metricsName);
+        List<RunningRule> runningRules = new ArrayList<>();
+        for (Map.Entry<String, Set<String>> entry : alarmRulesWatcher.getExprMetricsMap().entrySet()) {
+            if (entry.getValue().contains(metricsName)) {
+                List<RunningRule> found = alarmRulesWatcher.getRunningContext().get(entry.getKey());
+                if (found != null) {
+                    runningRules.addAll(found);
+                }
+            }
+        }
+        return runningRules.size() > 0 ? runningRules : null;
     }
 
     public void start(List<AlarmCallback> allCallbacks) {
@@ -76,15 +92,8 @@ public void start(List<AlarmCallback> allCallbacks) {
                 }
 
                 if (!alarmMessageList.isEmpty()) {
-                    if (!alarmRulesWatcher.getCompositeRules().isEmpty()) {
-                        List<AlarmMessage> messages = alarmRulesWatcher.getCompositeRuleEvaluator().evaluate(alarmRulesWatcher.getCompositeRules(), alarmMessageList);
-                        alarmMessageList.addAll(messages);
-                    }
-                    List<AlarmMessage> filteredMessages = alarmMessageList.stream().filter(msg -> !msg.isOnlyAsCondition()).collect(Collectors.toList());
-                    if (!filteredMessages.isEmpty()) {
-                        for (AlarmCallback callback : allCallbacks) {
-                            callback.doAlarm(filteredMessages);
-                        }
+                    for (AlarmCallback callback : allCallbacks) {
+                        callback.doAlarm(alarmMessageList);
                     }
                 }
             } catch (Exception e) {
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/MetricsValueType.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmEntity.java
similarity index 72%
rename from oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/MetricsValueType.java
rename to oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmEntity.java
index 23ad11528fd2..71276829b96c 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/MetricsValueType.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmEntity.java
@@ -18,6 +18,17 @@
 
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
-public enum MetricsValueType {
-    LONG, INT, DOUBLE, LABELED_LONG, MULTI_INTS
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Getter
+@EqualsAndHashCode
+@RequiredArgsConstructor
+public class AlarmEntity {
+    private final String scope;
+    private final int scopeId;
+    private final String name;
+    private final String id0;
+    private final String id1;
 }
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatter.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatter.java
index 1c64549347f8..87e5dc5fa381 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatter.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatter.java
@@ -20,7 +20,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import org.apache.skywalking.oap.server.core.alarm.MetaInAlarm;
 
 /**
  * This is a formatter especially for alarm message.
@@ -72,17 +71,17 @@ public AlarmMessageFormatter(String format) {
         while (match);
     }
 
-    public String format(MetaInAlarm meta) {
+    public String format(AlarmEntity alarmEntity) {
         StringBuilder message = new StringBuilder();
         for (int i = 0; i < formatSegments.size(); i++) {
             message.append(formatSegments.get(i));
             if (i != formatSegments.size() - 1) {
                 switch (valueFroms.get(i)) {
                     case ID:
-                        message.append(meta.getId0());
+                        message.append(alarmEntity.getId0());
                         break;
                     case NAME:
-                        message.append(meta.getName());
+                        message.append(alarmEntity.getName());
                 }
             }
         }
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
index 580cf7c2b57e..dd83ec26e3a5 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProvider.java
@@ -54,19 +54,8 @@ public ConfigCreator newConfigCreator() {
 
     @Override
     public void prepare() throws ServiceNotProvidedException, ModuleStartException {
-        Reader applicationReader;
-        try {
-            applicationReader = ResourceUtils.read("alarm-settings.yml");
-        } catch (FileNotFoundException e) {
-            throw new ModuleStartException("can't load alarm-settings.yml", e);
-        }
-        RulesReader reader = new RulesReader(applicationReader);
-        Rules rules = reader.readRules();
-
-        alarmRulesWatcher = new AlarmRulesWatcher(rules, this);
-
+        alarmRulesWatcher = new AlarmRulesWatcher(new Rules(), this);
         notifyHandler = new NotifyHandler(alarmRulesWatcher, getManager());
-        notifyHandler.init(new AlarmStandardPersistence(getManager()));
         this.registerServiceImplementation(MetricsNotify.class, notifyHandler);
     }
 
@@ -81,6 +70,16 @@ public void start() throws ServiceNotProvidedException, ModuleStartException {
 
     @Override
     public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
+        Reader applicationReader;
+        try {
+            applicationReader = ResourceUtils.read("alarm-settings.yml");
+        } catch (FileNotFoundException e) {
+            throw new ModuleStartException("can't load alarm-settings.yml", e);
+        }
+        RulesReader reader = new RulesReader(applicationReader);
+        Rules rules = reader.readRules();
+        alarmRulesWatcher.notify(rules);
+        notifyHandler.init(new AlarmStandardPersistence(getManager()));
     }
 
     @Override
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRule.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRule.java
index 0da0698ea323..02c8c78e4903 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRule.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRule.java
@@ -20,40 +20,92 @@
 
 import java.util.ArrayList;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
+import lombok.AccessLevel;
+import lombok.Data;
 import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.misc.ParseCancellationException;
+import org.antlr.v4.runtime.tree.ParseTree;
+import org.apache.skywalking.mqe.rt.exception.ParseErrorListener;
+import org.apache.skywalking.mqe.rt.grammar.MQELexer;
+import org.apache.skywalking.mqe.rt.grammar.MQEParser;
+import org.apache.skywalking.mqe.rt.type.ExpressionResult;
+import org.apache.skywalking.mqe.rt.type.ExpressionResultType;
+import org.apache.skywalking.oap.server.core.alarm.provider.expr.rt.AlarmMQEVerifyVisitor;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
 
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-@Setter
-@Getter
+@Data
 @ToString
 @EqualsAndHashCode
 public class AlarmRule {
     private String alarmRuleName;
-    private String metricsName;
+    private String expression;
+    @Setter(AccessLevel.NONE)
+    private Set<String> includeMetrics;
     private ArrayList<String> includeNames;
     private String includeNamesRegex;
     private ArrayList<String> excludeNames;
     private String excludeNamesRegex;
-    private ArrayList<String> includeLabels;
-    private String includeLabelsRegex;
-    private ArrayList<String> excludeLabels;
-    private String excludeLabelsRegex;
-    private String threshold;
-    private String op;
     private int period;
-    private int count;
     private int silencePeriod;
     private String message;
-    private boolean onlyAsCondition;
     private Map<String, String> tags;
     private Set<String> hooks;
+
+    /**
+     * Init includeMetrics and verify the expression.
+     * ValueColumnMetadata need init metrics info, don't invoke before the module finishes start.
+     */
+    public void setExpression(final String expression) {
+        MQELexer lexer = new MQELexer(CharStreams.fromString(expression));
+        lexer.addErrorListener(new ParseErrorListener());
+        MQEParser parser = new MQEParser(new CommonTokenStream(lexer));
+        ParseTree tree;
+        try {
+            tree = parser.expression();
+        } catch (ParseCancellationException e) {
+            throw new IllegalArgumentException("expression:" + expression + " error: " + e.getMessage());
+        }
+        AlarmMQEVerifyVisitor visitor = new AlarmMQEVerifyVisitor();
+        ExpressionResult parseResult = visitor.visit(tree);
+        if (StringUtil.isNotBlank(parseResult.getError())) {
+            throw new IllegalArgumentException("expression:" + expression + " error: " + parseResult.getError());
+        }
+        if (!parseResult.isBoolResult()) {
+            throw new IllegalArgumentException("expression:" + expression + " root operation is not a Compare Operation.");
+        }
+        if (ExpressionResultType.SINGLE_VALUE != parseResult.getType()) {
+            throw new IllegalArgumentException("expression:" + expression + " is not a SINGLE_VALUE result expression.");
+        }
+
+        verifyIncludeMetrics(visitor.getIncludeMetrics());
+        this.expression = expression;
+        this.includeMetrics = visitor.getIncludeMetrics();
+    }
+
+    private void verifyIncludeMetrics(Set<String> includeMetrics) {
+        includeMetrics.forEach(metricName -> {
+            Optional<ValueColumnMetadata.ValueColumn> valueColumn = ValueColumnMetadata.INSTANCE.readValueColumnDefinition(
+                metricName);
+            if (valueColumn.isEmpty()) {
+                throw new IllegalArgumentException("Metric: [" + metricName + "] dose not exist.");
+            }
+            Column.ValueDataType dataType = valueColumn.get().getDataType();
+            switch (dataType) {
+                case COMMON_VALUE:
+                case LABELED_VALUE:
+                    return;
+                default:
+                    throw new IllegalArgumentException(
+                        "Metric dose not supported in alarm, metric: [" + metricName + "] is not a common or labeled metric.");
+            }
+        });
+    }
 }
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcher.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcher.java
index f27ea3f4752a..96eed396c16a 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcher.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcher.java
@@ -23,14 +23,13 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
 import org.apache.skywalking.oap.server.core.alarm.AlarmModule;
 import org.apache.skywalking.oap.server.core.alarm.provider.dingtalk.DingtalkSettings;
 import org.apache.skywalking.oap.server.core.alarm.provider.discord.DiscordSettings;
-import org.apache.skywalking.oap.server.core.alarm.provider.expression.Expression;
-import org.apache.skywalking.oap.server.core.alarm.provider.expression.ExpressionContext;
 import org.apache.skywalking.oap.server.core.alarm.provider.feishu.FeishuSettings;
 import org.apache.skywalking.oap.server.core.alarm.provider.grpc.GRPCAlarmSetting;
 import org.apache.skywalking.oap.server.core.alarm.provider.pagerduty.PagerDutySettings;
@@ -51,18 +50,17 @@ public class AlarmRulesWatcher extends ConfigChangeWatcher {
     @Getter
     private volatile Map<String, List<RunningRule>> runningContext;
     private volatile Map<AlarmRule, RunningRule> alarmRuleRunningRuleMap;
+    @Getter
+    private volatile Map<String, Set<String>> exprMetricsMap;
     private volatile Rules rules;
     private volatile String settingsString;
-    @Getter
-    private final CompositeRuleEvaluator compositeRuleEvaluator;
 
     public AlarmRulesWatcher(Rules defaultRules, ModuleProvider provider) {
         super(AlarmModule.NAME, provider, "alarm-settings");
         this.runningContext = new HashMap<>();
         this.alarmRuleRunningRuleMap = new HashMap<>();
+        this.exprMetricsMap = new HashMap<>();
         this.settingsString = null;
-        Expression expression = new Expression(new ExpressionContext());
-        this.compositeRuleEvaluator = new CompositeRuleEvaluator(expression);
         notify(defaultRules);
     }
 
@@ -79,9 +77,13 @@ public void notify(ConfigChangeEvent value) {
         }
     }
 
-    void notify(Rules newRules) {
+    /**
+     * Don't invoke before the module finishes start
+     */
+    public void notify(Rules newRules) {
         Map<AlarmRule, RunningRule> newAlarmRuleRunningRuleMap = new HashMap<>();
         Map<String, List<RunningRule>> newRunningContext = new HashMap<>();
+        Map<String, Set<String>> newExprMetricsMap = new HashMap<>();
 
         newRules.getRules().forEach(rule -> {
             /*
@@ -92,9 +94,10 @@ void notify(Rules newRules) {
 
             newAlarmRuleRunningRuleMap.put(rule, runningRule);
 
-            String metricsName = rule.getMetricsName();
+            String expression = rule.getExpression();
+            newExprMetricsMap.put(expression, rule.getIncludeMetrics());
 
-            List<RunningRule> runningRules = newRunningContext.computeIfAbsent(metricsName, key -> new ArrayList<>());
+            List<RunningRule> runningRules = newRunningContext.computeIfAbsent(expression, key -> new ArrayList<>());
 
             runningRules.add(runningRule);
         });
@@ -102,6 +105,7 @@ void notify(Rules newRules) {
         this.rules = newRules;
         this.runningContext = newRunningContext;
         this.alarmRuleRunningRuleMap = newAlarmRuleRunningRuleMap;
+        this.exprMetricsMap = newExprMetricsMap;
         log.info("Update alarm rules to {}", rules);
     }
 
@@ -114,10 +118,6 @@ public List<AlarmRule> getRules() {
         return this.rules.getRules();
     }
 
-    public List<CompositeAlarmRule> getCompositeRules() {
-        return this.rules.getCompositeRules();
-    }
-
     public Map<String, WebhookSettings> getWebHooks() {
         return this.rules.getWebhookSettingsMap();
     }
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeAlarmRule.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeAlarmRule.java
deleted file mode 100644
index 3a8c1050b599..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeAlarmRule.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import java.util.Set;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-import lombok.ToString;
-
-import java.util.Map;
-
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-@Setter
-@Getter
-@ToString
-public class CompositeAlarmRule {
-    private String alarmRuleName;
-    private String expression;
-    private String message;
-    private Map<String, String> tags;
-    private Set<String> hooks;
-}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluator.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluator.java
deleted file mode 100644
index f0d0440b8e0f..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluator.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableListMultimap;
-import com.google.common.collect.Multimaps;
-import org.apache.skywalking.oap.server.core.Const;
-import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
-import org.apache.skywalking.oap.server.core.alarm.MetaInAlarm;
-import org.apache.skywalking.oap.server.core.alarm.provider.expression.Expression;
-import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-
-/**
- * Evaluate composite rule using expression eval
- *
- * @since 8.2.0
- */
-public class CompositeRuleEvaluator {
-
-    private Expression expression;
-    private Map<String, AlarmMessageFormatter> messageFormatterCache;
-
-    public CompositeRuleEvaluator(Expression expression) {
-        this.expression = expression;
-        this.messageFormatterCache = new ConcurrentHashMap<>();
-    }
-
-    /**
-     * Evaluate composite rule
-     *
-     * @param compositeAlarmRules compositeRules
-     * @param alarmMessages       triggered alarm messages
-     * @return
-     */
-    public List<AlarmMessage> evaluate(List<CompositeAlarmRule> compositeAlarmRules, List<AlarmMessage> alarmMessages) {
-        final List<AlarmMessage> compositeRuleMessages = new ArrayList<>();
-        ImmutableListMultimap<String, AlarmMessage> messageMap = Multimaps.index(alarmMessages, alarmMessage ->
-                Joiner.on(Const.ID_CONNECTOR).useForNull(Const.EMPTY_STRING).join(alarmMessage.getId0(), alarmMessage.getId1()));
-        for (CompositeAlarmRule compositeAlarmRule : compositeAlarmRules) {
-            String expr = compositeAlarmRule.getExpression();
-            Set<String> dependencyRules = expression.analysisInputs(expr);
-            Map<String, Object> dataContext = new HashMap<>();
-            messageMap.asMap().forEach((key, alarmMessageList) -> {
-                dependencyRules.forEach(ruleName -> dataContext.put(ruleName, false));
-                alarmMessageList.forEach(alarmMessage -> {
-                    if (dependencyRules.contains(alarmMessage.getRuleName())) {
-                        dataContext.put(alarmMessage.getRuleName(), true);
-                    }
-                });
-                Object matched = expression.eval(expr, dataContext);
-                if (matched instanceof Boolean && (Boolean) matched) {
-                    AlarmMessage headMsg = alarmMessageList.iterator().next();
-                    AlarmMessage message = new AlarmMessage();
-                    message.setOnlyAsCondition(false);
-                    message.setScopeId(headMsg.getScopeId());
-                    message.setScope(headMsg.getScope());
-                    message.setName(headMsg.getName());
-                    message.setId0(headMsg.getId0());
-                    message.setId1(headMsg.getId1());
-                    message.setStartTime(System.currentTimeMillis());
-                    message.setRuleName(compositeAlarmRule.getAlarmRuleName());
-                    String alarmMessage = formatMessage(message, compositeAlarmRule.getMessage(), compositeAlarmRule.getExpression());
-                    message.setAlarmMessage(alarmMessage);
-                    message.setPeriod(headMsg.getPeriod());
-                    message.setTags(compositeAlarmRule.getTags().entrySet().stream().map(e -> new Tag(e.getKey(), e.getValue())).collect(Collectors.toList()));
-                    message.getHooks().addAll(compositeAlarmRule.getHooks());
-                    compositeRuleMessages.add(message);
-                }
-            });
-        }
-        return compositeRuleMessages;
-    }
-
-    /**
-     * Format alarm message using {@link AlarmMessageFormatter}, only support name and id0 meta
-     */
-    private String formatMessage(AlarmMessage alarmMessage, String message, String metricName) {
-        return messageFormatterCache.computeIfAbsent(message, AlarmMessageFormatter::new).format(new MetaInAlarm() {
-            @Override
-            public String getScope() {
-                return alarmMessage.getScope();
-            }
-
-            @Override
-            public int getScopeId() {
-                return alarmMessage.getScopeId();
-            }
-
-            @Override
-            public String getName() {
-                return alarmMessage.getName();
-            }
-
-            @Override
-            public String getMetricsName() {
-                return metricName;
-            }
-
-            @Override
-            public String getId0() {
-                return alarmMessage.getId0();
-            }
-
-            @Override
-            public String getId1() {
-                return alarmMessage.getId1();
-            }
-        });
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/OP.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/OP.java
deleted file mode 100644
index 7ae660972b86..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/OP.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import static java.util.Objects.requireNonNull;
-
-public enum OP {
-    GT {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() > requireNonNull(expected, "expected").doubleValue();
-        }
-    },
-
-    GTE {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() >= requireNonNull(expected, "expected").doubleValue();
-        }
-    },
-
-    LT {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() < requireNonNull(expected, "expected").doubleValue();
-        }
-    },
-
-    LTE {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() <= requireNonNull(expected, "expected").doubleValue();
-        }
-    },
-
-    // NOTICE: double equal is not reliable in Java,
-    // match result is not predictable
-    EQ {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() == requireNonNull(expected, "expected").doubleValue();
-        }
-    },
-
-    NEQ {
-        @Override
-        public boolean test(final Number expected, final Number actual) {
-            return requireNonNull(actual, "actual").doubleValue() != requireNonNull(expected, "expected").doubleValue();
-        }
-    };
-
-    public static OP get(String op) {
-        switch (op) {
-            case ">":
-                return GT;
-            case ">=":
-                return GTE;
-            case "<":
-                return LT;
-            case "<=":
-                return LTE;
-            case "==":
-                return EQ;
-            case "!=":
-                return NEQ;
-            default:
-                throw new IllegalArgumentException("unknown op, " + op);
-        }
-    }
-
-    public abstract boolean test(final Number expected, final Number actual);
-}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java
index 9179d9ea2667..0ef71df7948f 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Rules.java
@@ -44,7 +44,6 @@ public class Rules {
     private Map<String, GRPCAlarmSetting> grpcAlarmSettingMap;
     private Map<String, SlackSettings> slackSettingsMap;
     private Map<String, WechatSettings> wechatSettingsMap;
-    private List<CompositeAlarmRule> compositeRules;
     private Map<String, DingtalkSettings> dingtalkSettingsMap;
     private Map<String, FeishuSettings> feishuSettingsMap;
     private Map<String, WeLinkSettings> weLinkSettingsMap;
@@ -54,7 +53,6 @@ public class Rules {
     public Rules() {
         this.rules = new ArrayList<>();
         this.webhookSettingsMap = new HashMap<>();
-        this.compositeRules = new ArrayList<>();
         this.grpcAlarmSettingMap = new HashMap<>();
         this.slackSettingsMap = new HashMap<>();
         this.wechatSettingsMap = new HashMap<>();
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java
index 4c1862395193..df288c6cf15d 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReader.java
@@ -72,7 +72,6 @@ public Rules readRules() {
             // Should read hooks config first.
             readHooksConfig(rules);
             readRulesConfig(rules);
-            readCompositeRuleConfig(rules);
         }
         return rules;
     }
@@ -91,29 +90,18 @@ private void readRulesConfig(Rules rules) {
                 AlarmRule alarmRule = new AlarmRule();
                 alarmRule.setAlarmRuleName((String) k);
                 Map settings = (Map) v;
-                Object metricsName = settings.get("metrics-name");
-                if (metricsName == null) {
-                    throw new IllegalArgumentException("metrics-name can't be null");
+                Object expression = settings.get("expression");
+                if (StringUtil.isEmpty((String) expression)) {
+                    throw new IllegalArgumentException("expression can't be empty");
                 }
-
-                alarmRule.setMetricsName((String) metricsName);
+                alarmRule.setExpression(expression.toString());
                 alarmRule.setIncludeNames((ArrayList) settings.getOrDefault("include-names", new ArrayList(0)));
                 alarmRule.setExcludeNames((ArrayList) settings.getOrDefault("exclude-names", new ArrayList(0)));
                 alarmRule.setIncludeNamesRegex((String) settings.getOrDefault("include-names-regex", ""));
                 alarmRule.setExcludeNamesRegex((String) settings.getOrDefault("exclude-names-regex", ""));
-                alarmRule.setIncludeLabels(
-                        (ArrayList) settings.getOrDefault("include-labels", new ArrayList(0)));
-                alarmRule.setExcludeLabels(
-                        (ArrayList) settings.getOrDefault("exclude-labels", new ArrayList(0)));
-                alarmRule.setIncludeLabelsRegex((String) settings.getOrDefault("include-labels-regex", ""));
-                alarmRule.setExcludeLabelsRegex((String) settings.getOrDefault("exclude-labels-regex", ""));
-                alarmRule.setThreshold(settings.get("threshold").toString());
-                alarmRule.setOp((String) settings.get("op"));
                 alarmRule.setPeriod((Integer) settings.getOrDefault("period", 1));
-                alarmRule.setCount((Integer) settings.getOrDefault("count", 1));
                 // How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
                 alarmRule.setSilencePeriod((Integer) settings.getOrDefault("silence-period", alarmRule.getPeriod()));
-                alarmRule.setOnlyAsCondition((Boolean) settings.getOrDefault("only-as-condition", false));
                 alarmRule.setMessage(
                         (String) settings.getOrDefault("message", "Alarm caused by Rule " + alarmRule
                                 .getAlarmRuleName()));
@@ -264,42 +252,6 @@ private void readWechatConfig(Map hooks, Rules rules) {
         });
     }
 
-    /**
-     * Read composite rule config into {@link CompositeAlarmRule}
-     */
-    @SuppressWarnings("unchecked")
-    private void readCompositeRuleConfig(Rules rules) {
-        Map compositeRulesData = (Map) yamlData.get("composite-rules");
-        if (compositeRulesData == null) {
-            return;
-        }
-        compositeRulesData.forEach((k, v) -> {
-            String ruleName = (String) k;
-            if (ruleName.endsWith("_rule")) {
-                Map settings = (Map) v;
-                CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule();
-                compositeAlarmRule.setAlarmRuleName(ruleName);
-                String expression = (String) settings.get("expression");
-                if (expression == null) {
-                    throw new IllegalArgumentException("expression can't be null");
-                }
-                compositeAlarmRule.setExpression(expression);
-                compositeAlarmRule.setMessage(
-                        (String) settings.getOrDefault("message", "Alarm caused by Rule " + ruleName));
-                compositeAlarmRule.setTags((Map) settings.getOrDefault("tags", new HashMap<String, String>(0)));
-
-                Set<String> specificHooks = new HashSet<>((ArrayList) settings.getOrDefault("hooks", new ArrayList<>()));
-                checkSpecificHooks(compositeAlarmRule.getAlarmRuleName(), specificHooks);
-                compositeAlarmRule.setHooks(specificHooks);
-                // If no specific hooks, use global hooks.
-                if (compositeAlarmRule.getHooks().isEmpty()) {
-                    compositeAlarmRule.getHooks().addAll(defaultHooks);
-                }
-                rules.getCompositeRules().add(compositeAlarmRule);
-            }
-        });
-    }
-
     /**
      * Read dingtalk hook config into {@link DingtalkSettings}
      */
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRule.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRule.java
index a9bbdaabe152..62cedae9ff16 100644
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRule.java
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRule.java
@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -33,6 +34,16 @@
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
+import org.antlr.v4.runtime.CharStreams;
+import org.antlr.v4.runtime.CommonTokenStream;
+import org.antlr.v4.runtime.tree.ParseTree;
+import org.apache.skywalking.mqe.rt.exception.ParseErrorListener;
+import org.apache.skywalking.mqe.rt.grammar.MQELexer;
+import org.apache.skywalking.mqe.rt.grammar.MQEParser;
+import org.apache.skywalking.mqe.rt.type.ExpressionResult;
+import org.apache.skywalking.mqe.rt.type.ExpressionResultType;
+import org.apache.skywalking.mqe.rt.type.MQEValues;
+import org.apache.skywalking.oap.server.core.alarm.provider.expr.rt.AlarmMQEVisitor;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
 import org.apache.skywalking.oap.server.core.alarm.MetaInAlarm;
@@ -59,61 +70,44 @@ public class RunningRule {
 
     private final String ruleName;
     private final int period;
-    private final String metricsName;
-    private final Threshold threshold;
-    private final OP op;
-    private final int countThreshold;
+    private final String expression;
     private final int silencePeriod;
-    private final Map<MetaInAlarm, Window> windows;
-    private volatile MetricsValueType valueType;
+    private final Map<AlarmEntity, Window> windows;
     private final List<String> includeNames;
     private final List<String> excludeNames;
     private final Pattern includeNamesRegex;
     private final Pattern excludeNamesRegex;
-    private final List<String> includeLabels;
-    private final List<String> excludeLabels;
-    private final Pattern includeLabelsRegex;
-    private final Pattern excludeLabelsRegex;
     private final AlarmMessageFormatter formatter;
-    private final boolean onlyAsCondition;
     private final List<Tag> tags;
     private final Set<String> hooks;
+    private final Set<String> includeMetrics;
+    private final ParseTree exprTree;
 
     public RunningRule(AlarmRule alarmRule) {
-        metricsName = alarmRule.getMetricsName();
+        expression = alarmRule.getExpression();
         this.ruleName = alarmRule.getAlarmRuleName();
-
+        this.includeMetrics = alarmRule.getIncludeMetrics();
         // Init the empty window for alarming rule.
         windows = new ConcurrentHashMap<>();
-
         period = alarmRule.getPeriod();
-
-        threshold = new Threshold(alarmRule.getAlarmRuleName(), alarmRule.getThreshold());
-        op = OP.get(alarmRule.getOp());
-
-        this.countThreshold = alarmRule.getCount();
         this.silencePeriod = alarmRule.getSilencePeriod();
-
         this.includeNames = alarmRule.getIncludeNames();
         this.excludeNames = alarmRule.getExcludeNames();
         this.includeNamesRegex = StringUtil.isNotEmpty(alarmRule.getIncludeNamesRegex()) ?
             Pattern.compile(alarmRule.getIncludeNamesRegex()) : null;
         this.excludeNamesRegex = StringUtil.isNotEmpty(alarmRule.getExcludeNamesRegex()) ?
             Pattern.compile(alarmRule.getExcludeNamesRegex()) : null;
-        this.includeLabels = alarmRule.getIncludeLabels();
-        this.excludeLabels = alarmRule.getExcludeLabels();
-        this.includeLabelsRegex = StringUtil.isNotEmpty(alarmRule.getIncludeLabelsRegex()) ?
-            Pattern.compile(alarmRule.getIncludeLabelsRegex()) : null;
-        this.excludeLabelsRegex = StringUtil.isNotEmpty(alarmRule.getExcludeLabelsRegex()) ?
-            Pattern.compile(alarmRule.getExcludeLabelsRegex()) : null;
         this.formatter = new AlarmMessageFormatter(alarmRule.getMessage());
-        this.onlyAsCondition = alarmRule.isOnlyAsCondition();
         this.tags = alarmRule.getTags()
                              .entrySet()
                              .stream()
                              .map(e -> new Tag(e.getKey(), e.getValue()))
                              .collect(Collectors.toList());
         this.hooks = alarmRule.getHooks();
+        MQELexer lexer = new MQELexer(CharStreams.fromString(alarmRule.getExpression()));
+        MQEParser parser = new MQEParser(new CommonTokenStream(lexer));
+        parser.addErrorListener(new ParseErrorListener());
+        this.exprTree = parser.expression();
     }
 
     /**
@@ -124,10 +118,10 @@ public RunningRule(AlarmRule alarmRule) {
      * @param metrics includes the values.
      */
     public void in(MetaInAlarm meta, Metrics metrics) {
-        if (!meta.getMetricsName().equals(metricsName)) {
+        if (!includeMetrics.contains(meta.getMetricsName())) {
             //Don't match rule, exit.
             if (log.isTraceEnabled()) {
-                log.trace("Metric names are inconsistent, {}-{}", meta.getMetricsName(), metricsName);
+                log.trace("Metric name not in the expression, {}-{}", expression, meta.getMetricsName());
             }
             return;
         }
@@ -137,42 +131,11 @@ public void in(MetaInAlarm meta, Metrics metrics) {
             return;
         }
 
-        if (valueType == null) {
-            if (metrics instanceof LongValueHolder) {
-                valueType = MetricsValueType.LONG;
-                threshold.setType(MetricsValueType.LONG);
-            } else if (metrics instanceof IntValueHolder) {
-                valueType = MetricsValueType.INT;
-                threshold.setType(MetricsValueType.INT);
-            } else if (metrics instanceof DoubleValueHolder) {
-                valueType = MetricsValueType.DOUBLE;
-                threshold.setType(MetricsValueType.DOUBLE);
-            } else if (metrics instanceof MultiIntValuesHolder) {
-                valueType = MetricsValueType.MULTI_INTS;
-                threshold.setType(MetricsValueType.MULTI_INTS);
-            } else if (metrics instanceof LabeledValueHolder) {
-                if (((LabeledValueHolder) metrics).getValue().keys().stream()
-                                                  .noneMatch(label -> validate(
-                                                      label,
-                                                      includeLabels,
-                                                      excludeLabels,
-                                                      includeLabelsRegex,
-                                                      excludeLabelsRegex
-                                                  ))) {
-                    return;
-                }
-                valueType = MetricsValueType.LABELED_LONG;
-                threshold.setType(MetricsValueType.LONG);
-            } else {
-                log.warn("Unsupported value type {}", valueType);
-                return;
-            }
-        }
+        AlarmEntity entity = new AlarmEntity(
+            meta.getScope(), meta.getScopeId(), meta.getName(), meta.getId0(), meta.getId1());
 
-        if (valueType != null) {
-            Window window = windows.computeIfAbsent(meta, ignored -> new Window(period));
-            window.add(metrics);
-        }
+        Window window = windows.computeIfAbsent(entity, ignored -> new Window(period));
+        window.add(meta.getMetricsName(), metrics);
     }
 
     /**
@@ -236,18 +199,17 @@ public void moveTo(LocalDateTime targetTime) {
     public List<AlarmMessage> check() {
         List<AlarmMessage> alarmMessageList = new ArrayList<>(30);
 
-        windows.forEach((meta, window) -> {
+        windows.forEach((alarmEntity, window) -> {
             Optional<AlarmMessage> alarmMessageOptional = window.checkAlarm();
             if (alarmMessageOptional.isPresent()) {
                 AlarmMessage alarmMessage = alarmMessageOptional.get();
-                alarmMessage.setScopeId(meta.getScopeId());
-                alarmMessage.setScope(meta.getScope());
-                alarmMessage.setName(meta.getName());
-                alarmMessage.setId0(meta.getId0());
-                alarmMessage.setId1(meta.getId1());
+                alarmMessage.setScopeId(alarmEntity.getScopeId());
+                alarmMessage.setScope(alarmEntity.getScope());
+                alarmMessage.setName(alarmEntity.getName());
+                alarmMessage.setId0(alarmEntity.getId0());
+                alarmMessage.setId1(alarmEntity.getId1());
                 alarmMessage.setRuleName(this.ruleName);
-                alarmMessage.setAlarmMessage(formatter.format(meta));
-                alarmMessage.setOnlyAsCondition(this.onlyAsCondition);
+                alarmMessage.setAlarmMessage(formatter.format(alarmEntity));
                 alarmMessage.setStartTime(System.currentTimeMillis());
                 alarmMessage.setPeriod(this.period);
                 alarmMessage.setTags(this.tags);
@@ -265,10 +227,9 @@ public List<AlarmMessage> check() {
      */
     public class Window {
         private LocalDateTime endTime;
-        private int period;
+        private final int period;
         private int silenceCountdown;
-
-        private LinkedList<Metrics> values;
+        private LinkedList<Map<String, Metrics>> values;
         private ReentrantLock lock = new ReentrantLock();
 
         public Window(int period) {
@@ -307,7 +268,7 @@ public void moveTo(LocalDateTime current) {
             }
         }
 
-        public void add(Metrics metrics) {
+        public void add(String metricsName, Metrics metrics) {
             long bucket = metrics.getTimeBucket();
 
             LocalDateTime timeBucket = TIME_BUCKET_FORMATTER.parseLocalDateTime(bucket + "");
@@ -319,6 +280,7 @@ public void add(Metrics metrics) {
                     this.endTime = timeBucket;
                 }
                 int minutes = Minutes.minutesBetween(timeBucket, this.endTime).getMinutes();
+                //timeBucket > endTime
                 if (minutes < 0) {
                     this.moveTo(timeBucket);
                     minutes = 0;
@@ -335,8 +297,15 @@ public void add(Metrics metrics) {
                     }
                     return;
                 }
-
-                this.values.set(values.size() - minutes - 1, metrics);
+                int index = values.size() - minutes - 1;
+                Map<String, Metrics> metricsMap = values.get(index);
+                if (metricsMap == null) {
+                    metricsMap = new HashMap<>();
+                    metricsMap.put(metricsName, metrics);
+                    values.set(index, metricsMap);
+                } else {
+                    metricsMap.put(metricsName, metrics);
+                }
             } finally {
                 this.lock.unlock();
             }
@@ -365,81 +334,53 @@ public Optional<AlarmMessage> checkAlarm() {
         }
 
         private boolean isMatch() {
-            int matchCount = 0;
-            for (Metrics metrics : values) {
-                if (metrics == null) {
-                    continue;
+            int isMatch = 0;
+            AlarmMQEVisitor visitor = new AlarmMQEVisitor(this.values, this.endTime);
+            ExpressionResult parseResult = visitor.visit(exprTree);
+            if (StringUtil.isNotBlank(parseResult.getError())) {
+                log.error("expression:" + expression + " error: " + parseResult.getError());
+                return false;
+            }
+            if (!parseResult.isBoolResult() ||
+                ExpressionResultType.SINGLE_VALUE != parseResult.getType() ||
+                CollectionUtils.isEmpty(parseResult.getResults())) {
+                return false;
+            }
+            if (!parseResult.isLabeledResult()) {
+                MQEValues mqeValues = parseResult.getResults().get(0);
+                if (mqeValues != null &&
+                    CollectionUtils.isNotEmpty(mqeValues.getValues()) &&
+                    mqeValues.getValues().get(0) != null) {
+                    isMatch = (int) mqeValues.getValues().get(0).getDoubleValue();
                 }
-
-                switch (valueType) {
-                    case LONG:
-                        long lvalue = ((LongValueHolder) metrics).getValue();
-                        long lexpected = RunningRule.this.threshold.getLongThreshold();
-                        if (op.test(lexpected, lvalue)) {
-                            matchCount++;
-                        }
-                        break;
-                    case INT:
-                        int ivalue = ((IntValueHolder) metrics).getValue();
-                        int iexpected = RunningRule.this.threshold.getIntThreshold();
-                        if (op.test(iexpected, ivalue)) {
-                            matchCount++;
-                        }
-                        break;
-                    case DOUBLE:
-                        double dvalue = ((DoubleValueHolder) metrics).getValue();
-                        double dexpected = RunningRule.this.threshold.getDoubleThreshold();
-                        if (op.test(dexpected, dvalue)) {
-                            matchCount++;
-                        }
-                        break;
-                    case MULTI_INTS:
-                        int[] ivalueArray = ((MultiIntValuesHolder) metrics).getValues();
-                        Integer[] iaexpected = RunningRule.this.threshold.getIntValuesThreshold();
-                        if (log.isTraceEnabled()) {
-                            log.trace("Value array is {}, expected array is {}", ivalueArray, iaexpected);
-                        }
-                        for (int i = 0; i < ivalueArray.length; i++) {
-                            ivalue = ivalueArray[i];
-                            Integer iNullableExpected = 0;
-                            if (iaexpected.length > i) {
-                                iNullableExpected = iaexpected[i];
-                                if (iNullableExpected == null) {
-                                    continue;
-                                }
-                            }
-                            if (op.test(iNullableExpected, ivalue)) {
-                                if (log.isTraceEnabled()) {
-                                    log.trace("Matched, expected {}, value {}", iNullableExpected, ivalue);
-                                }
-                                matchCount++;
-                                break;
-                            }
-                        }
-                        break;
-                    case LABELED_LONG:
-                        DataTable values = ((LabeledValueHolder) metrics).getValue();
-                        lexpected = RunningRule.this.threshold.getLongThreshold();
-                        if (values.keys().stream().anyMatch(label ->
-                                                                validate(
-                                                                    label,
-                                                                    RunningRule.this.includeLabels,
-                                                                    RunningRule.this.excludeLabels,
-                                                                    RunningRule.this.includeLabelsRegex,
-                                                                    RunningRule.this.excludeLabelsRegex
-                                                                )
-                                                                    && op.test(lexpected, values.get(label)))) {
-                            matchCount++;
+            } else {
+                // if the result has multiple labels, when there is one label match, then the result is match
+                // for example in 5 minutes, the sum(percentile{_='P50,P75'} > 1000) >= 3
+                // percentile{_='P50,P75'} result is:
+                // P50(1000,1100,1200,1000,500), > 1000 2 times
+                // P75(2000,1500,1200,1000,500), > 1000 3 times
+                // percentile{_='P50,P75'} > 1000 result is:
+                // P50(0,1,1,0,0)
+                // P75(1,1,1,0,0)
+                // sum(percentile{_='P50,P75'} > 1000) >= 3 result is:
+                // P50(0)
+                // P75(1)
+                // then the isMatch is 1
+                for (MQEValues mqeValues : parseResult.getResults()) {
+                    if (mqeValues != null &&
+                        CollectionUtils.isNotEmpty(mqeValues.getValues()) &&
+                        mqeValues.getValues().get(0) != null) {
+                        isMatch = (int) mqeValues.getValues().get(0).getDoubleValue();
+                        if (isMatch == 1) {
+                            break;
                         }
-                        break;
+                    }
                 }
             }
-
             if (log.isTraceEnabled()) {
-                log.trace("Match count is {}, threshold is {}", matchCount, countThreshold);
+                log.trace("Match expression is {}", expression);
             }
-            // Reach the threshold in current bucket.
-            return matchCount >= countThreshold;
+            return isMatch == 1;
         }
 
         private void init() {
@@ -450,37 +391,38 @@ private void init() {
         }
     }
 
-    private LinkedList<TraceLogMetric> transformValues(final LinkedList<Metrics> values) {
-        LinkedList<TraceLogMetric> r = new LinkedList<>();
-        values.forEach(m -> {
-            if (m == null) {
-                r.add(null);
-                return;
+    private LinkedList<Map<String, TraceLogMetric>> transformValues(LinkedList<Map<String, Metrics>> values) {
+        LinkedList<Map<String, TraceLogMetric>> result = new LinkedList<>();
+        for (Map<String, Metrics> value : values) {
+            if (value == null) {
+                result.add(null);
+                continue;
             }
-            switch (valueType) {
-                case LONG:
-                    r.add(new TraceLogMetric(m.getTimeBucket(), new Number[] {((LongValueHolder) m).getValue()}));
-                    break;
-                case INT:
-                    r.add(new TraceLogMetric(m.getTimeBucket(), new Number[] {((IntValueHolder) m).getValue()}));
-                    break;
-                case DOUBLE:
-                    r.add(new TraceLogMetric(m.getTimeBucket(), new Number[] {((DoubleValueHolder) m).getValue()}));
-                    break;
-                case MULTI_INTS:
+            value.forEach((name, m) -> {
+                Map<String, TraceLogMetric> r = new HashMap<>();
+                result.add(r);
+                if (m instanceof LongValueHolder) {
+                    r.put(name, new TraceLogMetric(m.getTimeBucket(), new Number[] {((LongValueHolder) m).getValue()}));
+                } else if (m instanceof IntValueHolder) {
+                    r.put(name, new TraceLogMetric(m.getTimeBucket(), new Number[] {((IntValueHolder) m).getValue()}));
+                } else if (m instanceof DoubleValueHolder) {
+                    r.put(name, new TraceLogMetric(m.getTimeBucket(), new Number[] {((DoubleValueHolder) m).getValue()}));
+                } else if (m instanceof MultiIntValuesHolder) {
                     int[] iArr = ((MultiIntValuesHolder) m).getValues();
-                    r.add(new TraceLogMetric(m.getTimeBucket(), Arrays.stream(iArr).boxed().toArray(Number[]::new)));
-                    break;
-                case LABELED_LONG:
+                    r.put(name, new TraceLogMetric(m.getTimeBucket(), Arrays.stream(iArr).boxed().toArray(Number[]::new)));
+                } else if (m instanceof LabeledValueHolder) {
                     DataTable dt = ((LabeledValueHolder) m).getValue();
                     TraceLogMetric l = new TraceLogMetric(
                         m.getTimeBucket(), dt.sortedValues(Comparator.naturalOrder())
                                              .toArray(new Number[0]));
                     l.labels = dt.sortedKeys(Comparator.naturalOrder()).toArray(new String[0]);
-                    r.add(l);
-            }
-        });
-        return r;
+                    r.put(name, l);
+                } else {
+                    log.warn("Unsupported metrics {}", m);
+                }
+            });
+        }
+        return result;
     }
 
     @RequiredArgsConstructor
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Threshold.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Threshold.java
deleted file mode 100644
index ba3299ce4766..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/Threshold.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class Threshold {
-    private static final Logger LOGGER = LoggerFactory.getLogger(Threshold.class);
-    private static final String NONE_THRESHOLD = "-";
-
-    private String alarmRuleName;
-    private final String threshold;
-    private int intThreshold;
-    private double doubleThreshold;
-    private long longThreshold;
-    private Integer[] intValuesThreshold;
-
-    public Threshold(String alarmRuleName, String threshold) {
-        this.alarmRuleName = alarmRuleName;
-        this.threshold = threshold;
-    }
-
-    public int getIntThreshold() {
-        return intThreshold;
-    }
-
-    public double getDoubleThreshold() {
-        return doubleThreshold;
-    }
-
-    public long getLongThreshold() {
-        return longThreshold;
-    }
-
-    public Integer[] getIntValuesThreshold() {
-        return intValuesThreshold;
-    }
-
-    public void setType(MetricsValueType type) {
-        try {
-            switch (type) {
-                case INT:
-                    intThreshold = Integer.parseInt(threshold);
-                    break;
-                case LONG:
-                    longThreshold = Long.parseLong(threshold);
-                    break;
-                case DOUBLE:
-                    doubleThreshold = Double.parseDouble(threshold);
-                    break;
-                case MULTI_INTS:
-                    String[] strings = threshold.split(",");
-                    intValuesThreshold = new Integer[strings.length];
-                    for (int i = 0; i < strings.length; i++) {
-                        String thresholdItem = strings[i].trim();
-                        if (NONE_THRESHOLD.equals(thresholdItem)) {
-                            intValuesThreshold[i] = null;
-                        } else {
-                            intValuesThreshold[i] = Integer.parseInt(thresholdItem);
-                        }
-                    }
-            }
-        } catch (NumberFormatException e) {
-            LOGGER.warn("Alarm rule {} threshold doesn't match the metrics type, expected type: {}", alarmRuleName, type);
-        }
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVerifyVisitor.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVerifyVisitor.java
new file mode 100644
index 000000000000..78c93afdd773
--- /dev/null
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVerifyVisitor.java
@@ -0,0 +1,60 @@
+/*
+ * 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.alarm.provider.expr.rt;
+
+import java.util.HashSet;
+import java.util.Set;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.mqe.rt.MQEVisitorBase;
+import org.apache.skywalking.mqe.rt.grammar.MQEParser;
+import org.apache.skywalking.mqe.rt.type.ExpressionResult;
+import org.apache.skywalking.mqe.rt.type.ExpressionResultType;
+import org.apache.skywalking.mqe.rt.type.MQEValue;
+import org.apache.skywalking.mqe.rt.type.MQEValues;
+
+/**
+ * Used for verify the alarm expression and get the metrics name when read the alarm rules.
+ */
+@Getter
+@Slf4j
+public class AlarmMQEVerifyVisitor extends MQEVisitorBase {
+    private final Set<String> includeMetrics = new HashSet<>();
+
+    @Override
+    public ExpressionResult visitMetric(MQEParser.MetricContext ctx) {
+        ExpressionResult result = new ExpressionResult();
+        String metricName = ctx.metricName().getText();
+        this.includeMetrics.add(metricName);
+
+        if (ctx.parent instanceof MQEParser.TopNOPContext) {
+            result.setType(ExpressionResultType.UNKNOWN);
+            result.setError("Unsupported operation: [top_n] in alarm expression.");
+            return result;
+        }
+
+        MQEValues mqeValues = new MQEValues();
+        MQEValue mqeValue = new MQEValue();
+        mqeValue.setEmptyValue(true);
+        mqeValues.getValues().add(mqeValue);
+        result.getResults().add(mqeValues);
+        result.setType(ExpressionResultType.TIME_SERIES_VALUES);
+        return result;
+    }
+}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVisitor.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVisitor.java
new file mode 100644
index 000000000000..98a32d791bcc
--- /dev/null
+++ b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expr/rt/AlarmMQEVisitor.java
@@ -0,0 +1,271 @@
+/*
+ * 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.alarm.provider.expr.rt;
+
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.mqe.rt.grammar.MQEParser;
+import org.apache.skywalking.mqe.rt.type.ExpressionResult;
+import org.apache.skywalking.mqe.rt.type.ExpressionResultType;
+import org.apache.skywalking.mqe.rt.type.MQEValue;
+import org.apache.skywalking.mqe.rt.type.MQEValues;
+import org.apache.skywalking.mqe.rt.type.Metadata;
+import org.apache.skywalking.oap.server.core.Const;
+import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
+import org.apache.skywalking.oap.server.core.analysis.metrics.DoubleValueHolder;
+import org.apache.skywalking.oap.server.core.analysis.metrics.IntValueHolder;
+import org.apache.skywalking.oap.server.core.analysis.metrics.LabeledValueHolder;
+import org.apache.skywalking.oap.server.core.analysis.metrics.LongValueHolder;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
+import org.apache.skywalking.mqe.rt.MQEVisitorBase;
+import org.apache.skywalking.oap.server.core.analysis.metrics.MultiIntValuesHolder;
+import org.apache.skywalking.oap.server.core.query.type.KeyValue;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.joda.time.LocalDateTime;
+
+@Slf4j
+public class AlarmMQEVisitor extends MQEVisitorBase {
+    private final LinkedList<Map<String, Metrics>> metricsValues;
+    private final Map<String, Map<String, Double>> commonValuesMap;
+    private final Map<String, Map<String, DataTable>> labeledValuesMap;
+    private final int windowSize;
+    private final LocalDateTime endTime;
+    private final ArrayList<String> windowTimes;
+
+    public AlarmMQEVisitor(final LinkedList<Map<String, Metrics>> metricsValues,
+                           final LocalDateTime endTime) {
+        this.metricsValues = metricsValues;
+        this.commonValuesMap = new HashMap<>();
+        this.labeledValuesMap = new HashMap<>();
+        this.endTime = endTime;
+        this.windowSize = metricsValues.size();
+        this.windowTimes = initWindowTimes();
+        this.initMetricsValues();
+    }
+
+    @Override
+    public ExpressionResult visitMetric(MQEParser.MetricContext ctx) {
+        ExpressionResult result = new ExpressionResult();
+        String metricName = ctx.metricName().getText();
+        Optional<ValueColumnMetadata.ValueColumn> valueColumn = ValueColumnMetadata.INSTANCE.readValueColumnDefinition(
+            metricName);
+        if (valueColumn.isEmpty()) {
+            result.setType(ExpressionResultType.UNKNOWN);
+            result.setError("Metric: [" + metricName + "] dose not exist.");
+            return result;
+        }
+        Column.ValueDataType dataType = valueColumn.get().getDataType();
+
+        //if no data, build empty value MQEValuesList for calculation
+        List<MQEValues> mqeValuesList;
+        if (dataType == Column.ValueDataType.COMMON_VALUE) {
+            Map<String, Double> timeValues = commonValuesMap.get(metricName);
+            if (CollectionUtils.isEmpty(timeValues)) {
+                mqeValuesList = buildEmptyMQEValuesList();
+            } else {
+                mqeValuesList = buildMqeValuesList(timeValues);
+            }
+        } else if (dataType == Column.ValueDataType.LABELED_VALUE) {
+            List<String> labelValues = Collections.emptyList();
+            if (ctx.label() != null) {
+                String labelValue = ctx.label().labelValue().getText();
+                String labelValueTrim = labelValue.substring(1, labelValue.length() - 1);
+                if (StringUtil.isNotBlank(labelValueTrim)) {
+                    labelValues = Arrays.asList(labelValueTrim.split(Const.COMMA));
+                }
+            }
+            Map<String, DataTable> timeValues = labeledValuesMap.get(metricName);
+            if (CollectionUtils.isEmpty(timeValues)) {
+                mqeValuesList = buildEmptyLabeledMQEValuesList(labelValues);
+            } else {
+                mqeValuesList = buildLabledMqeValuesList(timeValues, labelValues);
+            }
+            result.setLabeledResult(true);
+        } else {
+            result.setType(ExpressionResultType.UNKNOWN);
+            result.setError("Unsupported value type: " + dataType);
+            return result;
+        }
+
+        result.setResults(mqeValuesList);
+        result.setType(ExpressionResultType.TIME_SERIES_VALUES);
+        return result;
+    }
+
+    private ArrayList<String> initWindowTimes() {
+        ArrayList<String> windowTimes = new ArrayList<>();
+        for (int i = this.windowSize - 1; i >= 0; i--) {
+            windowTimes.add(endTime.minusMinutes(i).toString("yyyyMMddHHmm"));
+        }
+        return windowTimes;
+    }
+
+    private void initMetricsValues() {
+        for (Map<String, Metrics> metricsMap : metricsValues) {
+            if (metricsMap == null) {
+                continue;
+            }
+            for (Map.Entry<String, Metrics> entry : metricsMap.entrySet()) {
+                String metricName = entry.getKey();
+                Metrics metrics = entry.getValue();
+                if (metrics instanceof LongValueHolder) {
+                    initCommonMetricValues(metricName, ((LongValueHolder) metrics).getValue(), metrics.getTimeBucket());
+                } else if (metrics instanceof IntValueHolder) {
+                    initCommonMetricValues(metricName, ((IntValueHolder) metrics).getValue(), metrics.getTimeBucket());
+                } else if (metrics instanceof DoubleValueHolder) {
+                    initCommonMetricValues(metricName, ((DoubleValueHolder) metrics).getValue(), metrics.getTimeBucket());
+                } else if (metrics instanceof MultiIntValuesHolder) {
+                    int[] ivalueArray = ((MultiIntValuesHolder) metrics).getValues();
+                    DataTable values = new DataTable();
+                    for (int i = 0; i < ivalueArray.length; i++) {
+                        values.put(String.valueOf(i), (long) ivalueArray[i]);
+                    }
+                    initLabeledMetricValues(metricName, values, metrics.getTimeBucket());
+                } else if (metrics instanceof LabeledValueHolder) {
+                    DataTable values = ((LabeledValueHolder) metrics).getValue();
+                    initLabeledMetricValues(metricName, values, metrics.getTimeBucket());
+                } else {
+                    log.warn("Unsupported metrics {}", metricName);
+                    return;
+                }
+            }
+        }
+    }
+
+    private void initCommonMetricValues(String metricName, double value, long timeBucket) {
+        Map<String, Double> timeValues = commonValuesMap.computeIfAbsent(
+            metricName, v -> new HashMap<>());
+        timeValues.put(String.valueOf(timeBucket), value);
+    }
+
+    private void initLabeledMetricValues(String metricName,
+                                         DataTable values, long timeBucket) {
+        Map<String, DataTable> timeValues = labeledValuesMap.computeIfAbsent(
+            metricName, v -> new HashMap<>());
+        timeValues.put(String.valueOf(timeBucket), values);
+    }
+
+    private List<MQEValues> buildMqeValuesList(Map<String, Double> timeValues) {
+        List<MQEValues> mqeValuesList = new ArrayList<>();
+            MQEValues mqeValues = new MQEValues();
+            for (String time : windowTimes) {
+                Metadata metadata = new Metadata();
+                Double metricValue = timeValues.get(time);
+                MQEValue mqeValue = new MQEValue();
+                //use timeBucket as id here
+                mqeValue.setId(time);
+                if (metricValue != null) {
+                    mqeValue.setDoubleValue(metricValue);
+                } else {
+                    mqeValue.setEmptyValue(true);
+                }
+                mqeValues.setMetric(metadata);
+                mqeValues.getValues().add(mqeValue);
+            }
+            mqeValuesList.add(mqeValues);
+
+        return mqeValuesList;
+    }
+
+    private List<MQEValues> buildLabledMqeValuesList(Map<String, DataTable> timeValues, List<String> labelValues) {
+        List<MQEValues> mqeValuesList = new ArrayList<>();
+        List<String> allLabelValues;
+        if (CollectionUtils.isEmpty(labelValues) || labelValues.stream().allMatch(Strings::isNullOrEmpty)) {
+            allLabelValues = timeValues.values().stream()
+                             .flatMap(dataTable -> dataTable.keys().stream())
+                             .distinct().collect(Collectors.toList());
+        } else {
+            allLabelValues = labelValues;
+        }
+        for (String label : allLabelValues) {
+            MQEValues mqeValues = new MQEValues();
+            for (String time : windowTimes) {
+                Metadata metadata = new Metadata();
+                DataTable dataTable = timeValues.getOrDefault(time, new DataTable());
+                Long metricValue = dataTable.get(label);
+                MQEValue mqeValue = new MQEValue();
+                //use timeBucket as id here
+                mqeValue.setId(time);
+                if (metricValue != null) {
+                    mqeValue.setDoubleValue(metricValue);
+                } else {
+                    mqeValue.setEmptyValue(true);
+                }
+                KeyValue labelValue = new KeyValue(GENERAL_LABEL_NAME, label);
+                metadata.getLabels().add(labelValue);
+                mqeValues.setMetric(metadata);
+                mqeValues.getValues().add(mqeValue);
+            }
+            mqeValuesList.add(mqeValues);
+        }
+        return mqeValuesList;
+    }
+
+    //init MQEValues with empty value according to window size and end time
+    private MQEValues initMQEValues() {
+        MQEValues mqeValues = new MQEValues();
+        for (String times : windowTimes) {
+            MQEValue mqeValue = new MQEValue();
+            mqeValue.setEmptyValue(true);
+            mqeValue.setId(times);
+            mqeValues.getValues().add(mqeValue);
+        }
+        return mqeValues;
+    }
+
+    private ArrayList<MQEValues> buildEmptyMQEValuesList() {
+        ArrayList<MQEValues> mqeValuesList = new ArrayList<>();
+        mqeValuesList.add(initMQEValues());
+        return mqeValuesList;
+    }
+
+    private ArrayList<MQEValues> buildEmptyLabeledMQEValuesList(List<String> labelValues) {
+        ArrayList<MQEValues> mqeValuesList = new ArrayList<>();
+        if (CollectionUtils.isEmpty(labelValues)) {
+            MQEValues mqeValues = initMQEValues();
+            KeyValue label = new KeyValue(GENERAL_LABEL_NAME, GENERAL_LABEL_NAME);
+            Metadata metadata = new Metadata();
+            metadata.getLabels().add(label);
+            mqeValues.setMetric(metadata);
+            mqeValuesList.add(mqeValues);
+        } else {
+            for (String labelValue : labelValues) {
+                MQEValues mqeValues = initMQEValues();
+                Metadata metadata = new Metadata();
+                KeyValue label = new KeyValue(GENERAL_LABEL_NAME, labelValue);
+                metadata.getLabels().add(label);
+                mqeValues.setMetric(metadata);
+                mqeValuesList.add(mqeValues);
+            }
+        }
+        return mqeValuesList;
+    }
+}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/Expression.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/Expression.java
deleted file mode 100644
index 956f3e3d76c3..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/Expression.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.alarm.provider.expression;
-
-import lombok.extern.slf4j.Slf4j;
-import org.mvel2.MVEL;
-import org.mvel2.ParserContext;
-
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Expression support eval java basic expressions, just like groovy script
- * The internal detail is it first compile the expression to a parseTree then execute the parseTree with data
- * It caches the compiled expression for sake of performance
- */
-@Slf4j
-public class Expression {
-    private final Map<String, Object> expressionCache;
-    private final ExpressionContext context;
-
-    public Expression(ExpressionContext context) {
-        this.context = context;
-        this.expressionCache = new ConcurrentHashMap<>();
-    }
-
-    /**
-     * Eval the given expression using empty data context
-     */
-    public Object eval(String expression) {
-        return eval(expression, null);
-    }
-
-    /**
-     * Eval the given expression with data context
-     */
-    public Object eval(String expression, Map<String, Object> vars) {
-        try {
-            Object obj = compile(expression, context);
-            return MVEL.executeExpression(obj, vars);
-        } catch (Throwable e) {
-            log.error("eval expression {} error", expression, e);
-            return null;
-        }
-    }
-
-    /**
-     * Compile the given expression to a parseTree
-     */
-    public Object compile(String expression, ExpressionContext pctx) {
-        return expressionCache.computeIfAbsent(expression, s -> MVEL.compileExpression(expression, pctx.getContext()));
-    }
-
-    /**
-     * Analysis expression dependencies
-     */
-    public Set<String> analysisInputs(String expression) {
-        ParserContext pCtx = ParserContext.create();
-        MVEL.analysisCompile(expression, pCtx);
-        Map<String, Class> inputsMap = pCtx.getInputs();
-        return inputsMap.keySet();
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContext.java b/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContext.java
deleted file mode 100644
index 049853726af1..000000000000
--- a/oap-server/server-alarm-plugin/src/main/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContext.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.alarm.provider.expression;
-
-import lombok.Getter;
-import org.mvel2.ParserContext;
-import java.lang.reflect.Method;
-import java.lang.reflect.Modifier;
-
-/***
- * Expression context can support custom function in expression,
- * for example `md5(a) == '111111'`, the md5 function add register in the context
- */
-public class ExpressionContext {
-
-    @Getter
-    private ParserContext context;
-
-    public ExpressionContext() {
-        context = new ParserContext();
-    }
-
-    /**
-     * Register a single method in the context
-     */
-    public void registerFunc(String func, Method method) {
-        context.addImport(func, method);
-    }
-
-    /**
-     * Register hole class public static methods in the context
-     */
-    public void registerFunc(Class<?> clz) {
-        Method[] methods = clz.getDeclaredMethods();
-        for (Method method : methods) {
-            int mod = method.getModifiers();
-            if (Modifier.isStatic(mod) && Modifier.isPublic(mod)) {
-                registerFunc(method.getName(), method);
-            }
-        }
-    }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmMessage.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmMessage.java
index d3f629878f30..a2e25e2e42fd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmMessage.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmMessage.java
@@ -42,6 +42,5 @@ public class AlarmMessage {
     private List<Tag> tags;
     private long startTime;
     private transient int period;
-    private transient boolean onlyAsCondition;
     private Set<String> hooks = new HashSet<>();
 }
diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/mqe/rt/MQEVisitor.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/mqe/rt/MQEVisitor.java
index 0dd33bdc59f5..9f112429f62e 100644
--- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/mqe/rt/MQEVisitor.java
+++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/mqe/rt/MQEVisitor.java
@@ -60,8 +60,6 @@ public class MQEVisitor extends MQEVisitorBase {
     private final Entity entity;
     private final Duration duration;
 
-    private final static String GENERAL_LABEL_NAME = "_";
-
     public MQEVisitor(final MetricsQuery metricsQuery,
                       final RecordsQuery recordsQuery,
                       final Entity entity,
diff --git a/oap-server/server-starter/src/main/resources/alarm-settings.yml b/oap-server/server-starter/src/main/resources/alarm-settings.yml
index 95e9e0ff2994..4d726c932653 100755
--- a/oap-server/server-starter/src/main/resources/alarm-settings.yml
+++ b/oap-server/server-starter/src/main/resources/alarm-settings.yml
@@ -17,27 +17,21 @@
 rules:
   # Rule unique name, must be ended with `_rule`.
 #  endpoint_percent_rule:
-#    # Metrics value need to be long, double or int
-#    metrics-name: endpoint_percent
-#    threshold: 75
-#    op: <
+#    # A MQE expression, the result type must be `SINGLE_VALUE` and the root operation of the expression must be a Compare Operation
+     # which provides `1`(true) or `0`(false) result. When the result is `1`(true), the alarm will be triggered.
+#    expression: sum(endpoint_percent < 75) >= 3
 #    # The length of time to evaluate the metrics
 #    period: 10
-#    # How many times after the metrics match the condition, will trigger alarm
-#    count: 3
 #    # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
 #    silence-period: 10
 #    message: Successful rate of endpoint {name} is lower than 75%
   service_resp_time_rule:
-    metrics-name: service_resp_time
+    expression: sum(service_resp_time > 1000) >= 1
     # [Optional] Default, match all services in this metrics
     include-names:
       - dubbox-provider
       - dubbox-consumer
-    threshold: 1000
-    op: ">"
     period: 10
-    count: 1
     tags:
       level: WARNING
     # Binding the specific names of the hooks when the alarm is triggered. The name format is {hookType}.{hookName}


Test patch:
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatterTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatterTest.java
index d22cf0190d23..b1a4622880d2 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatterTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmMessageFormatterTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
-import org.apache.skywalking.oap.server.core.alarm.MetaInAlarm;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -26,38 +25,7 @@ public class AlarmMessageFormatterTest {
     @Test
     public void testStringFormatWithNoArg() {
         AlarmMessageFormatter formatter = new AlarmMessageFormatter("abc words {sdf");
-        String message = formatter.format(new MetaInAlarm() {
-
-            @Override
-            public String getScope() {
-                return "SERVICE";
-            }
-
-            @Override
-            public int getScopeId() {
-                return -1;
-            }
-
-            @Override
-            public String getName() {
-                return null;
-            }
-
-            @Override
-            public String getMetricsName() {
-                return null;
-            }
-
-            @Override
-            public String getId0() {
-                return "";
-            }
-
-            @Override
-            public String getId1() {
-                return "";
-            }
-        });
+        String message = formatter.format(new AlarmEntity("SERVICE", -1, null, "", ""));
 
         Assertions.assertEquals("abc words {sdf", message);
     }
@@ -65,38 +33,7 @@ public String getId1() {
     @Test
     public void testStringFormatWithArg() {
         AlarmMessageFormatter formatter = new AlarmMessageFormatter("abc} words {name} - {id} .. {");
-        String message = formatter.format(new MetaInAlarm() {
-
-            @Override
-            public String getScope() {
-                return "SERVICE";
-            }
-
-            @Override
-            public int getScopeId() {
-                return -1;
-            }
-
-            @Override
-            public String getName() {
-                return "service";
-            }
-
-            @Override
-            public String getMetricsName() {
-                return null;
-            }
-
-            @Override
-            public String getId0() {
-                return "1290";
-            }
-
-            @Override
-            public String getId1() {
-                return "";
-            }
-        });
+        String message = formatter.format(new AlarmEntity("SERVICE", -1, "service", "1290", ""));
         Assertions.assertEquals("abc} words service - 1290 .. {", message);
     }
 }
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
index 7b85106aecba..f63d6b167aa5 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmModuleProviderTest.java
@@ -23,6 +23,10 @@
 import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule;
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.alarm.AlarmModule;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.sql.Function;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
 import org.apache.skywalking.oap.server.library.module.ModuleProvider;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -49,6 +53,10 @@ public void setUp() throws Exception {
         moduleProvider.newConfigCreator();
 
         moduleProvider.prepare();
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "service_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Service.getScopeId());
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
     }
 
     @Test
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcherTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcherTest.java
index bc0acc0d0b88..ab9d6c8a1cd7 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcherTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/AlarmRulesWatcherTest.java
@@ -19,6 +19,10 @@
 package org.apache.skywalking.oap.server.core.alarm.provider;
 
 import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.sql.Function;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
 import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.library.util.ResourceUtils;
 import org.junit.jupiter.api.BeforeEach;
@@ -43,34 +47,13 @@ public class AlarmRulesWatcherTest {
     @Spy
     private AlarmRulesWatcher alarmRulesWatcher = new AlarmRulesWatcher(new Rules(), null);
 
-    private AlarmRule.AlarmRuleBuilder rulePrototypeBuilder = AlarmRule.builder()
-                                                                       .alarmRuleName("name1")
-                                                                       .count(1)
-                                                                       .includeNames(new ArrayList<String>() {
-                                                                           {
-                                                                               add("1");
-                                                                               add("2");
-                                                                           }
-                                                                       })
-                                                                       .excludeNames(new ArrayList<String>() {
-                                                                           {
-                                                                               add("3");
-                                                                               add("4");
-                                                                           }
-                                                                       })
-                                                                       .message("test")
-                                                                       .metricsName("metrics1")
-                                                                       .op(">")
-                                                                       .period(1)
-                                                                       .silencePeriod(2)
-                                                                       .tags(new HashMap<String, String>() {{
-                                                                           put("key", "value");
-                                                                       }})
-                                                                       .threshold("2");
-
     @BeforeEach
     public void setUp() {
         MockitoAnnotations.initMocks(this);
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "service_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Service.getScopeId());
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
     }
 
     @Test
@@ -83,11 +66,11 @@ public void shouldSetAlarmRulesOnEventChanged() throws IOException {
 
         alarmRulesWatcher.notify(new ConfigChangeWatcher.ConfigChangeEvent(new String(chars, 0, length), ConfigChangeWatcher.EventType.MODIFY));
 
-        assertEquals(3, alarmRulesWatcher.getRules().size());
+        assertEquals(5, alarmRulesWatcher.getRules().size());
         assertEquals(2, alarmRulesWatcher.getWebHooks().get(AlarmHooksType.webhook.name() + ".default").getUrls().size());
         assertNotNull(alarmRulesWatcher.getGrpchookSetting());
         assertEquals(9888, alarmRulesWatcher.getGrpchookSetting().get(AlarmHooksType.gRPC.name() + ".default").getTargetPort());
-        assertEquals(2, alarmRulesWatcher.getRunningContext().size());
+        assertEquals(4, alarmRulesWatcher.getRunningContext().size());
         assertNotNull(alarmRulesWatcher.getDingtalkSettings());
         assertNotNull(alarmRulesWatcher.getWechatSettings());
         assertEquals(2, alarmRulesWatcher.getSlackSettings().size());
@@ -111,82 +94,100 @@ public void shouldClearAlarmRulesOnEventDeleted() throws IOException {
 
     @Test
     public void shouldKeepExistedRunningRuleIfAlarmRuleExists() {
-        AlarmRule.AlarmRuleBuilder ruleBuilder = rulePrototypeBuilder;
-
-        AlarmRule rule = ruleBuilder.build();
+        AlarmRule rule = newAlarmRule("name1", "avg(service_percent) < 80");
         Rules rules = new Rules();
         rules.getRules().add(rule);
 
         alarmRulesWatcher = spy(new AlarmRulesWatcher(rules, null));
         assertEquals(1, alarmRulesWatcher.getRunningContext().size());
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).size());
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).size());
 
-        RunningRule runningRule = alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).get(0);
+        RunningRule runningRule = alarmRulesWatcher.getRunningContext().get(rule.getExpression()).get(0);
 
         Rules updatedRules = new Rules();
-        updatedRules.getRules().addAll(Arrays.asList(rule, ruleBuilder.alarmRuleName("name2").build()));
+        updatedRules.getRules().addAll(Arrays.asList(rule, newAlarmRule("name2", "avg(service_percent) < 80")));
 
         alarmRulesWatcher.notify(updatedRules);
 
         assertEquals(1, alarmRulesWatcher.getRunningContext().size());
-        assertEquals(2, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).size());
+        assertEquals(2, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).size());
         assertEquals(
-                runningRule, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).get(0),
+                runningRule, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).get(0),
                 "The same alarm rule should map to the same existed running rule");
     }
 
     @Test
     public void shouldRemoveRunningRuleIfAlarmRuleIsRemoved() {
-        AlarmRule.AlarmRuleBuilder ruleBuilder = rulePrototypeBuilder;
-
-        AlarmRule rule = ruleBuilder.build();
+        AlarmRule rule = newAlarmRule("name1", "avg(service_percent) < 80");
         Rules rules = new Rules();
         rules.getRules().add(rule);
 
         alarmRulesWatcher = spy(new AlarmRulesWatcher(rules, null));
         assertEquals(1, alarmRulesWatcher.getRunningContext().size());
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).size());
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).size());
 
-        RunningRule runningRule = alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).get(0);
+        RunningRule runningRule = alarmRulesWatcher.getRunningContext().get(rule.getExpression()).get(0);
 
         Rules updatedRules = new Rules();
-        updatedRules.getRules().add(ruleBuilder.alarmRuleName("name2").build());
+        updatedRules.getRules().add(newAlarmRule("name2", "avg(service_percent) < 80"));
 
         alarmRulesWatcher.notify(updatedRules);
 
         assertEquals(1, alarmRulesWatcher.getRunningContext().size());
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).size());
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).size());
         assertNotEquals(
-                runningRule, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).get(0),
+                runningRule, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).get(0),
                 "The new alarm rule should map to a different running rule");
     }
 
     @Test
     public void shouldReplaceRunningRuleIfAlarmRulesAreReplaced() {
-        AlarmRule.AlarmRuleBuilder ruleBuilder = rulePrototypeBuilder;
-
-        AlarmRule rule = ruleBuilder.build();
+        AlarmRule rule = newAlarmRule("name1", "avg(service_percent) < 80");
         Rules rules = new Rules();
         rules.getRules().add(rule);
 
         alarmRulesWatcher = spy(new AlarmRulesWatcher(rules, null));
         assertEquals(1, alarmRulesWatcher.getRunningContext().size());
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getMetricsName()).size());
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get(rule.getExpression()).size());
 
         Rules updatedRules = new Rules();
         // replace the original alarm rules
         updatedRules.getRules()
-                    .addAll(Arrays.asList(ruleBuilder.alarmRuleName("name2")
-                                                     .metricsName("metrics2")
-                                                     .build(), ruleBuilder.alarmRuleName("name3")
-                                                                          .metricsName("metrics3")
-                                                                          .build()));
+                    .addAll(Arrays.asList(
+                        newAlarmRule("name2", "avg(service_percent) < 90"),
+                        newAlarmRule("name3", "avg(service_percent) < 99")
+                    ));
 
         alarmRulesWatcher.notify(updatedRules);
 
         assertEquals(2, alarmRulesWatcher.getRunningContext().size());
-        assertNull(alarmRulesWatcher.getRunningContext().get("metrics1"));
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get("metrics2").size());
-        assertEquals(1, alarmRulesWatcher.getRunningContext().get("metrics3").size());
+        assertNull(alarmRulesWatcher.getRunningContext().get("avg(service_percent) < 80"));
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get("avg(service_percent) < 90").size());
+        assertEquals(1, alarmRulesWatcher.getRunningContext().get("avg(service_percent) < 99").size());
+    }
+
+    private AlarmRule newAlarmRule(String name, String expression) {
+       AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName(name);
+        alarmRule.setIncludeNames(new ArrayList<String>() {
+            {
+                add("1");
+                add("2");
+            }
+        });
+        alarmRule.setExcludeNames(new ArrayList<String>() {
+            {
+                add("3");
+                add("4");
+            }
+        });
+        alarmRule.setMessage("test");
+        alarmRule.setExpression(expression);
+        alarmRule.setPeriod(1);
+        alarmRule.setSilencePeriod(2);
+        alarmRule.setTags(new HashMap<String, String>() {{
+            put("key", "value");
+        }});
+        return alarmRule;
     }
 }
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluatorTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluatorTest.java
deleted file mode 100644
index 9ccddc07477d..000000000000
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/CompositeRuleEvaluatorTest.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import java.util.HashSet;
-import org.apache.skywalking.oap.server.core.alarm.AlarmMessage;
-import org.apache.skywalking.oap.server.core.alarm.provider.expression.Expression;
-import org.apache.skywalking.oap.server.core.alarm.provider.expression.ExpressionContext;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-public class CompositeRuleEvaluatorTest {
-
-    private CompositeRuleEvaluator ruleEvaluate;
-
-    @BeforeEach
-    public void init() {
-        Expression expression = new Expression(new ExpressionContext());
-        ruleEvaluate = new CompositeRuleEvaluator(expression);
-    }
-
-    @Test
-    public void testEvaluateMessageWithAndOp() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "a_rule && b_rule", "composite rule {name},{id} triggered!", new HashMap<String, String>() {{
-            put("key", "value");
-        }}, new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(1);
-        assertThat(compositeMsgs.get(0).getAlarmMessage()).isEqualTo("composite rule demo service,id0 triggered!");
-        assertThat(compositeMsgs.get(0).getRuleName()).isEqualTo("dummy");
-        assertThat(compositeMsgs.get(0).getId0()).isEqualTo("id0");
-        assertThat(compositeMsgs.get(0).getId1()).isEqualTo("id1");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getKey()).isEqualTo("key");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getValue()).isEqualTo("value");
-        assertThat(compositeMsgs.get(0).isOnlyAsCondition()).isEqualTo(false);
-    }
-
-    @Test
-    public void testEvaluateMessageWithFormatMessage() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "a_rule && b_rule", "composite rule {name} triggered!", new HashMap<String, String>() {{
-            put("key", "value");
-        }}, new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(1);
-        assertThat(compositeMsgs.get(0).getAlarmMessage()).isEqualTo("composite rule demo service triggered!");
-        assertThat(compositeMsgs.get(0).getRuleName()).isEqualTo("dummy");
-        assertThat(compositeMsgs.get(0).getId0()).isEqualTo("id0");
-        assertThat(compositeMsgs.get(0).getId1()).isEqualTo("id1");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getKey()).isEqualTo("key");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getValue()).isEqualTo("value");
-        assertThat(compositeMsgs.get(0).isOnlyAsCondition()).isEqualTo(false);
-    }
-
-    @Test
-    public void testEvaluateMessageWithNotExistsRule() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "a_rule && not_exist_rule", "composite rule triggered!", new HashMap<>(), new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(0);
-    }
-
-    @Test
-    public void testEvaluateMessageWithException() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "a_rule + b_rule", "composite rule triggered!", new HashMap<>(), new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(0);
-    }
-
-    private List<AlarmMessage> getAlarmMessages() {
-        List<AlarmMessage> alarmMessages = new ArrayList<>();
-        AlarmMessage alarmMessage = new AlarmMessage();
-        alarmMessage.setRuleName("a_rule");
-        alarmMessage.setOnlyAsCondition(true);
-        alarmMessage.setId0("id0");
-        alarmMessage.setId1("id1");
-        alarmMessage.setName("demo service");
-        alarmMessage.setScope("");
-        alarmMessage.setScopeId(1);
-        alarmMessages.add(alarmMessage);
-        alarmMessage = new AlarmMessage();
-        alarmMessage.setRuleName("b_rule");
-        alarmMessage.setOnlyAsCondition(true);
-        alarmMessage.setId0("id0");
-        alarmMessage.setId1("id1");
-        alarmMessage.setName("demo service");
-        alarmMessage.setScope("");
-        alarmMessage.setScopeId(1);
-        alarmMessages.add(alarmMessage);
-        alarmMessage = new AlarmMessage();
-        alarmMessage.setRuleName("c_rule");
-        alarmMessage.setOnlyAsCondition(true);
-        alarmMessage.setId0("id0");
-        alarmMessage.setId1("id1");
-        alarmMessage.setName("demo service");
-        alarmMessage.setScope("");
-        alarmMessage.setScopeId(1);
-        alarmMessages.add(alarmMessage);
-        return alarmMessages;
-    }
-
-    @Test
-    public void testEvaluateMessageWithOrOp() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "a_rule || b_rule", "composite rule triggered!", new HashMap<String, String>() {{
-            put("key", "value");
-        }}, new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        alarmMessages.remove(0);
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(1);
-        assertThat(compositeMsgs.get(0).getAlarmMessage()).isEqualTo("composite rule triggered!");
-        assertThat(compositeMsgs.get(0).getRuleName()).isEqualTo("dummy");
-        assertThat(compositeMsgs.get(0).getId0()).isEqualTo("id0");
-        assertThat(compositeMsgs.get(0).getId1()).isEqualTo("id1");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getKey()).isEqualTo("key");
-        assertThat(compositeMsgs.get(0).getTags().get(0).getValue()).isEqualTo("value");
-        assertThat(compositeMsgs.get(0).isOnlyAsCondition()).isEqualTo(false);
-    }
-
-    @Test
-    public void testEvaluateMessageWithParenthesisAndOp() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "(a_rule || b_rule) && c_rule", "composite rule triggered!", new HashMap<>(), new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        alarmMessages.remove(alarmMessages.size() - 1);
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(0);
-    }
-
-    @Test
-    public void testEvaluateMessageWithParenthesisAndOrOp() {
-        List<CompositeAlarmRule> compositeAlarmRules = new ArrayList<>();
-        CompositeAlarmRule compositeAlarmRule = new CompositeAlarmRule("dummy", "(a_rule && b_rule) || c_rule", "composite rule triggered!", new HashMap<>(), new HashSet<>());
-        compositeAlarmRules.add(compositeAlarmRule);
-        List<AlarmMessage> alarmMessages = getAlarmMessages();
-        List<AlarmMessage> compositeMsgs = ruleEvaluate.evaluate(compositeAlarmRules, alarmMessages);
-        assertThat(compositeMsgs.size()).isEqualTo(1);
-        assertThat(compositeMsgs.get(0).getAlarmMessage()).isEqualTo("composite rule triggered!");
-        assertThat(compositeMsgs.get(0).getRuleName()).isEqualTo("dummy");
-        assertThat(compositeMsgs.get(0).getId0()).isEqualTo("id0");
-        assertThat(compositeMsgs.get(0).getId1()).isEqualTo("id1");
-        assertThat(compositeMsgs.get(0).isOnlyAsCondition()).isEqualTo(false);
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
index 234c134de53d..784828295e4e 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/NotifyHandlerTest.java
@@ -109,7 +109,6 @@ public void testNotifyWithEndpointCatalog() {
         assertTrue(metaInAlarm instanceof EndpointMetaInAlarm);
         assertEquals("c2VydmljZS1pbnZlbnRvcnktbmFtZQ==.1_ZW5kcG9pbnQtaW52ZW50b3J5LW5hbWU=", metaInAlarm.getId0());
         assertEquals(DefaultScopeDefine.ENDPOINT_CATALOG_NAME, metaInAlarm.getScope());
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals(endpointInventoryName + " in " + serviceInventoryName, metaInAlarm.getName());
         assertEquals(DefaultScopeDefine.ENDPOINT, metaInAlarm.getScopeId());
 
@@ -135,7 +134,6 @@ public void testNotifyWithServiceInstanceCatalog() {
         MetaInAlarm metaInAlarm = metaCaptor.getValue();
 
         assertTrue(metaInAlarm instanceof ServiceInstanceMetaInAlarm);
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals("c2VydmljZQ==.1_aW5zdGFuY2UtaW52ZW50b3J5LW5hbWU=", metaInAlarm.getId0());
         assertEquals(DefaultScopeDefine.SERVICE_INSTANCE_CATALOG_NAME, metaInAlarm.getScope());
         assertEquals("instance-inventory-name of service", metaInAlarm.getName());
@@ -158,7 +156,6 @@ public void testNotifyWithServiceCatalog() {
         MetaInAlarm metaInAlarm = metaCaptor.getValue();
 
         assertTrue(metaInAlarm instanceof ServiceMetaInAlarm);
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals("c2VydmljZQ==.1", metaInAlarm.getId0());
         assertEquals(DefaultScopeDefine.SERVICE_CATALOG_NAME, metaInAlarm.getScope());
         assertEquals("service", metaInAlarm.getName());
@@ -184,7 +181,6 @@ public void testNotifyWithServiceRelationCatalog() {
         MetaInAlarm metaInAlarm = metaCaptor.getValue();
 
         assertTrue(metaInAlarm instanceof ServiceRelationMetaInAlarm);
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals("ZnJvbS1zZXJ2aWNl.1", metaInAlarm.getId0());
         assertEquals("ZGVzdC1zZXJ2aWNl.1", metaInAlarm.getId1());
         assertEquals(DefaultScopeDefine.SERVICE_RELATION_CATALOG_NAME, metaInAlarm.getScope());
@@ -211,7 +207,6 @@ public void testNotifyWithServiceInstanceRelationCatalog() {
         MetaInAlarm metaInAlarm = metaCaptor.getValue();
 
         assertTrue(metaInAlarm instanceof ServiceInstanceRelationMetaInAlarm);
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals("ZnJvbS1zZXJ2aWNl.1_ZnJvbS1zZXJ2aWNlLWluc3RhbmNl", metaInAlarm.getId0());
         assertEquals("ZGVzdC1zZXJ2aWNl.1_ZGVzdC1zZXJ2aWNlLWluc3RhbmNl", metaInAlarm.getId1());
         assertEquals(DefaultScopeDefine.SERVICE_INSTANCE_RELATION_CATALOG_NAME, metaInAlarm.getScope());
@@ -238,7 +233,6 @@ public void testNotifyWithEndpointRelationCatalog() {
         MetaInAlarm metaInAlarm = metaCaptor.getValue();
 
         assertTrue(metaInAlarm instanceof EndpointRelationMetaInAlarm);
-        assertEquals(metricsName, metaInAlarm.getMetricsName());
         assertEquals("ZnJvbS1zZXJ2aWNl.1_L3NvdXJjZS1wYXRo", metaInAlarm.getId0());
         assertEquals("ZGVzdC1zZXJ2aWNl.1_L2Rlc3QtcGF0aA==", metaInAlarm.getId1());
         assertEquals(DefaultScopeDefine.ENDPOINT_RELATION_CATALOG_NAME, metaInAlarm.getScope());
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/OPTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/OPTest.java
deleted file mode 100644
index 8051f344e672..000000000000
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/OPTest.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class OPTest {
-    @Test
-    public void test() {
-        assertTrue(OP.EQ.test(123, 123));
-        assertTrue(OP.EQ.test(123L, 123L));
-        assertTrue(OP.EQ.test(123.0D, 123.0D));
-
-        assertTrue(OP.NEQ.test(124, 123));
-        assertTrue(OP.NEQ.test(124L, 123L));
-        assertTrue(OP.NEQ.test(124.0D, 123.0D));
-
-        assertTrue(OP.GT.test(122, 123));
-        assertTrue(OP.GT.test(122L, 123L));
-        assertTrue(OP.GT.test(122.0D, 123.0D));
-
-        assertTrue(OP.GTE.test(122, 123));
-        assertTrue(OP.GTE.test(122L, 123L));
-        assertTrue(OP.GTE.test(122.0D, 123.0D));
-        assertTrue(OP.GTE.test(122, 122));
-        assertTrue(OP.GTE.test(122L, 122L));
-        assertTrue(OP.GTE.test(122.0D, 122.0D));
-
-        assertTrue(OP.LT.test(124, 123));
-        assertTrue(OP.LT.test(124L, 123L));
-        assertTrue(OP.LT.test(124.0D, 123.0D));
-
-        assertTrue(OP.LTE.test(124, 124));
-        assertTrue(OP.LTE.test(124L, 124L));
-        assertTrue(OP.LTE.test(124.0D, 124.0D));
-        assertTrue(OP.LTE.test(124, 123));
-        assertTrue(OP.LTE.test(124L, 123L));
-        assertTrue(OP.LTE.test(124.0D, 123.0D));
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReaderTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReaderTest.java
index 2a757c5406d8..6c0e6a84fe59 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReaderTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RulesReaderTest.java
@@ -26,17 +26,28 @@
 import org.apache.skywalking.oap.server.core.alarm.provider.webhook.WebhookSettings;
 import org.apache.skywalking.oap.server.core.alarm.provider.wechat.WechatSettings;
 import org.apache.skywalking.oap.server.core.alarm.provider.welink.WeLinkSettings;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.sql.Function;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-
 import java.util.List;
-
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class RulesReaderTest {
+    @BeforeEach
+    public void setUp() {
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "service_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Service.getScopeId());
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
+    }
+
     @Test
     public void testReadRules() {
         RulesReader reader = new RulesReader(this.getClass()
@@ -45,8 +56,8 @@ public void testReadRules() {
         Rules rules = reader.readRules();
 
         List<AlarmRule> ruleList = rules.getRules();
-        Assertions.assertEquals(3, ruleList.size());
-        Assertions.assertEquals("85", ruleList.get(1).getThreshold());
+        Assertions.assertEquals(5, ruleList.size());
+        Assertions.assertEquals("sum(service_percent < 85) >= 4", ruleList.get(1).getExpression());
         Assertions.assertEquals("endpoint_percent_rule", ruleList.get(0).getAlarmRuleName());
         Assertions.assertEquals(0, ruleList.get(0).getIncludeNames().size());
         Assertions.assertEquals(0, ruleList.get(0).getExcludeNames().size());
@@ -82,14 +93,6 @@ public void testReadRules() {
         assertThat(wechatSettings.getWebhooks().get(0)).isEqualTo("https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=dummy_key");
         assertThat(slackSettings.getTextTemplate()).isInstanceOfAny(String.class);
 
-        List<CompositeAlarmRule> compositeRules = rules.getCompositeRules();
-        Assertions.assertEquals(2, compositeRules.size());
-        Assertions.assertEquals("endpoint_percent_more_rule && endpoint_percent_rule", compositeRules.get(0).getExpression());
-        //comp1_rule's hooks
-        Assertions.assertEquals(8, compositeRules.get(0).getHooks().size());
-        //comp2_rule's hooks
-        Assertions.assertEquals(3, compositeRules.get(1).getHooks().size());
-
         DingtalkSettings dingtalkSettings = rules.getDingtalkSettingsMap().get(AlarmHooksType.dingtalk.name() + ".default");
         assertThat(dingtalkSettings.getTextTemplate()).isInstanceOfAny(String.class);
         List<DingtalkSettings.WebHookUrl> webHookUrls = dingtalkSettings.getWebhooks();
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java
index a272d2279f62..43dbe96fc7f4 100644
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java
+++ b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/RunningRuleTest.java
@@ -30,13 +30,18 @@
 import org.apache.skywalking.oap.server.core.analysis.metrics.LabeledValueHolder;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.metrics.MultiIntValuesHolder;
+import org.apache.skywalking.oap.server.core.query.enumeration.Scope;
+import org.apache.skywalking.oap.server.core.query.sql.Function;
 import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
 import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
 import org.apache.skywalking.oap.server.core.storage.StorageID;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
 import org.joda.time.LocalDateTime;
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.powermock.reflect.Whitebox;
 
@@ -54,14 +59,22 @@
 public class RunningRuleTest {
     private static DateTimeFormatter TIME_BUCKET_FORMATTER = DateTimeFormat.forPattern("yyyyMMddHHmm");
 
+    @BeforeEach
+    public void setup() {
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_percent", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_multiple_values", "testColumn", Column.ValueDataType.LABELED_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_cpm", "testColumn", Column.ValueDataType.COMMON_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
+    }
+
     @Test
     public void testInitAndStart() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("75");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_percent < 75) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(15);
         alarmRule.setTags(new HashMap<String, String>() {{
             put("key", "value");
@@ -71,9 +84,9 @@ public void testInitAndStart() {
         long timeInPeriod1 = 201808301434L;
         runningRule.in(getMetaInAlarm(123), getMetrics(timeInPeriod1, 70));
 
-        Map<MetaInAlarm, RunningRule.Window> windows = Whitebox.getInternalState(runningRule, "windows");
+        Map<AlarmEntity, RunningRule.Window> windows = Whitebox.getInternalState(runningRule, "windows");
 
-        RunningRule.Window window = windows.get(getMetaInAlarm(123));
+        RunningRule.Window window = windows.get(getAlarmEntity(123));
         LocalDateTime endTime = Whitebox.getInternalState(window, "endTime");
         int period = Whitebox.getInternalState(window, "period");
         LinkedList<Metrics> metricsBuffer = Whitebox.getInternalState(window, "values");
@@ -87,10 +100,8 @@ public void testInitAndStart() {
     public void testAlarm() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("75");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_percent < 75) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(15);
         alarmRule.setMessage("Successful rate of endpoint {name} is lower than 75%");
         alarmRule.setTags(new HashMap<String, String>() {{
@@ -121,10 +132,8 @@ public void testAlarm() {
     public void testMultipleValuesAlarm() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_multiple_values_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp(">");
-        alarmRule.setThreshold("50,60,70,-, 100");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_multiple_values > 50) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_multiple_values");
         alarmRule.setPeriod(15);
         alarmRule.setMessage("response percentile of endpoint {name} is lower than expected values");
         alarmRule.setTags(new HashMap<String, String>() {{
@@ -137,15 +146,15 @@ public void testMultipleValuesAlarm() {
         long timeInPeriod2 = 201808301436L;
         long timeInPeriod3 = 201808301438L;
 
-        runningRule.in(getMetaInAlarm(123), getMultipleValueMetrics(timeInPeriod1, 70, 60, 40, 40, 40));
-        runningRule.in(getMetaInAlarm(123), getMultipleValueMetrics(timeInPeriod2, 60, 60, 40, 40, 40));
+        runningRule.in(getMetaInAlarm(123, "endpoint_multiple_values"), getMultipleValueMetrics(timeInPeriod1, 70, 60, 40, 40, 40));
+        runningRule.in(getMetaInAlarm(123, "endpoint_multiple_values"), getMultipleValueMetrics(timeInPeriod2, 60, 60, 40, 40, 40));
 
         // check at 201808301440
         List<AlarmMessage> alarmMessages = runningRule.check();
         Assertions.assertEquals(0, alarmMessages.size());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
 
-        runningRule.in(getMetaInAlarm(123), getMultipleValueMetrics(timeInPeriod3, 74, 60, 40, 40, 40));
+        runningRule.in(getMetaInAlarm(123, "endpoint_multiple_values"), getMultipleValueMetrics(timeInPeriod3, 74, 60, 40, 40, 40));
 
         // check at 201808301440
         alarmMessages = runningRule.check();
@@ -155,28 +164,67 @@ public void testMultipleValuesAlarm() {
 
     @Test
     public void testLabeledAlarm() {
+        ValueColumnMetadata.INSTANCE.putIfAbsent(
+            "endpoint_labeled", "testColumn", Column.ValueDataType.LABELED_VALUE, Function.Avg, 0, Scope.Endpoint.getScopeId());
         AlarmRule alarmRule = new AlarmRule();
-        alarmRule.setIncludeLabels(Lists.newArrayList("95", "99"));
-        assertLabeled(alarmRule);
-        alarmRule = new AlarmRule();
-        alarmRule.setIncludeLabelsRegex("9\\d{1}");
-        assertLabeled(alarmRule);
-        alarmRule = new AlarmRule();
-        alarmRule.setExcludeLabels(Lists.newArrayList("50", "75"));
-        assertLabeled(alarmRule);
-        alarmRule = new AlarmRule();
-        alarmRule.setExcludeLabelsRegex("^[5-7][0-9]$");
-        assertLabeled(alarmRule);
+        alarmRule.setExpression("sum(endpoint_labeled{_='95,99'} > 10) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_labeled");
+        assertLabeled(alarmRule, "50,17|99,11", "75,15|95,12|99,12", "90,1|99,20", 1);
+        alarmRule.setExpression("sum(endpoint_labeled > 10) >= 3");
+        assertLabeled(alarmRule, "50,17|99,11", "75,15|95,12|99,12", "90,1|99,20", 1);
+        alarmRule.setExpression("sum(endpoint_labeled{_='50'} > 10) >= 3");
+        assertLabeled(alarmRule, "50,17|99,11", "75,15|95,12|99,12", "90,1|99,20", 0);
+    }
+
+    @Test
+    public void testMultipleMetricsAlarm() {
+        multipleMetricsAlarm("sum((endpoint_percent < 75) * (endpoint_cpm < 100)) >= 3", 1);
+    }
+
+    @Test
+    public void testMultipleMetricsNoAlarm() {
+        multipleMetricsAlarm("sum((endpoint_percent < 75) * (endpoint_cpm < 99)) >= 3", 0);
+    }
+
+    private void multipleMetricsAlarm(String expression, int alarmMsgSize) {
+        AlarmRule alarmRule = new AlarmRule();
+        alarmRule.setAlarmRuleName("endpoint_percent_rule");
+        alarmRule.setExpression(expression);
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
+        alarmRule.getIncludeMetrics().add("endpoint_cpm");
+        alarmRule.setPeriod(15);
+        alarmRule.setMessage("Successful rate of endpoint {name} is lower than 75% and cpm is lower than 100");
+        alarmRule.setTags(new HashMap<String, String>() {{
+            put("key", "value");
+        }});
+        RunningRule runningRule = new RunningRule(alarmRule);
+        LocalDateTime startTime = TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301440");
+
+        long timeInPeriod1 = 201808301434L;
+        long timeInPeriod2 = 201808301436L;
+        long timeInPeriod3 = 201808301438L;
+
+        runningRule.in(getMetaInAlarm(123), getMetrics(timeInPeriod1, 70));
+        runningRule.in(getMetaInAlarm(123), getMetrics(timeInPeriod2, 71));
+        runningRule.in(getMetaInAlarm(123, "endpoint_cpm"), getMetrics(timeInPeriod1, 50));
+        runningRule.in(getMetaInAlarm(123, "endpoint_cpm"), getMetrics(timeInPeriod2, 99));
+
+        List<AlarmMessage> alarmMessages = runningRule.check();
+        Assertions.assertEquals(0, alarmMessages.size());
+
+        runningRule.in(getMetaInAlarm(123), getMetrics(timeInPeriod3, 74));
+        runningRule.in(getMetaInAlarm(123, "endpoint_cpm"), getMetrics(timeInPeriod3, 60));
+
+        alarmMessages = runningRule.check();
+        Assertions.assertEquals(alarmMsgSize, alarmMessages.size());
     }
 
     @Test
     public void testNoAlarm() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp(">");
-        alarmRule.setThreshold("75");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_percent > 75) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(15);
         //alarmRule.setSilencePeriod(0);
         alarmRule.setTags(new HashMap<String, String>() {{
@@ -220,10 +268,8 @@ public void doAlarm(List<AlarmMessage> alarmMessage) {
     public void testSilence() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("75");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_percent < 75) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(15);
         alarmRule.setSilencePeriod(2);
         alarmRule.setTags(new HashMap<String, String>() {{
@@ -260,10 +306,8 @@ public void testSilence() {
     public void testExclude() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("75");
-        alarmRule.setCount(3);
+        alarmRule.setExpression("sum(endpoint_percent < 75) >= 3");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(15);
         alarmRule.setMessage("Successful rate of endpoint {name} is lower than 75%");
         alarmRule.setExcludeNames(Lists.newArrayList("Service_123"));
@@ -294,10 +338,8 @@ public void testExclude() {
     public void testIncludeNamesRegex() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("1000");
-        alarmRule.setCount(1);
+        alarmRule.setExpression("sum(endpoint_percent < 1000) >= 1");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(10);
         alarmRule.setMessage(
             "Response time of service instance {name} is more than 1000ms in 2 minutes of last 10 minutes");
@@ -329,10 +371,8 @@ public void testIncludeNamesRegex() {
     public void testExcludeNamesRegex() {
         AlarmRule alarmRule = new AlarmRule();
         alarmRule.setAlarmRuleName("endpoint_percent_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp("<");
-        alarmRule.setThreshold("1000");
-        alarmRule.setCount(1);
+        alarmRule.setExpression("sum(endpoint_percent < 1000) >= 1");
+        alarmRule.getIncludeMetrics().add("endpoint_percent");
         alarmRule.setPeriod(10);
         alarmRule.setMessage(
             "Response time of service instance {name} is more than 1000ms in 2 minutes of last 10 minutes");
@@ -361,6 +401,10 @@ public void testExcludeNamesRegex() {
     }
 
     private MetaInAlarm getMetaInAlarm(int id) {
+        return getMetaInAlarm(id, "endpoint_percent");
+    }
+
+    private MetaInAlarm getMetaInAlarm(int id, String metricName) {
         return new MetaInAlarm() {
             @Override
             public String getScope() {
@@ -379,7 +423,7 @@ public String getName() {
 
             @Override
             public String getMetricsName() {
-                return "endpoint_percent";
+                return metricName;
             }
 
             @Override
@@ -427,6 +471,13 @@ private Metrics getLabeledValueMetrics(long timeBucket, String values) {
         return mockLabeledValueMetrics;
     }
 
+    private AlarmEntity getAlarmEntity(int id) {
+        MetaInAlarm metaInAlarm = getMetaInAlarm(id);
+        return new AlarmEntity(metaInAlarm.getScope(), metaInAlarm.getScopeId(), metaInAlarm.getName(),
+                               metaInAlarm.getId0(), metaInAlarm.getId1()
+        );
+    }
+
     private class MockMetrics extends Metrics implements IntValueHolder {
         private int value;
 
@@ -580,12 +631,8 @@ public RemoteData.Builder serialize() {
         }
     }
 
-    private void assertLabeled(AlarmRule alarmRule) {
-        alarmRule.setAlarmRuleName("endpoint_percent_alarm_rule");
-        alarmRule.setMetricsName("endpoint_percent");
-        alarmRule.setOp(">");
-        alarmRule.setThreshold("10");
-        alarmRule.setCount(3);
+    private void assertLabeled(AlarmRule alarmRule, String value1, String value2, String value3, int alarmMsgSize) {
+        alarmRule.setAlarmRuleName("endpoint_labeled_alarm_rule");
         alarmRule.setPeriod(15);
         alarmRule.setMessage("response percentile of endpoint {name} is lower than expected value");
         alarmRule.setTags(new HashMap<String, String>() {{
@@ -598,18 +645,18 @@ private void assertLabeled(AlarmRule alarmRule) {
         long timeInPeriod2 = 201808301436L;
         long timeInPeriod3 = 201808301438L;
 
-        runningRule.in(getMetaInAlarm(123), getLabeledValueMetrics(timeInPeriod1, "50,17|99,11"));
-        runningRule.in(getMetaInAlarm(123), getLabeledValueMetrics(timeInPeriod2, "75,15|95,12"));
+        runningRule.in(getMetaInAlarm(123, "endpoint_labeled"), getLabeledValueMetrics(timeInPeriod1, value1));
+        runningRule.in(getMetaInAlarm(123, "endpoint_labeled"), getLabeledValueMetrics(timeInPeriod2, value2));
 
         List<AlarmMessage> alarmMessages = runningRule.check();
         Assertions.assertEquals(0, alarmMessages.size());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
 
-        runningRule.in(getMetaInAlarm(123), getLabeledValueMetrics(timeInPeriod3, "90,1|99,20"));
+        runningRule.in(getMetaInAlarm(123, "endpoint_labeled"), getLabeledValueMetrics(timeInPeriod3, value3));
 
         // check at 201808301440
         alarmMessages = runningRule.check();
-        Assertions.assertEquals(1, alarmMessages.size());
+        Assertions.assertEquals(alarmMsgSize, alarmMessages.size());
         runningRule.moveTo(TIME_BUCKET_FORMATTER.parseLocalDateTime("201808301441"));
     }
 }
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/ThresholdTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/ThresholdTest.java
deleted file mode 100644
index 053154e26e7d..000000000000
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/ThresholdTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.alarm.provider;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class ThresholdTest {
-
-    @Test
-    public void setType() {
-        Threshold threshold = new Threshold("my-rule", "75");
-        threshold.setType(MetricsValueType.DOUBLE);
-        assertEquals(0, Double.compare(75, threshold.getDoubleThreshold()));
-
-        threshold.setType(MetricsValueType.INT);
-        assertEquals(75, threshold.getIntThreshold());
-
-        threshold.setType(MetricsValueType.LONG);
-        assertEquals(75L, threshold.getLongThreshold());
-    }
-
-    @Test
-    public void setTypeMultipleValues() {
-        Threshold threshold = new Threshold("my-rule", "75,80, 90, -");
-        threshold.setType(MetricsValueType.MULTI_INTS);
-        assertArrayEquals(new Object[] {
-            75,
-            80,
-            90,
-            null
-        }, threshold.getIntValuesThreshold());
-
-    }
-
-    @Test
-    public void setTypeWithWrong() {
-        Threshold threshold = new Threshold("my-rule", "wrong");
-        threshold.setType(MetricsValueType.INT);
-        assertEquals(0, threshold.getIntThreshold());
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContextTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContextTest.java
deleted file mode 100644
index 1ff05a0caa99..000000000000
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionContextTest.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.alarm.provider.expression;
-
-import org.junit.jupiter.api.Test;
-
-import java.lang.reflect.Method;
-import java.util.Arrays;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-public class ExpressionContextTest {
-
-    @Test
-    public void testRegisterFuncWithMethod() throws NoSuchMethodException {
-        ExpressionContext expressionContext = new ExpressionContext();
-        Method[] methods = Math.class.getMethods();
-        Arrays.stream(methods).forEach(method -> {
-            if (method.getName().equalsIgnoreCase("sqrt")) {
-                expressionContext.registerFunc("sqrt", method);
-            }
-        });
-        Expression expression = new Expression(expressionContext);
-        Number number = (Number) expression.eval("sqrt(16)");
-        assertThat(number).isEqualTo(4.0);
-    }
-
-    @Test
-    public void testRegisterFuncWithClazz() throws NoSuchMethodException {
-        ExpressionContext expressionContext = new ExpressionContext();
-        expressionContext.registerFunc(Math.class);
-        Expression expression = new Expression(expressionContext);
-        Number number = (Number) expression.eval("sqrt(16)");
-        assertThat(number).isEqualTo(4.0);
-        number = (Number) expression.eval("abs(-12)");
-        assertThat(number).isEqualTo(12);
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionTest.java b/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionTest.java
deleted file mode 100644
index 3f81b0af2a8c..000000000000
--- a/oap-server/server-alarm-plugin/src/test/java/org/apache/skywalking/oap/server/core/alarm/provider/expression/ExpressionTest.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.alarm.provider.expression;
-
-import com.google.common.collect.Sets;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mvel2.CompileException;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-public class ExpressionTest {
-    private Expression expression;
-
-    @BeforeEach
-    public void init() {
-        expression = new Expression(new ExpressionContext());
-    }
-
-    @Test
-    public void testEval() {
-        String expr = " a && b ";
-        Map<String, Object> dataMap = new HashMap();
-        dataMap.put("a", Boolean.TRUE);
-        Object flag = expression.eval(expr, dataMap);
-        assertNull(flag);
-        dataMap.put("b", Boolean.TRUE);
-        flag = expression.eval(expr, dataMap);
-        assertThat(flag).isEqualTo(true);
-    }
-
-    @Test
-    public void testAnalysisInputs() {
-        String expr = " a && b ";
-        Set<String> inputs = expression.analysisInputs(expr);
-        assertThat(inputs.size()).isEqualTo(2);
-        assertThat(inputs).isEqualTo(Sets.newHashSet("a", "b"));
-    }
-
-    @Test
-    public void testEvalWithEmptyContext() {
-        String expr = " a && b ";
-        Object flag = expression.eval(expr);
-        assertNull(flag);
-        flag = expression.eval(" 1 > 0");
-        assertThat(flag).isEqualTo(true);
-    }
-
-    @Test
-    public void testCompile() {
-        String expr = " a && b ";
-        ExpressionContext context = new ExpressionContext();
-        Object compiledExpression = expression.compile(expr, context);
-        assertNotNull(compiledExpression);
-        Object sameExpression = expression.compile(expr, context);
-        assertThat(compiledExpression).isEqualTo(sameExpression);
-    }
-
-    @Test
-    public void testCompileWithException() {
-        assertThrows(CompileException.class, () -> {
-            String expr = " a && * b ";
-            ExpressionContext context = new ExpressionContext();
-            expression.compile(expr, context);
-        });
-    }
-}
diff --git a/oap-server/server-alarm-plugin/src/test/resources/alarm-settings.yml b/oap-server/server-alarm-plugin/src/test/resources/alarm-settings.yml
index 48f9b60f9a0c..f621021ea232 100755
--- a/oap-server/server-alarm-plugin/src/test/resources/alarm-settings.yml
+++ b/oap-server/server-alarm-plugin/src/test/resources/alarm-settings.yml
@@ -16,57 +16,43 @@
 rules:
   # Rule unique name, must be ended with `_rule`.
   endpoint_percent_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: endpoint_percent
-    threshold: 75
-    op: <
-    # The length of time to evaluate the metrics
+    expression: sum(endpoint_percent < 75) >= 3
     period: 10
-    # How many times after the metrics match the condition, will trigger alarm
-    count: 3
     # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
     silence-period: 10
-    only-as-condition: false
     message: Successful rate of endpoint {name} is lower than 75%
 
   service_percent_rule:
-    metrics-name: service_percent
+    expression: sum(service_percent < 85) >= 4
     # [Optional] Default, match all services in this metrics
     include-names:
       - service_a
       - service_b
     exclude-names:
       - service_c
-    threshold: 85
-    op: <
     period: 10
-    count: 4
-    only-as-condition: false
 
   endpoint_percent_more_rule:
-    # Metrics value need to be long, double or int
-    metrics-name: endpoint_percent
-    threshold: 60
-    op: ">"
+    expression: sum(endpoint_percent > 60) >= 3
     # The length of time to evaluate the metrics
     period: 10
-    # How many times after the metrics match the condition, will trigger alarm
-    count: 3
     # How many times of checks, the alarm keeps silence after alarm triggered, default as same as period.
     silence-period: 10
-    # Control if this rule is just as a composite rule condition. do not notification alone
-    only-as-condition: false
     message: Successful rate of endpoint {name} is higher than 60%
     hooks:
       - "slack.custom1"
       - "pagerduty.custom1"
 
-composite-rules:
   comp1_rule:
-    expression: endpoint_percent_more_rule && endpoint_percent_rule
+    expression: sum((endpoint_percent > 60) * (endpoint_percent < 75)) >= 3
+    period: 10
+    silence-period: 10
     message: xxxxx
+
   comp2_rule:
-    expression: endpoint_percent_more_rule && endpoint_percent_rule
+    expression: sum((endpoint_percent > 60) * (endpoint_percent < 75)) >= 3
+    period: 10
+    silence-period: 10
     message: xxxxx
     hooks:
       - "slack.default"
diff --git a/test/e2e-v2/cases/alarm/alarm-settings.yml b/test/e2e-v2/cases/alarm/alarm-settings.yml
index 58843ac9d951..b7ba39f52b78 100755
--- a/test/e2e-v2/cases/alarm/alarm-settings.yml
+++ b/test/e2e-v2/cases/alarm/alarm-settings.yml
@@ -16,14 +16,10 @@
 rules:
   # service response time > 10ms
   service_resp_time_rule:
-    metrics-name: service_resp_time
-    op: ">"
-    threshold: 10
+    expression: sum(service_resp_time > 10) >= 1
     period: 10
-    count: 1
     silence-period: 1
     message: Response time of service {name} is more than 10ms in 1 minutes of last 10 minutes.
-    only-as-condition: false
     tags:
       level: WARNING
       receivers: lisi
@@ -31,17 +27,13 @@ rules:
       - webhook.custom
   # service sla > 1%
   service_sla_rule:
-    metrics-name: service_sla
-    op: ">"
-    threshold: 100
+    expression: sum(service_sla > 100) >= 1
     period: 10
-    count: 1
     silence-period: 1
     message: Successful rate of service {name} is more than 1% in 1 minutes of last 10 minutes
-    only-as-condition: true
-composite-rules:
   comp_rule:
-    expression: service_resp_time_rule && service_sla_rule
+    expression: sum((service_resp_time > 10) * (service_sla > 100)) >= 1
+    period: 10
     message: Service {name} response time is more than 10ms and sla is more than 1%.
     tags:
       level: CRITICAL