apache · tomsun28 · Dec 24, 2024 · Dec 8, 2024 · Dec 8, 2024 · Dec 8, 2024
diff --git a/README.md b/README.md
@@ -160,7 +160,7 @@ Detailed config refer to [Install HertzBeat via Package](https://hertzbeat.apach
 ##### 3：Start via source code        
 
 1. Local source code debugging needs to start the back-end project `manager` and the front-end project `web-app`.  
-2. Backend：need `maven3+`, `java17`, `lombok`, start the `manager` service.  
+2. Backend：need `maven3+`, `java17`, `lombok`, add VM options in IDE: ` --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED `, then start the `manager` service. 
 3. Web：need `nodejs npm angular-cli` environment, Run `ng serve --open` in `web-app` directory after backend startup.  
 4. Access `http://localhost:4200` to start, default account: `admin/hertzbeat`  
 

diff --git a/README_CN.md b/README_CN.md
@@ -159,7 +159,7 @@
 ##### 方式三：本地代码启动
 
 1. 此为前后端分离项目，本地代码调试需要分别启动后端工程 `manager` 和前端工程 `web-app`
-2. 后端：需要 `maven3+`, `java17` 和 `lombok` 环境，修改 `YML` 配置信息并启动 `manager` 服务
+2. 后端：需要 `maven3+`, `java17` 和 `lombok` 环境，修改 `YML` 配置信息，添加JVM参数`--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED`后启动 `manager` 服务即可。
 3. 前端：需要 `nodejs npm angular-cli`环境，待本地后端启动后，在 `web-app` 目录下启动 `ng serve --open`
 4. 浏览器访问 `http://localhost:4200` 即可开始，默认账号密码 `admin/hertzbeat`
 

diff --git a/hertzbeat-alerter/pom.xml b/hertzbeat-alerter/pom.xml
@@ -73,6 +73,14 @@
             <version>${easy-poi.version}</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.arrow</groupId>
+            <artifactId>arrow-vector</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.arrow</groupId>
+            <artifactId>arrow-memory-netty</artifactId>
+        </dependency>
     </dependencies>
 
 </project>
diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/calculate/CalculateAlarm.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/calculate/CalculateAlarm.java
@@ -26,7 +26,10 @@
 import static org.apache.hertzbeat.common.constants.CommonConstants.TAG_MONITOR_APP;
 import static org.apache.hertzbeat.common.constants.CommonConstants.TAG_MONITOR_ID;
 import static org.apache.hertzbeat.common.constants.CommonConstants.TAG_MONITOR_NAME;
+
+import com.google.common.collect.Maps;
 import jakarta.persistence.criteria.Predicate;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -45,8 +48,12 @@
 import org.apache.hertzbeat.alert.service.AlertService;
 import org.apache.hertzbeat.alert.util.AlertTemplateUtil;
 import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.constants.MetricDataFieldConstants;
 import org.apache.hertzbeat.common.entity.alerter.Alert;
 import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReader;
+import org.apache.hertzbeat.common.entity.arrow.ArrowVectorReaderImpl;
+import org.apache.hertzbeat.common.entity.arrow.RowWrapper;
 import org.apache.hertzbeat.common.entity.manager.Monitor;
 import org.apache.hertzbeat.common.entity.manager.TagItem;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -155,94 +162,100 @@ private void calculate(CollectRep.MetricsData metricsData) {
         if (defineMap.isEmpty()) {
             return;
         }
-        List<CollectRep.Field> fields = metricsData.getFieldsList();
-        Map<String, Object> fieldValueMap = new HashMap<>(8);
-        int valueRowCount = metricsData.getValuesCount();
-        for (Map.Entry<String, List<AlertDefine>> entry : defineMap.entrySet()) {
-            List<AlertDefine> defines = entry.getValue();
-            for (AlertDefine define : defines) {
-                final String expr = define.getExpr();
-                if (StringUtils.isBlank(expr)) {
-                    continue;
-                }
-                if (expr.contains(SYSTEM_VALUE_ROW_COUNT) && metricsData.getValuesCount() == 0) {
-                    fieldValueMap.put(SYSTEM_VALUE_ROW_COUNT, valueRowCount);
-                    try {
-                        boolean match = execAlertExpression(fieldValueMap, expr);
-                        try {
-                            if (match) {
-                                // If the threshold rule matches, the number of times the threshold has been triggered is determined and an alarm is triggered
-                                afterThresholdRuleMatch(currentTimeMilli, monitorId, app, metrics, "", fieldValueMap, define);
-                                // if the threshold is triggered, ignore other data rows
-                                continue;
-                            } else {
-                                String alarmKey = String.valueOf(monitorId) + define.getId();
-                                triggeredAlertMap.remove(alarmKey);
-                                if (define.isRecoverNotice()) {
-                                    handleRecoveredAlert(currentTimeMilli, define, expr, alarmKey);
-                                }
-                            }
-                        } catch (Exception e) {
-                            log.error(e.getMessage(), e);
-                        }
-                    } catch (Exception ignored) {}
-                }
-                for (CollectRep.ValueRow valueRow : metricsData.getValuesList()) {
+        Map<String, Object> fieldValueMap = Maps.newHashMapWithExpectedSize(8);
 
-                    if (CollectionUtils.isEmpty(valueRow.getColumnsList())) {
+        try (ArrowVectorReader arrowVectorReader = new ArrowVectorReaderImpl(metricsData.getData().toByteArray())) {
+            for (Map.Entry<String, List<AlertDefine>> entry : defineMap.entrySet()) {
+                List<AlertDefine> defines = entry.getValue();
+                for (AlertDefine define : defines) {
+                    final String expr = define.getExpr();
+                    if (StringUtils.isBlank(expr)) {
                         continue;
                     }
-                    fieldValueMap.clear();
-                    fieldValueMap.put(SYSTEM_VALUE_ROW_COUNT, valueRowCount);
-                    StringBuilder tagBuilder = new StringBuilder();
-                    for (int index = 0; index < valueRow.getColumnsList().size(); index++) {
-                        String valueStr = valueRow.getColumns(index);
-                        if (CommonConstants.NULL_VALUE.equals(valueStr)) {
-                            continue;
-                        }
-
-                        final CollectRep.Field field = fields.get(index);
-                        final int fieldType = field.getType();
-
-                        if (fieldType == CommonConstants.TYPE_NUMBER) {
-                            final Double doubleValue;
-                            if ((doubleValue = CommonUtil.parseStrDouble(valueStr)) != null) {
-                                fieldValueMap.put(field.getName(), doubleValue);
-                            }
-                        } else if (fieldType == CommonConstants.TYPE_TIME) {
-                            final Integer integerValue;
-                            if ((integerValue = CommonUtil.parseStrInteger(valueStr)) != null) {
-                                fieldValueMap.put(field.getName(), integerValue);
+
+                    if (expr.contains(SYSTEM_VALUE_ROW_COUNT) && arrowVectorReader.getRowCount() == 0L) {
+                        fieldValueMap.put(SYSTEM_VALUE_ROW_COUNT, arrowVectorReader.getRowCount());
+                        try {
+                            boolean match = execAlertExpression(fieldValueMap, expr);
+                            try {
+                                if (match) {
+                                    // If the threshold rule matches, the number of times the threshold has been triggered is determined and an alarm is triggered
+                                    afterThresholdRuleMatch(currentTimeMilli, monitorId, app, metrics, "", fieldValueMap, define);
+                                    // if the threshold is triggered, ignore other data rows
+                                    continue;
+                                } else {
+                                    String alarmKey = String.valueOf(monitorId) + define.getId();
+                                    triggeredAlertMap.remove(alarmKey);
+                                    if (define.isRecoverNotice()) {
+                                        handleRecoveredAlert(currentTimeMilli, define, expr, alarmKey);
+                                    }
+                                }
+                            } catch (Exception e) {
+                                log.error(e.getMessage(), e);
                             }
-                        } else {
-                            if (StringUtils.isNotEmpty(valueStr)) {
-                                fieldValueMap.put(field.getName(), valueStr);
+                        } catch (Exception ignored) {}
+                    }
+
+                    RowWrapper rowWrapper = arrowVectorReader.readRow();
+                    while (rowWrapper.hasNextRow()) {
+                        rowWrapper = rowWrapper.nextRow();
+
+                        String finalApp = app;
+                        rowWrapper.foreachCell(cell -> {
+                            fieldValueMap.clear();
+                            fieldValueMap.put(SYSTEM_VALUE_ROW_COUNT, arrowVectorReader.getRowCount());
+                            StringBuilder tagBuilder = new StringBuilder();
+
+                            String valueStr = cell.getValue();
+                            if (CommonConstants.NULL_VALUE.equals(valueStr)) {
+                                return;
                             }
-                        }
 
-                        if (field.getLabel()) {
-                            tagBuilder.append("-").append(valueStr);
-                        }
-                    }
-                    try {
-                        boolean match = execAlertExpression(fieldValueMap, expr);
-                        try {
-                            if (match) {
-                                // If the threshold rule matches, the number of times the threshold has been triggered is determined and an alarm is triggered
-                                afterThresholdRuleMatch(currentTimeMilli, monitorId, app, metrics, tagBuilder.toString(), fieldValueMap, define);
+                            final String fieldName = cell.getField().getName();
+                            final int fieldType = cell.getIntMetaData(MetricDataFieldConstants.TYPE);
+
+                            if (fieldType == CommonConstants.TYPE_NUMBER) {
+                                final Double doubleValue;
+                                if ((doubleValue = CommonUtil.parseStrDouble(valueStr)) != null) {
+                                    fieldValueMap.put(fieldName, doubleValue);
+                                }
+                            } else if (fieldType == CommonConstants.TYPE_TIME) {
+                                final Integer integerValue;
+                                if ((integerValue = CommonUtil.parseStrInteger(valueStr)) != null) {
+                                    fieldValueMap.put(fieldName, integerValue);
+                                }
                             } else {
-                                String alarmKey = String.valueOf(monitorId) + define.getId() + tagBuilder;
-                                triggeredAlertMap.remove(alarmKey);
-                                if (define.isRecoverNotice()) {
-                                    handleRecoveredAlert(currentTimeMilli, define, expr, alarmKey);
+                                if (StringUtils.isNotEmpty(valueStr)) {
+                                    fieldValueMap.put(fieldName, valueStr);
                                 }
                             }
-                        } catch (Exception e) {
-                            log.error(e.getMessage(), e);
-                        }
-                    } catch (Exception ignored) {}
+
+                            if (cell.getBooleanMetaData(MetricDataFieldConstants.LABEL)) {
+                                tagBuilder.append("-").append(valueStr);
+                            }
+                            try {
+                                boolean match = execAlertExpression(fieldValueMap, expr);
+                                try {
+                                    if (match) {
+                                        // If the threshold rule matches, the number of times the threshold has been triggered is determined and an alarm is triggered
+                                        afterThresholdRuleMatch(currentTimeMilli, monitorId, finalApp, metrics, tagBuilder.toString(), fieldValueMap, define);
+                                    } else {
+                                        String alarmKey = String.valueOf(monitorId) + define.getId() + tagBuilder;
+                                        triggeredAlertMap.remove(alarmKey);
+                                        if (define.isRecoverNotice()) {
+                                            handleRecoveredAlert(currentTimeMilli, define, expr, alarmKey);
+                                        }
+                                    }
+                                } catch (Exception e) {
+                                    log.error(e.getMessage(), e);
+                                }
+                            } catch (Exception ignored) {}
+                        });
+                    }
                 }
             }
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
         }
     }