gfwleak/wanglihui-ip-learning-graph
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改边属性逻辑错误

Browse Source
This commit is contained in:
wanglihui
2020-07-01 14:44:45 +08:00
parent 7e8f4d763e
commit 2d543b3df9
5 changed files with 95 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
View File

@@ -66,12 +66,9 @@ public class BaseClickhouseData {
DruidPooledConnection connection = manger.getConnection(); DruidPooledConnection connection = manger.getConnection();
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql); ResultSet resultSet = statement.executeQuery(sql);
// HashSet<String> fqdnSet = new HashSet<>();
while (resultSet.next()) { while (resultSet.next()) {
// String commonSchemaType = resultSet.getString("common_schema_type");
// String fqdnName = commonSchemaGetFqdn(commonSchemaType,resultSet);
String fqdnName = resultSet.getString("FQDN"); String fqdnName = resultSet.getString("FQDN");
// fqdnSet.add(fqdnName); if (isDomain(fqdnName)){
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME"); long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME"); long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
BaseDocument newDoc = new BaseDocument(); BaseDocument newDoc = new BaseDocument();
@@ -83,6 +80,7 @@ public class BaseClickhouseData {
ArrayList<BaseDocument> documentList = vFqdnMap.getOrDefault(i, new ArrayList<>()); ArrayList<BaseDocument> documentList = vFqdnMap.getOrDefault(i, new ArrayList<>());
documentList.add(newDoc); documentList.add(newDoc);
} }
}
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse v_FQDN时间：" + (last - start)); LOG.info(sql + "\n读取clickhouse v_FQDN时间：" + (last - start));
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) { for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
@@ -154,6 +152,7 @@ public class BaseClickhouseData {
while (resultSet.next()) { while (resultSet.next()) {
String commonSchemaType = resultSet.getString("common_schema_type"); String commonSchemaType = resultSet.getString("common_schema_type");
String vFqdn = resultSet.getString("FQDN"); String vFqdn = resultSet.getString("FQDN");
if (isDomain(vFqdn)){
String vIp = resultSet.getString("common_server_ip"); String vIp = resultSet.getString("common_server_ip");
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME"); long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME"); long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
@@ -178,6 +177,7 @@ public class BaseClickhouseData {
schemaHashMap.put(commonSchemaType, newDoc); schemaHashMap.put(commonSchemaType, newDoc);
documentHashMap.put(key, schemaHashMap); documentHashMap.put(key, schemaHashMap);
} }
}
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse EFqdnAddressIp时间：" + (last - start)); LOG.info(sql + "\n读取clickhouse EFqdnAddressIp时间：" + (last - start));
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) { for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
@@ -203,6 +203,7 @@ public class BaseClickhouseData {
String commonSchemaType = resultSet.getString("common_schema_type"); String commonSchemaType = resultSet.getString("common_schema_type");
String vIp = resultSet.getString("common_client_ip"); String vIp = resultSet.getString("common_client_ip");
String vFqdn = resultSet.getString("FQDN"); String vFqdn = resultSet.getString("FQDN");
if (isDomain(vFqdn)){
String key = vIp + "-" + vFqdn; String key = vIp + "-" + vFqdn;
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME"); long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME"); long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
@@ -222,6 +223,7 @@ public class BaseClickhouseData {
schemaHashMap.put(commonSchemaType, newDoc); schemaHashMap.put(commonSchemaType, newDoc);
documentHashMap.put(key, schemaHashMap); documentHashMap.put(key, schemaHashMap);
} }
}
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse EIpVisitFqdn时间：" + (last - start)); LOG.info(sql + "\n读取clickhouse EIpVisitFqdn时间：" + (last - start));
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) { for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
@@ -235,6 +237,7 @@ public class BaseClickhouseData {
} }
} }
@Deprecated
private static String commonSchemaGetFqdn(String commonSchemaType, ResultSet resultSet) { private static String commonSchemaGetFqdn(String commonSchemaType, ResultSet resultSet) {
String vFqdn = ""; String vFqdn = "";
try { try {

54
IP-learning-graph/src/main/java/cn/ac/iie/etl/BaseUpdateEtl.java
View File

@@ -10,37 +10,45 @@ public class BaseUpdateEtl {
BaseEdgeDocument newBaseEdgeDocument = new BaseEdgeDocument(); BaseEdgeDocument newBaseEdgeDocument = new BaseEdgeDocument();
Set<String> schemaSets = newEdgeDocumentSchemaMap.keySet(); Set<String> schemaSets = newEdgeDocumentSchemaMap.keySet();
Map<String, Object> properties = newBaseEdgeDocument.getProperties();
for (String schema : schemaSets) { for (String schema : schemaSets) {
BaseEdgeDocument schemaEdgeDoc = newEdgeDocumentSchemaMap.get(schema); BaseEdgeDocument schemaEdgeDoc = newEdgeDocumentSchemaMap.get(schema);
setSchemaCnt(schema,schemaEdgeDoc,newBaseEdgeDocument); if (!properties.isEmpty()){
if (newBaseEdgeDocument.getKey() != null){
Map<String, Object> properties = newBaseEdgeDocument.getProperties();
setFoundTime(properties,schemaEdgeDoc); setFoundTime(properties,schemaEdgeDoc);
setDistinctClientIpBySchema(properties,schemaEdgeDoc); setDistinctClientIpBySchema(properties,schemaEdgeDoc);
}else { }else {
Map<String, Object> properties = schemaEdgeDoc.getProperties();
properties.remove("COUNT_TOTAL");
newBaseEdgeDocument = schemaEdgeDoc; newBaseEdgeDocument = schemaEdgeDoc;
properties = schemaEdgeDoc.getProperties();
} }
setSchemaCnt(schema,schemaEdgeDoc,properties);
} }
properties.remove("COUNT_TOTAL");
addSchemaProperty(properties);
newBaseEdgeDocument.setProperties(properties);
return newBaseEdgeDocument; return newBaseEdgeDocument;
} }
public static BaseEdgeDocument mergeIp2FqdnBySchema(HashMap<String, BaseEdgeDocument> newEdgeDocumentMap){ public static BaseEdgeDocument mergeIp2FqdnBySchema(HashMap<String, BaseEdgeDocument> newEdgeDocumentMap){
BaseEdgeDocument newBaseEdgeDocument = new BaseEdgeDocument(); BaseEdgeDocument newBaseEdgeDocument = new BaseEdgeDocument();
Set<String> schemaSets = newEdgeDocumentMap.keySet(); Set<String> schemaSets = newEdgeDocumentMap.keySet();
Map<String, Object> properties = newBaseEdgeDocument.getProperties();
for (String schema : schemaSets) { for (String schema : schemaSets) {
BaseEdgeDocument schemaEdgeDoc = newEdgeDocumentMap.get(schema); BaseEdgeDocument schemaEdgeDoc = newEdgeDocumentMap.get(schema);
setSchemaCnt(schema,schemaEdgeDoc,newBaseEdgeDocument); if (!properties.isEmpty()){
if (newBaseEdgeDocument.getKey() != null){
Map<String, Object> properties = newBaseEdgeDocument.getProperties();
setFoundTime(properties,schemaEdgeDoc); setFoundTime(properties,schemaEdgeDoc);
}else { }else {
Map<String, Object> properties = schemaEdgeDoc.getProperties();
properties.remove("COUNT_TOTAL");
newBaseEdgeDocument = schemaEdgeDoc; newBaseEdgeDocument = schemaEdgeDoc;
properties = schemaEdgeDoc.getProperties();
} }
setSchemaCnt(schema,schemaEdgeDoc,properties);
} }
properties.remove("COUNT_TOTAL");
addSchemaProperty(properties);
newBaseEdgeDocument.setProperties(properties);
return newBaseEdgeDocument; return newBaseEdgeDocument;
} }
@@ -53,11 +61,21 @@ public class BaseUpdateEtl {
setDistinctClientIpByHistory(newEdgeDocument,edgeDocument); setDistinctClientIpByHistory(newEdgeDocument,edgeDocument);
} }
private static void addSchemaProperty(Map<String, Object> properties){
if (!properties.containsKey("TLS_CNT_TOTAL")){
properties.put("TLS_CNT_TOTAL",0L);
properties.put("TLS_CNT_RECENT",new long[7]);
}else if (!properties.containsKey("HTTP_CNT_TOTAL")){
properties.put("HTTP_CNT_TOTAL",0L);
properties.put("HTTP_CNT_RECENT",new long[7]);
}
}
private static void setDistinctClientIpByHistory(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){ private static void setDistinctClientIpByHistory(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){
ArrayList<String> distCipTotal = (ArrayList<String>) edgeDocument.getAttribute("DIST_CIP_TOTAL"); ArrayList<String> distCipTotal = (ArrayList<String>) edgeDocument.getAttribute("DIST_CIP_TOTAL");
String[] distCipTotalsSrc = distCipTotal.toArray(new String[distCipTotal.size()]); String[] distCipTotalsSrc = distCipTotal.toArray(new String[distCipTotal.size()]);
String[] distCipRecentsSrc = (String[]) newEdgeDocument.getAttribute("DIST_CIP_RECENT"); Object[] distCipRecentsSrc = (Object[])newEdgeDocument.getAttribute("DIST_CIP_RECENT");
if (distCipTotalsSrc.length == 30) { if (distCipTotalsSrc.length == 30) {
Object[] distCipTotals = mergeClientIp(distCipTotalsSrc, distCipRecentsSrc); Object[] distCipTotals = mergeClientIp(distCipTotalsSrc, distCipRecentsSrc);
edgeDocument.addAttribute("DIST_CIP_TOTAL", distCipTotals); edgeDocument.addAttribute("DIST_CIP_TOTAL", distCipTotals);
@@ -88,8 +106,8 @@ public class BaseUpdateEtl {
edgeDocument.addAttribute(totalSchema, countTotal + updateCountTotal); edgeDocument.addAttribute(totalSchema, countTotal + updateCountTotal);
} }
private static Object[] mergeClientIp(String[] distCipTotalsSrc,String[] distCipRecentsSrc){ private static Object[] mergeClientIp(Object[] distCipTotalsSrc,Object[] distCipRecentsSrc){
HashSet<String> dIpSet = new HashSet<>(); HashSet<Object> dIpSet = new HashSet<>();
dIpSet.addAll(Arrays.asList(distCipRecentsSrc)); dIpSet.addAll(Arrays.asList(distCipRecentsSrc));
dIpSet.addAll(Arrays.asList(distCipTotalsSrc)); dIpSet.addAll(Arrays.asList(distCipTotalsSrc));
Object[] distCipTotals = dIpSet.toArray(); Object[] distCipTotals = dIpSet.toArray();
@@ -116,21 +134,21 @@ public class BaseUpdateEtl {
properties.put("LAST_FOUND_TIME",schemaLastFoundTime>lastFoundTime?schemaLastFoundTime:lastFoundTime); properties.put("LAST_FOUND_TIME",schemaLastFoundTime>lastFoundTime?schemaLastFoundTime:lastFoundTime);
} }
private static void setSchemaCnt(String schema,BaseEdgeDocument schemaEdgeDoc,BaseEdgeDocument newBaseEdgeDocument){ private static void setSchemaCnt(String schema,BaseEdgeDocument schemaEdgeDoc,Map<String, Object> properties){
switch (schema) { switch (schema) {
case "HTTP": case "HTTP":
long httpCntTotal = Long.parseLong(schemaEdgeDoc.getAttribute("COUNT_TOTAL").toString()); long httpCntTotal = Long.parseLong(schemaEdgeDoc.getAttribute("COUNT_TOTAL").toString());
newBaseEdgeDocument.addAttribute("HTTP_CNT_TOTAL", httpCntTotal); properties.put("HTTP_CNT_TOTAL", httpCntTotal);
long[] httpCntRecentsDst = new long[7]; long[] httpCntRecentsDst = new long[7];
httpCntRecentsDst[0] = httpCntTotal; httpCntRecentsDst[0] = httpCntTotal;
newBaseEdgeDocument.addAttribute("HTTP_CNT_RECENT", httpCntRecentsDst); properties.put("HTTP_CNT_RECENT", httpCntRecentsDst);
break; break;
case "SSL": case "SSL":
long tlsCntTotal = Long.parseLong(schemaEdgeDoc.getAttribute("COUNT_TOTAL").toString()); long tlsCntTotal = Long.parseLong(schemaEdgeDoc.getAttribute("COUNT_TOTAL").toString());
newBaseEdgeDocument.addAttribute("TLS_CNT_TOTAL", tlsCntTotal); properties.put("TLS_CNT_TOTAL", tlsCntTotal);
long[] tlsCntRecentsDst = new long[7]; long[] tlsCntRecentsDst = new long[7];
tlsCntRecentsDst[0] = tlsCntTotal; tlsCntRecentsDst[0] = tlsCntTotal;
newBaseEdgeDocument.addAttribute("TLS_CNT_RECENT", tlsCntRecentsDst); properties.put("TLS_CNT_RECENT", tlsCntRecentsDst);
break; break;
} }
} }

3
IP-learning-graph/src/main/java/cn/ac/iie/etl/fqdn2ip/UpdateEFqdnAddressIp.java
View File

@@ -50,7 +50,8 @@ public class UpdateEFqdnAddressIp implements Runnable {
LOG.info("更新R_LOCATE_FQDN2IP:" + i); LOG.info("更新R_LOCATE_FQDN2IP:" + i);
} }
} catch (Exception e) { } catch (Exception e) {
LOG.error(e.getMessage()); e.printStackTrace();
LOG.error(e.toString());
} }
} }
} }

2
IP-learning-graph/src/main/java/cn/ac/iie/etl/ip2fqdn/UpdateEIpVisitFqdn.java
View File

@@ -54,7 +54,7 @@ public class UpdateEIpVisitFqdn implements Runnable {
LOG.info("更新R_VISIT_IP2FQDN:" + i); LOG.info("更新R_VISIT_IP2FQDN:" + i);
} }
} catch (Exception e) { } catch (Exception e) {
LOG.error(e.getMessage()); LOG.error(e.toString());
} }
} }
} }

4
IP-learning-graph/src/main/resources/application.properties
View File

@@ -1,5 +1,5 @@
#arangoDB参数配置 #arangoDB参数配置
arangoDB.host=192.168.40.127 arangoDB.host=192.168.40.182
arangoDB.port=8529 arangoDB.port=8529
arangoDB.user=root arangoDB.user=root
arangoDB.password=111111 arangoDB.password=111111
@@ -13,5 +13,5 @@ update.arango.batch=10000
thread.pool.number=10 thread.pool.number=10
thread.await.termination.time=10 thread.await.termination.time=10
read.clickhouse.max.time=1593162456 read.clickhouse.max.time=1593582211
read.clickhouse.min.time=1592879247 read.clickhouse.min.time=1592879247