refactor: fix manager query count and agent print log

ozhera-app/app-server/src/test/java/com/xiaomi/mone/test/SpringBootFeatTest.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2020 Xiaomi
+ *
+ *  Licensed 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 com.xiaomi.mone.test;
+
+import com.xiaomi.mone.app.AppBootstrap;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import javax.annotation.Resource;
+import javax.validation.ConstraintViolation;
+import javax.validation.Validation;
+import javax.validation.Validator;
+import javax.validation.ValidatorFactory;
+import java.util.Set;
+
+/**
+ *
+ * @description
+ * @version 1.0
+ * @author wtt
+ * @date 2024/4/28 10:02
+ *
+ */
+//@SpringBootTest(classes = AppBootstrap.class)
+public class SpringBootFeatTest {
+
+    @Resource
+    private Validator validator;
+
+    @Test
+    public void validatorSpringTest() {
+//        validator(validator);
+    }
+
+    private void validator(Validator validator) {
+
+        UserBean userBean = new UserBean();
+        userBean.setUserName("");
+        userBean.setAge(17);
+
+        Set<ConstraintViolation<UserBean>> constraintViolations = validator.validate(userBean);
+        System.out.println("validate 校验对象属性：");
+        System.out.println(constraintViolations);
+        System.out.println();
+
+        constraintViolations = validator.validateProperty(userBean, "age");
+        System.out.println("validate 校验对象属性age：");
+        System.out.println(constraintViolations);
+        System.out.println();
+
+        constraintViolations = validator.validateValue(UserBean.class, "age", 15);
+        System.out.println("validate 校验对象属性age：");
+        System.out.println(constraintViolations);
+    }
+
+    @Test
+    public void validatorJavaTest() {
+        Validator validator1 = ValidationUtils.getValidator();
+        validator(validator1);
+    }
+
+    public static class ValidationUtils {
+
+        public static Validator getValidator() {
+            ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
+            return validatorFactory.getValidator();
+        }
+    }
+
+}

ozhera-app/app-server/src/test/java/com/xiaomi/mone/test/UserBean.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2020 Xiaomi
+ *
+ *  Licensed 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 com.xiaomi.mone.test;
+
+import lombok.Data;
+
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotBlank;
+
+/**
+ *
+ * @description
+ * @version 1.0
+ * @author wtt
+ * @date 2024/4/28 10:10
+ *
+ */
+@Data
+public class UserBean {
+
+    @NotBlank
+    private String userName;
+
+    @Min(value = 18, message = "age最小值为18")
+    private Integer age;
+}

ozhera-log/log-agent/src/main/java/com/xiaomi/mone/log/agent/channel/ChannelServiceImpl.java

@@ -382,9 +382,8 @@ private void readFile(String patternCode, String ip, String filePath, Long chann
 
         ReadListener listener = initFileReadListener(mLog, patternCode, usedIp, filePath);
         Map<String, ChannelMemory.FileProgress> fileProgressMap = channelMemory.getFileProgressMap();
-        if (!collectOnce) {
-            log.info("fileProgressMap:{}", gson.toJson(fileProgressMap));
-        }
+        printMapToJson(fileProgressMap, collectOnce);
+
         ILogFile logFile = getLogFile(filePath, listener, fileProgressMap);
         if (null == logFile) {
             log.warn("file:{} marked stop to collect", filePath);
@@ -420,6 +419,25 @@ private void stopOldCurrentFileThread(String filePath) {
         }
     }
 
+    private void printMapToJson(Map<String, ChannelMemory.FileProgress> map, boolean collectOnce) {
+        if (map == null || map.isEmpty()) {
+            return;
+        }
+
+        Map<String, ChannelMemory.FileProgress> snapshot;
+        try {
+            snapshot = new HashMap<>(map);
+        } catch (ConcurrentModificationException e) {
+            log.error("Failed to create snapshot of fileProgressMap", e);
+            return;
+        }
+
+        if (!collectOnce && !snapshot.isEmpty()) {
+            String jsonMap = gson.toJson(snapshot);
+            log.info("fileProgressMap: {}", jsonMap);
+        }
+    }
+
 
     private ILogFile getLogFile(String filePath, ReadListener listener, Map<String, ChannelMemory.FileProgress> fileProgressMap) {
 

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/dao/MilogAppMiddlewareRelDao.java

@@ -136,8 +136,8 @@ public List<MilogAppMiddlewareRel> getAppRelByLimit(int offset, int rows) {
     }
 
     public Integer queryCountByTopicName(String topicName) {
-        Sql sql = Sqls.queryRecord("SELECT count(1) as count FROM `milog_app_middleware_rel` where config like '%' || @topicName || '%'");
-        sql.params().set("topicName", topicName);
+        Sql sql = Sqls.queryRecord("SELECT count(1) as count FROM `milog_app_middleware_rel` where config like @topicName ");
+        sql.params().set("topicName", "%" + topicName + "%");
         LinkedList<Record> records = (LinkedList<Record>) dao.execute(sql).getResult();
         int access = records.get(0).getInt("count");
         return access;

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/dao/MilogLogTailDao.java

@@ -280,12 +280,10 @@ public List<MilogLogTailDo> queryStoreIdByRegionNameEN(String nameEn) {
     }
 
     public List<MilogLogTailDo> queryTailNameExists(String tailName, String machineRoom, Long spaceId) {
-        Sql sql = Sqls.queryEntity("SELECT la.* FROM milog_logstail la LEFT JOIN milog_logstore lt ON la.store_id = lt.id WHERE la.tail = @tailName AND lt.machine_room = @machineRoom");
+        Sql sql = Sqls.queryEntity("SELECT la.* FROM milog_logstail la LEFT JOIN milog_logstore lt ON la.store_id = lt.id WHERE la.tail = @tailName AND lt.machine_room = @machineRoom and la.space_id = @spaceId");
         sql.params().set("tailName", tailName);
         sql.params().set("machineRoom", machineRoom);
-        if (null != spaceId) {
-            sql.params().set("store_id", spaceId);
-        }
+        sql.params().set("spaceId", spaceId);
         sql.setEntity(dao.getEntity(MilogLogTailDo.class));
         dao.execute(sql);
         return sql.getList(MilogLogTailDo.class);

ozhera-log/log-stream/src/main/java/com/xiaomi/mone/log/stream/plugin/es/EsPlugin.java

@@ -139,26 +139,30 @@ public void beforeBulk(long executionId, BulkRequest request) {
 
             @Override
             public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
-                log.debug("success send to es,desc:{}", request.getDescription());
-                AtomicInteger count = new AtomicInteger();
-                response.spliterator().forEachRemaining(x -> {
-                    if (x.isFailed()) {
-                        BulkItemResponse.Failure failure = x.getFailure();
-                        String msg = String.format(
-                                "Index:[%s], type:[%s], id:[%s], itemId:[%s], opt:[%s], version:[%s], errMsg:%s"
-                                , x.getIndex()
-                                , x.getType()
-                                , x.getId()
-                                , x.getItemId()
-                                , x.getOpType().getLowercase()
-                                , x.getVersion()
-                                , failure.getCause().getMessage()
-                        );
-                        log.error("Bulk executionId:[{}] has error messages:\t{}", executionId, msg);
-                        count.incrementAndGet();
-                    }
-                });
-                log.debug("Finished handling bulk commit executionId:[{}] for {} requests with {} errors", executionId, request.numberOfActions(), count.intValue());
+                if (response.hasFailures()) {
+                    AtomicInteger count = new AtomicInteger();
+                    response.spliterator().forEachRemaining(x -> {
+                        if (x.isFailed()) {
+                            BulkItemResponse.Failure failure = x.getFailure();
+                            String msg = String.format(
+                                    "Index:[%s], type:[%s], id:[%s], itemId:[%s], opt:[%s], version:[%s], errMsg:%s"
+                                    , x.getIndex()
+                                    , x.getType()
+                                    , x.getId()
+                                    , x.getItemId()
+                                    , x.getOpType().getLowercase()
+                                    , x.getVersion()
+                                    , failure.getCause().getMessage()
+                            );
+                            log.error("Bulk executionId:[{}] has error messages:{}", executionId, msg);
+                            count.incrementAndGet();
+                        }
+                    });
+                    log.debug("Finished handling bulk commit executionId:[{}] for {} requests with {} errors", executionId, request.numberOfActions(), count.intValue());
+                    sendMessageToTopic(request, esInfo, onFailedConsumer);
+                } else {
+                    log.debug("success send to es,desc:{}", request.getDescription());
+                }
             }
 
             @Override
@@ -172,35 +176,39 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
                         , clazz.getTypeName()
                         , clazz.getCanonicalName()
                         , failure.getMessage());
-                MqMessageDTO MqMessageDTO = new MqMessageDTO();
-                MqMessageDTO.setEsInfo(esInfo);
-                List<MqMessageDTO.CompensateMqDTO> compensateMqDTOS = Lists.newArrayList();
-                request.requests().stream().filter(x -> x instanceof IndexRequest)
-                        .forEach(x -> {
-                            Map source = ((IndexRequest) x).sourceAsMap();
-                            log.error("Failure to handle index:[{}], type:[{}],id:[{}] data:[{}]", x.index(), x.type(), x.id(), JSON.toJSONString(source));
-                            MqMessageDTO.CompensateMqDTO compensateMqDTO = new MqMessageDTO.CompensateMqDTO();
-                            compensateMqDTO.setMsg(JSON.toJSONString(source));
-                            compensateMqDTO.setEsIndex(x.index());
-                            compensateMqDTOS.add(compensateMqDTO);
-                        });
-                //The message is sent to mq for consumption - the data cannot be larger than 10M, otherwise it cannot be written, divided into 2 parts
-                int length = JSON.toJSONString(compensateMqDTOS).getBytes().length;
-                if (length > SINGLE_MESSAGE_BYTES_MAXIMAL) {
-                    List<List<MqMessageDTO.CompensateMqDTO>> splitList = ListUtil.partition(compensateMqDTOS, 2);
-                    for (List<MqMessageDTO.CompensateMqDTO> mqDTOS : splitList) {
-                        MqMessageDTO.setCompensateMqDTOS(mqDTOS);
-                        onFailedConsumer.accept(MqMessageDTO);
-                    }
-                } else {
-                    MqMessageDTO.setCompensateMqDTOS(compensateMqDTOS);
-                    onFailedConsumer.accept(MqMessageDTO);
-                }
+                sendMessageToTopic(request, esInfo, onFailedConsumer);
             }
         }));
         return esProcessor;
     }
 
+    private static void sendMessageToTopic(BulkRequest request, StorageInfo esInfo, Consumer<MqMessageDTO> onFailedConsumer) {
+        MqMessageDTO MqMessageDTO = new MqMessageDTO();
+        MqMessageDTO.setEsInfo(esInfo);
+        List<MqMessageDTO.CompensateMqDTO> compensateMqDTOS = Lists.newArrayList();
+        request.requests().stream().filter(x -> x instanceof IndexRequest)
+                .forEach(x -> {
+                    Map source = ((IndexRequest) x).sourceAsMap();
+                    log.error("Failure to handle index:[{}], type:[{}],id:[{}] data:[{}]", x.index(), x.type(), x.id(), JSON.toJSONString(source));
+                    MqMessageDTO.CompensateMqDTO compensateMqDTO = new MqMessageDTO.CompensateMqDTO();
+                    compensateMqDTO.setMsg(JSON.toJSONString(source));
+                    compensateMqDTO.setEsIndex(x.index());
+                    compensateMqDTOS.add(compensateMqDTO);
+                });
+        //The message is sent to mq for consumption - the data cannot be larger than 10M, otherwise it cannot be written, divided into 2 parts
+        int length = JSON.toJSONString(compensateMqDTOS).getBytes().length;
+        if (length > SINGLE_MESSAGE_BYTES_MAXIMAL) {
+            List<List<MqMessageDTO.CompensateMqDTO>> splitList = ListUtil.partition(compensateMqDTOS, 2);
+            for (List<MqMessageDTO.CompensateMqDTO> mqDTOS : splitList) {
+                MqMessageDTO.setCompensateMqDTOS(mqDTOS);
+                onFailedConsumer.accept(MqMessageDTO);
+            }
+        } else {
+            MqMessageDTO.setCompensateMqDTOS(compensateMqDTOS);
+            onFailedConsumer.accept(MqMessageDTO);
+        }
+    }
+
     private static String cacheKey(StorageInfo esInfo) {
         StringBuilder stringBuilder = new StringBuilder();
         stringBuilder.append(esInfo.getId()).append(",");