Skip to content

[fix] keep one-row metric data when command output is short #4308

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
orangeCatDeveloper wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix] keep one-row metric data when command output is short #4308

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 113 additions & 0 deletions ...ic/src/main/java/org/apache/hertzbeat/collector/collect/common/OneRowResponseSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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());
}
}
79 changes: 46 additions & 33 deletions ...-basic/src/main/java/org/apache/hertzbeat/collector/collect/script/ScriptCollectImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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";
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -207,28 +230,7 @@ private void parseResponseDataByNetcat(String result, List<String> aliasFields,
}

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,
Expand Down Expand Up @@ -261,4 +263,15 @@ private void parseResponseDataByMulti(String result, List<String> 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();
}
}
49 changes: 21 additions & 28 deletions ...lector-basic/src/main/java/org/apache/hertzbeat/collector/collect/ssh/SshCollectImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;

/**
Expand All @@ -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";
Expand Down Expand Up @@ -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<ClientChannelEvent> list = new ArrayList<>();
list.add(ClientChannelEvent.CLOSED);
Expand All @@ -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);
Expand Down Expand Up @@ -244,28 +258,7 @@ private void parseResponseDataByNetcat(String result, List<String> aliasFields,
}

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,
Expand Down
Loading
Loading