This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new e12a1ddead [fix] keep one-row metric data when command output is short
(#4308)
e12a1ddead is described below
commit e12a1ddead349ca6aa39f7a9c6bb648d81589a42
Author: NekoPunch <[email protected]>
AuthorDate: Sun Aug 16 18:45:54 2026 -0700
[fix] keep one-row metric data when command output is short (#4308)
Co-authored-by: aias00 <[email protected]>
---
.../collect/common/OneRowResponseSupport.java | 113 +++++++++++++++++
.../collect/script/ScriptCollectImpl.java | 79 +++++++-----
.../collector/collect/ssh/SshCollectImpl.java | 49 ++++----
.../collect/common/OneRowResponseSupportTest.java | 123 +++++++++++++++++++
.../collect/script/ScriptCollectImplTest.java | 82 +++++++++++++
.../collector/collect/ssh/SshCollectImplTest.java | 136 +++++++++++++++++++++
.../common/entity/job/protocol/SshProtocol.java | 5 +
7 files changed, 526 insertions(+), 61 deletions(-)
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupport.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupport.java
new file mode 100644
index 0000000000..537105377b
--- /dev/null
+++
b/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());
+ }
+}
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImpl.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImpl.java
index 3d797b53a1..25fbc061ef 100644
---
a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImpl.java
+++
b/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();
+ }
}
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImpl.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImpl.java
index 41b2987446..26c8408fc7 100644
---
a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImpl.java
+++
b/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,
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupportTest.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupportTest.java
new file mode 100644
index 0000000000..a504a820fb
--- /dev/null
+++
b/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"));
+ }
+}
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImplTest.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImplTest.java
index f51fbd5c74..5af641d271 100644
---
a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImplTest.java
+++
b/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
diff --git
a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImplTest.java
b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImplTest.java
index 2810bb354d..bdb088edd1 100644
---
a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImplTest.java
+++
b/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")
diff --git
a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/protocol/SshProtocol.java
b/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/protocol/SshProtocol.java
index 1b79e45528..b2f98a3e05 100644
---
a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/protocol/SshProtocol.java
+++
b/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
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
