gfwleak/galaxy-tsg-olap-sip-rtp-correlation
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[GAL-684] fix some bugs and update field mapping values

Browse Source
This commit is contained in:
chaochaoc
2024-10-30 18:07:49 +08:00
parent 728e3407e8
commit 45dafb9dbe
11 changed files with 117 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
pom.xml
View File

@@ -62,8 +62,8 @@
</dependency>
<dependency>
<groupId>com.zdjizhi</groupId>
<artifactId>galaxy</artifactId>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
</dependency>
<dependency>
<groupId>xyz.downgoon</groupId>
@@ -239,23 +239,9 @@
<!-- Common -->
<dependency>
<groupId>com.zdjizhi</groupId>
<artifactId>galaxy</artifactId>
<version>1.1.3</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</exclusion>
</exclusions>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.8.32</version>
</dependency>
<!-- Easy Stream-->
@@ -588,7 +574,8 @@
<groupId>${project.groupId}</groupId>
<artifactId>easy-stream-common</artifactId>
<version>${project.version}</version>
<remoteRepositories>${project.distributionManagement.repository.url}</remoteRepositories>
<remoteRepositories>${project.distributionManagement.repository.url}
</remoteRepositories>
</configuration>
</execution>
</executions>

1
src/main/java/com/geedgenetworks/flink/easy/application/voip/VoipUDFFactory.java
View File

@@ -22,6 +22,7 @@ public class VoipUDFFactory implements UDFFactory {
put("STREAM_DIR", new StreamDir());
put("STREAM_DIR_SET", new StreamDirSet());
put("FIND_NOT_BLANK", new FindNotBlank());
put("DISTINCT_CONCAT", new DistinctConcat());
put("SORT_ADDRESS", new SortAddress());
put("SNOWFLAKE_ID", new SnowflakeID());

18
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/DistinctConcat.java Normal file
View File

@@ -0,0 +1,18 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class DistinctConcat extends ScalarFunction {
public @DataTypeHint("STRING") String eval(String s1, String s2) {
return Stream.of(s1, s2).filter(StringUtils::isNotBlank)
.map(StringUtils::trim)
.distinct()
.collect(Collectors.joining(","));
}
}

8
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasIpAddress.java
View File

@@ -1,17 +1,17 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import com.zdjizhi.utils.IPUtil;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
public class HasIpAddress extends ScalarFunction {
public class HasIpAddress extends IpAddressScalarFunction {
public @DataTypeHint("BOOLEAN") Boolean eval(String... ipaddr) {
if (null == ipaddr) {
return false;
}
for (var ip : ipaddr) {
return ip != null && IPUtil.isIPAddress(ip);
if (ip != null && isIpAddress(ip)) {
return true;
}
}
return false;
}

25
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IpAddressScalarFunction.java Normal file
View File

@@ -0,0 +1,25 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import cn.hutool.core.lang.Validator;
import cn.hutool.core.net.NetUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.functions.ScalarFunction;
public abstract class IpAddressScalarFunction extends ScalarFunction {
protected static boolean isIpAddress(String ipaddr) {
return isIpv4(ipaddr) || isIpv6(ipaddr);
}
protected static boolean isIpv4(String ipaddr) {
return Validator.isIpv4(StringUtils.trim(ipaddr));
}
protected static boolean isIpv6(String ipaddr) {
return Validator.isIpv6(StringUtils.trim(ipaddr));
}
protected static boolean isInternalIpAddress(String ipaddr) {
return NetUtil.isInnerIP(StringUtils.trim(ipaddr));
}
}

10
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsExternalIpAddress.java
View File

@@ -1,17 +1,13 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import com.zdjizhi.utils.IPUtil;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
import static com.zdjizhi.utils.IPUtil.isIPAddress;
public class IsExternalIpAddress extends ScalarFunction {
public class IsExternalIpAddress extends IpAddressScalarFunction {
public @DataTypeHint("BOOLEAN") Boolean eval(String ipaddr) {
if (ipaddr == null || !isIPAddress(ipaddr)) {
if (ipaddr == null || !isIpAddress(ipaddr)) {
return false;
}
return !IPUtil.internalIp(ipaddr);
return !isInternalIpAddress(ipaddr);
}
}

10
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsInternalIpAddress.java
View File

@@ -1,17 +1,13 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import com.zdjizhi.utils.IPUtil;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
import static com.zdjizhi.utils.IPUtil.isIPAddress;
public class IsInternalIpAddress extends ScalarFunction {
public class IsInternalIpAddress extends IpAddressScalarFunction {
public @DataTypeHint("BOOLEAN") Boolean eval(String ipaddr) {
if (!isIPAddress(ipaddr)) {
if (!isIpAddress(ipaddr)) {
return false;
}
return IPUtil.internalIp(ipaddr);
return isInternalIpAddress(ipaddr);
}
}

6
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsIpAddress.java
View File

@@ -1,15 +1,13 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import com.zdjizhi.utils.IPUtil;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
public class IsIpAddress extends ScalarFunction {
public class IsIpAddress extends IpAddressScalarFunction {
public @DataTypeHint("BOOLEAN") Boolean eval(String ipaddr) {
if (null == ipaddr) {
return false;
}
return IPUtil.isIPAddress(ipaddr);
return isIpAddress(ipaddr);
}
}

31
src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SortAddress.java
View File

@@ -1,13 +1,14 @@
package com.geedgenetworks.flink.easy.application.voip.udf;
import com.google.common.collect.Lists;
import com.zdjizhi.utils.IPUtil;
import cn.hutool.core.net.NetUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.functions.ScalarFunction;
public class SortAddress extends ScalarFunction {
import java.util.ArrayList;
import java.util.List;
public class SortAddress extends IpAddressScalarFunction {
public @DataTypeHint("STRING")
String eval(
@@ -17,14 +18,14 @@ public class SortAddress extends ScalarFunction {
public static String of(
Tuple2<String, Integer> a1, Tuple2<String, Integer> a2) {
var list = Lists.newArrayList(a1, a2);
if (a1.f1 == null || a2.f1 == null || StringUtils.isAnyEmpty(a1.f0, a2.f0)) {
return "";
var list = new ArrayList<>(List.of(a1, a2));
if (a1.f1 == null || a2.f1 == null || StringUtils.isAnyEmpty(a1.f0, a2.f0)
|| !isIpAddress(a1.f0) || !isIpAddress(a2.f0)) {
return a1.f0 + ":" + a1.f1 + "," + a2.f0 + ":" + a2.f1;
}
list.sort((a, b) -> {
if (a.f1.equals(b.f1)) {
return Long.compare(
IPUtil.getIpDesimal(a.f0), IPUtil.getIpDesimal(b.f0));
return compareAddress(a.f0, b.f0);
} else {
return a.f1.compareTo(b.f1);
}
@@ -32,4 +33,16 @@ public class SortAddress extends ScalarFunction {
return String.format("%s:%s,%s:%s",
list.get(0).f0, list.get(0).f1, list.get(1).f0, list.get(1).f1);
}
private static int compareAddress(String ip1, String ip2) {
try {
var v1 = isIpv4(ip1) ? NetUtil.ipv4ToLong(ip1) :
NetUtil.ipv6ToBigInteger(ip1).longValue();
var v2 = isIpv4(ip2) ? NetUtil.ipv4ToLong(ip2) :
NetUtil.ipv6ToBigInteger(ip2).longValue();
return Long.compare(v1, v2);
} catch (Exception e) {
return 0;
}
}
}

50
src/main/resources/jobs/job.yml
View File

@@ -16,7 +16,7 @@ source:
group.id: sip-rtp-correlation
security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="example" password="example";
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="EXAMPLE-KAFKA-USERNAME" password="EXAMPLE-KAFKA-PASSWORD";
format: json
schema:
## General
@@ -301,7 +301,7 @@ sink:
bootstrap.servers: localhost:9092
security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="example" password="example";
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="EXAMPLE-KAFKA-USERNAME" password="EXAMPLE-KAFKA-PASSWORD";
format: json
# 关联成功的 VOIP
- name: only-voip-records
@@ -313,7 +313,7 @@ sink:
bootstrap.servers: localhost:9092
security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="example" password="example";
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="EXAMPLE-KAFKA-USERNAME" password="EXAMPLE-KAFKA-PASSWORD";
format: json
# 没有关联成功的 SIP 和 RTP
- name: fusion-fail-records
@@ -325,7 +325,7 @@ sink:
bootstrap.servers: localhost:9092
security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="example" password="example";
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="EXAMPLE-KAFKA-USERNAME" password="EXAMPLE-KAFKA-PASSWORD";
format: json
pipeline:
@@ -345,7 +345,7 @@ pipeline:
- name: error4-records
where: decoded_as == 'SIP' && STREAM_DIR(flags) == 3 && ( NOT( IS_IP_ADDRESS(sip_originator_sdp_connect_ip) ) || NOT( IS_IP_ADDRESS(sip_responder_sdp_connect_ip) ) )
### Notes: If internal IP address correlate is needed, please uncomment the following two items
### Notes: If internal IP address correlate is needed, please comment the following two items
# # Invalid: SIP one-way stream and internal network address
# - name: internal-error1-records
# where: decoded_as == 'SIP' && NOT(HAS_EXTERNAL_IP_ADDRESS(sip_originator_sdp_connect_ip, sip_responder_sdp_connect_ip))
@@ -643,9 +643,17 @@ pipeline:
- if: STREAM_DIR(flags) != 3 && @v1.isNotNull && STREAM_DIR(@v1.$flags) != STREAM_DIR(flags)
then:
- |-
OUTPUT ok FROM withColumns(recv_time to t_vsys_id),
OUTPUT ok FROM withColumns(recv_time to device_tag),
DISTINCT_CONCAT(@v1.$data_center, data_center) AS data_center,
DISTINCT_CONCAT(@v1.$device_group, device_group) AS device_group,
withColumns(sled_ip to t_vsys_id),
STREAM_DIR_SET(flags) AS flags,
withColumns(flags_identify_info to sip_call_id),
withColumns(flags_identify_info to decoded_path),
@v1.$sent_pkts + sent_pkts AS sent_pkts,
@v1.$received_pkts + received_pkts AS received_pkts,
@v1.$sent_bytes + sent_bytes AS sent_bytes,
@v1.$received_bytes + received_bytes AS received_bytes ,
withColumns(tcp_c2s_ip_fragments to sip_call_id),
FIND_NOT_BLANK(@v1.$sip_originator_description, sip_originator_description) AS sip_originator_description,
FIND_NOT_BLANK(@v1.$sip_responder_description, sip_responder_description) AS sip_responder_description,
FIND_NOT_BLANK(@v1.$sip_user_agent, sip_user_agent) AS sip_user_agent,
@@ -674,7 +682,7 @@ pipeline:
then:
- |-
OUTPUT ok FROM withColumns(recv_time to rtp_originator_dir)
- SCHEDULING USING EVENT TIME FOR NOW + 60 * 1000
- SCHEDULING USING PROCESS TIME FOR NOW + 60 * 1000
schedule:
- if: '@v1.isNotNull'
then:
@@ -1383,14 +1391,14 @@ pipeline:
@i.$out_link_id AS out_link_id,
@i.$in_link_id AS in_link_id,
@i.$device_tag AS device_tag,
@i.$data_center AS data_center,
@i.$device_group AS device_group,
DISTINCT_CONCAT(@i.$data_center, data_center) AS data_center,
DISTINCT_CONCAT(@i.$device_group, device_group) AS device_group,
@i.$sled_ip AS sled_ip,
@i.$address_type AS address_type,
@i.$direction AS direction,
@i.$vsys_id AS vsys_id,
@i.$t_vsys_id AS t_vsys_id,
@i.$flags AS flags,
flags AS flags,
@i.$flags_identify_info AS flags_identify_info,
@i.$c2s_ttl AS c2s_ttl,
@@ -1418,10 +1426,10 @@ pipeline:
@i.$ip_protocol AS ip_protocol,
@i.$sent_pkts AS sent_pkts,
@i.$received_pkts AS received_pkts,
@i.$sent_bytes AS sent_bytes,
@i.$received_bytes AS received_bytes,
@i.$sent_pkts + sent_pkts AS sent_pkts,
@i.$received_pkts + received_pkts AS received_pkts,
@i.$sent_bytes + sent_bytes AS sent_bytes,
@i.$received_bytes + received_bytes AS received_bytes,
withColumns(sip_call_id to sip_bye_reason),
@@ -1462,8 +1470,8 @@ pipeline:
@i.$direction AS direction,
@i.$vsys_id AS vsys_id,
@i.$t_vsys_id AS t_vsys_id,
@i.$flags AS flags,
@i.$flags_identify_info AS flags_identify_info,
@sip.$flags AS flags,
@sip.$flags_identify_info AS flags_identify_info,
@i.$c2s_ttl AS c2s_ttl,
@i.$s2c_ttl AS s2c_ttl,
@@ -1490,10 +1498,10 @@ pipeline:
@i.$ip_protocol AS ip_protocol,
@i.$sent_pkts AS sent_pkts,
@i.$received_pkts AS received_pkts,
@i.$sent_bytes AS sent_bytes,
@i.$received_bytes AS received_bytes,
@i.$sent_pkts + sent_pkts AS sent_pkts,
@i.$received_pkts + received_pkts AS received_pkts,
@i.$sent_bytes + sent_bytes AS sent_bytes,
@i.$received_bytes + received_bytes AS received_bytes,
@sip.$sip_call_id AS sip_call_id,
@sip.$sip_originator_description AS sip_originator_description,

3
tools/maven/spotbugs-exclude.xml
View File

@@ -6,6 +6,9 @@
<Match>
<Bug pattern="EI_EXPOSE_REP2"/>
</Match>
<Match>
<Bug pattern="REC_CATCH_EXCEPTION"/>
</Match>
<Match>
<Bug pattern="SE_NO_SERIALVERSIONID"/>
</Match>