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

Compare commits

merge into: gfwleak:ip-learning-graph-YSP
Branches Tags
gfwleak:master gfwleak:tsg-v2412 gfwleak:tsg-v2402 gfwleak:tsg-v2311 gfwleak:tsg-v2210 gfwleak:tsg-v2110 gfwleak:tsg-v2 gfwleak:ip-learning-graph-tsg gfwleak:ip-learning-graph-report gfwleak:ip-learning-graph-zx gfwleak:ip-learning-graph-YSP
...
pull from: gfwleak:ip-learning-graph-zx
Branches Tags
gfwleak:tsg-v2412 gfwleak:tsg-v2402 gfwleak:tsg-v2311 gfwleak:tsg-v2210 gfwleak:tsg-v2110 gfwleak:tsg-v2 gfwleak:master gfwleak:ip-learning-graph-tsg gfwleak:ip-learning-graph-report gfwleak:ip-learning-graph-zx gfwleak:ip-learning-graph-YSP

12 Commits
ip-learnin ... ip-learnin

Author SHA1 Message Date
wanglihui
191092f210 中心测试版本，修改为在clickhouse计算后读取结果 2020-08-28 17:42:54 +08:00
wanglihui
a563591051 修改recv_time为end_time 2020-08-27 10:04:09 +08:00
wanglihui
6136635b7b 修改recv_time为end_time 2020-08-27 10:03:49 +08:00
wanglihui
e5f30f5bfd 修改读取client IP空指针异常bug 2020-08-26 10:51:24 +08:00
wanglihui
9c2831013e 修改读取client IP空指针异常bug 2020-08-26 10:51:08 +08:00
wanglihui
77b4d1e758 中心测试版本 2020-08-25 10:31:35 +08:00
wanglihui
ad1bef2466 中心测试版本 2020-08-25 10:31:22 +08:00
wanglihui
abb3b4162b 修改读取arango方式为分页读取 2020-08-24 18:14:08 +08:00
wanglihui
0faaeee7c2 修改读取arangoDb方式为分页读取。 2020-08-24 18:10:18 +08:00
wanglihui
86b484e7b4 IP Learning spark中心测试版本 2020-08-17 10:05:52 +08:00
wanglihui
1750549c7d 修改DIST_CIP未更新bug 2020-08-13 17:37:17 +08:00
wanglihui
5a039bb492 ignore scala.xml 2020-08-12 15:07:04 +08:00
48 changed files with 168 additions and 11251 deletions
Expand all files Collapse all files

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

@@ -41,9 +41,11 @@ public class BaseArangoData {
map.put(i,new ConcurrentHashMap<>()); map.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);
Long total = getCountTotal(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);
String sql = getQuerySql(total, i, table);
ReadHistoryArangoData<T> readHistoryArangoData = ReadHistoryArangoData<T> readHistoryArangoData =
new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch); new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
threadPool.executor(readHistoryArangoData); threadPool.executor(readHistoryArangoData);
@@ -56,6 +58,30 @@ public class BaseArangoData {
} }
} }
private Long getCountTotal(String table){
long start = System.currentTimeMillis();
Long cnt = 0L;
String sql = "RETURN LENGTH("+table+")";
try {
ArangoCursor<Long> longs = arangoDBConnect.executorQuery(sql, Long.class);
while (longs.hasNext()){
cnt = longs.next();
}
}catch (Exception e){
LOG.error(sql +"执行异常");
}
long last = System.currentTimeMillis();
LOG.info(sql+" 结果："+cnt+" 执行时间："+(last-start));
return cnt;
}
private String getQuerySql(Long cnt,int threadNumber, String table){
long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER + 1;
long offsetNum = threadNumber * sepNum;
return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
}
private long[] getTimeRange(String table){ private long[] getTimeRange(String table){
long minTime = 0L; long minTime = 0L;
long maxTime = 0L; long maxTime = 0L;

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

@@ -44,29 +44,25 @@ public class UpdateGraphData {
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", // updateDocument(newVertexSubscriberMap,historyVertexSubscriberMap,"SUBSCRIBER", Subscriber.class,BaseDocument.class,
Subscriber.class,BaseDocument.class, // ReadClickhouseData::getVertexSubscriberSql,ReadClickhouseData::getVertexSubscriberDocument);
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(newRelationSubsciberLocateIpMap,historyRelationSubsciberLocateIpMap,"R_LOCATE_SUBSCRIBER2IP",
LocateSubscriber2Ip.class,BaseEdgeDocument.class, // LocateSubscriber2Ip.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument); // ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument);
long last = System.currentTimeMillis(); long last = System.currentTimeMillis();

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

@@ -264,15 +264,15 @@ public class ReadClickhouseData {
public static String getVertexIpSql() { public static String getVertexIpSql() {
String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime; 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) 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 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_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 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";
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_recv_time >= " + minTime + " AND common_recv_time < " + maxTime; String where = " common_end_time >= " + minTime + " AND common_end_time < " + maxTime;
String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_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"; String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_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";
String httpSql = "SELECT http_host AS FQDN,common_server_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_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"; String httpSql = "SELECT http_host AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_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 != ''"; return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
} }

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

@@ -50,6 +50,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
long s = System.currentTimeMillis(); long s = System.currentTimeMillis();
ArangoCursor<T> docs = arangoConnect.executorQuery(query, type); ArangoCursor<T> docs = arangoConnect.executorQuery(query, type);
if (docs != null) { if (docs != null) {
ArrayList<T> list = new ArrayList<>();
List<T> baseDocuments = docs.asListRemaining(); List<T> baseDocuments = docs.asListRemaining();
int i = 0; int i = 0;
for (T doc : baseDocuments) { for (T doc : baseDocuments) {
@@ -58,9 +59,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
case "R_LOCATE_FQDN2IP": case "R_LOCATE_FQDN2IP":
updateProtocolDocument(doc); updateProtocolDocument(doc);
deleteDistinctClientIpByTime(doc); deleteDistinctClientIpByTime(doc);
break; list.add(doc);
case "R_VISIT_IP2FQDN":
updateProtocolDocument(doc);
break; break;
default: default:
} }
@@ -69,6 +68,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
tmpMap.put(key, doc); tmpMap.put(key, doc);
i++; i++;
} }
arangoConnect.overwrite(list,table);
long l = System.currentTimeMillis(); long l = System.currentTimeMillis();
LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s)); LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
} }
@@ -97,12 +97,18 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
private void deleteDistinctClientIpByTime(T doc) { private void deleteDistinctClientIpByTime(T doc) {
ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP"); ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS"); ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
if (distCip == null || distCip.isEmpty()){
doc.updateAttribute("DIST_CIP", new String[0]);
doc.updateAttribute("DIST_CIP_TS", new long[0]);
return;
}
distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600); distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
Collections.sort(distCipTs); Collections.sort(distCipTs);
Collections.reverse(distCipTs);
int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600); int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);
String[] distCipArr = new String[index]; String[] distCipArr = new String[index];
long[] disCipTsArr = new long[index]; long[] disCipTsArr = new long[index];
if (distCip.size() + 1 == distCipTs.size()){ if (index != 0 && distCip.size() + 1 == distCipTs.size()){
for (int i = 0; i < index; i++) { for (int i = 0; i < index; i++) {
distCipArr[i] = distCip.get(i); distCipArr[i] = distCip.get(i);
disCipTsArr[i] = distCipTs.get(i); disCipTsArr[i] = distCipTs.get(i);

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

@@ -4,7 +4,7 @@ arangoDB.host=192.168.40.182
arangoDB.port=8529 arangoDB.port=8529
arangoDB.user=upsert arangoDB.user=upsert
arangoDB.password=ceiec2018 arangoDB.password=ceiec2018
arangoDB.DB.name=ip-learning-test arangoDB.DB.name=ip-learning-test-0
#arangoDB.DB.name=tsg_galaxy_v3 #arangoDB.DB.name=tsg_galaxy_v3
arangoDB.batch=100000 arangoDB.batch=100000
arangoDB.ttl=3600 arangoDB.ttl=3600
@@ -12,16 +12,16 @@ arangoDB.ttl=3600
arangoDB.read.limit= arangoDB.read.limit=
update.arango.batch=10000 update.arango.batch=10000
thread.pool.number=10 thread.pool.number=80
thread.await.termination.time=10 thread.await.termination.time=10
#读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围 #读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
time.limit.type=1 time.limit.type=1
read.clickhouse.max.time=1596684142 read.clickhouse.max.time=1598246519
read.clickhouse.min.time=1596425769 read.clickhouse.min.time=1597161600
update.interval=3600 update.interval=3600
distinct.client.ip.num=10000 distinct.client.ip.num=1
recent.count.hour=24 recent.count.hour=24

7
IP-learning-graph/src/test/java/cn/ac/iie/TestList.java
View File

@@ -11,6 +11,7 @@ import java.util.List;
public class TestList { public class TestList {
public static void main(String[] args) { public static void main(String[] args) {
/*
ArangoDBConnect arangoConnect = ArangoDBConnect.getInstance(); ArangoDBConnect arangoConnect = ArangoDBConnect.getInstance();
ArangoCursor<BaseEdgeDocument> documents = arangoConnect.executorQuery("FOR doc IN R_LOCATE_FQDN2IP filter doc.FIRST_FOUND_TIME >= 1596080839 and doc.FIRST_FOUND_TIME <= 1596395473 RETURN doc", BaseEdgeDocument.class); ArangoCursor<BaseEdgeDocument> documents = arangoConnect.executorQuery("FOR doc IN R_LOCATE_FQDN2IP filter doc.FIRST_FOUND_TIME >= 1596080839 and doc.FIRST_FOUND_TIME <= 1596395473 RETURN doc", BaseEdgeDocument.class);
List<BaseEdgeDocument> baseEdgeDocuments = documents.asListRemaining(); List<BaseEdgeDocument> baseEdgeDocuments = documents.asListRemaining();
@@ -18,8 +19,8 @@ public class TestList {
doc.updateAttribute("PROTOCOL_TYPE","123"); doc.updateAttribute("PROTOCOL_TYPE","123");
} }
*/
/*
ArrayList<Integer> integers = new ArrayList<>(); ArrayList<Integer> integers = new ArrayList<>();
integers.add(10); integers.add(10);
integers.add(8); integers.add(8);
@@ -39,7 +40,9 @@ public class TestList {
integers.add(5); integers.add(5);
Collections.sort(integers); Collections.sort(integers);
System.out.println(integers); System.out.println(integers);
Collections.reverse(integers);
System.out.println(integers);
System.out.println(integers.indexOf(5)); System.out.println(integers.indexOf(5));
*/
} }
} }

9
ip-learning-java-test/.gitignore vendored
View File

@@ -1,9 +0,0 @@
# Created by .ignore support plugin (hsz.mobi)
### Example user template template
### Example user template
# IntelliJ project files
.idea
*.iml
target
logs/

90
ip-learning-java-test/pom.xml
View File

@@ -1,90 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.ac.iie</groupId>
<artifactId>ip-learning-java-test</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>ru.yandex.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.2.4</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.10</version>
</dependency>
<dependency>
<groupId>com.typesafe</groupId>
<artifactId>config</artifactId>
<version>1.2.1</version>
</dependency>
<dependency>
<groupId>com.arangodb</groupId>
<artifactId>arangodb-java-driver</artifactId>
<version>6.6.3</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.6</version>
<configuration>
<archive>
<manifest>
<mainClass>cn.ac.iie.test.IpLearningApplicationTest</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>

39
ip-learning-java-test/src/main/java/cn/ac/iie/config/ApplicationConfig.java
View File

@@ -1,39 +0,0 @@
package cn.ac.iie.config;
import cn.ac.iie.utils.ConfigUtils;
public class ApplicationConfig {
public static final String ARANGODB_HOST = ConfigUtils.getStringProperty( "arangoDB.host");
public static final Integer ARANGODB_PORT = ConfigUtils.getIntProperty("arangoDB.port");
public static final String ARANGODB_USER = ConfigUtils.getStringProperty( "arangoDB.user");
public static final String ARANGODB_PASSWORD = ConfigUtils.getStringProperty( "arangoDB.password");
public static final String ARANGODB_DB_NAME = ConfigUtils.getStringProperty( "arangoDB.DB.name");
public static final Integer ARANGODB_TTL = ConfigUtils.getIntProperty( "arangoDB.ttl");
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 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 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 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 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 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");
}

103
ip-learning-java-test/src/main/java/cn/ac/iie/dao/BaseArangoData.java
View File

@@ -1,103 +0,0 @@
package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.service.read.ReadHistoryArangoData;
import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.ArangoCursor;
import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
/**
* 获取arangoDB历史数据
*
* @author wlh
*/
public class BaseArangoData {
private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexFqdnMap = 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, BaseEdgeDocument>> historyRelationFqdnAddressIpMap = 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>> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>();
private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
<T extends BaseDocument> void readHistoryData(String table,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
Class<T> type) {
try {
LOG.info("开始更新" + table);
long start = System.currentTimeMillis();
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
historyMap.put(i, new ConcurrentHashMap<>());
}
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
long[] timeRange = getTimeRange(table);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
String sql = getQuerySql(timeRange, i, table);
ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
threadPool.executor(readHistoryArangoData);
}
countDownLatch.await();
long last = System.currentTimeMillis();
LOG.info("读取" + table + " arangoDB 共耗时：" + (last - start));
} catch (Exception e) {
e.printStackTrace();
}
}
private long[] getTimeRange(String table) {
long minTime = 0L;
long maxTime = 0L;
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)}";
switch (ApplicationConfig.ARANGO_TIME_LIMIT_TYPE) {
case 0:
ArangoCursor<BaseDocument> timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class);
try {
if (timeDoc != null) {
while (timeDoc.hasNext()) {
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();
}
break;
case 1:
maxTime = ApplicationConfig.READ_ARANGO_MAX_TIME;
minTime = ApplicationConfig.READ_ARANGO_MIN_TIME;
break;
default:
}
long lastTime = System.currentTimeMillis();
LOG.info(sql + "\n查询最大最小时间用时：" + (lastTime - startTime));
return new long[]{minTime, maxTime};
}
private String getQuerySql(long[] timeRange, int threadNumber, String table) {
long minTime = timeRange[0];
long maxTime = timeRange[1];
long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER;
long maxThreadTime = minTime + (threadNumber + 1) * 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";
}
}

79
ip-learning-java-test/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
View File

@@ -1,79 +0,0 @@
package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.ClickhouseConnect;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.function.Function;
import java.util.function.Supplier;
import static cn.ac.iie.service.read.ReadClickhouseData.putMapByHashcode;
/**
* 读取clickhouse数据，封装到map
* @author wlh
*/
public class BaseClickhouseData {
private static final Logger LOG = LoggerFactory.getLogger(BaseClickhouseData.class);
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>>> newVertexSubscriberMap = 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>>> newRelationSubsciberLocateIpMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> newRelationFqdnSameFqdnMap = new HashMap<>();
private static ClickhouseConnect manger = ClickhouseConnect.getInstance();
private DruidPooledConnection connection;
private Statement statement;
<T extends BaseDocument> void baseDocumentFromClickhouse(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap,
Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc){
long start = System.currentTimeMillis();
initializeMap(newMap);
String sql = getSqlSupplier.get();
LOG.info(sql);
try {
connection = manger.getConnection();
statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
int i = 0;
while (resultSet.next()) {
T newDoc = formatResultFunc.apply(resultSet);
if (newDoc != null) {
i+=1;
putMapByHashcode(newDoc, newMap);
}
}
long last = System.currentTimeMillis();
LOG.info("读取"+i+"条数据，运行时间：" + (last - start));
}catch (Exception e){
e.printStackTrace();
}finally {
manger.clear(statement,connection);
}
}
private <T extends BaseDocument> void initializeMap(HashMap<Integer, HashMap<String,ArrayList<T>>> map){
try {
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
map.put(i, new HashMap<>(16));
}
}catch (Exception e){
e.printStackTrace();
LOG.error("初始化数据失败");
}
}
}

116
ip-learning-java-test/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
View File

@@ -1,116 +0,0 @@
package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.service.read.ReadClickhouseData;
import cn.ac.iie.service.update.Document;
import cn.ac.iie.service.update.relationship.LocateFqdn2Ip;
import cn.ac.iie.service.update.relationship.SameFqdn2Fqdn;
import cn.ac.iie.service.update.relationship.VisitIp2Fqdn;
import cn.ac.iie.service.update.vertex.Fqdn;
import cn.ac.iie.service.update.vertex.Ip;
import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Constructor;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.function.Function;
import java.util.function.Supplier;
import static cn.ac.iie.dao.BaseArangoData.*;
import static cn.ac.iie.dao.BaseClickhouseData.*;
/**
* 更新图数据库业务类
* @author wlh
*/
public class UpdateGraphData {
private static final Logger LOG = LoggerFactory.getLogger(UpdateGraphData.class);
private static ExecutorThreadPool pool = ExecutorThreadPool.getInstance();
private static ArangoDBConnect arangoManger = ArangoDBConnect.getInstance();
private static BaseArangoData baseArangoData = new BaseArangoData();
private static BaseClickhouseData baseClickhouseData = new BaseClickhouseData();
public void updateArango(){
long start = System.currentTimeMillis();
try {
updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN",
Fqdn.class,BaseDocument.class,
ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
updateDocument(newVertexIpMap,historyVertexIpMap,"IP",
Ip.class,BaseDocument.class,
ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);
updateDocument(newRelationFqdnAddressIpMap,historyRelationFqdnAddressIpMap,"R_LOCATE_FQDN2IP",
LocateFqdn2Ip.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipFqdnAddressIpSql,ReadClickhouseData::getRelationFqdnAddressIpDocument);
updateDocument(newRelationIpVisitFqdnMap,historyRelationIpVisitFqdnMap,"R_VISIT_IP2FQDN",
VisitIp2Fqdn.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument);
updateDocument(newRelationFqdnSameFqdnMap,historyRelationFqdnSameFqdnMap,"R_SAME_ORIGIN_FQDN2FQDN",
SameFqdn2Fqdn.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipFqdnSameFqdnSql,ReadClickhouseData::getRelationshipFqdnSameFqdnDocument);
long last = System.currentTimeMillis();
LOG.info("更新图数据库时间共计："+(last - start));
}catch (Exception e){
e.printStackTrace();
}finally {
arangoManger.clean();
pool.shutdown();
}
}
private <T extends BaseDocument> void updateDocument(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
String collection,
Class<? extends Document<T>> taskType,
Class<T> docmentType,
Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc) {
try {
baseArangoData.readHistoryData(collection,historyMap,docmentType);
LOG.info(collection+" 读取clickhouse,封装结果集");
baseClickhouseData.baseDocumentFromClickhouse(newMap, getSqlSupplier,formatResultFunc);
LOG.info(collection+" 开始更新");
long start = System.currentTimeMillis();
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++){
HashMap<String, ArrayList<T>> tmpNewMap = newMap.get(i);
ConcurrentHashMap<String, T> tmpHisMap = historyMap.get(i);
Constructor constructor = taskType.getConstructor(
HashMap.class,
ArangoDBConnect.class,
String.class,
ConcurrentHashMap.class,
CountDownLatch.class);
Document docTask = (Document)constructor.newInstance(tmpNewMap, arangoManger, collection, tmpHisMap, countDownLatch);
pool.executor(docTask);
}
countDownLatch.await();
long last = System.currentTimeMillis();
LOG.info(collection+" 更新完毕,共耗时："+(last-start));
}catch (Exception e){
e.printStackTrace();
}finally {
newMap.clear();
historyMap.clear();
}
}
}

338
ip-learning-java-test/src/main/java/cn/ac/iie/service/read/ReadClickhouseData.java
View File

@@ -1,338 +0,0 @@
package cn.ac.iie.service.read;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.TopDomainUtils;
import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.regex.Pattern;
/**
* @author wlh
*/
public class ReadClickhouseData {
public static long currentHour = System.currentTimeMillis() / (60 * 60 * 1000) * 60 * 60;
private static Pattern pattern = Pattern.compile("^[\\d]*$");
private static final Logger LOG = LoggerFactory.getLogger(ReadClickhouseData.class);
private static long[] timeLimit = getTimeLimit();
private static long maxTime = timeLimit[0];
private static long minTime = timeLimit[1];
public static final Integer DISTINCT_CLIENT_IP_NUM = ApplicationConfig.DISTINCT_CLIENT_IP_NUM;
static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR;
public static final HashSet<String> PROTOCOL_SET;
static {
PROTOCOL_SET = new HashSet<>();
PROTOCOL_SET.add("HTTP");
PROTOCOL_SET.add("TLS");
PROTOCOL_SET.add("DNS");
}
public static BaseDocument getVertexFqdnDocument(ResultSet resultSet){
BaseDocument newDoc = null;
try {
String fqdnOrReferer = resultSet.getString("FQDN");
String fqdnName = TopDomainUtils.getDomainFromUrl(fqdnOrReferer);
if (isDomain(fqdnName)) {
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
newDoc = new BaseDocument();
newDoc.setKey(fqdnName);
newDoc.addAttribute("FQDN_NAME", fqdnName);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
}
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static BaseDocument getVertexIpDocument(ResultSet resultSet){
BaseDocument newDoc = new BaseDocument();
try {
String ip = resultSet.getString("IP");
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long sessionCount = resultSet.getLong("SESSION_COUNT");
long bytesSum = resultSet.getLong("BYTES_SUM");
String ipType = resultSet.getString("ip_type");
newDoc.setKey(ip);
newDoc.addAttribute("IP", ip);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
switch (ipType) {
case "client":
newDoc.addAttribute("CLIENT_SESSION_COUNT", sessionCount);
newDoc.addAttribute("CLIENT_BYTES_SUM", bytesSum);
newDoc.addAttribute("SERVER_SESSION_COUNT", 0L);
newDoc.addAttribute("SERVER_BYTES_SUM", 0L);
break;
case "server":
newDoc.addAttribute("SERVER_SESSION_COUNT", sessionCount);
newDoc.addAttribute("SERVER_BYTES_SUM", bytesSum);
newDoc.addAttribute("CLIENT_SESSION_COUNT", 0L);
newDoc.addAttribute("CLIENT_BYTES_SUM", 0L);
break;
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", "");
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static BaseDocument getVertexSubscriberDocument(ResultSet resultSet){
BaseDocument newDoc = new BaseDocument();
try {
String subscriberId = resultSet.getString("common_subscriber_id");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
newDoc.setKey(subscriberId);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static BaseEdgeDocument getRelationshipSubsciberLocateIpDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = new BaseEdgeDocument();
try {
String subscriberId = resultSet.getString("common_subscriber_id");
String framedIp = resultSet.getString("radius_framed_ip");
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long countTotal = resultSet.getLong("COUNT_TOTAL");
String key = subscriberId + "-" + framedIp;
newDoc.setKey(key);
newDoc.setFrom("SUBSCRIBER/" + subscriberId);
newDoc.setTo("IP/" + framedIp);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("COUNT_TOTAL", countTotal);
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static BaseEdgeDocument getRelationFqdnAddressIpDocument(ResultSet resultSet){
BaseEdgeDocument newDoc = null;
try {
String vFqdn = resultSet.getString("FQDN");
if (isDomain(vFqdn)) {
String vIp = resultSet.getString("common_server_ip");
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long countTotal = resultSet.getLong("COUNT_TOTAL");
String[] distCipRecents = (String[]) resultSet.getArray("DIST_CIP_RECENT").getArray();
long[] clientIpTs = new long[distCipRecents.length];
for (int i = 0; i < clientIpTs.length; i++) {
clientIpTs[i] = currentHour;
}
String key = vFqdn + "-" + vIp;
newDoc = new BaseEdgeDocument();
newDoc.setKey(key);
newDoc.setFrom("FQDN/" + vFqdn);
newDoc.setTo("IP/" + vIp);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
newDoc.addAttribute("CNT_TOTAL",countTotal);
newDoc.addAttribute("DIST_CIP", distCipRecents);
newDoc.addAttribute("DIST_CIP_TS", clientIpTs);
}
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
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;
try {
String vFqdn = resultSet.getString("FQDN");
if (isDomain(vFqdn)) {
String vIp = resultSet.getString("common_client_ip");
String key = vIp + "-" + vFqdn;
long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
long countTotal = resultSet.getLong("COUNT_TOTAL");
newDoc = new BaseEdgeDocument();
newDoc.setKey(key);
newDoc.setFrom("IP/" + vIp);
newDoc.setTo("FQDN/" + vFqdn);
newDoc.addAttribute("CNT_TOTAL",countTotal);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
}
}catch (Exception e){
e.printStackTrace();
}
return newDoc;
}
public static <T extends BaseDocument> void putMapByHashcode(T newDoc, HashMap<Integer, HashMap<String, ArrayList<T>>> map) {
if (newDoc != null) {
String key = newDoc.getKey();
int i = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER;
HashMap<String, ArrayList<T>> documentHashMap = map.getOrDefault(i, new HashMap<>());
ArrayList<T> documentArrayList = documentHashMap.getOrDefault(key, new ArrayList<>());
documentArrayList.add(newDoc);
documentHashMap.put(key, documentArrayList);
}
}
private static boolean isDomain(String fqdn) {
try {
if (fqdn == null || fqdn.length() == 0){
return false;
}
if (fqdn.contains(":")){
String s = fqdn.split(":")[0];
if (s.contains(":")){
return false;
}
}
String[] fqdnArr = fqdn.split("\\.");
if (fqdnArr.length < 4 || fqdnArr.length > 4) {
return true;
}
for (String f : fqdnArr) {
if (pattern.matcher(f).matches()) {
long i = Long.parseLong(f);
if (i < 0 || i > 255) {
return true;
}
} else {
return true;
}
}
} catch (Exception e) {
LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString());
}
return false;
}
private static void checkSchemaProperty(BaseEdgeDocument newDoc, String schema, long countTotal) {
long[] recentCnt = new long[RECENT_COUNT_HOUR];
recentCnt[0] = countTotal;
for (String protocol: PROTOCOL_SET){
String protocolRecent = protocol +"_CNT_RECENT";
String protocolTotal = protocol + "_CNT_TOTAL";
if (protocol.equals(schema)){
newDoc.addAttribute(protocolTotal, countTotal);
newDoc.addAttribute(protocolRecent, recentCnt);
}else {
newDoc.addAttribute(protocolTotal, 0L);
newDoc.addAttribute(protocolRecent, new long[RECENT_COUNT_HOUR]);
}
}
}
public static String getVertexFqdnSql() {
String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime;
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 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 * FROM((" + mediaDomainSql + ") UNION ALL (" + refererSql + "))";
}
public static String getVertexIpSql() {
String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
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 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 + "))";
}
public static String getRelationshipFqdnAddressIpSql() {
String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime +" AND s1_domain != '' AND s1_d_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";
}
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() {
String where = "recv_time >= "+minTime+" and recv_time <= "+maxTime+" AND s1_s_ip != '' AND s1_domain != '' ";
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";
}
public static String getVertexSubscriberSql() {
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_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() {
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_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() {
long maxTime = 0L;
long minTime = 0L;
switch (ApplicationConfig.CLICKHOUSE_TIME_LIMIT_TYPE) {
case 0:
maxTime = currentHour;
minTime = maxTime - ApplicationConfig.UPDATE_INTERVAL;
break;
case 1:
maxTime = ApplicationConfig.READ_CLICKHOUSE_MAX_TIME;
minTime = ApplicationConfig.READ_CLICKHOUSE_MIN_TIME;
break;
default:
}
return new long[]{maxTime, minTime};
}
}

85
ip-learning-java-test/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java
View File

@@ -1,85 +0,0 @@
package cn.ac.iie.service.read;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.ArangoCursor;
import com.arangodb.entity.BaseDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import static cn.ac.iie.service.read.ReadClickhouseData.RECENT_COUNT_HOUR;
/**
* @author wlh
* 多线程全量读取arangoDb历史数据，封装到map
*/
public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
private static final Logger LOG = LoggerFactory.getLogger(ReadHistoryArangoData.class);
private ArangoDBConnect arangoConnect;
private String query;
private ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map;
private Class<T> type;
private String table;
private CountDownLatch countDownLatch;
public ReadHistoryArangoData(ArangoDBConnect arangoConnect,
String query,
ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,
Class<T> type,
String table,
CountDownLatch countDownLatch) {
this.arangoConnect = arangoConnect;
this.query = query;
this.map = map;
this.type = type;
this.table = table;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
try {
long s = System.currentTimeMillis();
ArangoCursor<T> docs = arangoConnect.executorQuery(query, type);
if (docs != null) {
List<T> baseDocuments = docs.asListRemaining();
int i = 0;
for (T doc : baseDocuments) {
String key = doc.getKey();
int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER;
ConcurrentHashMap<String, T> tmpMap = map.get(hashCode);
tmpMap.put(key, doc);
i++;
}
long l = System.currentTimeMillis();
LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
}
}catch (Exception e){
e.printStackTrace();
}finally {
countDownLatch.countDown();
LOG.info("本线程读取完毕，剩余线程数量："+countDownLatch.getCount());
}
}
private void updateProtocolDocument(T doc) {
if (doc.getProperties().containsKey("PROTOCOL_TYPE")) {
for (String protocol : ReadClickhouseData.PROTOCOL_SET) {
String protocolRecent = protocol + "_CNT_RECENT";
ArrayList<Long> cntRecent = (ArrayList<Long>) doc.getAttribute(protocolRecent);
Long[] cntRecentsSrc = cntRecent.toArray(new Long[cntRecent.size()]);
Long[] cntRecentsDst = new Long[RECENT_COUNT_HOUR];
System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1);
cntRecentsDst[0] = 0L;
doc.addAttribute(protocolRecent, cntRecentsDst);
}
}
}
}

128
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/Document.java
View File

@@ -1,128 +0,0 @@
package cn.ac.iie.service.update;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class Document<T extends BaseDocument> extends Thread{
private static final Logger LOG = LoggerFactory.getLogger(Document.class);
private HashMap<String, ArrayList<T>> newDocumentMap;
private ArangoDBConnect arangoManger;
private String collectionName;
private ConcurrentHashMap<String, T> historyDocumentMap;
private CountDownLatch countDownLatch;
Document(HashMap<String, ArrayList<T>> newDocumentMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, T> historyDocumentMap,
CountDownLatch countDownLatch) {
this.newDocumentMap = newDocumentMap;
this.arangoManger = arangoManger;
this.collectionName = collectionName;
this.historyDocumentMap = historyDocumentMap;
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
long start = System.currentTimeMillis();
LOG.info("新读取数据"+newDocumentMap.size()+"条，历史数据"+historyDocumentMap.size()+"条");
try {
Set<String> keySet = newDocumentMap.keySet();
ArrayList<T> resultDocumentList = new ArrayList<>();
int i = 0;
for (String key : keySet) {
ArrayList<T> newDocumentSchemaList = newDocumentMap.getOrDefault(key, null);
if (newDocumentSchemaList != null) {
T newDocument = mergeDocument(newDocumentSchemaList);
i += 1;
T historyDocument = historyDocumentMap.getOrDefault(key, null);
updateDocument(newDocument,historyDocument,resultDocumentList);
if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
arangoManger.overwrite(resultDocumentList, collectionName);
LOG.info("更新"+collectionName+":" + i);
i = 0;
}
}
}
if (i != 0) {
arangoManger.overwrite(resultDocumentList, collectionName);
LOG.info("更新"+collectionName+":" + i);
}
} catch (Exception e) {
e.printStackTrace();
LOG.error(e.toString());
}finally {
countDownLatch.countDown();
long last = System.currentTimeMillis();
LOG.info("本线程更新完毕，用时："+(last-start)+"，剩余线程数量："+countDownLatch.getCount());
}
}
private void updateDocument(T newDocument, T historyDocument, ArrayList<T> resultDocumentList) {
if (historyDocument != null){
updateFunction(newDocument,historyDocument);
resultDocumentList.add(historyDocument);
}else {
resultDocumentList.add(newDocument);
}
}
protected void updateFunction(T newDocument, T historyDocument) {
Object lastFoundTime = newDocument.getAttribute("LAST_FOUND_TIME");
historyDocument.addAttribute("LAST_FOUND_TIME",lastFoundTime);
}
private T mergeDocument(ArrayList<T> newDocumentSchemaList){
if (newDocumentSchemaList == null || newDocumentSchemaList.isEmpty()){
return null;
}else if (newDocumentSchemaList.size() == 1){
return newDocumentSchemaList.get(0);
}else {
T newDocument = null;
for (T lastDoc:newDocumentSchemaList){
if (newDocument == null){
newDocument = lastDoc;
}else {
mergeFunction(lastDoc,newDocument);
}
}
return newDocument;
}
}
protected void mergeFunction(T lastDoc,T newDocument) {
putMinAttribute(lastDoc,newDocument,"FIRST_FOUND_TIME");
putMaxAttribute(lastDoc,newDocument,"LAST_FOUND_TIME");
}
protected void putMinAttribute(T firstDoc,T lastDoc,String attribute){
long firstMinAttribute = Long.parseLong(firstDoc.getAttribute(attribute).toString());
long lastMinAttribute = Long.parseLong(lastDoc.getAttribute(attribute).toString());
lastDoc.addAttribute(attribute,firstMinAttribute<lastMinAttribute? firstMinAttribute:lastMinAttribute);
}
protected void putMaxAttribute(T firstDoc,T lastDoc,String attribute){
long firstMaxAttribute = Long.parseLong(firstDoc.getAttribute(attribute).toString());
long lastMaxAttribute = Long.parseLong(lastDoc.getAttribute(attribute).toString());
lastDoc.addAttribute(attribute,firstMaxAttribute>lastMaxAttribute? firstMaxAttribute:lastMaxAttribute);
}
protected void putSumAttribute(T firstDoc,T lastDoc,String attribute){
long firstSumAttribute = Long.parseLong(firstDoc.getAttribute(attribute).toString());
long lastSumAttribute = Long.parseLong(lastDoc.getAttribute(attribute).toString());
lastDoc.addAttribute(attribute,firstSumAttribute+lastSumAttribute);
}
}

76
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/Relationship.java
View File

@@ -1,76 +0,0 @@
package cn.ac.iie.service.update;
import cn.ac.iie.service.read.ReadClickhouseData;
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 Relationship extends Document<BaseEdgeDocument> {
public Relationship(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);
}
protected void updateProcotol(BaseEdgeDocument historyEdgeDocument, String schema, BaseEdgeDocument newEdgeDocument){
String recentSchema = schema +"_CNT_RECENT";
String totalSchema = schema + "_CNT_TOTAL";
long countTotal = Long.parseLong(newEdgeDocument.getAttribute(totalSchema).toString());
if (countTotal > 0L){
long updateCountTotal = Long.parseLong(historyEdgeDocument.getAttribute(totalSchema).toString());
Long[] cntRecent = (Long[]) historyEdgeDocument.getAttribute(recentSchema);
cntRecent[0] = countTotal;
historyEdgeDocument.addAttribute(recentSchema, cntRecent);
historyEdgeDocument.addAttribute(totalSchema, countTotal + updateCountTotal);
String hisProtocolType = historyEdgeDocument.getAttribute("PROTOCOL_TYPE").toString();
if (!hisProtocolType.contains(schema)){
hisProtocolType = hisProtocolType + "," + schema;
historyEdgeDocument.addAttribute("PROTOCOL_TYPE",hisProtocolType);
}
}
}
@Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
super.mergeFunction(lastDoc, newDocument);
}
protected void mergeProtocol(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument) {
String schema = lastDoc.getAttribute("PROTOCOL_TYPE").toString();
if (ReadClickhouseData.PROTOCOL_SET.contains(schema)){
setProtocolProperties(schema,newDocument,lastDoc);
}
}
private void setProtocolProperties(String protocol,BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){
String protocolRecent = protocol +"_CNT_RECENT";
String protocolTotal = protocol + "_CNT_TOTAL";
putSumAttribute(lastDoc,newDocument,protocolTotal);
long[] cntRecents = (long[]) lastDoc.getAttribute(protocolRecent);
newDocument.addAttribute(protocolRecent, cntRecents);
String protocolType = newDocument.getAttribute("PROTOCOL_TYPE").toString();
newDocument.addAttribute("PROTOCOL_TYPE",addProcotolType(protocolType,protocol));
}
private String addProcotolType(String protocolType,String schema){
if (!protocolType.contains(schema)){
protocolType = protocolType + "," + schema;
}
return protocolType;
}
}

39
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/Vertex.java
View File

@@ -1,39 +0,0 @@
package cn.ac.iie.service.update;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseDocument;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
/**
* @author wlh
* 多线程更新vertex数据
*/
public class Vertex extends Document<BaseDocument> {
public Vertex(HashMap<String, ArrayList<BaseDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseDocument> historyDocumentMap,
CountDownLatch 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();
}
}

94
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java
View File

@@ -1,94 +0,0 @@
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.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import static cn.ac.iie.service.read.ReadClickhouseData.*;
public class LocateFqdn2Ip extends Relationship {
public LocateFqdn2Ip(HashMap<String, ArrayList<BaseEdgeDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseEdgeDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap,countDownLatch);
}
@Override
protected void mergeFunction(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){
super.mergeFunction(lastDoc, newDocument);
mergeDistinctClientIp(lastDoc, newDocument);
putSumAttribute(lastDoc, newDocument,"CNT_TOTAL");
}
private void mergeDistinctClientIp(BaseEdgeDocument lastDoc,BaseEdgeDocument newDocument){
HashSet<String> clientIpSet = new HashSet<>();
String[] distCips = (String[]) newDocument.getAttribute("DIST_CIP");
String[] lastDistCips = (String[]) lastDoc.getAttribute("DIST_CIP");
clientIpSet.addAll(Arrays.asList(distCips));
clientIpSet.addAll(Arrays.asList(lastDistCips));
long[] clientIpTs = new long[clientIpSet.size()];
for (int i = 0; i < clientIpTs.length; i++) {
clientIpTs[i] = currentHour;
}
newDocument.addAttribute("DIST_CIP", clientIpSet.toArray());
newDocument.addAttribute("DIST_CIP_TS", clientIpTs);
}
@Override
protected void updateFunction(BaseEdgeDocument newEdgeDocument, BaseEdgeDocument historyEdgeDocument) {
super.updateFunction(newEdgeDocument, historyEdgeDocument);
updateDistinctClientIp(newEdgeDocument, historyEdgeDocument);
putSumAttribute(newEdgeDocument, historyEdgeDocument,"CNT_TOTAL");
}
private void updateDistinctClientIp(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){
ArrayList<String> distCip = (ArrayList<String>) edgeDocument.getAttribute("DIST_CIP");
ArrayList<Long> distCipTs = (ArrayList<Long>) edgeDocument.getAttribute("DIST_CIP_TS");
HashMap<String, Long> distCipToTs = new HashMap<>();
if (distCip.size() == distCipTs.size()){
for (int i = 0;i < distCip.size();i++){
distCipToTs.put(distCip.get(i),distCipTs.get(i));
}
}
Object[] distCipRecent = (Object[])newEdgeDocument.getAttribute("DIST_CIP");
for (Object cip:distCipRecent){
distCipToTs.put(cip.toString(), currentHour);
}
Map<String, Long> sortDistCip = sortMapByValue(distCipToTs);
edgeDocument.addAttribute("DIST_CIP",sortDistCip.keySet().toArray());
edgeDocument.addAttribute("DIST_CIP_TS",sortDistCip.values().toArray());
}
/**
* 使用 Map按value进行排序
*/
private Map<String, Long> sortMapByValue(Map<String, Long> oriMap) {
if (oriMap == null || oriMap.isEmpty()) {
return null;
}
Map<String, Long> sortedMap = new LinkedHashMap<>();
List<Map.Entry<String, Long>> entryList = new ArrayList<>(oriMap.entrySet());
entryList.sort((o1, o2) -> o2.getValue().compareTo(o1.getValue()));
if(entryList.size() > DISTINCT_CLIENT_IP_NUM){
for(Map.Entry<String, Long> set:entryList.subList(0, DISTINCT_CLIENT_IP_NUM)){
sortedMap.put(set.getKey(), set.getValue());
}
}else {
for(Map.Entry<String, Long> set:entryList){
sortedMap.put(set.getKey(), set.getValue());
}
}
return sortedMap;
}
}

21
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/relationship/LocateSubscriber2Ip.java
View File

@@ -1,21 +0,0 @@
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 LocateSubscriber2Ip extends Relationship {
public LocateSubscriber2Ip(HashMap<String, ArrayList<BaseEdgeDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseEdgeDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
}
}

34
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/relationship/SameFqdn2Fqdn.java
View File

@@ -1,34 +0,0 @@
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");
}
}

32
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/relationship/VisitIp2Fqdn.java
View File

@@ -1,32 +0,0 @@
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 VisitIp2Fqdn extends Relationship {
public VisitIp2Fqdn(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");
}
}

21
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/vertex/Fqdn.java
View File

@@ -1,21 +0,0 @@
package cn.ac.iie.service.update.vertex;
import cn.ac.iie.service.update.Vertex;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseDocument;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class Fqdn extends Vertex {
public Fqdn(HashMap<String, ArrayList<BaseDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap,countDownLatch);
}
}

48
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/vertex/Ip.java
View File

@@ -1,48 +0,0 @@
package cn.ac.iie.service.update.vertex;
import cn.ac.iie.service.update.Vertex;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseDocument;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class Ip extends Vertex {
public Ip(HashMap<String, ArrayList<BaseDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
}
@Override
protected void updateFunction(BaseDocument newDocument, BaseDocument historyDocument) {
super.updateFunction(newDocument, historyDocument);
updateIpByType(newDocument, historyDocument);
}
@Override
protected void mergeFunction(BaseDocument lastDoc, BaseDocument newDocument) {
super.mergeFunction(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) {
putSumAttribute(newDocument, historyDocument, "CLIENT_SESSION_COUNT");
putSumAttribute(newDocument, historyDocument, "CLIENT_BYTES_SUM");
putSumAttribute(newDocument, historyDocument, "SERVER_SESSION_COUNT");
putSumAttribute(newDocument, historyDocument, "SERVER_BYTES_SUM");
}
}

21
ip-learning-java-test/src/main/java/cn/ac/iie/service/update/vertex/Subscriber.java
View File

@@ -1,21 +0,0 @@
package cn.ac.iie.service.update.vertex;
import cn.ac.iie.service.update.Vertex;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.entity.BaseDocument;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class Subscriber extends Vertex {
public Subscriber(HashMap<String, ArrayList<BaseDocument>> newDocumentHashMap,
ArangoDBConnect arangoManger,
String collectionName,
ConcurrentHashMap<String, BaseDocument> historyDocumentMap,
CountDownLatch countDownLatch) {
super(newDocumentHashMap, arangoManger, collectionName, historyDocumentMap, countDownLatch);
}
}

18
ip-learning-java-test/src/main/java/cn/ac/iie/test/IpLearningApplicationTest.java
View File

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

116
ip-learning-java-test/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java
View File

@@ -1,116 +0,0 @@
package cn.ac.iie.utils;
import cn.ac.iie.config.ApplicationConfig;
import com.arangodb.ArangoCollection;
import com.arangodb.ArangoCursor;
import com.arangodb.ArangoDB;
import com.arangodb.ArangoDatabase;
import com.arangodb.entity.DocumentCreateEntity;
import com.arangodb.entity.ErrorEntity;
import com.arangodb.entity.MultiDocumentEntity;
import com.arangodb.model.AqlQueryOptions;
import com.arangodb.model.DocumentCreateOptions;
import com.arangodb.util.MapBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
public class ArangoDBConnect {
private static final Logger LOG = LoggerFactory.getLogger(ArangoDBConnect.class);
private static ArangoDB arangoDB = null;
private static ArangoDBConnect conn = null;
static {
getArangoDatabase();
}
private static void getArangoDatabase(){
arangoDB = new ArangoDB.Builder()
.maxConnections(ApplicationConfig.THREAD_POOL_NUMBER)
.host(ApplicationConfig.ARANGODB_HOST, ApplicationConfig.ARANGODB_PORT)
.user(ApplicationConfig.ARANGODB_USER)
.password(ApplicationConfig.ARANGODB_PASSWORD)
.build();
}
public static synchronized ArangoDBConnect getInstance(){
if (null == conn){
conn = new ArangoDBConnect();
}
return conn;
}
private ArangoDatabase getDatabase(){
return arangoDB.db(ApplicationConfig.ARANGODB_DB_NAME);
}
public void clean(){
try {
if (arangoDB != null){
arangoDB.shutdown();
}
}catch (Exception e){
e.printStackTrace();
}
}
public <T> ArangoCursor<T> executorQuery(String query,Class<T> type){
ArangoDatabase database = getDatabase();
Map<String, Object> bindVars = new MapBuilder().get();
AqlQueryOptions options = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL);
try {
return database.query(query, bindVars, options, type);
}catch (Exception e){
e.printStackTrace();
return null;
}finally {
bindVars.clear();
}
}
@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();
try {
ArangoCollection collection = database.collection(collectionName);
if (!docOverwrite.isEmpty()){
DocumentCreateOptions documentCreateOptions = new DocumentCreateOptions();
documentCreateOptions.overwrite(true);
documentCreateOptions.silent(true);
MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
for (ErrorEntity errorEntity:errors){
LOG.debug("写入arangoDB异常："+errorEntity.getErrorMessage());
}
}
}catch (Exception e){
System.out.println("更新失败："+e.toString());
}finally {
docOverwrite.clear();
}
}
}

103
ip-learning-java-test/src/main/java/cn/ac/iie/utils/ClickhouseConnect.java
View File

@@ -1,103 +0,0 @@
package cn.ac.iie.utils;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import java.sql.*;
import java.util.Properties;
public class ClickhouseConnect {
private static DruidDataSource dataSource = null;
private static ClickhouseConnect dbConnect = null;
private static Properties props = new Properties();
static {
getDbConnect();
}
private static void getDbConnect() {
try {
if (dataSource == null) {
dataSource = new DruidDataSource();
props.load(ClickhouseConnect.class.getClassLoader().getResourceAsStream("clickhouse.properties"));
//设置连接参数
dataSource.setUrl("jdbc:clickhouse://" + props.getProperty("db.id"));
dataSource.setDriverClassName(props.getProperty("drivers"));
dataSource.setUsername(props.getProperty("mdb.user"));
dataSource.setPassword(props.getProperty("mdb.password"));
//配置初始化大小、最小、最大
dataSource.setInitialSize(Integer.parseInt(props.getProperty("initialsize")));
dataSource.setMinIdle(Integer.parseInt(props.getProperty("minidle")));
dataSource.setMaxActive(Integer.parseInt(props.getProperty("maxactive")));
//配置获取连接等待超时的时间
dataSource.setMaxWait(30000);
//配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
dataSource.setTimeBetweenEvictionRunsMillis(2000);
//防止过期
dataSource.setValidationQuery("SELECT 1");
dataSource.setTestWhileIdle(true);
dataSource.setTestOnBorrow(true);
dataSource.setKeepAlive(true);
}
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 数据库连接池单例
*
* @return dbConnect
*/
public static synchronized ClickhouseConnect getInstance() {
if (null == dbConnect) {
dbConnect = new ClickhouseConnect();
}
return dbConnect;
}
/**
* 返回druid数据库连接
*
* @return 连接
* @throws SQLException sql异常
*/
public DruidPooledConnection getConnection() throws SQLException {
return dataSource.getConnection();
}
/**
* 清空PreparedStatement、Connection对象，未定义的置空。
*
* @param pstmt PreparedStatement对象
* @param connection Connection对象
*/
public void clear(Statement pstmt, Connection connection) {
try {
if (pstmt != null) {
pstmt.close();
}
if (connection != null) {
connection.close();
}
} catch (SQLException e) {
e.printStackTrace();
}
}
public ResultSet executorQuery(String query,Connection connection,Statement pstm){
// Connection connection = null;
// Statement pstm = null;
try {
connection = getConnection();
pstm = connection.createStatement();
return pstm.executeQuery(query);
}catch (Exception e){
e.printStackTrace();
return null;
}
}
}

36
ip-learning-java-test/src/main/java/cn/ac/iie/utils/ConfigUtils.java
View File

@@ -1,36 +0,0 @@
package cn.ac.iie.utils;
import java.util.Properties;
public class ConfigUtils {
private static Properties propCommon = new Properties();
public static String getStringProperty(String key) {
return propCommon.getProperty(key);
}
public static Integer getIntProperty(String key) {
return Integer.parseInt(propCommon.getProperty(key));
}
public static Long getLongProperty(String key) {
return Long.parseLong(propCommon.getProperty(key));
}
public static Boolean getBooleanProperty(String key) {
return "true".equals(propCommon.getProperty(key).toLowerCase().trim());
}
static {
try {
propCommon.load(ConfigUtils.class.getClassLoader().getResourceAsStream("application.properties"));
System.out.println("application.properties加载成功");
} catch (Exception e) {
propCommon = null;
System.err.println("配置加载失败");
}
}
}

67
ip-learning-java-test/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
View File

@@ -1,67 +0,0 @@
package cn.ac.iie.utils;
import cn.ac.iie.config.ApplicationConfig;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
* 线程池管理
* @author wlh
*/
public class ExecutorThreadPool {
private static ExecutorService pool = null ;
private static ExecutorThreadPool poolExecutor = null;
static {
getThreadPool();
}
private static void getThreadPool(){
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("iplearning-application-pool-%d").build();
//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());
// pool = Executors.newFixedThreadPool(ApplicationConfig.THREAD_POOL_NUMBER);
}
public static ExecutorThreadPool getInstance(){
if (null == poolExecutor){
poolExecutor = new ExecutorThreadPool();
}
return poolExecutor;
}
public void executor(Runnable command){
pool.execute(command);
}
@Deprecated
public void awaitThreadTask(){
try {
while (!pool.awaitTermination(ApplicationConfig.THREAD_AWAIT_TERMINATION_TIME, TimeUnit.SECONDS)) {
System.out.println("线程池没有关闭");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void shutdown(){
pool.shutdown();
}
@Deprecated
public static Long getThreadNumber(){
String name = Thread.currentThread().getName();
String[] split = name.split("-");
return Long.parseLong(split[3]);
}
}

158
ip-learning-java-test/src/main/java/cn/ac/iie/utils/TopDomainUtils.java
View File

@@ -1,158 +0,0 @@
package cn.ac.iie.utils;
import cn.ac.iie.config.ApplicationConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
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;
}
}

34
ip-learning-java-test/src/main/resources/application.properties
View File

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

8
ip-learning-java-test/src/main/resources/clickhouse.properties
View File

@@ -1,8 +0,0 @@
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.password=111111
initialsize=1
minidle=1
maxactive=50

25
ip-learning-java-test/src/main/resources/log4j.properties
View File

@@ -1,25 +0,0 @@
######################### logger ##############################
log4j.logger.org.apache.http=OFF
log4j.logger.org.apache.http.wire=OFF
#Log4j
log4j.rootLogger=info,console,file
# <20><><EFBFBD><EFBFBD>̨<EFBFBD><CCA8>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Threshold=info
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
# <20>ļ<EFBFBD><C4BC><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
log4j.appender.file.Threshold=info
log4j.appender.file.encoding=UTF-8
log4j.appender.file.Append=true
#·<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><D8B2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6>Ŀ<EFBFBD><C4BF>
#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=./logs/ip-learning-application.log
log4j.appender.file.DatePattern='.'yyyy-MM-dd
log4j.appender.file.layout=org.apache.log4j.PatternLayout
#log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss} %X{ip} [%t] %5p %c{1} %m%n
log4j.appender.file.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] %X{ip} [Thread\:%t] %l %x - %m%n

8911
ip-learning-java-test/src/main/resources/topDomain.txt
View File

File diff suppressed because it is too large Load Diff

29
ip-learning-java-test/src/test/java/cn/ac/iie/Test.java
View File

@@ -1,29 +0,0 @@
package cn.ac.iie;
import java.lang.reflect.Array;
import java.util.ArrayList;
public class Test {
public static void main(String args[]) throws Exception {
Tester test = new Tester();
Tester[] tests = new Tester[0];
// ArrayList<Tester> testers = new ArrayList<>();
// testers.add(test);
Class<?> c1 = tests.getClass().getComponentType();
Class<?> c2 = Tester.class;
Class<?> c3 = test.getClass();
System.out.println(c1.getName());
System.out.println(c2.getName());
System.out.println(c3.getName());
// Tester[] newTesters = (Tester[]) Array.newInstance(c1, 10);
// Tester newTester = (Tester) c1.newInstance();
// System.out.println(newTesters.length);
}
}
class Tester {
private String name;
private String mem;
}

37
ip-learning-java-test/src/test/java/cn/ac/iie/TestArango.java
View File

@@ -1,37 +0,0 @@
package cn.ac.iie;
import cn.ac.iie.utils.ArangoDBConnect;
import com.arangodb.ArangoCursor;
import com.arangodb.ArangoDatabase;
import com.arangodb.entity.BaseEdgeDocument;
import java.util.ArrayList;
import java.util.List;
public class TestArango {
public static void main(String[] args) {
ArangoDBConnect instance = ArangoDBConnect.getInstance();
/*
String query = "FOR doc IN IP filter doc.FIRST_FOUND_TIME >= 1592996080 and doc.FIRST_FOUND_TIME <= 1593112913 RETURN doc";
ArangoCursor<BaseEdgeDocument> baseEdgeDocuments = instance.executorQuery(query, BaseEdgeDocument.class);
while (baseEdgeDocuments.hasNext()){
BaseEdgeDocument next = baseEdgeDocuments.next();
System.out.println(next.toString());
}
*/
BaseEdgeDocument baseEdgeDocument = new BaseEdgeDocument();
baseEdgeDocument.setKey("192.168.50.6-www.liftopia.com");
baseEdgeDocument.setFrom("IP/192.168.50.6");
baseEdgeDocument.setTo("FQDN/www.liftopia.com");
baseEdgeDocument.addAttribute("HTTP_CNT_TOTAL",3L);
baseEdgeDocument.addAttribute("DNS_CNT_RECENT",new long[24]);
baseEdgeDocument.addAttribute("PROTOCOL_TYPE","HTTP");
ArrayList<BaseEdgeDocument> baseEdgeDocuments = new ArrayList<>();
baseEdgeDocuments.add(baseEdgeDocument);
instance.overwrite(baseEdgeDocuments,"R_LOCATE_FQDN2IP");
instance.clean();
}
}

30
ip-learning-java-test/src/test/java/cn/ac/iie/TestReadLine.java
View File

@@ -1,30 +0,0 @@
package cn.ac.iie;
import java.io.*;
import java.net.URL;
import java.util.TreeMap;
import java.util.regex.Pattern;
public class TestReadLine {
public static void main(String[] args) throws Exception {
Pattern pattern = Pattern.compile("^[^.]*$");
String encoding = "UTF-8";
// File file = new File("C:\\Users\\94976\\Desktop\\test.txt");
URL url = TestReadLine.class.getClassLoader().getResource("topDomain.txt");
assert url != null;
File file = new File(url.getFile());
InputStreamReader read = new InputStreamReader(
new FileInputStream(file), encoding);
BufferedReader bufferedReader = new BufferedReader(read);
String lineTxt;
int cnt = 0;
while ((lineTxt = bufferedReader.readLine()) != null){
if (pattern.matcher(lineTxt).matches()){
cnt += 1;
System.out.println(lineTxt);
}
}
System.out.println(cnt);
System.out.println(url.getFile());
}
}

20
ip-learning-java-test/src/test/java/cn/ac/iie/TestReflect.java
View File

@@ -1,20 +0,0 @@
package cn.ac.iie;
import cn.ac.iie.service.update.vertex.Fqdn;
import cn.ac.iie.utils.ArangoDBConnect;
import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class TestReflect {
public static void main(String[] args) throws Exception {
Class<Fqdn> fqdnClass = Fqdn.class;
// for(Constructor constructor : fqdnClass.getConstructors()){
// System.out.println(constructor);
// }
Constructor<Fqdn> constructor = fqdnClass.getConstructor(HashMap.class, ArangoDBConnect.class, String.class, ConcurrentHashMap.class, CountDownLatch.class);
System.out.println(constructor);
}
}

10
ip-learning-java-test/src/test/java/cn/ac/iie/TestRegex.java
View File

@@ -1,10 +0,0 @@
package cn.ac.iie;
import java.util.regex.Pattern;
public class TestRegex {
public static void main(String[] args) {
Pattern pattern = Pattern.compile("^[^.]*$");
System.out.println(pattern.matcher("com.dz").matches());
}
}

162
ip-learning-spark/.idea/libraries/scala_sdk_2_11_7.xml generated
View File

@@ -1,162 +0,0 @@
<component name="libraryTable">
<library name="scala-sdk-2.11.7" type="Scala">
<properties>
<language-level>Scala_2_11</language-level>
<compiler-classpath>
<root url="file://D:/soft/scala/scala-2.11.7/lib/scala-compiler.jar" />
<root url="file://D:/soft/scala/scala-2.11.7/lib/scala-library.jar" />
<root url="file://D:/soft/scala/scala-2.11.7/lib/scala-reflect.jar" />
</compiler-classpath>
</properties>
<CLASSES>
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-actors-2.11.0.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-actors-migration_2.11-1.1.0.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-library.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-parser-combinators_2.11-1.0.4.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-reflect.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-swing_2.11-1.0.2.jar!/" />
<root url="jar://D:/soft/scala/scala-2.11.7/lib/scala-xml_2.11-1.0.4.jar!/" />
</CLASSES>
<JAVADOC>
<root url="http://www.scala-lang.org/api/2.11.7/" />
</JAVADOC>
<SOURCES>
<root url="file://D:/tar/scala-2.11.7" />
<root url="file://D:/tar/scala-2.11.7/src/actors" />
<root url="file://D:/tar/scala-2.11.7/src/forkjoin" />
<root url="file://D:/tar/scala-2.11.7/src/library" />
<root url="file://D:/tar/scala-2.11.7/src/partest-extras" />
<root url="file://D:/tar/scala-2.11.7/src/partest-javaagent" />
<root url="file://D:/tar/scala-2.11.7/src/repl" />
<root url="file://D:/tar/scala-2.11.7/test/disabled/pos/t1737" />
<root url="file://D:/tar/scala-2.11.7/test/disabled/presentation/akka/src" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/deprecation" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/duration-java" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t1143-2" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t1342" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t1464" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t2163" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t2470" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t2570" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t2585" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t3003" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t3415" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/t7253" />
<root url="file://D:/tar/scala-2.11.7/test/files/jvm/varargs" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/abstract-class-error" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/primitive-sigs-1" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/raw-types-stubs" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t0673" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t2442" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t4851" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t6013" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t6289" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t750" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t750b" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t8244" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t8244e" />
<root url="file://D:/tar/scala-2.11.7/test/files/neg/t8376" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/ilya2" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/super" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t0695" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1101" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1102" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1150" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1152" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1176" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1186" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1196" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1197" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1230" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1231" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1232" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1235" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1254" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1409" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1642" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1711" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1745" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1751" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1782" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t1836" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2377" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2409" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2433" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2464" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t294" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2940" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t2956" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3120" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3249" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3349" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3404" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3429" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3486" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3521" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3567" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3642" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3938" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t3943" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t4603" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t4744" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t5165" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t5703" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t5957" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t6169" />
<root url="file://D:/tar/scala-2.11.7/test/files/pos/t942" />
<root url="file://D:/tar/scala-2.11.7/test/files/presentation/ide-bug-1000469/src" />
<root url="file://D:/tar/scala-2.11.7/test/files/presentation/ide-bug-1000531/src" />
<root url="file://D:/tar/scala-2.11.7/test/files/res/t6613" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/bcodeInlinerMixed" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/reflection-fancy-java-classes" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/reflection-java-annotations" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/reflection-java-crtp" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452a" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452b-bcode" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452d" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452e" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3452g" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t3897" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4238" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4317" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4729" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4788" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4788-separate-compilation" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t4891" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t6168" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t6168b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t6240a" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t6240b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t6548" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7008" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7246" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7246b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7359" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7374" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7439" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7455" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7741a" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t7741b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t8442" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t8601e" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t9268" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t9298" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t9298b" />
<root url="file://D:/tar/scala-2.11.7/test/files/run/t9359" />
<root url="file://D:/tar/scala-2.11.7/test/files/t8449" />
<root url="file://D:/tar/scala-2.11.7/test/flaky/pos/t2868" />
<root url="file://D:/tar/scala-2.11.7/test/instrumented/library" />
<root url="file://D:/tar/scala-2.11.7/test/junit" />
<root url="file://D:/tar/scala-2.11.7/test/pending/jvm/t2705" />
<root url="file://D:/tar/scala-2.11.7/test/pending/pos/misc" />
<root url="file://D:/tar/scala-2.11.7/test/pending/pos/t3943" />
<root url="file://D:/tar/scala-2.11.7/test/pending/pos/t7778" />
<root url="file://D:/tar/scala-2.11.7/test/pending/run/t3899" />
<root url="file://D:/tar/scala-2.11.7/test/pending/run/t4713" />
<root url="file://D:/tar/scala-2.11.7/test/support/annotations" />
</SOURCES>
</library>
</component>

35
ip-learning-spark/src/main/java/cn/ac/iie/dao/BaseArangoData.java
View File

@@ -33,9 +33,8 @@ public class BaseArangoData {
private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance(); private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
public <T extends BaseDocument> void readHistoryData(String table, public <T extends BaseDocument> ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> readHistoryData(String table, Class<T> type) {
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap, ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap = new ConcurrentHashMap<>();
Class<T> type) {
try { try {
LOG.warn("开始更新" + table); LOG.warn("开始更新" + table);
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
@@ -43,9 +42,11 @@ public class BaseArangoData {
historyMap.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);
Long total = getCountTotal(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);
String sql = getQuerySql(total, i, table);
ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch); ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
threadPool.executor(readHistoryArangoData); threadPool.executor(readHistoryArangoData);
} }
@@ -55,6 +56,30 @@ public class BaseArangoData {
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} }
return historyMap;
}
private Long getCountTotal(String table){
long start = System.currentTimeMillis();
Long cnt = 0L;
String sql = "RETURN LENGTH("+table+")";
try {
ArangoCursor<Long> longs = arangoDBConnect.executorQuery(sql, Long.class);
while (longs.hasNext()){
cnt = longs.next();
}
}catch (Exception e){
LOG.error(sql +"执行异常");
}
long last = System.currentTimeMillis();
LOG.warn(sql+" 结果："+cnt+" 执行时间："+(last-start));
return cnt;
}
private String getQuerySql(Long cnt,int threadNumber, String table){
long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER() + 1;
long offsetNum = threadNumber * sepNum;
return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
} }
private long[] getTimeRange(String table) { private long[] getTimeRange(String table) {

13
ip-learning-spark/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java
View File

@@ -58,6 +58,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
public void run() { public void run() {
try { try {
long s = System.currentTimeMillis(); long s = System.currentTimeMillis();
LOG.warn(query+" \n 开始查询");
ArangoCursor<T> docs = arangoConnect.executorQuery(query, type); ArangoCursor<T> docs = arangoConnect.executorQuery(query, type);
if (docs != null) { if (docs != null) {
List<T> baseDocuments = docs.asListRemaining(); List<T> baseDocuments = docs.asListRemaining();
@@ -69,9 +70,9 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
updateProtocolDocument(doc); updateProtocolDocument(doc);
deleteDistinctClientIpByTime(doc); deleteDistinctClientIpByTime(doc);
break; break;
case "R_VISIT_IP2FQDN": // case "R_VISIT_IP2FQDN":
updateProtocolDocument(doc); // updateProtocolDocument(doc);
break; // break;
default: default:
} }
int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER(); int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER();
@@ -107,6 +108,12 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
private void deleteDistinctClientIpByTime(T doc) { private void deleteDistinctClientIpByTime(T doc) {
ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP"); ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS"); ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
if (distCip == null || distCip.isEmpty()){
doc.updateAttribute("DIST_CIP", new String[0]);
doc.updateAttribute("DIST_CIP_TS", new long[0]);
return;
}
distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600); distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
Collections.sort(distCipTs); Collections.sort(distCipTs);
int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600); int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);

23
ip-learning-spark/src/main/resources/application.properties
View File

@@ -7,13 +7,16 @@ repartitionNumber=36
spark.serializer=org.apache.spark.serializer.KryoSerializer spark.serializer=org.apache.spark.serializer.KryoSerializer
master=local[*] master=local[*]
#spark读取clickhouse配置 #spark读取clickhouse配置
spark.read.clickhouse.url=jdbc:clickhouse://192.168.40.186:8123/tsg_galaxy_v3 #spark.read.clickhouse.url=jdbc:clickhouse://192.168.40.186:8123/tsg_galaxy_v3
spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.12:8123/tsg_galaxy_v3
spark.read.clickhouse.driver=ru.yandex.clickhouse.ClickHouseDriver spark.read.clickhouse.driver=ru.yandex.clickhouse.ClickHouseDriver
spark.read.clickhouse.user=default spark.read.clickhouse.user=default
spark.read.clickhouse.password=111111 #spark.read.clickhouse.password=111111
spark.read.clickhouse.password=ceiec2019
spark.read.clickhouse.numPartitions=144 spark.read.clickhouse.numPartitions=144
spark.read.clickhouse.fetchsize=10000 spark.read.clickhouse.fetchsize=10000
spark.read.clickhouse.partitionColumn=common_recv_time #spark.read.clickhouse.partitionColumn=common_end_time
spark.read.clickhouse.partitionColumn=FIRST_FOUND_TIME
clickhouse.socket.timeout=300000 clickhouse.socket.timeout=300000
#arangoDB配置 #arangoDB配置
arangoDB.host=192.168.40.182 arangoDB.host=192.168.40.182
@@ -27,19 +30,19 @@ arangoDB.ttl=3600
thread.pool.number=5 thread.pool.number=5
#读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围 #读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
clickhouse.time.limit.type=0 clickhouse.time.limit.type=1
read.clickhouse.max.time=1571245220 read.clickhouse.max.time=1598246519
read.clickhouse.min.time=1571245210 read.clickhouse.min.time=1597161600
#读取arangoDB时间范围方式，0：正常读；1：指定时间范围 #读取arangoDB时间范围方式，0：正常读；1：指定时间范围
arango.time.limit.type=0 arango.time.limit.type=0
read.arango.max.time=1571245320 read.arango.max.time=1598246519
read.arango.min.time=1571245200 read.arango.min.time=1597161600
arangoDB.read.limit= arangoDB.read.limit=
update.arango.batch=10000 update.arango.batch=10000
distinct.client.ip.num=10000 distinct.client.ip.num=1
recent.count.hour=24 recent.count.hour=24
update.interval=10800 update.interval=3600

41
ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala
View File

@@ -11,7 +11,7 @@ object BaseClickhouseData {
val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60 val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60
private val timeLimit: (Long, Long) = getTimeLimit private val timeLimit: (Long, Long) = getTimeLimit
private def initClickhouseData(sql:String): Unit ={ private def initClickhouseData(sql:String): DataFrame ={
val dataFrame: DataFrame = spark.read.format("jdbc") val dataFrame: DataFrame = spark.read.format("jdbc")
.option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL) .option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL)
@@ -28,14 +28,15 @@ object BaseClickhouseData {
.load() .load()
dataFrame.printSchema() dataFrame.printSchema()
dataFrame.createOrReplaceGlobalTempView("dbtable") dataFrame.createOrReplaceGlobalTempView("dbtable")
dataFrame
} }
def loadConnectionDataFromCk(): Unit ={ def loadConnectionDataFromCk(): Unit ={
val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1 val where = "common_end_time >= " + timeLimit._2 + " AND common_end_time < " + timeLimit._1 + " and common_schema_type != 'BASE'"
val sql = val sql =
s""" s"""
|(SELECT |(SELECT
| ssl_sni,http_host,common_client_ip,common_server_ip,common_recv_time,common_c2s_byte_num,common_s2c_byte_num,common_schema_type | ssl_sni,http_host,common_client_ip,common_server_ip,common_end_time,common_c2s_byte_num,common_s2c_byte_num,common_schema_type
|FROM |FROM
| connection_record_log | connection_record_log
|WHERE $where) as dbtable |WHERE $where) as dbtable
@@ -45,6 +46,28 @@ object BaseClickhouseData {
initClickhouseData(sql) initClickhouseData(sql)
} }
def getRelationFqdnLocateIpDf(): DataFrame ={
val where = "common_end_time >= " + timeLimit._2 + " AND common_end_time < " + timeLimit._1 + " and common_schema_type != 'BASE'"
val sql =
s"""
|(SELECT * FROM
|((SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
|toString(groupUniqArray(${ApplicationConfig.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)
|UNION ALL
|(SELECT http_host AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
|toString(groupUniqArray(${ApplicationConfig.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))
|WHERE FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
private def loadRadiusDataFromCk(): Unit ={ private def loadRadiusDataFromCk(): Unit ={
val where = val where =
s""" s"""
@@ -146,6 +169,7 @@ object BaseClickhouseData {
vertexIpDf vertexIpDf
} }
/*
def getRelationFqdnLocateIpDf: DataFrame ={ def getRelationFqdnLocateIpDf: DataFrame ={
loadConnectionDataFromCk() loadConnectionDataFromCk()
val sslSql = val sslSql =
@@ -153,10 +177,9 @@ object BaseClickhouseData {
|SELECT |SELECT
| ssl_sni AS FQDN, | ssl_sni AS FQDN,
| common_server_ip, | common_server_ip,
| MAX(common_recv_time) AS LAST_FOUND_TIME, | MAX(common_end_time) AS LAST_FOUND_TIME,
| MIN(common_recv_time) AS FIRST_FOUND_TIME, | MIN(common_end_time) AS FIRST_FOUND_TIME,
| COUNT(*) AS COUNT_TOTAL, | COUNT(*) AS COUNT_TOTAL,
| collect_set(common_client_ip) AS DIST_CIP_RECENT,
| 'TLS' AS schema_type | 'TLS' AS schema_type
|FROM |FROM
| global_temp.dbtable | global_temp.dbtable
@@ -171,10 +194,9 @@ object BaseClickhouseData {
|SELECT |SELECT
| http_host AS FQDN, | http_host AS FQDN,
| common_server_ip, | common_server_ip,
| MAX(common_recv_time) AS LAST_FOUND_TIME, | MAX(common_end_time) AS LAST_FOUND_TIME,
| MIN(common_recv_time) AS FIRST_FOUND_TIME, | MIN(common_end_time) AS FIRST_FOUND_TIME,
| COUNT(*) AS COUNT_TOTAL, | COUNT(*) AS COUNT_TOTAL,
| collect_set(common_client_ip) AS DIST_CIP_RECENT,
| 'HTTP' AS schema_type | 'HTTP' AS schema_type
|FROM |FROM
| global_temp.dbtable | global_temp.dbtable
@@ -190,6 +212,7 @@ object BaseClickhouseData {
relationFqdnLocateIpDf.printSchema() relationFqdnLocateIpDf.printSchema()
relationFqdnLocateIpDf relationFqdnLocateIpDf
} }
*/
private def getTimeLimit: (Long,Long) ={ private def getTimeLimit: (Long,Long) ={
var maxTime = 0L var maxTime = 0L

13
ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala
View File

@@ -35,7 +35,7 @@ object MergeDataFrame {
} }
def mergeRelationFqdnLocateIp(): RDD[Row] ={ def mergeRelationFqdnLocateIp(): RDD[Row] ={
val frame = BaseClickhouseData.getRelationFqdnLocateIpDf.filter(row => isDomain(row.getAs[String]("FQDN"))) val frame = BaseClickhouseData.getRelationFqdnLocateIpDf().filter(row => isDomain(row.getAs[String]("FQDN")))
.groupBy("FQDN", "common_server_ip") .groupBy("FQDN", "common_server_ip")
.agg( .agg(
min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"), min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
@@ -58,13 +58,8 @@ object MergeDataFrame {
if (fqdn == null || fqdn.length == 0) { if (fqdn == null || fqdn.length == 0) {
return false return false
} }
if (fqdn.contains(":")) { val domain = fqdn.split(":")(0)
val s = fqdn.split(":")(0) val fqdnArr = domain.split("\\.")
if (s.contains(":")){
return false
}
}
val fqdnArr = fqdn.split("\\.")
if (fqdnArr.length < 4 || fqdnArr.length > 4){ if (fqdnArr.length < 4 || fqdnArr.length > 4){
return true return true
} }
@@ -80,7 +75,7 @@ object MergeDataFrame {
} }
} catch { } catch {
case e: Exception => case e: Exception =>
LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString) LOG.warn("解析域名 " + fqdn + " 失败：\n" + e.toString)
} }
false false
} }

10
ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocHandler.scala
View File

@@ -93,8 +93,14 @@ object UpdateDocHandler {
doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",","")) doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",",""))
} }
def mergeDistinctIp(distCipRecent:ofRef[ofRef[String]]): Array[String] ={ def mergeDistinctIp(distCipRecent:ofRef[String]): Array[String] ={
distCipRecent.flatten.distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray distCipRecent.flatMap(str => {
str.replaceAll("\\[", "")
.replaceAll("\\]", "")
.replaceAll("'", "")
.split(",")
}).distinct.toArray
// distCipRecent.flatten.distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
} }
def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={ def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={

17
ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala
View File

@@ -26,9 +26,7 @@ object UpdateDocument {
def update(): Unit = { def update(): Unit = {
try { try {
updateDocument("FQDN", historyVertexFqdnMap, getVertexFqdnRow, classOf[BaseDocument], mergeVertexFqdn) updateDocument("R_LOCATE_FQDN2IP", getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
updateDocument("IP", historyVertexIpMap, getVertexIpRow, classOf[BaseDocument], mergeVertexIp)
updateDocument("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
} catch { } catch {
case e: Exception => e.printStackTrace() case e: Exception => e.printStackTrace()
} finally { } finally {
@@ -39,13 +37,13 @@ object UpdateDocument {
} }
private def updateDocument[T <: BaseDocument](collName: String, private def updateDocument[T <: BaseDocument](collName: String,
historyMap: ConcurrentHashMap[Integer, ConcurrentHashMap[String, T]],
getDocumentRow: (Row, ConcurrentHashMap[String, T]) => T, getDocumentRow: (Row, ConcurrentHashMap[String, T]) => T,
clazz: Class[T], clazz: Class[T],
getNewDataRdd: () => RDD[Row] getNewDataRdd: () => RDD[Row]
): Unit = { ): Unit = {
baseArangoData.readHistoryData(collName, historyMap, clazz) val historyMap = baseArangoData.readHistoryData(collName, clazz)
val hisBc = spark.sparkContext.broadcast(historyMap) val hisBc = spark.sparkContext.broadcast(historyMap)
LOG.warn("广播变量发送完毕")
try { try {
val start = System.currentTimeMillis() val start = System.currentTimeMillis()
val newDataRdd = getNewDataRdd() val newDataRdd = getNewDataRdd()
@@ -133,17 +131,18 @@ object UpdateDocument {
val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME") val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
val countTotalList = row.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST") val countTotalList = row.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST")
val schemaTypeList = row.getAs[ofRef[AnyRef]]("schema_type_list") val schemaTypeList = row.getAs[ofRef[AnyRef]]("schema_type_list")
val distCipRecent = row.getAs[ofRef[ofRef[String]]]("DIST_CIP_RECENT")
val distCipRecent = row.getAs[ofRef[String]]("DIST_CIP_RECENT")
val disCips = mergeDistinctIp(distCipRecent)
val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList) val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
val key = fqdn.concat("-" + serverIp) val key = fqdn.concat("-" + serverIp)
var document = dictionaryMap.getOrDefault(key, null) var document = dictionaryMap.getOrDefault(key, null)
if (document != null) { if (document != null) {
updateMaxAttribute(document, lastFoundTime, "LAST_FOUND_TIME") updateMaxAttribute(document, lastFoundTime, "LAST_FOUND_TIME")
updateProtocolAttritube(document, sepAttritubeMap) updateProtocolAttritube(document, sepAttritubeMap)
updateDistinctIp(document, distinctIp) updateDistinctIp(document,disCips)
} else { } else {
document = new BaseEdgeDocument() document = new BaseEdgeDocument()
document.setKey(key) document.setKey(key)
@@ -151,8 +150,8 @@ object UpdateDocument {
document.setTo("IP/" + serverIp) document.setTo("IP/" + serverIp)
document.addAttribute("FIRST_FOUND_TIME", firstFoundTime) document.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
document.addAttribute("LAST_FOUND_TIME", lastFoundTime) document.addAttribute("LAST_FOUND_TIME", lastFoundTime)
putDistinctIp(document,disCips)
putProtocolAttritube(document, sepAttritubeMap) putProtocolAttritube(document, sepAttritubeMap)
putDistinctIp(document, distinctIp)
} }
document document
} }

35
ip-learning-spark/src/test/scala/cn/ac/iie/service/update/UpdateDocumentTest.scala
View File

@@ -1,35 +0,0 @@
package cn.ac.iie.service.update
import java.util
import java.util.ArrayList
import java.util.concurrent.ConcurrentHashMap
import cn.ac.iie.dao.BaseArangoData
import cn.ac.iie.dao.BaseArangoData._
import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
import scala.collection.mutable.WrappedArray.ofRef
object UpdateDocumentTest {
def main(args: Array[String]): Unit = {
val baseArangoData = new BaseArangoData()
baseArangoData.readHistoryData("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, classOf[BaseEdgeDocument])
val value = BaseArangoData.historyRelationFqdnAddressIpMap.keys()
while (value.hasMoreElements) {
val integer: Integer = value.nextElement()
val map: ConcurrentHashMap[String, BaseEdgeDocument] = historyRelationFqdnAddressIpMap.get(integer)
val unit = map.keys()
while (unit.hasMoreElements) {
val key = unit.nextElement()
val edgeDocument = map.get(key)
// val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[util.ArrayList[Long]]
// val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[util.ArrayList[String]]
val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[Array[String]]
val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[Array[java.lang.Long]]
println(longs.toString + "---" + strings.toString)
}
}
}
}