diff --git a/collector/pom.xml b/collector/pom.xml index a22b4a6..5b1eacb 100644 --- a/collector/pom.xml +++ b/collector/pom.xml @@ -31,7 +31,7 @@ com.usthe.tancloud common - 1.0-SNAPSHOT + 1.0 @@ -84,6 +84,13 @@ aviator 5.2.7 + + + + mysql + mysql-connector-java + 8.0.27 + \ No newline at end of file diff --git a/collector/src/main/java/com/usthe/collector/collect/database/JdbcCommonCollect.java b/collector/src/main/java/com/usthe/collector/collect/database/JdbcCommonCollect.java index d9883ad..8f5e6f7 100644 --- a/collector/src/main/java/com/usthe/collector/collect/database/JdbcCommonCollect.java +++ b/collector/src/main/java/com/usthe/collector/collect/database/JdbcCommonCollect.java @@ -1,8 +1,14 @@ package com.usthe.collector.collect.database; +import com.usthe.collector.collect.AbstractCollect; import com.usthe.collector.common.cache.CacheIdentifier; import com.usthe.collector.common.cache.CommonCache; import com.usthe.collector.common.cache.support.CommonJdbcConnect; +import com.usthe.collector.util.CollectorConstants; +import com.usthe.common.entity.job.Metrics; +import com.usthe.common.entity.job.protocol.JdbcProtocol; +import com.usthe.common.entity.message.CollectRep; +import com.usthe.common.util.CommonConstants; import lombok.extern.slf4j.Slf4j; import java.sql.Connection; @@ -11,8 +17,8 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.HashMap; -import java.util.LinkedList; import java.util.List; +import java.util.Objects; import java.util.Optional; /** @@ -21,10 +27,61 @@ import java.util.Optional; * @date 2021/9/1 21:37 */ @Slf4j -public class JdbcCommonCollect { +public class JdbcCommonCollect extends AbstractCollect { + + private static final String QUERY_TYPE_ONE_ROW = "oneRow"; + private static final String QUERY_TYPE_MULTI_ROW = "multiRow"; + private static final String QUERY_TYPE_COLUMNS = "columns"; + + private JdbcCommonCollect(){} + + public static JdbcCommonCollect getInstance() { + return Singleton.INSTANCE; + } + + @Override + public void collect(CollectRep.MetricsData.Builder builder, long appId, String app, Metrics metrics) { + long startTime = System.currentTimeMillis(); + // 简单校验必有参数 + if (metrics == null || metrics.getJdbc() == null) { + builder.setCode(CollectRep.Code.FAIL); + builder.setMsg("DATABASE collect must has jdbc params"); + return; + } + JdbcProtocol jdbcProtocol = metrics.getJdbc(); + String databaseUrl = constructDatabaseUrl(jdbcProtocol); + try { + Statement statement = getConnection(jdbcProtocol.getUsername(), + jdbcProtocol.getPassword(), databaseUrl); + switch (jdbcProtocol.getQueryType()) { + case QUERY_TYPE_ONE_ROW: + queryOneRow(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime); + break; + case QUERY_TYPE_MULTI_ROW: + queryMultiRow(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime); + break; + case QUERY_TYPE_COLUMNS: + queryOneRowByMatchTwoColumns(statement, jdbcProtocol.getSql(), metrics.getAliasFields(), builder, startTime); + break; + default: + builder.setCode(CollectRep.Code.FAIL); + builder.setMsg("Not support database query type: " + jdbcProtocol.getQueryType()); + break; + } + } catch (SQLException sqlException) { + log.error("Jdbc sql error: {}, code: {}.", sqlException.getMessage(), + sqlException.getErrorCode(), sqlException); + builder.setCode(CollectRep.Code.FAIL); + builder.setMsg("Query Error: " + sqlException.getMessage() + " Code: " + sqlException.getErrorCode()); + } catch (Exception e) { + log.error("Jdbc error: {}.", e.getMessage(), e); + builder.setCode(CollectRep.Code.FAIL); + builder.setMsg("Query Error: " + e.getMessage()); + } + } - private Statement getConnection(String username, String password, String url) { + private Statement getConnection(String username, String password, String url) throws Exception { CacheIdentifier identifier = CacheIdentifier.builder() .ip(url) .username(username).password(password).build(); @@ -56,20 +113,14 @@ public class JdbcCommonCollect { return statement; } // 复用失败则新建连接 - try { - Connection connection = DriverManager.getConnection(url, username, password); - statement = connection.createStatement(); - // 设置查询超时时间10秒 - statement.setQueryTimeout(10); - // 设置查询最大行数1000行 - statement.setMaxRows(1000); - CommonJdbcConnect jdbcConnect = new CommonJdbcConnect(connection); - CommonCache.getInstance().addCache(identifier, jdbcConnect, 10000L); - } catch (SQLException sqlException) { - log.error("Jdbc sql error: {}, code: {}.", sqlException.getMessage(), sqlException.getErrorCode(), sqlException); - } catch (Exception e) { - log.error("Jdbc error: {}.", e.getMessage(), e); - } + Connection connection = DriverManager.getConnection(url, username, password); + statement = connection.createStatement(); + // 设置查询超时时间10秒 + statement.setQueryTimeout(10); + // 设置查询最大行数1000行 + statement.setMaxRows(1000); + CommonJdbcConnect jdbcConnect = new CommonJdbcConnect(connection); + CommonCache.getInstance().addCache(identifier, jdbcConnect, 10000L); return statement; } @@ -83,30 +134,28 @@ public class JdbcCommonCollect { * @param columns 查询的列头(一般是数据库表字段，也可能包含特殊字段,eg: responseTime) * @throws Exception when error happen */ - private List> queryOneRow(Statement statement, String sql, List columns) throws Exception { - long startTime = System.currentTimeMillis(); + private void queryOneRow(Statement statement, String sql, List columns, + CollectRep.MetricsData.Builder builder, long startTime) throws Exception { statement.setMaxRows(1); ResultSet resultSet = statement.executeQuery(sql); - List> rowList = new LinkedList<>(); try { if (resultSet.next()) { - LinkedList rows = new LinkedList<>(); + CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); for (String column : columns) { - if ("responseTime".equals(column)) { + if (CollectorConstants.RESPONSE_TIME.equals(column)) { long time = System.currentTimeMillis() - startTime; - rows.add(String.valueOf(time)); + valueRowBuilder.addColumns(String.valueOf(time)); } else { String value = resultSet.getString(column); - value = value == null ? "" : value; - rows.add(value); + value = value == null ? CommonConstants.NULL_VALUE : value; + valueRowBuilder.addColumns(value); } } - rowList.add(rows); + builder.addValues(valueRowBuilder.build()); } } finally { resultSet.close(); } - return rowList; } /** @@ -120,10 +169,9 @@ public class JdbcCommonCollect { * @param columns 查询的列头(一般是数据库表字段，也可能包含特殊字段,eg: responseTime) * @throws Exception when error happen */ - private List> queryOneRowByMatchTwoColumns(Statement statement, String sql, List columns) throws Exception { - long startTime = System.currentTimeMillis(); + private void queryOneRowByMatchTwoColumns(Statement statement, String sql, List columns, + CollectRep.MetricsData.Builder builder, long startTime) throws Exception { ResultSet resultSet = statement.executeQuery(sql); - List> rowList = new LinkedList<>(); try { HashMap values = new HashMap<>(columns.size()); while (resultSet.next()) { @@ -131,22 +179,21 @@ public class JdbcCommonCollect { values.put(resultSet.getString(1).toLowerCase(), resultSet.getString(2)); } } - LinkedList rows = new LinkedList<>(); + CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); for (String column : columns) { - if ("responseTime".equals(column)) { + if (CollectorConstants.RESPONSE_TIME.equals(column)) { long time = System.currentTimeMillis() - startTime; - rows.add(String.valueOf(time)); + valueRowBuilder.addColumns(String.valueOf(time)); } else { String value = values.get(column.toLowerCase()); - value = value == null ? "" : value; - rows.add(value); + value = value == null ? CommonConstants.NULL_VALUE : value; + valueRowBuilder.addColumns(value); } } - rowList.add(rows); + builder.addValues(valueRowBuilder.build()); } finally { resultSet.close(); } - return rowList; } /** @@ -159,30 +206,56 @@ public class JdbcCommonCollect { * @param columns 查询的列头(一般是数据库表字段，也可能包含特殊字段,eg: responseTime) * @throws Exception when error happen */ - private List> queryMultiRow(Statement statement, String sql, List columns) throws Exception { - long startTime = System.currentTimeMillis(); + private void queryMultiRow(Statement statement, String sql, List columns, + CollectRep.MetricsData.Builder builder, long startTime) throws Exception { ResultSet resultSet = statement.executeQuery(sql); - List> rowList = new LinkedList<>(); try { while (resultSet.next()) { - LinkedList rows = new LinkedList<>(); + CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); for (String column : columns) { - if ("responseTime".equals(column)) { + if (CollectorConstants.RESPONSE_TIME.equals(column)) { long time = System.currentTimeMillis() - startTime; - rows.add(String.valueOf(time)); + valueRowBuilder.addColumns(String.valueOf(time)); } else { String value = resultSet.getString(column); - value = value == null ? "" : value; - rows.add(value); + value = value == null ? CommonConstants.NULL_VALUE : value; + valueRowBuilder.addColumns(value); } } - rowList.add(rows); + builder.addValues(valueRowBuilder.build()); } } finally { resultSet.close(); } - return rowList; } + /** + * 根据jdbc入参构造数据库URL + * @param jdbcProtocol jdbc + * @return URL + */ + private String constructDatabaseUrl(JdbcProtocol jdbcProtocol) { + if (Objects.nonNull(jdbcProtocol.getUrl()) + && !Objects.equals("", jdbcProtocol.getUrl()) + && jdbcProtocol.getUrl().startsWith("jdbc")) { + // 入参数URL有效 则优先级最高返回 + return jdbcProtocol.getUrl(); + } + String url; + switch (jdbcProtocol.getPlatform()) { + case "mysql": + url = "jdbc:mysql://" + jdbcProtocol.getHost() + ":" + jdbcProtocol.getPort() + + "/" + (jdbcProtocol.getDatabase() == null ? "" : jdbcProtocol.getDatabase()) + + "?useUnicode=true&characterEncoding=utf-8&useSSL=false"; + break; + default: + throw new IllegalArgumentException("Not support database platform: " + jdbcProtocol.getPlatform()); + } + return url; + } + + private static class Singleton { + private static final JdbcCommonCollect INSTANCE = new JdbcCommonCollect(); + } } diff --git a/collector/src/main/java/com/usthe/collector/dispatch/MetricsCollect.java b/collector/src/main/java/com/usthe/collector/dispatch/MetricsCollect.java index dc130eb..ab19586 100644 --- a/collector/src/main/java/com/usthe/collector/dispatch/MetricsCollect.java +++ b/collector/src/main/java/com/usthe/collector/dispatch/MetricsCollect.java @@ -3,6 +3,7 @@ package com.usthe.collector.dispatch; import com.googlecode.aviator.AviatorEvaluator; import com.googlecode.aviator.Expression; import com.usthe.collector.collect.AbstractCollect; +import com.usthe.collector.collect.database.JdbcCommonCollect; import com.usthe.collector.collect.http.HttpCollectImpl; import com.usthe.collector.collect.icmp.IcmpCollectImpl; import com.usthe.collector.collect.telnet.TelnetCollectImpl; @@ -107,6 +108,9 @@ public class MetricsCollect implements Runnable, Comparable { case DispatchConstants.PROTOCOL_TELNET: abstractCollect = TelnetCollectImpl.getInstance(); break; + case DispatchConstants.PROTOCOL_JDBC: + abstractCollect = JdbcCommonCollect.getInstance(); + break; // todo default: break; } diff --git a/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java b/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java index 88e78b7..25ef9bb 100644 --- a/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java +++ b/collector/src/main/java/com/usthe/collector/util/JsonPathParser.java @@ -32,7 +32,7 @@ public class JsonPathParser { * 使用jsonPath来解析json内容 * @param content json内容 * @param jsonPath jsonPath脚本 - * @return 解析后的内容 + * @return 解析后的内容 [{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'}] */ public static List> parseContentWithJsonPath(String content, String jsonPath) { if (content == null || jsonPath == null || "".equals(content) || "".equals(jsonPath)) { diff --git a/collector/src/main/java/com/usthe/collector/util/PrometheusTextParser.java b/collector/src/main/java/com/usthe/collector/util/PrometheusTextParser.java new file mode 100644 index 0000000..49d750d --- /dev/null +++ b/collector/src/main/java/com/usthe/collector/util/PrometheusTextParser.java @@ -0,0 +1,31 @@ +package com.usthe.collector.util; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * prometheus-format-text parser + * @author tom + * @date 2022/1/9 14:12 + */ +public class PrometheusTextParser { + + /** + * 解析prometheusText + * @param content 待解析文本内容 + * @return eg:[{'name': 'tom', 'speed': '433'},{'name': 'lili', 'speed': '543'},{'name': 'sam', 'speed': '643'}] + */ + public static Map>> parsePrometheusText(String content) { + String[] lines = content.split("\n"); + Map>> parseResult = new HashMap<>(8); + for (String lineTmp : lines) { + String line = lineTmp.trim(); + if (line.length() == 0 || line.startsWith("#")) { + continue; + } + + } + return null; + } +} diff --git a/common/src/main/java/com/usthe/common/entity/dto/Field.java b/common/src/main/java/com/usthe/common/entity/dto/Field.java index 82a5eda..c911fab 100644 --- a/common/src/main/java/com/usthe/common/entity/dto/Field.java +++ b/common/src/main/java/com/usthe/common/entity/dto/Field.java @@ -29,6 +29,6 @@ public class Field { private String unit; @ApiModelProperty(value = "是否是实例字段", position = 3) - private boolean instance; + private Boolean instance; } diff --git a/common/src/main/java/com/usthe/common/entity/dto/Message.java b/common/src/main/java/com/usthe/common/entity/dto/Message.java index 6738d85..5afebaa 100644 --- a/common/src/main/java/com/usthe/common/entity/dto/Message.java +++ b/common/src/main/java/com/usthe/common/entity/dto/Message.java @@ -49,6 +49,11 @@ public class Message { this.msg = msg; } + public Message(byte code, String msg) { + this.code = code; + this.msg = msg; + } + public Message(T data) { this.data = data; } diff --git a/common/src/main/java/com/usthe/common/entity/dto/MetricsHistoryData.java b/common/src/main/java/com/usthe/common/entity/dto/MetricsHistoryData.java new file mode 100644 index 0000000..6c124a2 --- /dev/null +++ b/common/src/main/java/com/usthe/common/entity/dto/MetricsHistoryData.java @@ -0,0 +1,39 @@ +package com.usthe.common.entity.dto; + +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.List; +import java.util.Map; + +/** + * 历史单指标数据 + * @author tom + * @date 2022/1/21 09:58 + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +@ApiModel(description = "历史单指标数据") +public class MetricsHistoryData { + + @ApiModelProperty(value = "监控ID", position = 0) + private Long id; + + @ApiModelProperty(value = "监控类型", position = 1) + private String app; + + @ApiModelProperty(value = "监控指标组", position = 2) + private String metric; + + @ApiModelProperty(value = "监控指标", position = 4) + private Field field; + + @ApiModelProperty(value = "监控指标历史值 instance<==>values", position = 5) + private Map> values; +} diff --git a/common/src/main/java/com/usthe/common/entity/dto/Value.java b/common/src/main/java/com/usthe/common/entity/dto/Value.java index f0ad60c..8239bbe 100644 --- a/common/src/main/java/com/usthe/common/entity/dto/Value.java +++ b/common/src/main/java/com/usthe/common/entity/dto/Value.java @@ -23,18 +23,26 @@ public class Value { this.origin = origin; } + public Value(String origin, long time) { + this.origin = origin; + this.time = time; + } + @ApiModelProperty(value = "原始值", position = 0) private String origin; @ApiModelProperty(value = "平均值", position = 1) private String mean; - @ApiModelProperty(value = "中位数值", position = 0) + @ApiModelProperty(value = "中位数值,暂不支持", position = 2) private String median; - @ApiModelProperty(value = "最小值", position = 0) + @ApiModelProperty(value = "最小值", position = 3) private String min; - @ApiModelProperty(value = "最大值", position = 0) + @ApiModelProperty(value = "最大值", position = 4) private String max; + + @ApiModelProperty(value = "数据采集时间,此字段查历史数据时有效", position = 5) + private Long time; } diff --git a/common/src/main/java/com/usthe/common/entity/job/protocol/JdbcProtocol.java b/common/src/main/java/com/usthe/common/entity/job/protocol/JdbcProtocol.java index 4c61915..19a4685 100644 --- a/common/src/main/java/com/usthe/common/entity/job/protocol/JdbcProtocol.java +++ b/common/src/main/java/com/usthe/common/entity/job/protocol/JdbcProtocol.java @@ -22,7 +22,7 @@ public class JdbcProtocol { /** * 端口号 */ - private Integer port; + private String port; /** * 数据库用户名(可选) */ @@ -32,7 +32,23 @@ public class JdbcProtocol { */ private String password; /** - * 数据库链接url eg: jdbc:mysql://localhost:3306 + * 数据库 + */ + private String database; + /** + * 数据库类型 mysql oracle ... + */ + private String platform; + /** + * SQL查询方式： oneRow, multiRow, columns + */ + private String queryType; + /** + * sql + */ + private String sql; + /** + * 数据库链接url eg: jdbc:mysql://localhost:3306/usthe */ private String url; }