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

根据server与client IP类型区分链路信息

Browse Source
This commit is contained in:
wanglihui
2020-08-19 10:48:08 +08:00
parent 1750549c7d
commit cbeba6372b
5 changed files with 19 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java
View File

@@ -66,9 +66,11 @@ public class ReadClickhouseData {
long bytesSum = resultSet.getLong("BYTES_SUM");
String ipType = resultSet.getString("ip_type");
String[] commonLinkInfos = (String[]) resultSet.getArray("common_link_info").getArray();
String commonLinkInfo = "";
if (commonLinkInfos.length > 1){
String commonLinkInfo;
if (commonLinkInfos.length > 1 && !commonLinkInfos[1].equals("")){
commonLinkInfo = commonLinkInfos[1];
}else {
commonLinkInfo = commonLinkInfos[0];
}
newDoc.setKey(ip);
newDoc.addAttribute("IP", ip);
@@ -264,8 +266,8 @@ public class ReadClickhouseData {
public static String getVertexIpSql() {
String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
return "SELECT * FROM((" + clientIpSql + ") UNION ALL (" + serverIpSql + "))";
}

6
IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java
View File

@@ -8,6 +8,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -73,7 +74,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
}
} catch (Exception e) {
e.printStackTrace();
LOG.error(Arrays.toString(e.getStackTrace()));
} finally {
countDownLatch.countDown();
LOG.info("本线程读取完毕，剩余线程数量：" + countDownLatch.getCount());
@@ -97,6 +98,9 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
private void deleteDistinctClientIpByTime(T doc) {
ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
if (distCip == null || distCip.isEmpty()){
return;
}
distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
Collections.sort(distCipTs);
Collections.reverse(distCipTs);