From f3557b2859d8cd630317b0f0911212b750e89450 Mon Sep 17 00:00:00 2001 From: tomsun28 Date: Thu, 3 Mar 2022 11:09:14 +0800 Subject: [PATCH] =?UTF-8?q?[manager,collector]=E6=96=B0=E5=A2=9Emysql?= =?UTF-8?q?=E6=8C=87=E6=A0=87=EF=BC=8C=E9=87=87=E9=9B=86=E5=99=A8=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E7=AC=AC0=E4=BC=98=E5=85=88=E7=BA=A7=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E5=88=99=E5=8F=96=E6=B6=88=E5=90=8E=E7=BB=AD=E7=9A=84?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../collect/database/JdbcCommonCollect.java | 8 +- .../collector/dispatch/CommonDispatcher.java | 5 +- .../src/main/resources/define/app/mysql.yml | 73 +++++++++++++++++++ .../warehouse/store/MemoryDataStorage.java | 2 +- .../warehouse/store/RedisDataStorage.java | 2 +- 5 files changed, 85 insertions(+), 5 deletions(-) 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 9441d49..0a7eaea 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,5 +1,6 @@ package com.usthe.collector.collect.database; +import com.mysql.cj.jdbc.exceptions.CommunicationsException; import com.usthe.collector.collect.AbstractCollect; import com.usthe.collector.collect.common.cache.CacheIdentifier; import com.usthe.collector.collect.common.cache.CommonCache; @@ -68,9 +69,12 @@ public class JdbcCommonCollect extends AbstractCollect { builder.setMsg("Not support database query type: " + jdbcProtocol.getQueryType()); break; } + } catch (CommunicationsException communicationsException) { + log.warn("Jdbc sql error: {}, code: {}.", communicationsException.getMessage(), communicationsException.getErrorCode()); + builder.setCode(CollectRep.Code.UN_REACHABLE); + builder.setMsg("Query Error: " + communicationsException.getMessage() + " Code: " + communicationsException.getErrorCode()); } catch (SQLException sqlException) { - log.error("Jdbc sql error: {}, code: {}.", sqlException.getMessage(), - sqlException.getErrorCode(), sqlException); + log.warn("Jdbc sql error: {}, code: {}.", sqlException.getMessage(), sqlException.getErrorCode()); builder.setCode(CollectRep.Code.FAIL); builder.setMsg("Query Error: " + sqlException.getMessage() + " Code: " + sqlException.getErrorCode()); } catch (Exception e) { diff --git a/collector/src/main/java/com/usthe/collector/dispatch/CommonDispatcher.java b/collector/src/main/java/com/usthe/collector/dispatch/CommonDispatcher.java index e73a6aa..cc2878f 100644 --- a/collector/src/main/java/com/usthe/collector/dispatch/CommonDispatcher.java +++ b/collector/src/main/java/com/usthe/collector/dispatch/CommonDispatcher.java @@ -144,7 +144,10 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc if (job.isCyclic()) { // 若是异步的周期性循环任务,直接发送指标组的采集数据到消息中间件 kafkaDataExporter.send(metricsData); - if (metricsSet == null) { + // 若metricsSet为null表示执行完成 + // 或判断采集指标组是否优先级为0，即为可用性采集指标组 若可用性采集失败 则取消后面的指标组调度直接进入下一轮调度 + if (metricsSet == null + || (metrics.getPriority() == (byte)0 && metricsData.getCode() != CollectRep.Code.SUCCESS)) { // 此Job所有指标组采集执行完成 // 周期性任务再次将任务push到时间轮 // 先判断此次任务执行时间与任务采集间隔时间 diff --git a/manager/src/main/resources/define/app/mysql.yml b/manager/src/main/resources/define/app/mysql.yml index 1a86da9..bd5aa6c 100644 --- a/manager/src/main/resources/define/app/mysql.yml +++ b/manager/src/main/resources/define/app/mysql.yml @@ -64,4 +64,77 @@ metrics: queryType: columns # sql sql: show global variables where Variable_name like 'version%' or Variable_name = 'max_connections' or Variable_name = 'datadir' or Variable_name = 'port'; + url: ^_^url^_^ + + - name: status + priority: 1 + fields: + # 指标信息 包括 field名称 type字段类型:0-number数字,1-string字符串 instance是否为实例主键 unit:指标单位 + - field: threads_created + type: 0 + - field: threads_connected + type: 0 + - field: threads_cached + type: 0 + - field: threads_running + type: 0 + # (非必须)监控指标别名，与上面的指标名映射。用于采集接口数据字段不直接是最终指标名称,需要此别名做映射转换 + aliasFields: + - threads_created + - threads_connected + - threads_cached + - threads_running + # (非必须)指标计算表达式,与上面的别名一起作用,计算出最终需要的指标值 + # eg: cores=core1+core2, usage=usage, waitTime=allTime-runningTime + calculates: + - threads_created=threads_created + - threads_connected=threads_connected + - threads_cached=threads_cached + - threads_running=threads_running + protocol: jdbc + jdbc: + # 主机host: ipv4 ipv6 域名 + host: ^_^host^_^ + # 端口 + port: ^_^port^_^ + platform: mysql + username: ^_^username^_^ + password: ^_^password^_^ + database: ^_^database^_^ + # SQL查询方式： oneRow, multiRow, columns + queryType: columns + # sql + sql: show global status where Variable_name like 'thread%' or Variable_name = 'com_commit' or Variable_name = 'com_rollback' or Variable_name = 'questions' or Variable_name = 'uptime'; + url: ^_^url^_^ + + - name: innodb + priority: 2 + fields: + # 指标信息 包括 field名称 type字段类型:0-number数字,1-string字符串 instance是否为实例主键 unit:指标单位 + - field: innodb_data_reads + type: 0 + unit: 次数 + - field: innodb_data_writes + type: 0 + unit: 次数 + - field: innodb_data_read + type: 0 + unit: kb + - field: innodb_data_written + type: 0 + unit: kb + protocol: jdbc + jdbc: + # 主机host: ipv4 ipv6 域名 + host: ^_^host^_^ + # 端口 + port: ^_^port^_^ + platform: mysql + username: ^_^username^_^ + password: ^_^password^_^ + database: ^_^database^_^ + # SQL查询方式： oneRow, multiRow, columns + queryType: columns + # sql + sql: show global status where Variable_name like 'innodb%'; url: ^_^url^_^ \ No newline at end of file diff --git a/warehouse/src/main/java/com/usthe/warehouse/store/MemoryDataStorage.java b/warehouse/src/main/java/com/usthe/warehouse/store/MemoryDataStorage.java index 95225b1..ec4812a 100644 --- a/warehouse/src/main/java/com/usthe/warehouse/store/MemoryDataStorage.java +++ b/warehouse/src/main/java/com/usthe/warehouse/store/MemoryDataStorage.java @@ -62,7 +62,7 @@ public class MemoryDataStorage implements DisposableBean { private void saveData(CollectRep.MetricsData metricsData) { String hashKey = metricsData.getId() + metricsData.getMetrics(); if (metricsData.getValuesList().isEmpty()) { - log.info("[warehouse] redis flush metrics data {} is null, ignore.", hashKey); + log.debug("[warehouse memory] redis flush metrics data {} is null, ignore.", hashKey); return; } metricsDataMap.put(hashKey, metricsData); diff --git a/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java b/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java index af1f9b9..e4b6f51 100644 --- a/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java +++ b/warehouse/src/main/java/com/usthe/warehouse/store/RedisDataStorage.java @@ -70,7 +70,7 @@ public class RedisDataStorage implements DisposableBean { String key = String.valueOf(metricsData.getId()); String hashKey = metricsData.getMetrics(); if (metricsData.getValuesList().isEmpty()) { - log.info("[warehouse] redis flush metrics data {}:{} is null, ignore.", key, hashKey); + log.info("[warehouse redis] redis flush metrics data {} - {} is null, ignore.", key, hashKey); return; } RedisAsyncCommands commands = connection.async();