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 00000000000..537105377b5 --- /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 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 aliasFields, + CollectRep.MetricsData.Builder builder, Long responseTime) { + List 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 aliasFields, CollectRep.MetricsData.Builder builder, Long responseTime) { + List 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 3d797b53a15..25fbc061ef7 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.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 void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) { 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 @@ private void parseResponseDataByNetcat(String result, List aliasFields, } private void parseResponseDataByOne(String result, List 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 aliasFields, @@ -261,4 +263,15 @@ private void parseResponseDataByMulti(String result, List aliasFields, 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 41b29874465..26c8408fc73 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.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.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.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 @@ @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 void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) { } 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 list = new ArrayList<>(); list.add(ClientChannelEvent.CLOSED); @@ -107,16 +109,28 @@ public void collect(CollectRep.MetricsData.Builder builder, Metrics metrics) { 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 @@ private void parseResponseDataByNetcat(String result, List aliasFields, } private void parseResponseDataByOne(String result, List 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 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 00000000000..a504a820fb2 --- /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 f51fbd5c743..5af641d2712 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 @@ 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 @@ void collect() { 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 2810bb354d1..bdb088edd11 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.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 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 @@ void collectClosesExactSessionWhenChannelCannotClose() throws Exception { 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 = 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 = 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 = 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 = 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 out = new AtomicReference<>(); + AtomicReference 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 1b79e45528e..b2f98a3e057 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 */