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

修改spark离线任务，IP-Learning 相关学习功能，WAN NAT Pools功能（暂时停止任务） TSG-17496

Browse Source
This commit is contained in:
zhanghongqing
2023-11-02 18:04:32 +08:00
parent 272df41aec
commit 219e66cea1
5 changed files with 43 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -256,30 +256,30 @@ public class ReadClickhouseData {
}
public static String getVertexFqdnSql() {
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.session_record 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.session_record WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host";
String where = "recv_time >= " + minTime + " AND recv_time < " + maxTime;
String sslSql = "SELECT ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.session_record WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni";
String httpSql = "SELECT http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.session_record WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host";
return "SELECT FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME FROM ((" + sslSql + ") UNION ALL (" + httpSql + ")) GROUP BY FQDN HAVING FQDN != ''";
}
public static String getVertexIpSql() {
String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
String clientIpSql = "SELECT common_client_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
String serverIpSql = "SELECT common_server_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
return "SELECT * FROM((" + clientIpSql + ") UNION ALL (" + serverIpSql + "))";
}
public static String getRelationshipFqdnAddressIpSql() {
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.session_record 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.session_record WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip";
String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(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.session_record 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(recv_time) AS LAST_FOUND_TIME,MIN(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.session_record WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip";
return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
}
public static String getRelationshipIpVisitFqdnSql() {
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.session_record 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.session_record WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
String httpSql = "SELECT http_host AS FQDN,common_client_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'HTTP' AS schema_type FROM tsg_galaxy_v3.session_record 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(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'TLS' AS schema_type FROM tsg_galaxy_v3.session_record WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
}

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

@@ -36,21 +36,21 @@ object BaseClickhouseData {
}
def getVertexFqdnDf: DataFrame = {
val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
val sql =
s"""
|(SELECT
| FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
| FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
|FROM
| ((SELECT
| ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
| ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
| WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_vsys_id
| WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,vsys_id
| )UNION ALL
| (SELECT
| http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
| http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
| WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_vsys_id))
| WHERE $where and decoded_as = 'HTTP' GROUP BY http_host,vsys_id))
|GROUP BY FQDN,VSYS_ID HAVING FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)
@@ -60,32 +60,32 @@ object BaseClickhouseData {
}
def getVertexIpDf: DataFrame = {
val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
val sql =
s"""
|(SELECT * FROM
|((SELECT common_client_ip AS IP,MIN(common_recv_time) AS FIRST_FOUND_TIME,
|MAX(common_recv_time) AS LAST_FOUND_TIME,
|((SELECT client_ip AS IP,MIN(recv_time) AS FIRST_FOUND_TIME,
|MAX(recv_time) AS LAST_FOUND_TIME,
|count(*) as SESSION_COUNT,
|SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
|groupUniqArray(2)(common_link_info_c2s)[2] as common_link_info,
|SUM(sent_bytes+received_bytes) as BYTES_SUM,
|'' as common_link_info,
|'client' as ip_type
|,common_vsys_id AS VSYS_ID
|,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_client_ip,common_vsys_id)
|group by client_ip,vsys_id)
|UNION ALL
|(SELECT common_server_ip AS IP,
|MIN(common_recv_time) AS FIRST_FOUND_TIME,
|MAX(common_recv_time) AS LAST_FOUND_TIME,
|(SELECT server_ip AS IP,
|MIN(recv_time) AS FIRST_FOUND_TIME,
|MAX(recv_time) AS LAST_FOUND_TIME,
|count(*) as SESSION_COUNT,
|SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
|groupUniqArray(2)(common_link_info_s2c)[2] as common_link_info,
|SUM(sent_bytes+received_bytes) as BYTES_SUM,
|'' as common_link_info,
|'server' as ip_type
|,common_vsys_id AS VSYS_ID
|,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_server_ip,common_vsys_id))) as dbtable
|group by server_ip,vsys_id))) as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
@@ -95,19 +95,19 @@ object BaseClickhouseData {
def getRelationFqdnLocateIpDf: DataFrame = {
val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
val sql =
s"""
|(SELECT * FROM
|((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,
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'TLS' AS schema_type,common_vsys_id AS VSYS_ID
|((SELECT ssl_sni AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'TLS' AS schema_type,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip,common_vsys_id)
|WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,server_ip,vsys_id)
|UNION ALL
|(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,
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'HTTP' AS schema_type,common_vsys_id AS VSYS_ID
|(SELECT http_host AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'HTTP' AS schema_type,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip,common_vsys_id))
|WHERE $where and decoded_as = 'HTTP' GROUP BY http_host,server_ip,vsys_id))
|WHERE FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)

2
ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala
View File

@@ -1,7 +1,7 @@
package cn.ac.iie.main
import cn.ac.iie.service.update.UpdateDocument
@deprecated
object IpRecommendApplication {
def main(args: Array[String]): Unit = {

2
ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala
View File

@@ -1,7 +1,7 @@
package cn.ac.iie.main
import cn.ac.iie.service.update.UpdateDocument
@deprecated
object SubscriberRecommendApplication {
def main(args: Array[String]): Unit = {

4
ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala
View File

@@ -15,7 +15,7 @@ object BaseClickhouseDataTest {
|FROM
| (
| (SELECT
| ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
| ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME
| FROM
| global_temp.dbtable
| WHERE
@@ -23,7 +23,7 @@ object BaseClickhouseDataTest {
| )
| UNION ALL
| (SELECT
| http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
| http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME
| FROM
| global_temp.dbtable
| WHERE