修改distinct client IP存储时间为24小时

10 files changed, 81 insertions, 41 deletions

diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java b/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java
index 55d917c..fa5e88b 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java
@@ -31,7 +31,9 @@ public class BaseArangoData {
 
     private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
 
-    <T extends BaseDocument> void readHistoryData(String table,ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,Class<T> type){
+    <T extends BaseDocument> void readHistoryData(String table,
+                                                  ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,
+                                                  Class<T> type) {
         try {
             LOG.info("开始更新"+table);
             long start = System.currentTimeMillis();
@@ -42,7 +44,8 @@ public class BaseArangoData {
             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, map,type,table,countDownLatch);
+                ReadHistoryArangoData<T> readHistoryArangoData =
+                        new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
                 threadPool.executor(readHistoryArangoData);
             }
             countDownLatch.await();
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java b/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
index 2bfc5e3..7c89a18 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
@@ -35,7 +35,9 @@ public class BaseClickhouseData {
     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){
+    <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();
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java b/IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
index dbbd81f..8012f85 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
@@ -60,9 +60,9 @@ public class UpdateGraphData {
                     LocateFqdn2Ip.class,BaseEdgeDocument.class,
                     ReadClickhouseData::getRelationshipFqdnAddressIpSql,ReadClickhouseData::getRelationFqdnAddressIpDocument);
 
-            updateDocument(newRelationIpVisitFqdnMap,historyRelationIpVisitFqdnMap,"R_VISIT_IP2FQDN",
-                    VisitIp2Fqdn.class,BaseEdgeDocument.class,
-                    ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument);
+//            updateDocument(newRelationIpVisitFqdnMap,historyRelationIpVisitFqdnMap,"R_VISIT_IP2FQDN",
+//                    VisitIp2Fqdn.class,BaseEdgeDocument.class,
+//                    ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument);
 
             updateDocument(newRelationSubsciberLocateIpMap,historyRelationSubsciberLocateIpMap,"R_LOCATE_SUBSCRIBER2IP",
                     LocateSubscriber2Ip.class,BaseEdgeDocument.class,
@@ -106,7 +106,7 @@ public class UpdateGraphData {
                         String.class,
                         ConcurrentHashMap.class,
                         CountDownLatch.class);
-                Document<T> docTask = (Document<T>)constructor.newInstance(tmpNewMap, arangoManger, collection, tmpHisMap, countDownLatch);
+                Document docTask = (Document)constructor.newInstance(tmpNewMap, arangoManger, collection, tmpHisMap, countDownLatch);
                 pool.executor(docTask);
             }
             countDownLatch.await();
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java b/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java
index 5700cb8..05c28ea 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java
@@ -23,9 +23,11 @@ public class ReadClickhouseData {
     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;
-    public static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR;
+    static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR;
     public static final HashSet<String> PROTOCOL_SET;
 
     static {
@@ -248,8 +250,6 @@ public class ReadClickhouseData {
     }
 
     public static String getVertexFqdnSql() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         String where = "common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
         String sslSql = "SELECT ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni";
         String httpSql = "SELECT http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host";
@@ -257,8 +257,6 @@ public class ReadClickhouseData {
     }
 
     public static String getVertexIpSql() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         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,'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,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
@@ -266,8 +264,6 @@ public class ReadClickhouseData {
     }
 
     public static String getRelationshipFqdnAddressIpSql() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         String where = " common_recv_time >= " + minTime + " AND common_recv_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 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";
@@ -275,8 +271,6 @@ public class ReadClickhouseData {
     }
 
     public static String getRelationshipIpVisitFqdnSql() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
         String httpSql = "SELECT http_host AS FQDN,common_client_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_client_ip";
         String sslSql = "SELECT ssl_sni AS FQDN,common_client_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'TLS' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
@@ -284,15 +278,11 @@ public class ReadClickhouseData {
     }
 
     public static String getVertexSubscriberSql() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
         return "SELECT common_subscriber_id,MAX(common_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() {
-        long maxTime = timeLimit[0];
-        long minTime = timeLimit[1];
         String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_framed_ip != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
         return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_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";
     }
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java b/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java
index 9b26fdc..a4541f9 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java
@@ -8,11 +8,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 
-import static cn.ac.iie.service.ingestion.ReadClickhouseData.RECENT_COUNT_HOUR;
+import static cn.ac.iie.service.ingestion.ReadClickhouseData.*;
 
 /**
  * @author wlh
@@ -23,14 +24,14 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
 
     private ArangoDBConnect arangoConnect;
     private String query;
-    private ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map;
+    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,
+                                 ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> map,
                                  Class<T> type,
                                  String table,
                                  CountDownLatch countDownLatch) {
@@ -55,6 +56,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                     switch (table) {
                         case "R_LOCATE_FQDN2IP":
                             updateProtocolDocument(doc);
+                            deleteDistinctClientIpByTime(doc);
                             break;
                         case "R_VISIT_IP2FQDN":
                             updateProtocolDocument(doc);
@@ -69,11 +71,11 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                 long l = System.currentTimeMillis();
                 LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
             }
-        }catch (Exception e){
+        } catch (Exception e) {
             e.printStackTrace();
-        }finally {
+        } finally {
             countDownLatch.countDown();
-            LOG.info("本线程读取完毕，剩余线程数量："+countDownLatch.getCount());
+            LOG.info("本线程读取完毕，剩余线程数量：" + countDownLatch.getCount());
         }
     }
 
@@ -91,4 +93,22 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
         }
     }
 
+    private void deleteDistinctClientIpByTime(T doc) {
+        ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
+        ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
+        distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
+        Collections.sort(distCipTs);
+        int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);
+        String[] distCipArr = new String[index];
+        long[] disCipTsArr = new long[index];
+        if (distCip.size() + 1 == distCipTs.size()){
+            for (int i = 0; i < index; i++) {
+                distCipArr[i] = distCip.get(i);
+                disCipTsArr[i] = distCipTs.get(i);
+            }
+        }
+        doc.updateAttribute("DIST_CIP", distCipArr);
+        doc.updateAttribute("DIST_CIP_TS", disCipTsArr);
+    }
+
 }
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java b/IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java
index c8dca13..82d1a1a 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java
@@ -53,12 +53,12 @@ public class LocateFqdn2Ip extends Relationship {
     }
 
     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");
+        String[] distCip = (String[]) edgeDocument.getAttribute("DIST_CIP");
+        long[] distCipTs = (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));
+        if (distCip.length == distCipTs.length){
+            for (int i = 0;i < distCip.length;i++){
+                distCipToTs.put(distCip[i],distCipTs[i]);
             }
         }
         Object[] distCipRecent = (Object[])newEdgeDocument.getAttribute("DIST_CIP");
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java b/IP-learning-graph/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
index b47e796..79ef293 100644
--- a/IP-learning-graph/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
+++ b/IP-learning-graph/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
@@ -19,7 +19,7 @@ public class ExecutorThreadPool {
     private static void getThreadPool(){
         ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                 .setNameFormat("iplearning-application-pool-%d").build();
-        pool = new ThreadPoolExecutor(5, ApplicationConfig.THREAD_POOL_NUMBER,
+        pool = new ThreadPoolExecutor(ApplicationConfig.THREAD_POOL_NUMBER, ApplicationConfig.THREAD_POOL_NUMBER,
                 0L, TimeUnit.SECONDS,
                 new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
     }
diff --git a/IP-learning-graph/src/main/resources/application.properties b/IP-learning-graph/src/main/resources/application.properties
index e639891..c09d5d5 100644
--- a/IP-learning-graph/src/main/resources/application.properties
+++ b/IP-learning-graph/src/main/resources/application.properties
@@ -1,10 +1,11 @@
 #arangoDB参数配置
-arangoDB.host=192.168.40.182
+#arangoDB.host=192.168.40.182
+arangoDB.host=192.168.40.224
 arangoDB.port=8529
 arangoDB.user=root
 arangoDB.password=111111
-#arangoDB.DB.name=ip-learning-test
-arangoDB.DB.name=ip-learning-test-0
+arangoDB.DB.name=ip-learning-test
+#arangoDB.DB.name=ip-learning-test-0
 arangoDB.batch=100000
 arangoDB.ttl=3600
 
@@ -16,8 +17,8 @@ thread.await.termination.time=10
 
 #读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
 time.limit.type=1
-read.clickhouse.max.time=1595489408
-read.clickhouse.min.time=1593878400
+read.clickhouse.max.time=1596097930
+read.clickhouse.min.time=1593676894
 
 update.interval=3600
 distinct.client.ip.num=10000
diff --git a/IP-learning-graph/src/main/resources/clickhouse.properties b/IP-learning-graph/src/main/resources/clickhouse.properties
index 1a84107..3b0ddbb 100644
--- a/IP-learning-graph/src/main/resources/clickhouse.properties
+++ b/IP-learning-graph/src/main/resources/clickhouse.properties
@@ -1,9 +1,9 @@
 drivers=ru.yandex.clickhouse.ClickHouseDriver
-#db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=300000
-db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
+db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=300000
+#db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
 mdb.user=default
-#mdb.password=ceiec2019
-mdb.password=111111
+mdb.password=ceiec2019
+#mdb.password=111111
 initialsize=1
 minidle=1
 maxactive=50
diff --git a/IP-learning-graph/src/test/java/cn/ac/iie/TestList.java b/IP-learning-graph/src/test/java/cn/ac/iie/TestList.java
new file mode 100644
index 0000000..83ce03c
--- /dev/null
+++ b/IP-learning-graph/src/test/java/cn/ac/iie/TestList.java
@@ -0,0 +1,24 @@
+package cn.ac.iie;
+
+import java.util.ArrayList;
+import java.util.Collections;
+
+public class TestList {
+    public static void main(String[] args) {
+        ArrayList<Integer> integers = new ArrayList<>();
+        integers.add(10);
+        integers.add(8);
+        integers.add(11);
+        integers.add(5);
+        integers.add(5);
+        integers.add(4);
+        integers.add(4);
+        integers.add(12);
+        Collections.sort(integers);
+        System.out.println(integers);
+        integers.add(5);
+        Collections.sort(integers);
+        System.out.println(integers);
+        System.out.println(integers.indexOf(5));
+    }
+}