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

整合YSP项目

Browse Source
This commit is contained in:
wanglihui
2020-08-12 14:42:32 +08:00
parent 55879a2c32
commit 885be09053
23 changed files with 9382 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
IP-learning-graph/src/main/java/cn/ac/iie/config/ApplicationConfig.java
View File

@@ -13,8 +13,7 @@ public class ApplicationConfig {
public static final Integer ARANGODB_TTL = ConfigUtils.getIntProperty( "arangoDB.ttl"); public static final Integer ARANGODB_TTL = ConfigUtils.getIntProperty( "arangoDB.ttl");
public static final Integer ARANGODB_BATCH = ConfigUtils.getIntProperty( "arangoDB.batch"); public static final Integer ARANGODB_BATCH = ConfigUtils.getIntProperty( "arangoDB.batch");
public static final Integer UPDATE_ARANGO_BATCH = ConfigUtils.getIntProperty("update.arango.batch"); public static final Integer UPDATE_ARANGO_BATCH =ConfigUtils.getIntProperty("update.arango.batch");
public static final String ARANGODB_READ_LIMIT = ConfigUtils.getStringProperty("arangoDB.read.limit");
public static final Integer THREAD_POOL_NUMBER = ConfigUtils.getIntProperty( "thread.pool.number"); public static final Integer THREAD_POOL_NUMBER = ConfigUtils.getIntProperty( "thread.pool.number");
public static final Integer THREAD_AWAIT_TERMINATION_TIME = ConfigUtils.getIntProperty( "thread.await.termination.time"); public static final Integer THREAD_AWAIT_TERMINATION_TIME = ConfigUtils.getIntProperty( "thread.await.termination.time");
@@ -22,11 +21,19 @@ public class ApplicationConfig {
public static final Long READ_CLICKHOUSE_MAX_TIME = ConfigUtils.getLongProperty("read.clickhouse.max.time"); public static final Long READ_CLICKHOUSE_MAX_TIME = ConfigUtils.getLongProperty("read.clickhouse.max.time");
public static final Long READ_CLICKHOUSE_MIN_TIME = ConfigUtils.getLongProperty("read.clickhouse.min.time"); public static final Long READ_CLICKHOUSE_MIN_TIME = ConfigUtils.getLongProperty("read.clickhouse.min.time");
public static final Integer TIME_LIMIT_TYPE = ConfigUtils.getIntProperty("time.limit.type"); public static final Integer CLICKHOUSE_TIME_LIMIT_TYPE = ConfigUtils.getIntProperty("clickhouse.time.limit.type");
public static final Integer UPDATE_INTERVAL = ConfigUtils.getIntProperty("update.interval"); public static final Integer UPDATE_INTERVAL = ConfigUtils.getIntProperty("update.interval");
public static final Integer DISTINCT_CLIENT_IP_NUM = ConfigUtils.getIntProperty("distinct.client.ip.num"); public static final Integer DISTINCT_CLIENT_IP_NUM = ConfigUtils.getIntProperty("distinct.client.ip.num");
public static final Integer RECENT_COUNT_HOUR = ConfigUtils.getIntProperty("recent.count.hour"); public static final Integer RECENT_COUNT_HOUR = ConfigUtils.getIntProperty("recent.count.hour");
public static final String TOP_DOMAIN_FILE_NAME = ConfigUtils.getStringProperty("top.domain.file.name");
public static final String ARANGODB_READ_LIMIT = ConfigUtils.getStringProperty("arangoDB.read.limit");
public static final Integer ARANGO_TIME_LIMIT_TYPE = ConfigUtils.getIntProperty("arango.time.limit.type");
public static final Long READ_ARANGO_MAX_TIME = ConfigUtils.getLongProperty("read.arango.max.time");
public static final Long READ_ARANGO_MIN_TIME = ConfigUtils.getLongProperty("read.arango.min.time");
} }

80
IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java
View File

@@ -1,7 +1,7 @@
package cn.ac.iie.dao; package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig; import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.service.ingestion.ReadHistoryArangoData; import cn.ac.iie.service.read.ReadHistoryArangoData;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool; import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.ArangoCursor; import com.arangodb.ArangoCursor;
@@ -15,79 +15,89 @@ import java.util.concurrent.CountDownLatch;
/** /**
* 获取arangoDB历史数据 * 获取arangoDB历史数据
*
* @author wlh
*/ */
public class BaseArangoData { public class BaseArangoData {
private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class); private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseDocument>> historyVertexFqdnMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexFqdnMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseDocument>> historyVertexIpMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexIpMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseDocument>> historyVertexSubscriberMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexSubscriberMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnAddressIpMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnAddressIpMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationIpVisitFqdnMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationIpVisitFqdnMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnSameFqdnMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnSameFqdnMap = new ConcurrentHashMap<>();
static ConcurrentHashMap<Integer,ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>(); static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>();
private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance(); private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance(); private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
<T extends BaseDocument> void readHistoryData(String table, <T extends BaseDocument> void readHistoryData(String table,
ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map, ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
Class<T> type) { Class<T> type) {
try { try {
LOG.info("开始更新"+table); LOG.info("开始更新" + table);
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++){ for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
map.put(i,new ConcurrentHashMap<>()); historyMap.put(i, new ConcurrentHashMap<>());
} }
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER); CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
long[] timeRange = getTimeRange(table); long[] timeRange = getTimeRange(table);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) { for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
String sql = getQuerySql(timeRange, i, table); String sql = getQuerySql(timeRange, i, table);
ReadHistoryArangoData<T> readHistoryArangoData = ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
threadPool.executor(readHistoryArangoData); threadPool.executor(readHistoryArangoData);
} }
countDownLatch.await(); countDownLatch.await();
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info("读取"+table+" arangoDB 共耗时："+(last-start)); LOG.info("读取" + table + " arangoDB 共耗时：" + (last - start));
}catch (Exception e){ } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
} }
private long[] getTimeRange(String table){ private long[] getTimeRange(String table) {
long minTime = 0L; long minTime = 0L;
long maxTime = 0L; long maxTime = 0L;
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
String sql = "LET doc = (FOR doc IN "+table+" RETURN doc) return {max_time:MAX(doc[*].FIRST_FOUND_TIME),min_time:MIN(doc[*].FIRST_FOUND_TIME)}"; String sql = "LET doc = (FOR doc IN " + table + " RETURN doc) return {max_time:MAX(doc[*].FIRST_FOUND_TIME),min_time:MIN(doc[*].FIRST_FOUND_TIME)}";
ArangoCursor<BaseDocument> timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class); switch (ApplicationConfig.ARANGO_TIME_LIMIT_TYPE) {
try { case 0:
if (timeDoc != null){ ArangoCursor<BaseDocument> timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class);
while (timeDoc.hasNext()) { try {
BaseDocument doc = timeDoc.next(); if (timeDoc != null) {
maxTime = Long.parseLong(doc.getAttribute("max_time").toString()) + ApplicationConfig.THREAD_POOL_NUMBER; while (timeDoc.hasNext()) {
minTime = Long.parseLong(doc.getAttribute("min_time").toString()); BaseDocument doc = timeDoc.next();
maxTime = Long.parseLong(doc.getAttribute("max_time").toString()) + ApplicationConfig.THREAD_POOL_NUMBER;
minTime = Long.parseLong(doc.getAttribute("min_time").toString());
}
} else {
LOG.warn("获取ArangoDb时间范围为空");
}
} catch (Exception e) {
e.printStackTrace();
} }
long lastTime = System.currentTimeMillis(); break;
LOG.info(sql+"\n查询最大最小时间用时：" + (lastTime - startTime)); case 1:
}else { maxTime = ApplicationConfig.READ_ARANGO_MAX_TIME;
LOG.warn("获取ArangoDb时间范围为空"); minTime = ApplicationConfig.READ_ARANGO_MIN_TIME;
} break;
}catch (Exception e){ default:
e.printStackTrace();
} }
long lastTime = System.currentTimeMillis();
LOG.info(sql + "\n查询最大最小时间用时：" + (lastTime - startTime));
return new long[]{minTime, maxTime}; return new long[]{minTime, maxTime};
} }
private String getQuerySql(long[] timeRange,int threadNumber,String table){ private String getQuerySql(long[] timeRange, int threadNumber, String table) {
long minTime = timeRange[0]; long minTime = timeRange[0];
long maxTime = timeRange[1]; long maxTime = timeRange[1];
long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER; long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER;
long maxThreadTime = minTime + (threadNumber + 1)* diffTime; long maxThreadTime = minTime + (threadNumber + 1) * diffTime;
long minThreadTime = minTime + threadNumber * diffTime; long minThreadTime = minTime + threadNumber * diffTime;
return "FOR doc IN "+table+" filter doc.FIRST_FOUND_TIME >= "+minThreadTime+" and doc.FIRST_FOUND_TIME <= "+maxThreadTime+" " + ApplicationConfig.ARANGODB_READ_LIMIT + " RETURN doc"; return "FOR doc IN " + table + " filter doc.FIRST_FOUND_TIME >= " + minThreadTime + " and doc.FIRST_FOUND_TIME <= " + maxThreadTime + " " + ApplicationConfig.ARANGODB_READ_LIMIT + " RETURN doc";
} }
} }

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

@@ -15,7 +15,7 @@ import java.util.HashMap;
import java.util.function.Function; import java.util.function.Function;
import java.util.function.Supplier; import java.util.function.Supplier;
import static cn.ac.iie.service.ingestion.ReadClickhouseData.*; import static cn.ac.iie.service.read.ReadClickhouseData.putMapByHashcode;
/** /**
* 读取clickhouse数据，封装到map * 读取clickhouse数据，封装到map
@@ -24,23 +24,25 @@ import static cn.ac.iie.service.ingestion.ReadClickhouseData.*;
public class BaseClickhouseData { public class BaseClickhouseData {
private static final Logger LOG = LoggerFactory.getLogger(BaseClickhouseData.class); private static final Logger LOG = LoggerFactory.getLogger(BaseClickhouseData.class);
private static ClickhouseConnect manger = ClickhouseConnect.getInstance();
static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> newVertexFqdnMap = new HashMap<>(); static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> newVertexFqdnMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> newVertexIpMap = new HashMap<>(); static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> newVertexIpMap = new HashMap<>();
static HashMap<Integer, HashMap<String,ArrayList<BaseDocument>>> newVertexSubscriberMap = new HashMap<>(); static HashMap<Integer, HashMap<String,ArrayList<BaseDocument>>> newVertexSubscriberMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationFqdnAddressIpMap = new HashMap<>(); static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationFqdnAddressIpMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationIpVisitFqdnMap = new HashMap<>(); static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationIpVisitFqdnMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationSubsciberLocateIpMap = new HashMap<>(); static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationSubsciberLocateIpMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationFqdnSameFqdnMap = new HashMap<>();
private static ClickhouseConnect manger = ClickhouseConnect.getInstance();
private DruidPooledConnection connection; private DruidPooledConnection connection;
private Statement statement; private Statement statement;
<T extends BaseDocument> void baseDocumentFromClickhouse(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap, <T extends BaseDocument> void baseDocumentFromClickhouse(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap,
Supplier<String> getSqlSupplier, Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc) { Function<ResultSet,T> formatResultFunc){
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
initializeMap(newMap); initializeMap(newMap);
String sql = getSqlSupplier.get(); String sql = getSqlSupplier.get();
LOG.info(sql);
try { try {
connection = manger.getConnection(); connection = manger.getConnection();
statement = connection.createStatement(); statement = connection.createStatement();
@@ -54,7 +56,7 @@ public class BaseClickhouseData {
} }
} }
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info(sql + "\n读取"+i+"条数据，运行时间：" + (last - start)); LOG.info("读取"+i+"条数据，运行时间：" + (last - start));
}catch (Exception e){ }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
}finally { }finally {
@@ -65,7 +67,7 @@ public class BaseClickhouseData {
private <T extends BaseDocument> void initializeMap(HashMap<Integer, HashMap<String,ArrayList<T>>> map){ private <T extends BaseDocument> void initializeMap(HashMap<Integer, HashMap<String,ArrayList<T>>> map){
try { try {
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) { for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
map.put(i, new HashMap<>()); map.put(i, new HashMap<>(16));
} }
}catch (Exception e){ }catch (Exception e){
e.printStackTrace(); e.printStackTrace();

36
IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
View File

@@ -1,14 +1,13 @@
package cn.ac.iie.dao; package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig; import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.service.ingestion.ReadClickhouseData; import cn.ac.iie.service.read.ReadClickhouseData;
import cn.ac.iie.service.update.Document; import cn.ac.iie.service.update.Document;
import cn.ac.iie.service.update.relationship.LocateFqdn2Ip; import cn.ac.iie.service.update.relationship.LocateFqdn2Ip;
import cn.ac.iie.service.update.relationship.LocateSubscriber2Ip; import cn.ac.iie.service.update.relationship.SameFqdn2Fqdn;
import cn.ac.iie.service.update.relationship.VisitIp2Fqdn; import cn.ac.iie.service.update.relationship.VisitIp2Fqdn;
import cn.ac.iie.service.update.vertex.Fqdn; import cn.ac.iie.service.update.vertex.Fqdn;
import cn.ac.iie.service.update.vertex.Ip; import cn.ac.iie.service.update.vertex.Ip;
import cn.ac.iie.service.update.vertex.Subscriber;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool; import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.entity.BaseDocument; import com.arangodb.entity.BaseDocument;
@@ -36,41 +35,36 @@ public class UpdateGraphData {
private static final Logger LOG = LoggerFactory.getLogger(UpdateGraphData.class); private static final Logger LOG = LoggerFactory.getLogger(UpdateGraphData.class);
private static ExecutorThreadPool pool = ExecutorThreadPool.getInstance(); private static ExecutorThreadPool pool = ExecutorThreadPool.getInstance();
private static ArangoDBConnect arangoManger = ArangoDBConnect.getInstance(); private static ArangoDBConnect arangoManger = ArangoDBConnect.getInstance();
private static BaseArangoData baseArangoData = new BaseArangoData(); private static BaseArangoData baseArangoData = new BaseArangoData();
private static BaseClickhouseData baseClickhouseData = new BaseClickhouseData(); private static BaseClickhouseData baseClickhouseData = new BaseClickhouseData();
public void updateArango(){ public void updateArango(){
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
try { try {
updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN", updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN",
Fqdn.class,BaseDocument.class, Fqdn.class,BaseDocument.class,
ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument); ReadClickhouseData::getVertexFqdnSql, ReadClickhouseData::getVertexFqdnDocument);
updateDocument(newVertexIpMap,historyVertexIpMap,"IP", updateDocument(newVertexIpMap,historyVertexIpMap,"IP",
Ip.class,BaseDocument.class, Ip.class,BaseDocument.class,
ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument); ReadClickhouseData::getVertexIpSql, ReadClickhouseData::getVertexIpDocument);
// updateDocument(newVertexSubscriberMap,historyVertexSubscriberMap,"SUBSCRIBER",
// Subscriber.class,BaseDocument.class,
// ReadClickhouseData::getVertexSubscriberSql,ReadClickhouseData::getVertexSubscriberDocument);
updateDocument(newRelationFqdnAddressIpMap,historyRelationFqdnAddressIpMap,"R_LOCATE_FQDN2IP", updateDocument(newRelationFqdnAddressIpMap,historyRelationFqdnAddressIpMap,"R_LOCATE_FQDN2IP",
LocateFqdn2Ip.class,BaseEdgeDocument.class, LocateFqdn2Ip.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipFqdnAddressIpSql,ReadClickhouseData::getRelationFqdnAddressIpDocument); ReadClickhouseData::getRelationshipFqdnAddressIpSql, ReadClickhouseData::getRelationFqdnAddressIpDocument);
// updateDocument(newRelationIpVisitFqdnMap,historyRelationIpVisitFqdnMap,"R_VISIT_IP2FQDN", updateDocument(newRelationIpVisitFqdnMap,historyRelationIpVisitFqdnMap,"R_VISIT_IP2FQDN",
// VisitIp2Fqdn.class,BaseEdgeDocument.class, VisitIp2Fqdn.class,BaseEdgeDocument.class,
// ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument); ReadClickhouseData::getRelationshipIpVisitFqdnSql, ReadClickhouseData::getRelationIpVisitFqdnDocument);
// updateDocument(newRelationSubsciberLocateIpMap,historyRelationSubsciberLocateIpMap,"R_LOCATE_SUBSCRIBER2IP", updateDocument(newRelationFqdnSameFqdnMap,historyRelationFqdnSameFqdnMap,"R_SAME_ORIGIN_FQDN2FQDN",
// LocateSubscriber2Ip.class,BaseEdgeDocument.class, SameFqdn2Fqdn.class,BaseEdgeDocument.class,
// ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument); ReadClickhouseData::getRelationshipFqdnSameFqdnSql, ReadClickhouseData::getRelationshipFqdnSameFqdnDocument);
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();
LOG.info("iplearning application运行完毕，用时："+(last - start)); LOG.info("更新图数据库时间共计："+(last - start));
}catch (Exception e){ }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
}finally { }finally {
@@ -79,15 +73,13 @@ public class UpdateGraphData {
} }
} }
private <T extends BaseDocument> void updateDocument(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap, private <T extends BaseDocument> void updateDocument(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap, ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
String collection, String collection,
Class<? extends Document<T>> taskType, Class<? extends Document<T>> taskType,
Class<T> docmentType, Class<T> docmentType,
Supplier<String> getSqlSupplier, Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc Function<ResultSet,T> formatResultFunc) {
) {
try { try {
baseArangoData.readHistoryData(collection,historyMap,docmentType); baseArangoData.readHistoryData(collection,historyMap,docmentType);

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

@@ -1,6 +1,7 @@
package cn.ac.iie.service.ingestion; package cn.ac.iie.service.read;
import cn.ac.iie.config.ApplicationConfig; import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.TopDomainUtils;
import com.arangodb.entity.BaseDocument; import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument; import com.arangodb.entity.BaseEdgeDocument;
import org.slf4j.Logger; import org.slf4j.Logger;
@@ -22,10 +23,10 @@ public class ReadClickhouseData {
private static Pattern pattern = Pattern.compile("^[\\d]*$"); private static Pattern pattern = Pattern.compile("^[\\d]*$");
private static final Logger LOG = LoggerFactory.getLogger(ReadClickhouseData.class); private static final Logger LOG = LoggerFactory.getLogger(ReadClickhouseData.class);
private static long[] timeLimit = getTimeLimit(); private static long[] timeLimit = getTimeLimit();
private static long maxTime = timeLimit[0]; private static long maxTime = timeLimit[0];
private static long minTime = timeLimit[1]; private static long minTime = timeLimit[1];
public static final Integer DISTINCT_CLIENT_IP_NUM = ApplicationConfig.DISTINCT_CLIENT_IP_NUM; public static final Integer DISTINCT_CLIENT_IP_NUM = ApplicationConfig.DISTINCT_CLIENT_IP_NUM;
static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR; static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR;
public static final HashSet<String> PROTOCOL_SET; public static final HashSet<String> PROTOCOL_SET;
@@ -37,10 +38,11 @@ public class ReadClickhouseData {
PROTOCOL_SET.add("DNS"); PROTOCOL_SET.add("DNS");
} }
public static BaseDocument getVertexFqdnDocument(ResultSet resultSet) { public static BaseDocument getVertexFqdnDocument(ResultSet resultSet){
BaseDocument newDoc = null; BaseDocument newDoc = null;
try { try {
String fqdnName = resultSet.getString("FQDN"); String fqdnOrReferer = resultSet.getString("FQDN");
String fqdnName = TopDomainUtils.getDomainFromUrl(fqdnOrReferer);
if (isDomain(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");
@@ -50,13 +52,13 @@ public class ReadClickhouseData {
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime); newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
} }
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
} }
public static BaseDocument getVertexIpDocument(ResultSet resultSet) { public static BaseDocument getVertexIpDocument(ResultSet resultSet){
BaseDocument newDoc = new BaseDocument(); BaseDocument newDoc = new BaseDocument();
try { try {
String ip = resultSet.getString("IP"); String ip = resultSet.getString("IP");
@@ -65,11 +67,6 @@ public class ReadClickhouseData {
long sessionCount = resultSet.getLong("SESSION_COUNT"); long sessionCount = resultSet.getLong("SESSION_COUNT");
long bytesSum = resultSet.getLong("BYTES_SUM"); long bytesSum = resultSet.getLong("BYTES_SUM");
String ipType = resultSet.getString("ip_type"); String ipType = resultSet.getString("ip_type");
String[] commonLinkInfos = (String[]) resultSet.getArray("common_link_info").getArray();
String commonLinkInfo = "";
if (commonLinkInfos.length > 1){
commonLinkInfo = commonLinkInfos[1];
}
newDoc.setKey(ip); newDoc.setKey(ip);
newDoc.addAttribute("IP", ip); newDoc.addAttribute("IP", ip);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
@@ -88,15 +85,20 @@ public class ReadClickhouseData {
newDoc.addAttribute("CLIENT_BYTES_SUM", 0L); newDoc.addAttribute("CLIENT_BYTES_SUM", 0L);
break; break;
default: default:
newDoc.addAttribute("SERVER_SESSION_COUNT", 0L);
newDoc.addAttribute("SERVER_BYTES_SUM", 0L);
newDoc.addAttribute("CLIENT_SESSION_COUNT", 0L);
newDoc.addAttribute("CLIENT_BYTES_SUM", 0L);
break;
} }
newDoc.addAttribute("COMMON_LINK_INFO", commonLinkInfo); // newDoc.addAttribute("COMMON_LINK_INFO", "");
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
} }
public static BaseDocument getVertexSubscriberDocument(ResultSet resultSet) { public static BaseDocument getVertexSubscriberDocument(ResultSet resultSet){
BaseDocument newDoc = new BaseDocument(); BaseDocument newDoc = new BaseDocument();
try { try {
String subscriberId = resultSet.getString("common_subscriber_id"); String subscriberId = resultSet.getString("common_subscriber_id");
@@ -105,14 +107,13 @@ public class ReadClickhouseData {
newDoc.setKey(subscriberId); newDoc.setKey(subscriberId);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime); newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("SUBSCRIBER_ID",subscriberId); }catch (Exception e){
} catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
} }
public static BaseEdgeDocument getRelationshipSubsciberLocateIpDocument(ResultSet resultSet) { public static BaseEdgeDocument getRelationshipSubsciberLocateIpDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = new BaseEdgeDocument(); BaseEdgeDocument newDoc = new BaseEdgeDocument();
try { try {
String subscriberId = resultSet.getString("common_subscriber_id"); String subscriberId = resultSet.getString("common_subscriber_id");
@@ -128,14 +129,14 @@ public class ReadClickhouseData {
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime); newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("COUNT_TOTAL", countTotal); newDoc.addAttribute("COUNT_TOTAL", countTotal);
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
} }
public static BaseEdgeDocument getRelationFqdnAddressIpDocument(ResultSet resultSet) { public static BaseEdgeDocument getRelationFqdnAddressIpDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = null; BaseEdgeDocument newDoc = null;
try { try {
String vFqdn = resultSet.getString("FQDN"); String vFqdn = resultSet.getString("FQDN");
@@ -144,7 +145,6 @@ public class ReadClickhouseData {
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");
long countTotal = resultSet.getLong("COUNT_TOTAL"); long countTotal = resultSet.getLong("COUNT_TOTAL");
String schemaType = resultSet.getString("schema_type");
String[] distCipRecents = (String[]) resultSet.getArray("DIST_CIP_RECENT").getArray(); String[] distCipRecents = (String[]) resultSet.getArray("DIST_CIP_RECENT").getArray();
long[] clientIpTs = new long[distCipRecents.length]; long[] clientIpTs = new long[distCipRecents.length];
for (int i = 0; i < clientIpTs.length; i++) { for (int i = 0; i < clientIpTs.length; i++) {
@@ -158,18 +158,43 @@ public class ReadClickhouseData {
newDoc.setTo("IP/" + vIp); newDoc.setTo("IP/" + vIp);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime); newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("CNT_TOTAL",countTotal);
newDoc.addAttribute("DIST_CIP", distCipRecents); newDoc.addAttribute("DIST_CIP", distCipRecents);
newDoc.addAttribute("DIST_CIP_TS", clientIpTs); newDoc.addAttribute("DIST_CIP_TS", clientIpTs);
newDoc.addAttribute("PROTOCOL_TYPE", schemaType);
checkSchemaProperty(newDoc, schemaType, countTotal);
} }
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
} }
public static BaseEdgeDocument getRelationIpVisitFqdnDocument(ResultSet resultSet) { public static BaseEdgeDocument getRelationshipFqdnSameFqdnDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = null;
try {
String domainFqdn = resultSet.getString("domainFqdn");
String referer = resultSet.getString("referer");
String refererFqdn = TopDomainUtils.getDomainFromUrl(referer);
if (isDomain(refererFqdn) && isDomain(domainFqdn)){
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long countTotal = resultSet.getLong("COUNT_TOTAL");
String key = domainFqdn + "-" + refererFqdn;
newDoc = new BaseEdgeDocument();
newDoc.setKey(key);
newDoc.setFrom("FQDN/" + domainFqdn);
newDoc.setTo("FQDN/" + refererFqdn);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("CNT_TOTAL",countTotal);
}
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static BaseEdgeDocument getRelationIpVisitFqdnDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = null; BaseEdgeDocument newDoc = null;
try { try {
String vFqdn = resultSet.getString("FQDN"); String vFqdn = resultSet.getString("FQDN");
@@ -179,18 +204,16 @@ public class ReadClickhouseData {
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");
long countTotal = resultSet.getLong("COUNT_TOTAL"); long countTotal = resultSet.getLong("COUNT_TOTAL");
String schemaType = resultSet.getString("schema_type");
newDoc = new BaseEdgeDocument(); newDoc = new BaseEdgeDocument();
newDoc.setKey(key); newDoc.setKey(key);
newDoc.setFrom("IP/" + vIp); newDoc.setFrom("IP/" + vIp);
newDoc.setTo("FQDN/" + vFqdn); newDoc.setTo("FQDN/" + vFqdn);
newDoc.addAttribute("CNT_TOTAL",countTotal);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime); newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime); newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("PROTOCOL_TYPE", schemaType);
checkSchemaProperty(newDoc, schemaType, countTotal);
} }
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} }
return newDoc; return newDoc;
@@ -199,8 +222,8 @@ public class ReadClickhouseData {
public static <T extends BaseDocument> void putMapByHashcode(T newDoc, HashMap<Integer, HashMap<String, ArrayList<T>>> map) { public static <T extends BaseDocument> void putMapByHashcode(T newDoc, HashMap<Integer, HashMap<String, ArrayList<T>>> map) {
if (newDoc != null) { if (newDoc != null) {
String key = newDoc.getKey(); String key = newDoc.getKey();
int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER; int i = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER;
HashMap<String, ArrayList<T>> documentHashMap = map.getOrDefault(hashCode, new HashMap<>()); HashMap<String, ArrayList<T>> documentHashMap = map.getOrDefault(i, new HashMap<>());
ArrayList<T> documentArrayList = documentHashMap.getOrDefault(key, new ArrayList<>()); ArrayList<T> documentArrayList = documentHashMap.getOrDefault(key, new ArrayList<>());
documentArrayList.add(newDoc); documentArrayList.add(newDoc);
documentHashMap.put(key, documentArrayList); documentHashMap.put(key, documentArrayList);
@@ -239,16 +262,17 @@ public class ReadClickhouseData {
return false; return false;
} }
private static void checkSchemaProperty(BaseEdgeDocument newDoc, String schema, long countTotal) { private static void checkSchemaProperty(BaseEdgeDocument newDoc, String schema, long countTotal) {
long[] recentCnt = new long[RECENT_COUNT_HOUR]; long[] recentCnt = new long[RECENT_COUNT_HOUR];
recentCnt[0] = countTotal; recentCnt[0] = countTotal;
for (String protocol : PROTOCOL_SET) { for (String protocol: PROTOCOL_SET){
String protocolRecent = protocol + "_CNT_RECENT"; String protocolRecent = protocol +"_CNT_RECENT";
String protocolTotal = protocol + "_CNT_TOTAL"; String protocolTotal = protocol + "_CNT_TOTAL";
if (protocol.equals(schema)) { if (protocol.equals(schema)){
newDoc.addAttribute(protocolTotal, countTotal); newDoc.addAttribute(protocolTotal, countTotal);
newDoc.addAttribute(protocolRecent, recentCnt); newDoc.addAttribute(protocolRecent, recentCnt);
} else { }else {
newDoc.addAttribute(protocolTotal, 0L); newDoc.addAttribute(protocolTotal, 0L);
newDoc.addAttribute(protocolRecent, new long[RECENT_COUNT_HOUR]); newDoc.addAttribute(protocolRecent, new long[RECENT_COUNT_HOUR]);
} }
@@ -256,47 +280,48 @@ public class ReadClickhouseData {
} }
public static String getVertexFqdnSql() { public static String getVertexFqdnSql() {
String where = "common_start_time >= " + minTime + " AND common_start_time < " + maxTime; String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime;
String sslSql = "SELECT ssl_sni AS FQDN,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni"; String mediaDomainSql = "SELECT s1_domain AS FQDN,MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME FROM media_expire_patch WHERE "+where+" and s1_domain != '' GROUP BY s1_domain";
String httpSql = "SELECT http_host AS FQDN,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host"; String refererSql = "SELECT s1_referer AS FQDN,MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME FROM media_expire_patch WHERE "+where+" and s1_referer != '' GROUP BY s1_referer";
return "SELECT FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME FROM ((" + sslSql + ") UNION ALL (" + httpSql + ")) GROUP BY FQDN HAVING FQDN != ''"; return "SELECT * FROM((" + mediaDomainSql + ") UNION ALL (" + refererSql + "))";
} }
public static String getVertexIpSql() { public static String getVertexIpSql() {
String where = " common_start_time >= " + minTime + " AND common_start_time < " + maxTime; String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_start_time) AS FIRST_FOUND_TIME,MAX(common_start_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_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 clientIpSql = "SELECT s1_s_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(media_len) as BYTES_SUM,'client' as ip_type FROM media_expire_patch where " + where + " group by IP";
String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_start_time) AS FIRST_FOUND_TIME,MAX(common_start_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(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 serverIpSql = "SELECT s1_d_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(media_len) as BYTES_SUM,'server' as ip_type FROM media_expire_patch where " + where + " group by IP";
return "SELECT * FROM((" + clientIpSql + ") UNION ALL (" + serverIpSql + "))"; return "SELECT * FROM((" + clientIpSql + ") UNION ALL (" + serverIpSql + "))";
} }
public static String getRelationshipFqdnAddressIpSql() { public static String getRelationshipFqdnAddressIpSql() {
String where = " common_start_time >= " + minTime + " AND common_start_time < " + maxTime; String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime +" AND s1_domain != '' AND s1_d_ip != '' ";
String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'TLS' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip"; return "SELECT s1_domain AS FQDN,s1_d_ip AS common_server_ip,MIN( recv_time ) AS FIRST_FOUND_TIME,MAX( recv_time ) AS LAST_FOUND_TIME,COUNT( * ) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(s1_s_ip) AS DIST_CIP_RECENT FROM media_expire_patch WHERE "+where+" GROUP BY s1_d_ip,s1_domain";
String httpSql = "SELECT http_host AS FQDN,common_server_ip,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip"; }
return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
public static String getRelationshipFqdnSameFqdnSql(){
String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime +" AND s1_domain != '' AND s1_referer != '' ";
return "SELECT s1_domain AS domainFqdn,s1_referer AS referer,MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL FROM media_expire_patch where "+where+" GROUP BY s1_domain,s1_referer";
} }
public static String getRelationshipIpVisitFqdnSql() { public static String getRelationshipIpVisitFqdnSql() {
String where = " common_start_time >= " + minTime + " AND common_start_time < " + maxTime; String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime+" AND s1_s_ip != '' AND s1_domain != '' ";
String httpSql = "SELECT http_host AS FQDN,common_client_ip,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_client_ip"; return "SELECT s1_s_ip AS common_client_ip,s1_domain AS FQDN,MIN( recv_time ) AS FIRST_FOUND_TIME,MAX( recv_time ) AS LAST_FOUND_TIME,COUNT( * ) AS COUNT_TOTAL FROM media_expire_patch WHERE "+where+" GROUP BY s1_s_ip,s1_domain";
String sslSql = "SELECT ssl_sni AS FQDN,common_client_ip,MAX(common_start_time) AS LAST_FOUND_TIME,MIN(common_start_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'TLS' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
} }
public static String getVertexSubscriberSql() { public static String getVertexSubscriberSql() {
String where = " common_start_time >= " + minTime + " AND common_start_time < " + maxTime + " AND common_subscriber_id != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1"; String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
return "SELECT common_subscriber_id,MAX(common_start_time) as LAST_FOUND_TIME,MIN(common_start_time) as FIRST_FOUND_TIME FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id"; return "SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id";
} }
public static String getRelationshipSubsciberLocateIpSql() { public static String getRelationshipSubsciberLocateIpSql() {
String where = " common_start_time >= " + minTime + " AND common_start_time < " + maxTime + " AND common_subscriber_id != '' AND radius_framed_ip != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1"; String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_framed_ip != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_start_time) as LAST_FOUND_TIME,MIN(common_start_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip"; return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip";
} }
private static long[] getTimeLimit() { private static long[] getTimeLimit() {
long maxTime = 0L; long maxTime = 0L;
long minTime = 0L; long minTime = 0L;
switch (ApplicationConfig.TIME_LIMIT_TYPE) { switch (ApplicationConfig.CLICKHOUSE_TIME_LIMIT_TYPE) {
case 0: case 0:
maxTime = currentHour; maxTime = currentHour;
minTime = maxTime - ApplicationConfig.UPDATE_INTERVAL; minTime = maxTime - ApplicationConfig.UPDATE_INTERVAL;

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

@@ -1,4 +1,4 @@
package cn.ac.iie.service.ingestion; package cn.ac.iie.service.read;
import cn.ac.iie.config.ApplicationConfig; import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
@@ -8,31 +8,29 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import static cn.ac.iie.service.ingestion.ReadClickhouseData.*; import static cn.ac.iie.service.read.ReadClickhouseData.RECENT_COUNT_HOUR;
/** /**
* @author wlh * @author wlh
* 多线程全量读取arangoDb历史数据，封装到map * 多线程全量读取arangoDb历史数据，封装到map
*/ */
@SuppressWarnings("unchecked")
public class ReadHistoryArangoData<T extends BaseDocument> extends Thread { public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
private static final Logger LOG = LoggerFactory.getLogger(ReadHistoryArangoData.class); private static final Logger LOG = LoggerFactory.getLogger(ReadHistoryArangoData.class);
private ArangoDBConnect arangoConnect; private ArangoDBConnect arangoConnect;
private String query; private String query;
private ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> map; private ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map;
private Class<T> type; private Class<T> type;
private String table; private String table;
private CountDownLatch countDownLatch; private CountDownLatch countDownLatch;
public ReadHistoryArangoData(ArangoDBConnect arangoConnect, public ReadHistoryArangoData(ArangoDBConnect arangoConnect,
String query, String query,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> map, ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,
Class<T> type, Class<T> type,
String table, String table,
CountDownLatch countDownLatch) { CountDownLatch countDownLatch) {
@@ -54,16 +52,6 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
int i = 0; int i = 0;
for (T doc : baseDocuments) { for (T doc : baseDocuments) {
String key = doc.getKey(); String key = doc.getKey();
switch (table) {
case "R_LOCATE_FQDN2IP":
updateProtocolDocument(doc);
deleteDistinctClientIpByTime(doc);
break;
case "R_VISIT_IP2FQDN":
updateProtocolDocument(doc);
break;
default:
}
int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER; int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER;
ConcurrentHashMap<String, T> tmpMap = map.get(hashCode); ConcurrentHashMap<String, T> tmpMap = map.get(hashCode);
tmpMap.put(key, doc); tmpMap.put(key, doc);
@@ -72,11 +60,11 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
long l = System.currentTimeMillis(); long l = System.currentTimeMillis();
LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s)); LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
} }
} catch (Exception e) { }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
} finally { }finally {
countDownLatch.countDown(); countDownLatch.countDown();
LOG.info("本线程读取完毕，剩余线程数量：" + countDownLatch.getCount()); LOG.info("本线程读取完毕，剩余线程数量："+countDownLatch.getCount());
} }
} }
@@ -85,7 +73,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
for (String protocol : ReadClickhouseData.PROTOCOL_SET) { for (String protocol : ReadClickhouseData.PROTOCOL_SET) {
String protocolRecent = protocol + "_CNT_RECENT"; String protocolRecent = protocol + "_CNT_RECENT";
ArrayList<Long> cntRecent = (ArrayList<Long>) doc.getAttribute(protocolRecent); ArrayList<Long> cntRecent = (ArrayList<Long>) doc.getAttribute(protocolRecent);
Long[] cntRecentsSrc = cntRecent.toArray(new Long[0]); Long[] cntRecentsSrc = cntRecent.toArray(new Long[cntRecent.size()]);
Long[] cntRecentsDst = new Long[RECENT_COUNT_HOUR]; Long[] cntRecentsDst = new Long[RECENT_COUNT_HOUR];
System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1); System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1);
cntRecentsDst[0] = 0L; cntRecentsDst[0] = 0L;
@@ -94,22 +82,4 @@ 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");
distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
Collections.sort(distCipTs);
int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);
String[] distCipArr = new String[index];
long[] disCipTsArr = new long[index];
if (distCip.size() + 1 == distCipTs.size()){
for (int i = 0; i < index; i++) {
distCipArr[i] = distCip.get(i);
disCipTsArr[i] = distCipTs.get(i);
}
}
doc.updateAttribute("DIST_CIP", distCipArr);
doc.updateAttribute("DIST_CIP_TS", disCipTsArr);
}
} }

5
IP-learning-graph/src/main/java/cn/ac/iie/service/update/Document.java
View File

@@ -124,10 +124,5 @@ public class Document<T extends BaseDocument> extends Thread{
lastDoc.addAttribute(attribute,firstSumAttribute+lastSumAttribute); lastDoc.addAttribute(attribute,firstSumAttribute+lastSumAttribute);
} }
protected void replaceAttribute(T firstDoc,T lastDoc,String attribute){
Object attributeObj = firstDoc.getAttribute(attribute);
lastDoc.addAttribute(attribute,attributeObj);
}
} }

12
IP-learning-graph/src/main/java/cn/ac/iie/service/update/Relationship.java
View File

@@ -1,6 +1,6 @@
package cn.ac.iie.service.update; package cn.ac.iie.service.update;
import cn.ac.iie.service.ingestion.ReadClickhouseData; import cn.ac.iie.service.read.ReadClickhouseData;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseEdgeDocument; import com.arangodb.entity.BaseEdgeDocument;
@@ -24,11 +24,6 @@ public class Relationship extends Document<BaseEdgeDocument> {
super.updateFunction(newEdgeDocument,historyEdgeDocument); super.updateFunction(newEdgeDocument,historyEdgeDocument);
} }
@Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
super.mergeFunction(lastDoc, newDocument);
}
protected void updateProcotol(BaseEdgeDocument historyEdgeDocument, String schema, BaseEdgeDocument newEdgeDocument){ protected void updateProcotol(BaseEdgeDocument historyEdgeDocument, String schema, BaseEdgeDocument newEdgeDocument){
String recentSchema = schema +"_CNT_RECENT"; String recentSchema = schema +"_CNT_RECENT";
String totalSchema = schema + "_CNT_TOTAL"; String totalSchema = schema + "_CNT_TOTAL";
@@ -49,6 +44,11 @@ public class Relationship extends Document<BaseEdgeDocument> {
} }
} }
@Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
super.mergeFunction(lastDoc, newDocument);
}
protected void mergeProtocol(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) { protected void mergeProtocol(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
String schema = lastDoc.getAttribute("PROTOCOL_TYPE").toString(); String schema = lastDoc.getAttribute("PROTOCOL_TYPE").toString();
if (ReadClickhouseData.PROTOCOL_SET.contains(schema)){ if (ReadClickhouseData.PROTOCOL_SET.contains(schema)){

15
IP-learning-graph/src/main/java/cn/ac/iie/service/update/Vertex.java
View File

@@ -22,4 +22,19 @@ public class Vertex extends Document<BaseDocument> {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch); super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
} }
@Override
protected void updateFunction(BaseDocument newDocument, BaseDocument historyDocument) {
super.updateFunction(newDocument, historyDocument);
}
@Override
protected void mergeFunction(BaseDocument lastDoc,BaseDocument newDocument) {
super.mergeFunction(lastDoc, newDocument);
}
@Override
public void run() {
super.run();
}
} }

21
IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java
View File

@@ -1,6 +1,5 @@
package cn.ac.iie.service.update.relationship; package cn.ac.iie.service.update.relationship;
import cn.ac.iie.service.ingestion.ReadClickhouseData;
import cn.ac.iie.service.update.Relationship; import cn.ac.iie.service.update.Relationship;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseEdgeDocument; import com.arangodb.entity.BaseEdgeDocument;
@@ -9,8 +8,8 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import static cn.ac.iie.service.ingestion.ReadClickhouseData.DISTINCT_CLIENT_IP_NUM; import static cn.ac.iie.service.read.ReadClickhouseData.DISTINCT_CLIENT_IP_NUM;
import static cn.ac.iie.service.ingestion.ReadClickhouseData.currentHour; import static cn.ac.iie.service.read.ReadClickhouseData.currentHour;
public class LocateFqdn2Ip extends Relationship { public class LocateFqdn2Ip extends Relationship {
@@ -26,7 +25,7 @@ public class LocateFqdn2Ip extends Relationship {
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){ protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){
super.mergeFunction(lastDoc, newDocument); super.mergeFunction(lastDoc, newDocument);
mergeDistinctClientIp(lastDoc, newDocument); mergeDistinctClientIp(lastDoc, newDocument);
mergeProtocol(lastDoc, newDocument); putSumAttribute(lastDoc, newDocument,"CNT_TOTAL");
} }
private void mergeDistinctClientIp(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){ private void mergeDistinctClientIp(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){
@@ -46,19 +45,17 @@ public class LocateFqdn2Ip extends Relationship {
@Override @Override
protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) { protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) {
super.updateFunction(newEdgeDocument, historyEdgeDocument); super.updateFunction(newEdgeDocument, historyEdgeDocument);
for (String schema:ReadClickhouseData.PROTOCOL_SET){
updateProcotol(historyEdgeDocument,schema,newEdgeDocument);
}
updateDistinctClientIp(newEdgeDocument, historyEdgeDocument); updateDistinctClientIp(newEdgeDocument, historyEdgeDocument);
putSumAttribute(newEdgeDocument, historyEdgeDocument,"CNT_TOTAL");
} }
private void updateDistinctClientIp(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){ private void updateDistinctClientIp(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){
String[] distCip = (String[]) edgeDocument.getAttribute("DIST_CIP"); ArrayList<String> distCip = (ArrayList<String>) edgeDocument.getAttribute("DIST_CIP");
long[] distCipTs = (long[]) edgeDocument.getAttribute("DIST_CIP_TS"); ArrayList<Long> distCipTs = (ArrayList<Long>) edgeDocument.getAttribute("DIST_CIP_TS");
HashMap<String, Long> distCipToTs = new HashMap<>(); HashMap<String, Long> distCipToTs = new HashMap<>();
if (distCip.length == distCipTs.length){ if (distCip.size() == distCipTs.size()){
for (int i = 0;i < distCip.length;i++){ for (int i = 0;i < distCip.size();i++){
distCipToTs.put(distCip[i],distCipTs[i]); distCipToTs.put(distCip.get(i),distCipTs.get(i));
} }
} }
Object[] distCipRecent = (Object[])newEdgeDocument.getAttribute("DIST_CIP"); Object[] distCipRecent = (Object[])newEdgeDocument.getAttribute("DIST_CIP");

34
IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/SameFqdn2Fqdn.java Normal file
View File

@@ -0,0 +1,34 @@
package cn.ac.iie.service.update.relationship;
import cn.ac.iie.service.update.Relationship;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseEdgeDocument;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class SameFqdn2Fqdn extends Relationship {
public SameFqdn2Fqdn(HashMap<String, ArrayList<BaseEdgeDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseEdgeDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
}
@Override
protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) {
super.updateFunction(newEdgeDocument, historyEdgeDocument);
putSumAttribute(newEdgeDocument,historyEdgeDocument,"CNT_TOTAL");
}
@Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
super.mergeFunction(lastDoc, newDocument);
putSumAttribute(lastDoc,newDocument,"CNT_TOTAL");
}
}

8
IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/VisitIp2Fqdn.java
View File

@@ -1,13 +1,11 @@
package cn.ac.iie.service.update.relationship; package cn.ac.iie.service.update.relationship;
import cn.ac.iie.service.ingestion.ReadClickhouseData;
import cn.ac.iie.service.update.Relationship; import cn.ac.iie.service.update.Relationship;
import cn.ac.iie.utils.ArangoDBConnect; import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseEdgeDocument; import com.arangodb.entity.BaseEdgeDocument;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
@@ -23,14 +21,12 @@ public class VisitIp2Fqdn extends Relationship {
@Override @Override
protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) { protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) {
super.updateFunction(newEdgeDocument, historyEdgeDocument); super.updateFunction(newEdgeDocument, historyEdgeDocument);
for (String schema: ReadClickhouseData.PROTOCOL_SET){ putSumAttribute(newEdgeDocument,historyEdgeDocument,"CNT_TOTAL");
updateProcotol(historyEdgeDocument,schema,newEdgeDocument);
}
} }
@Override @Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) { protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
super.mergeFunction(lastDoc, newDocument); super.mergeFunction(lastDoc, newDocument);
mergeProtocol(lastDoc,newDocument); putSumAttribute(lastDoc,newDocument,"CNT_TOTAL");
} }
} }

12
IP-learning-graph/src/main/java/cn/ac/iie/service/update/vertex/Ip.java
View File

@@ -16,20 +16,26 @@ public class Ip extends Vertex {
String collectionName, String collectionName,
ConcurrentHashMap<String, BaseDocument> historyDocumentMap, ConcurrentHashMap<String, BaseDocument> historyDocumentMap,
CountDownLatch countDownLatch) { CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap,countDownLatch); super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
} }
@Override @Override
protected void updateFunction(BaseDocument newDocument, BaseDocument historyDocument) { protected void updateFunction(BaseDocument newDocument, BaseDocument historyDocument) {
super.updateFunction(newDocument, historyDocument); super.updateFunction(newDocument, historyDocument);
updateIpByType(newDocument, historyDocument); updateIpByType(newDocument, historyDocument);
super.replaceAttribute(newDocument,historyDocument,"COMMON_LINK_INFO");
} }
@Override @Override
protected void mergeFunction(BaseDocument lastDoc, BaseDocument newDocument) { protected void mergeFunction(BaseDocument lastDoc, BaseDocument newDocument) {
super.mergeFunction(lastDoc, newDocument); super.mergeFunction(lastDoc, newDocument);
updateIpByType(lastDoc, newDocument); mergeIpByType(lastDoc, newDocument);
}
private void mergeIpByType(BaseDocument lastDoc, BaseDocument newDocument) {
putSumAttribute(lastDoc,newDocument,"CLIENT_SESSION_COUNT");
putSumAttribute(lastDoc,newDocument,"CLIENT_BYTES_SUM");
putSumAttribute(lastDoc,newDocument,"SERVER_SESSION_COUNT");
putSumAttribute(lastDoc,newDocument,"SERVER_BYTES_SUM");
} }
private void updateIpByType(BaseDocument newDocument, BaseDocument historyDocument) { private void updateIpByType(BaseDocument newDocument, BaseDocument historyDocument) {

6
IP-learning-graph/src/main/java/cn/ac/iie/test/IpLearningApplicationTest.java
View File

@@ -2,9 +2,15 @@ package cn.ac.iie.test;
import cn.ac.iie.dao.UpdateGraphData; import cn.ac.iie.dao.UpdateGraphData;
/**
* iplearning程序入口
* @author wlh
*/
public class IpLearningApplicationTest { public class IpLearningApplicationTest {
public static void main(String[] args) { public static void main(String[] args) {
UpdateGraphData updateGraphData = new UpdateGraphData(); UpdateGraphData updateGraphData = new UpdateGraphData();
updateGraphData.updateArango(); updateGraphData.updateArango();

40
IP-learning-graph/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java
View File

@@ -16,7 +16,6 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.List;
import java.util.Map; import java.util.Map;
public class ArangoDBConnect { public class ArangoDBConnect {
@@ -24,10 +23,10 @@ public class ArangoDBConnect {
private static ArangoDB arangoDB = null; private static ArangoDB arangoDB = null;
private static ArangoDBConnect conn = null; private static ArangoDBConnect conn = null;
static { static {
getArangoDB(); getArangoDatabase();
} }
private static void getArangoDB(){ private static void getArangoDatabase(){
arangoDB = new ArangoDB.Builder() arangoDB = new ArangoDB.Builder()
.maxConnections(ApplicationConfig.THREAD_POOL_NUMBER) .maxConnections(ApplicationConfig.THREAD_POOL_NUMBER)
.host(ApplicationConfig.ARANGODB_HOST, ApplicationConfig.ARANGODB_PORT) .host(ApplicationConfig.ARANGODB_HOST, ApplicationConfig.ARANGODB_PORT)
@@ -53,26 +52,45 @@ public class ArangoDBConnect {
arangoDB.shutdown(); arangoDB.shutdown();
} }
}catch (Exception e){ }catch (Exception e){
LOG.error(e.getMessage()); e.printStackTrace();
} }
} }
public <T> ArangoCursor<T> executorQuery(String query,Class<T> type){ public <T> ArangoCursor<T> executorQuery(String query,Class<T> type){
ArangoDatabase database = getDatabase(); ArangoDatabase database = getDatabase();
Map<String, Object> bindVars = new MapBuilder().get(); Map<String, Object> bindVars = new MapBuilder().get();
AqlQueryOptions options = new AqlQueryOptions() AqlQueryOptions options = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL);
.ttl(ApplicationConfig.ARANGODB_TTL);
try { try {
return database.query(query, bindVars, options, type); return database.query(query, bindVars, options, type);
}catch (Exception e){ }catch (Exception e){
LOG.error(e.getMessage()); e.printStackTrace();
return null; return null;
}finally { }finally {
bindVars.clear(); bindVars.clear();
} }
} }
public <T> void overwrite(List<T> docOverwrite, String collectionName){ @Deprecated
public <T> void insertAndUpdate(ArrayList<T> docInsert,ArrayList<T> docUpdate,String collectionName){
ArangoDatabase database = getDatabase();
try {
ArangoCollection collection = database.collection(collectionName);
if (!docInsert.isEmpty()){
collection.importDocuments(docInsert);
}
if (!docUpdate.isEmpty()){
collection.replaceDocuments(docUpdate);
}
}catch (Exception e){
System.out.println("更新失败");
e.printStackTrace();
}finally {
docInsert.clear();
docInsert.clear();
}
}
public <T> void overwrite(ArrayList<T> docOverwrite,String collectionName){
ArangoDatabase database = getDatabase(); ArangoDatabase database = getDatabase();
try { try {
ArangoCollection collection = database.collection(collectionName); ArangoCollection collection = database.collection(collectionName);
@@ -83,14 +101,16 @@ public class ArangoDBConnect {
MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions); MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors(); Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
for (ErrorEntity errorEntity:errors){ for (ErrorEntity errorEntity:errors){
LOG.error("写入arangoDB异常："+errorEntity.getErrorMessage()); LOG.debug("写入arangoDB异常："+errorEntity.getErrorMessage());
} }
} }
}catch (Exception e){ }catch (Exception e){
LOG.error("更新失败："+e.toString()); System.out.println("更新失败："+e.toString());
}finally { }finally {
docOverwrite.clear(); docOverwrite.clear();
} }
} }
} }

9
IP-learning-graph/src/main/java/cn/ac/iie/utils/ClickhouseConnect.java
View File

@@ -2,8 +2,6 @@ package cn.ac.iie.utils;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection; import com.alibaba.druid.pool.DruidPooledConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection; import java.sql.Connection;
import java.sql.ResultSet; import java.sql.ResultSet;
@@ -12,7 +10,6 @@ import java.sql.Statement;
import java.util.Properties; import java.util.Properties;
public class ClickhouseConnect { public class ClickhouseConnect {
private static final Logger LOG = LoggerFactory.getLogger(ClickhouseConnect.class);
private static DruidDataSource dataSource = null; private static DruidDataSource dataSource = null;
private static ClickhouseConnect dbConnect = null; private static ClickhouseConnect dbConnect = null;
private static Properties props = new Properties(); private static Properties props = new Properties();
@@ -46,7 +43,7 @@ public class ClickhouseConnect {
dataSource.setKeepAlive(true); dataSource.setKeepAlive(true);
} }
} catch (Exception e) { } catch (Exception e) {
LOG.error(e.getMessage()); e.printStackTrace();
} }
} }
@@ -88,7 +85,7 @@ public class ClickhouseConnect {
connection.close(); connection.close();
} }
} catch (SQLException e) { } catch (SQLException e) {
LOG.error(e.getMessage()); e.printStackTrace();
} }
} }
@@ -101,7 +98,7 @@ public class ClickhouseConnect {
pstm = connection.createStatement(); pstm = connection.createStatement();
return pstm.executeQuery(query); return pstm.executeQuery(query);
}catch (Exception e){ }catch (Exception e){
LOG.error(e.getMessage()); e.printStackTrace();
return null; return null;
} }
} }

8
IP-learning-graph/src/main/java/cn/ac/iie/utils/ConfigUtils.java
View File

@@ -1,12 +1,8 @@
package cn.ac.iie.utils; package cn.ac.iie.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties; import java.util.Properties;
public class ConfigUtils { public class ConfigUtils {
private static final Logger LOG = LoggerFactory.getLogger(ConfigUtils.class);
private static Properties propCommon = new Properties(); private static Properties propCommon = new Properties();
public static String getStringProperty(String key) { public static String getStringProperty(String key) {
@@ -29,12 +25,12 @@ public class ConfigUtils {
static { static {
try { try {
propCommon.load(ConfigUtils.class.getClassLoader().getResourceAsStream("application.properties")); propCommon.load(ConfigUtils.class.getClassLoader().getResourceAsStream("application.properties"));
LOG.info("application.properties加载成功"); System.out.println("application.properties加载成功");
} catch (Exception e) { } catch (Exception e) {
propCommon = null; propCommon = null;
LOG.error("配置加载失败"); System.err.println("配置加载失败");
} }
} }
} }

17
IP-learning-graph/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
View File

@@ -2,13 +2,14 @@ package cn.ac.iie.utils;
import cn.ac.iie.config.ApplicationConfig; import cn.ac.iie.config.ApplicationConfig;
import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*; import java.util.concurrent.*;
/**
* 线程池管理
* @author wlh
*/
public class ExecutorThreadPool { public class ExecutorThreadPool {
private static final Logger LOG = LoggerFactory.getLogger(ExecutorThreadPool.class);
private static ExecutorService pool = null ; private static ExecutorService pool = null ;
private static ExecutorThreadPool poolExecutor = null; private static ExecutorThreadPool poolExecutor = null;
@@ -19,9 +20,13 @@ public class ExecutorThreadPool {
private static void getThreadPool(){ private static void getThreadPool(){
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder() ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("iplearning-application-pool-%d").build(); .setNameFormat("iplearning-application-pool-%d").build();
pool = new ThreadPoolExecutor(ApplicationConfig.THREAD_POOL_NUMBER, ApplicationConfig.THREAD_POOL_NUMBER,
0L, TimeUnit.SECONDS, //Common Thread Pool
pool = new ThreadPoolExecutor(ApplicationConfig.THREAD_POOL_NUMBER, ApplicationConfig.THREAD_POOL_NUMBER*2,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy()); new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
// pool = Executors.newFixedThreadPool(ApplicationConfig.THREAD_POOL_NUMBER);
} }
public static ExecutorThreadPool getInstance(){ public static ExecutorThreadPool getInstance(){
@@ -39,7 +44,7 @@ public class ExecutorThreadPool {
public void awaitThreadTask(){ public void awaitThreadTask(){
try { try {
while (!pool.awaitTermination(ApplicationConfig.THREAD_AWAIT_TERMINATION_TIME, TimeUnit.SECONDS)) { while (!pool.awaitTermination(ApplicationConfig.THREAD_AWAIT_TERMINATION_TIME, TimeUnit.SECONDS)) {
LOG.warn("线程池没有关闭"); System.out.println("线程池没有关闭");
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();

161
IP-learning-graph/src/main/java/cn/ac/iie/utils/TopDomainUtils.java Normal file
View File

@@ -0,0 +1,161 @@
package cn.ac.iie.utils;
import cn.ac.iie.config.ApplicationConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.net.URL;
import java.util.HashMap;
public class TopDomainUtils {
private static Logger logger = LoggerFactory.getLogger(TopDomainUtils.class);
public static String getSecDomain(String urlDomain, HashMap<String, HashMap<String, String>> maps) {
String[] split = urlDomain.split("\\.");
String secDomain = null;
for (int i = split.length - 1; i >= 0; i--) {
int mapsIndex = split.length - (i + 1);
HashMap<String, String> innerMap = maps.get("map_id_" + mapsIndex);
HashMap<String, String> fullTop = maps.get("full");
if (!(innerMap.containsKey(split[i]))) {
StringBuilder strSec = new StringBuilder();
for (int j = i; j < split.length; j++) {
strSec.append(split[j]).append(".");
}
secDomain = strSec.substring(0, strSec.length() - 1);
if (fullTop.containsKey(getTopDomainFromSecDomain(secDomain))) {
break;
} else {
while (!fullTop.containsKey(getTopDomainFromSecDomain(secDomain)) && getTopDomainFromSecDomain(secDomain).contains(".")) {
secDomain = getTopDomainFromSecDomain(secDomain);
}
break;
}
}
}
return secDomain;
}
private static String getTopDomainFromSecDomain(String secDomain) {
String quFirstDian = secDomain;
if (secDomain.contains(".")) {
quFirstDian = secDomain.substring(secDomain.indexOf(".")).substring(1);
}
return quFirstDian;
}
private static File getTopDomainFile(){
URL url = TopDomainUtils.class.getClassLoader().getResource(ApplicationConfig.TOP_DOMAIN_FILE_NAME);
File file = null;
if (url!=null){
file = new File(url.getFile());
}
if (file != null && file.isFile() && file.exists()){
return file;
}
return null;
}
public static HashMap<String, HashMap<String, String>> readTopDomainFile() {
URL url = TopDomainUtils.class.getClassLoader().getResource(ApplicationConfig.TOP_DOMAIN_FILE_NAME);
assert url != null;
HashMap<String, HashMap<String, String>> maps = makeHashMap(url.getFile());
try {
String encoding = "UTF-8";
File file = new File(url.getFile());
if (file.isFile() && file.exists()) {
InputStreamReader read = new InputStreamReader(
new FileInputStream(file), encoding);
BufferedReader bufferedReader = new BufferedReader(read);
String lineTxt;
while ((lineTxt = bufferedReader.readLine()) != null) {
HashMap<String, String> fullTop = maps.get("full");
fullTop.put(lineTxt, lineTxt);
maps.put("full", fullTop);
String[] split = lineTxt.split("\\.");
for (int i = split.length - 1; i >= 0; i--) {
int mapsIndex = split.length - (i + 1);
HashMap<String, String> innerMap = maps.get("map_id_" + mapsIndex);
innerMap.put(split[i], split[i]);
maps.put("map_id_" + mapsIndex, innerMap);
}
}
read.close();
}
} catch (Exception e) {
logger.error("TopDomainUtils>=>readTopDomainFile get filePathData error--->{" + e + "}<---");
e.printStackTrace();
}
return maps;
}
private static int getMaxLength(String filePath) {
int lengthDomain = 0;
try {
String encoding = "UTF-8";
File file = new File(filePath);
if (file.isFile() && file.exists()) {
InputStreamReader read = new InputStreamReader(
new FileInputStream(file), encoding);
BufferedReader bufferedReader = new BufferedReader(read);
String lineTxt;
while ((lineTxt = bufferedReader.readLine()) != null) {
String[] split = lineTxt.split("\\.");
if (split.length > lengthDomain) {
lengthDomain = split.length;
}
}
read.close();
} else {
logger.error("TopDomainUtils>>getMaxLength filePath is wrong--->{" + filePath + "}<---");
}
} catch (Exception e) {
logger.error("TopDomainUtils>=>getMaxLength get filePathData error--->{" + e + "}<---");
e.printStackTrace();
}
return lengthDomain;
}
private static HashMap<String, HashMap<String, String>> makeHashMap(String filePath) {
int maxLength = getMaxLength(filePath);
HashMap<String, HashMap<String, String>> maps = new HashMap<>();
for (int i = 0; i < maxLength; i++) {
maps.put("map_id_" + i, new HashMap<String, String>());
}
maps.put("full", new HashMap<String, String>());
return maps;
}
/**
* 通用方法,传入url,返回domain,这里的domain不包含端口号,含有:一定是v6
* @param oriUrl
* @return
*/
public static String getDomainFromUrl(String oriUrl) {
//先按照?切分,排除后续干扰
String url = oriUrl.split("[?]")[0];
//排除http://或https://干扰
url = url.replaceAll("https://", "").replaceAll("http://", "");
String domain;
//获取domain
if (url.split("/")[0].split(":").length <= 2) {
//按照:切分后最终长度为1或2,说明是v4
domain = url
//按照/切分,索引0包含domain
.split("/")[0]
//v4按照:切分去除domain上的端口号后,索引0为最终域名
.split(":")[0];
} else {
//按照:切分后长度>2,说明是v6地址,v6地址不包含端口号(暂定),只需要先切分//再切分/
domain = url.split("/")[0];
}
return domain;
}
}

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

@@ -1,27 +1,34 @@
#arangoDB参数配置 #arangoDB参数配置
arangoDB.host=192.168.40.182 arangoDB.host=192.168.40.182
#arangoDB.host=192.168.40.224
arangoDB.port=8529 arangoDB.port=8529
arangoDB.user=root arangoDB.user=root
arangoDB.password=111111 arangoDB.password=111111
#arangoDB.DB.name=ip-learning-test #arangoDB.DB.name=ip-learning-test
arangoDB.DB.name=tsg_galaxy_v3 arangoDB.DB.name=insert_iplearn_index
arangoDB.batch=100000 arangoDB.batch=100000
arangoDB.ttl=3600 arangoDB.ttl=3600
arangoDB.read.limit=
update.arango.batch=10000 update.arango.batch=10000
thread.pool.number=50 thread.pool.number=10
thread.await.termination.time=10 thread.await.termination.time=10
#读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围 #读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
time.limit.type=1 clickhouse.time.limit.type=1
read.clickhouse.max.time=1596684142 read.clickhouse.max.time=1571245220
read.clickhouse.min.time=1596425769 read.clickhouse.min.time=1571245210
#读取arangoDB时间范围方式，0：正常读，1：指定时间范围
arango.time.limit.type=1
read.arango.max.time=1571245220
read.arango.min.time=1571245210
update.interval=3600 update.interval=3600
distinct.client.ip.num=10000 distinct.client.ip.num=10000
recent.count.hour=24 recent.count.hour=24
top.domain.file.name=topDomain.txt
arangoDB.read.limit=

5
IP-learning-graph/src/main/resources/clickhouse.properties
View File

@@ -1,9 +1,8 @@
drivers=ru.yandex.clickhouse.ClickHouseDriver drivers=ru.yandex.clickhouse.ClickHouseDriver
db.id=192.168.40.193:8123/av_miner?socket_timeout=300000
#db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
mdb.user=default mdb.user=default
db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
mdb.password=111111 mdb.password=111111
#db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=300000
#mdb.password=ceiec2019
initialsize=1 initialsize=1
minidle=1 minidle=1
maxactive=50 maxactive=50

6
IP-learning-graph/src/main/resources/log4j.properties
View File

@@ -4,18 +4,18 @@ log4j.logger.org.apache.http.wire=OFF
#Log4j #Log4j
log4j.rootLogger=info,console,file log4j.rootLogger=info,console,file
# <20><><EFBFBD><EFBFBD>̨<EFBFBD><CCA8>־<EFBFBD><D6BE><EFBFBD><EFBFBD> # <20><><EFBFBD><EFBFBD>̨<EFBFBD><CCA8>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
log4j.appender.console=org.apache.log4j.ConsoleAppender log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Threshold=info log4j.appender.console.Threshold=info
log4j.appender.console.layout=org.apache.log4j.PatternLayout log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] [Thread\:%t] %l %x - <%m>%n log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] [Thread\:%t] %l %x - <%m>%n
# <20>ļ<EFBFBD><C4BC><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD> # <20>ļ<EFBFBD><C4BC><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
log4j.appender.file=org.apache.log4j.DailyRollingFileAppender log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
log4j.appender.file.Threshold=info log4j.appender.file.Threshold=info
log4j.appender.file.encoding=UTF-8 log4j.appender.file.encoding=UTF-8
log4j.appender.file.Append=true log4j.appender.file.Append=true
#·<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><EFBFBD>Ŀ<EFBFBD><EFBFBD> #·<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><EFBFBD>Ŀ<EFBFBD><EFBFBD>
#log4j.appender.file.file=/home/ceiec/iplearning/logs/ip-learning-application.log #log4j.appender.file.file=/home/ceiec/iplearning/logs/ip-learning-application.log
#log4j.appender.file.file=/home/ceiec/iplearning/testLog/ip-learning-application.log #log4j.appender.file.file=/home/ceiec/iplearning/testLog/ip-learning-application.log
log4j.appender.file.file=./logs/ip-learning-application.log log4j.appender.file.file=./logs/ip-learning-application.log

8911
IP-learning-graph/src/main/resources/topDomain.txt Normal file
View File

File diff suppressed because it is too large Load Diff