[GAL-602] refactor: impl correlate jobs through config yml

11 files changed, 1354 insertions, 72 deletions

diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/VoipUDFFactory.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/VoipUDFFactory.java
index c615d44..20d0746 100644
--- a/src/main/java/com/geedgenetworks/flink/easy/application/voip/VoipUDFFactory.java
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/VoipUDFFactory.java
@@ -1,8 +1,6 @@
 package com.geedgenetworks.flink.easy.application.voip;
 
-import com.geedgenetworks.flink.easy.application.voip.udf.IsInternalIPAddress;
-import com.geedgenetworks.flink.easy.application.voip.udf.IsIpAddress;
-import com.geedgenetworks.flink.easy.application.voip.udf.StreamDir;
+import com.geedgenetworks.flink.easy.application.voip.udf.*;
 import com.geedgenetworks.flink.easy.common.api.UDFFactory;
 import org.apache.flink.table.functions.UserDefinedFunction;
 
@@ -14,8 +12,18 @@ public class VoipUDFFactory implements UDFFactory {
     private static final Map<String, UserDefinedFunction> R =
             new HashMap<>() {{
                 put("IS_IP_ADDRESS", new IsIpAddress());
-                put("IS_INTERNAL_IP_ADDRESS", new IsInternalIPAddress());
+
+                put("IS_INTERNAL_IP_ADDRESS", new IsInternalIpAddress());
+                put("IS_EXTERNAL_IP_ADDRESS", new IsExternalIpAddress());
+
+                put("HAS_IP_ADDRESS", new HasIpAddress());
+                put("HAS_EXTERNAL_IP_ADDRESS", new HasExternalIpAddress());
+
                 put("STREAM_DIR", new StreamDir());
+                put("FIND_NOT_BLANK", new FindNotBlank());
+                put("SORT_ADDRESS", new SortAddress());
+
+                put("SNOWFLAKE_ID", new SnowflakeID());
             }};
 
     @Override
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/FindNotBlank.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/FindNotBlank.java
new file mode 100644
index 0000000..a018ce2
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/FindNotBlank.java
@@ -0,0 +1,15 @@
+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;
+
+public class FindNotBlank extends ScalarFunction {
+
+    public @DataTypeHint("STRING") String eval(String s1, String s2) {
+        if (StringUtils.isBlank(s1) && StringUtils.isNotBlank(s2)) {
+            return s2;
+        }
+        return s1;
+    }
+}
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasExternalIpAddress.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasExternalIpAddress.java
new file mode 100644
index 0000000..2ce254a
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasExternalIpAddress.java
@@ -0,0 +1,19 @@
+package com.geedgenetworks.flink.easy.application.voip.udf;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.functions.ScalarFunction;
+
+public class HasExternalIpAddress extends ScalarFunction {
+
+    private final IsExternalIpAddress isExternalIpAddress = new IsExternalIpAddress();
+
+    public @DataTypeHint("BOOLEAN") Boolean eval(String... ipaddr) {
+        if (null == ipaddr) {
+            return false;
+        }
+        for (var ip : ipaddr) {
+            return isExternalIpAddress.eval(ip);
+        }
+        return false;
+    }
+}
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasIpAddress.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasIpAddress.java
new file mode 100644
index 0000000..7dddbc7
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/HasIpAddress.java
@@ -0,0 +1,18 @@
+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 @DataTypeHint("BOOLEAN") Boolean eval(String... ipaddr) {
+        if (null == ipaddr) {
+            return false;
+        }
+        for (var ip : ipaddr) {
+            return ip != null && IPUtil.isIPAddress(ip);
+        }
+        return false;
+    }
+}
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsExternalIpAddress.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsExternalIpAddress.java
new file mode 100644
index 0000000..a2970a6
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsExternalIpAddress.java
@@ -0,0 +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;
+
+import static com.zdjizhi.utils.IPUtil.isIPAddress;
+
+public class IsExternalIpAddress extends ScalarFunction {
+
+    public @DataTypeHint("BOOLEAN") Boolean eval(String ipaddr) {
+        if (ipaddr == null || !isIPAddress(ipaddr)) {
+            return false;
+        }
+        return !IPUtil.internalIp(ipaddr);
+    }
+}
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsInternalIPAddress.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsInternalIpAddress.java
index 578c507..55839ba 100644
--- a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsInternalIPAddress.java
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/IsInternalIpAddress.java
@@ -6,7 +6,7 @@ import org.apache.flink.table.functions.ScalarFunction;
 
 import static com.zdjizhi.utils.IPUtil.isIPAddress;
 
-public class IsInternalIPAddress extends ScalarFunction {
+public class IsInternalIpAddress extends ScalarFunction {
 
     public @DataTypeHint("BOOLEAN") Boolean eval(String ipaddr) {
         if (!isIPAddress(ipaddr)) {
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SnowflakeID.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SnowflakeID.java
new file mode 100644
index 0000000..abaea8f
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SnowflakeID.java
@@ -0,0 +1,14 @@
+package com.geedgenetworks.flink.easy.application.voip.udf;
+
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.functions.ScalarFunction;
+import xyz.downgoon.snowflake.Snowflake;
+
+public class SnowflakeID extends ScalarFunction {
+
+    private static final Snowflake SNOWFLAKE = new Snowflake(1, 1);
+
+    public @DataTypeHint("BIGINT") Long eval() {
+        return SNOWFLAKE.nextId();
+    }
+}
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SortAddress.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SortAddress.java
new file mode 100644
index 0000000..b689275
--- /dev/null
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/SortAddress.java
@@ -0,0 +1,26 @@
+package com.geedgenetworks.flink.easy.application.voip.udf;
+
+import com.google.common.collect.Lists;
+import com.zdjizhi.utils.IPUtil;
+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 {
+
+    public @DataTypeHint("STRING") String eval(String ip1, Integer port1, String ip2, Integer port2) {
+        return of(Tuple2.of(ip1, port1), Tuple2.of(ip2, port2));
+    }
+
+    public static String of(Tuple2<String, Integer> a1, Tuple2<String, Integer> a2) {
+        var list = Lists.newArrayList(a1, a2);
+        list.sort((a, b) -> {
+            if (a.f1.equals(b.f1)) {
+                return Long.compare(IPUtil.getIpDesimal(a.f0), IPUtil.getIpDesimal(b.f0));
+            } else {
+                return a.f1.compareTo(b.f1);
+            }
+        });
+        return list.get(0).f0 + list.get(0).f1 + list.get(1).f0 + list.get(1).f1;
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/StreamDir.java b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/StreamDir.java
index 016bfe9..3f5166f 100644
--- a/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/StreamDir.java
+++ b/src/main/java/com/geedgenetworks/flink/easy/application/voip/udf/StreamDir.java
@@ -7,6 +7,9 @@ public class StreamDir extends ScalarFunction {
 
     public @DataTypeHint("INT") Integer eval(Long flags) {
         int v = 0;
+        if (flags == null) {
+            return v;
+        }
         if ((flags & 8192) == 8192) {
             v += 1;
         }
@@ -15,4 +18,9 @@ public class StreamDir extends ScalarFunction {
         }
         return v;
     }
+
+    public static void main(String[] args) {
+        System.out.println(8192L + 16384L);
+        System.out.println(new StreamDir().eval(8192L + 16384L));
+    }
 }
\ No newline at end of file
diff --git a/src/main/resources/job.yml b/src/main/resources/job.yml
index 6d838d0..f44f17d 100644
--- a/src/main/resources/job.yml
+++ b/src/main/resources/job.yml
@@ -3,48 +3,58 @@ job:
   parallelism: 1
   active-pipeline:
     - console
-    - console1
 
 source:
   - name: session-records
-    #    type: kafka
+    type: kafka
+    option:
+      topic: SESSION-RECORD
+      properties:
+        bootstrap.servers: 192.168.44.12:9092
+        group.id: easy-stream-tester9
+        client.id: easy-stream-tester9
+    #    type: file
     #    option:
-    #      topic: SESSION-RECORD
-    #      properties:
-    #        bootstrap.servers: 192.168.44.12:9092
-    #        group.id: easy-stream-tester9
-    #        client.id: easy-stream-tester9
-#    type: file
+    #      path: E:\java-workspace\sip-rtp-correlation\feature\easy-refactor\src\main\resources\session-records.txt
+#    type: socket
 #    option:
-#      path: E:\java-workspace\sip-rtp-correlation\feature\easy-refactor\src\main\resources\session-records.txt
-    type: socket
-    option:
-      hostname: localhost
-      port: 9999
+#      hostname: localhost
+#      port: 9999
     format: json
     schema:
+      ## General
+      - name: recv_time
+        data-type: BIGINT NOT NULL
+      - name: log_id
+        data-type: BIGINT NOT NULL
+      - name: decoded_as
+        data-type: STRING NOT NULL
       - name: session_id
         data-type: BIGINT NOT NULL
       - name: start_timestamp_ms
         data-type: BIGINT NOT NULL
-        # row-time:
+      # row-time:
       - name: start_timestamp
         for: TO_TIMESTAMP_LTZ(start_timestamp_ms, 3)
         watermark: start_timestamp - INTERVAL '5' MINUTE
       - name: end_timestamp_ms
-        data-type: BIGINT NOT NULL
-      - name: decoded_as
-        data-type: STRING NOT NULL
+        data-type: BIGINT
       - name: duration_ms
-        data-type: BIGINT NOT NULL
+        data-type: INT
       - name: tcp_handshake_latency_ms
+        data-type: INT
+      - name: ingestion_time
+        data-type: BIGINT
+      - name: processing_time
+        data-type: BIGINT
+      - name: insert_time
         data-type: BIGINT
       - name: device_id
-        data-type: STRING NOT NULL
+        data-type: STRING
       - name: out_link_id
-        data-type: BIGINT
+        data-type: INT
       - name: in_link_id
-        data-type: BIGINT
+        data-type: INT
       - name: device_tag
         data-type: STRING
       - name: data_center
@@ -56,15 +66,19 @@ source:
       - name: address_type
         data-type: INT
       - name: direction
-        data-type: INT
+        data-type: STRING
       - name: vsys_id
-        data-type: BIGINT
+        data-type: INT
       - name: t_vsys_id
-        data-type: BIGINT
+        data-type: INT
       - name: flags
         data-type: BIGINT
       - name: flags_identify_info
         data-type: STRING
+      - name: c2s_ttl
+        data-type: INT
+      - name: s2c_ttl
+        data-type: INT
       ## Treatment
       - name: security_rule_list
         data-type: ARRAY<BIGINT>
@@ -74,16 +88,16 @@ source:
         data-type: ARRAY<BIGINT>
       - name: shaping_rule_list
         data-type: ARRAY<BIGINT>
-      - name: sc_rule_list
+      - name: proxy_rule_list
         data-type: ARRAY<BIGINT>
       - name: statistics_rule_list
         data-type: ARRAY<BIGINT>
+      - name: sc_rule_list
+        data-type: ARRAY<BIGINT>
       - name: sc_rsp_raw
         data-type: ARRAY<BIGINT>
       - name: sc_rsp_decrypted
         data-type: ARRAY<BIGINT>
-      - name: proxy_rule_list
-        data-type: ARRAY<BIGINT>
       - name: proxy_action
         data-type: STRING
       - name: proxy_pinning_status
@@ -93,9 +107,9 @@ source:
       - name: proxy_passthrough_reason
         data-type: STRING
       - name: proxy_client_side_latency_ms
-        data-type: BIGINT
+        data-type: INT
       - name: proxy_server_side_latency_ms
-        data-type: BIGINT
+        data-type: INT
       - name: proxy_client_side_version
         data-type: STRING
       - name: proxy_server_side_version
@@ -161,12 +175,14 @@ source:
       - name: server_domain
         data-type: STRING
       - name: fqdn_category_list
-        data-type: ARRAY<BIGINT>
+        data-type: ARRAY<INT>
       ## Application
       - name: app_transition
         data-type: STRING
       - name: app
         data-type: STRING
+      - name: app_category
+        data-type: STRING
       - name: app_debug_info
         data-type: STRING
       - name: app_content
@@ -178,6 +194,62 @@ source:
         data-type: STRING
       - name: decoded_path
         data-type: STRING
+      ## Transmission
+      - name: sent_pkts
+        data-type: BIGINT
+      - name: received_pkts
+        data-type: BIGINT
+      - name: sent_bytes
+        data-type: BIGINT
+      - name: received_bytes
+        data-type: BIGINT
+      - name: tcp_c2s_ip_fragments
+        data-type: BIGINT
+      - name: tcp_s2c_ip_fragments
+        data-type: BIGINT
+      - name: tcp_c2s_lost_bytes
+        data-type: BIGINT
+      - name: tcp_s2c_lost_bytes
+        data-type: BIGINT
+      - name: tcp_c2s_o3_pkts
+        data-type: BIGINT
+      - name: tcp_s2c_o3_pkts
+        data-type: BIGINT
+      - name: tcp_c2s_rtx_pkts
+        data-type: BIGINT
+      - name: tcp_s2c_rtx_pkts
+        data-type: BIGINT
+      - name: tcp_c2s_rtx_bytes
+        data-type: BIGINT
+      - name: tcp_s2c_rtx_bytes
+        data-type: BIGINT
+      - name: tcp_rtt_ms
+        data-type: INT
+      - name: tcp_client_isn
+        data-type: BIGINT
+      - name: tcp_server_isn
+        data-type: BIGINT
+      ## Other
+      - name: packet_capture_file
+        data-type: STRING
+      - name: in_src_mac
+        data-type: STRING
+      - name: out_src_mac
+        data-type: STRING
+      - name: in_dest_mac
+        data-type: STRING
+      - name: out_dest_mac
+        data-type: STRING
+      - name: encapsulation
+        data-type: STRING
+      - name: dup_traffic_flag
+        data-type: INT
+      - name: tunnel_id_list
+        data-type: ARRAY<BIGINT>
+      - name: tunnel_endpoint_a_desc
+        data-type: STRING
+      - name: tunnel_endpoint_b_desc
+        data-type: STRING
       ## SIP
       - name: sip_call_id
         data-type: STRING
@@ -218,16 +290,11 @@ source:
         data-type: STRING
       - name: rtp_originator_dir
         data-type: INT
-
 pipeline:
   - name: console
     category: PRINT
-#    on: sip-records
-    on: sip-double-way-records.ok
-    use-err: true
-  - name: console1
-    category: PRINT
-    on: sip-records
+    #    on: sip-records
+    on: voip-fusion.ok
     use-err: true
   - name: split-for-valid
     category: SPLIT
@@ -262,32 +329,43 @@ pipeline:
         where: decoded_as == 'SIP'
   - name: sip-double-way-records
     category: CORRELATE
-    on: sip-records
     cache:
       - name: v1
         type: VALUE
         ttl: 1 minute
         schema:
+          ## General
+          - name: recv_time
+            data-type: BIGINT NOT NULL
+          - name: log_id
+            data-type: BIGINT NOT NULL
+          - name: decoded_as
+            data-type: STRING NOT NULL
           - name: session_id
             data-type: BIGINT NOT NULL
           - name: start_timestamp_ms
             data-type: BIGINT NOT NULL
+          # row-time:
           - name: start_timestamp
             data-type: TIMESTAMP_LTZ(3)
           - name: end_timestamp_ms
-            data-type: BIGINT NOT NULL
-          - name: decoded_as
-            data-type: STRING NOT NULL
+            data-type: BIGINT
           - name: duration_ms
-            data-type: BIGINT NOT NULL
+            data-type: INT
           - name: tcp_handshake_latency_ms
+            data-type: INT
+          - name: ingestion_time
+            data-type: BIGINT
+          - name: processing_time
+            data-type: BIGINT
+          - name: insert_time
             data-type: BIGINT
           - name: device_id
-            data-type: STRING NOT NULL
+            data-type: STRING
           - name: out_link_id
-            data-type: BIGINT
+            data-type: INT
           - name: in_link_id
-            data-type: BIGINT
+            data-type: INT
           - name: device_tag
             data-type: STRING
           - name: data_center
@@ -299,15 +377,19 @@ pipeline:
           - name: address_type
             data-type: INT
           - name: direction
-            data-type: INT
+            data-type: STRING
           - name: vsys_id
-            data-type: BIGINT
+            data-type: INT
           - name: t_vsys_id
-            data-type: BIGINT
+            data-type: INT
           - name: flags
             data-type: BIGINT
           - name: flags_identify_info
             data-type: STRING
+          - name: c2s_ttl
+            data-type: INT
+          - name: s2c_ttl
+            data-type: INT
           ## Treatment
           - name: security_rule_list
             data-type: ARRAY<BIGINT>
@@ -317,16 +399,16 @@ pipeline:
             data-type: ARRAY<BIGINT>
           - name: shaping_rule_list
             data-type: ARRAY<BIGINT>
-          - name: sc_rule_list
+          - name: proxy_rule_list
             data-type: ARRAY<BIGINT>
           - name: statistics_rule_list
             data-type: ARRAY<BIGINT>
+          - name: sc_rule_list
+            data-type: ARRAY<BIGINT>
           - name: sc_rsp_raw
             data-type: ARRAY<BIGINT>
           - name: sc_rsp_decrypted
             data-type: ARRAY<BIGINT>
-          - name: proxy_rule_list
-            data-type: ARRAY<BIGINT>
           - name: proxy_action
             data-type: STRING
           - name: proxy_pinning_status
@@ -336,9 +418,9 @@ pipeline:
           - name: proxy_passthrough_reason
             data-type: STRING
           - name: proxy_client_side_latency_ms
-            data-type: BIGINT
+            data-type: INT
           - name: proxy_server_side_latency_ms
-            data-type: BIGINT
+            data-type: INT
           - name: proxy_client_side_version
             data-type: STRING
           - name: proxy_server_side_version
@@ -404,12 +486,14 @@ pipeline:
           - name: server_domain
             data-type: STRING
           - name: fqdn_category_list
-            data-type: ARRAY<BIGINT>
+            data-type: ARRAY<INT>
           ## Application
           - name: app_transition
             data-type: STRING
           - name: app
             data-type: STRING
+          - name: app_category
+            data-type: STRING
           - name: app_debug_info
             data-type: STRING
           - name: app_content
@@ -421,6 +505,62 @@ pipeline:
             data-type: STRING
           - name: decoded_path
             data-type: STRING
+          ## Transmission
+          - name: sent_pkts
+            data-type: BIGINT
+          - name: received_pkts
+            data-type: BIGINT
+          - name: sent_bytes
+            data-type: BIGINT
+          - name: received_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_ip_fragments
+            data-type: BIGINT
+          - name: tcp_s2c_ip_fragments
+            data-type: BIGINT
+          - name: tcp_c2s_lost_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_lost_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_o3_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_o3_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_rtt_ms
+            data-type: INT
+          - name: tcp_client_isn
+            data-type: BIGINT
+          - name: tcp_server_isn
+            data-type: BIGINT
+          ## Other
+          - name: packet_capture_file
+            data-type: STRING
+          - name: in_src_mac
+            data-type: STRING
+          - name: out_src_mac
+            data-type: STRING
+          - name: in_dest_mac
+            data-type: STRING
+          - name: out_dest_mac
+            data-type: STRING
+          - name: encapsulation
+            data-type: STRING
+          - name: dup_traffic_flag
+            data-type: INT
+          - name: tunnel_id_list
+            data-type: ARRAY<BIGINT>
+          - name: tunnel_endpoint_a_desc
+            data-type: STRING
+          - name: tunnel_endpoint_b_desc
+            data-type: STRING
           ## SIP
           - name: sip_call_id
             data-type: STRING
@@ -467,10 +607,8 @@ pipeline:
         process:
           - if: STREAM_DIR(flags) != 3 && @v1.isNotNull && STREAM_DIR(@v1.$flags) != STREAM_DIR(flags)
             then:
-              - |- 
-                OUTPUT ok FROM session_id, 
-                      start_timestamp_ms,
-                      withColumns(end_timestamp_ms to sip_call_id),
+              - |-
+                OUTPUT ok FROM withColumns(recv_time 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,
@@ -490,26 +628,31 @@ pipeline:
                       rtp_pcap_path,
                       rtp_originator_dir
               - TRUNCATE v1
-          - if:  STREAM_DIR(flags) != 3 && @v1.isNull
+          - if: STREAM_DIR(flags) != 3 && @v1.isNull
             then:
               - |-
-                SET v1 FROM withColumns(session_id to rtp_originator_dir)
+                SET v1 FROM withColumns(recv_time to rtp_originator_dir)
           - if: STREAM_DIR(flags) == 3
             then:
               - |-
-                OUTPUT ok FROM session_id,
-                            start_timestamp_ms, withColumns(end_timestamp_ms to rtp_originator_dir)
+                OUTPUT ok FROM withColumns(recv_time to rtp_originator_dir)
           - SCHEDULING USING EVENT TIME FOR NOW + 60 * 1000
     schedule:
       - if: '@v1.isNotNull'
         then:
           - |-
-            OUTPUT fail FROM @v1.$session_id AS session_id,
+            OUTPUT fail FROM @v1.$recv_time AS recv_time,
+                          @v1.$log_id AS log_id,
+                          @v1.$decoded_as AS decoded_as,
+                          @v1.$session_id AS session_id,
                           @v1.$start_timestamp_ms AS start_timestamp_ms,
+                          @v1.$start_timestamp AS start_timestamp,
                           @v1.$end_timestamp_ms AS end_timestamp_ms,
-                          @v1.$decoded_as AS decoded_as,
                           @v1.$duration_ms AS duration_ms,
                           @v1.$tcp_handshake_latency_ms AS tcp_handshake_latency_ms,
+                          @v1.$ingestion_time AS ingestion_time,
+                          @v1.$processing_time AS processing_time,
+                          @v1.$insert_time AS insert_time,
                           @v1.$device_id AS device_id,
                           @v1.$out_link_id AS out_link_id,
                           @v1.$in_link_id AS in_link_id,
@@ -523,15 +666,17 @@ pipeline:
                           @v1.$t_vsys_id AS t_vsys_id,
                           @v1.$flags AS flags,
                           @v1.$flags_identify_info AS flags_identify_info,
+                          @v1.$c2s_ttl AS c2s_ttl,
+                          @v1.$s2c_ttl AS s2c_ttl,
                           @v1.$security_rule_list AS security_rule_list,
                           @v1.$security_action AS security_action,
                           @v1.$monitor_rule_list AS monitor_rule_list,
                           @v1.$shaping_rule_list AS shaping_rule_list,
-                          @v1.$sc_rule_list AS sc_rule_list,
+                          @v1.$proxy_rule_list AS proxy_rule_list,
                           @v1.$statistics_rule_list AS statistics_rule_list,
+                          @v1.$sc_rule_list AS sc_rule_list,
                           @v1.$sc_rsp_raw AS sc_rsp_raw,
                           @v1.$sc_rsp_decrypted AS sc_rsp_decrypted,
-                          @v1.$proxy_rule_list AS proxy_rule_list,
                           @v1.$proxy_action AS proxy_action,
                           @v1.$proxy_pinning_status AS proxy_pinning_status,
                           @v1.$proxy_intercept_status AS proxy_intercept_status,
@@ -572,11 +717,39 @@ pipeline:
                           @v1.$fqdn_category_list AS fqdn_category_list,
                           @v1.$app_transition AS app_transition,
                           @v1.$app AS app,
+                          @v1.$app_category AS app_category,
                           @v1.$app_debug_info AS app_debug_info,
                           @v1.$app_content AS app_content,
                           @v1.$app_extra_info AS app_extra_info,
                           @v1.$ip_protocol AS ip_protocol,
                           @v1.$decoded_path AS decoded_path,
+                          @v1.$sent_pkts AS sent_pkts,
+                          @v1.$received_pkts AS received_pkts,
+                          @v1.$sent_bytes AS sent_bytes,
+                          @v1.$received_bytes AS received_bytes,
+                          @v1.$tcp_c2s_ip_fragments AS tcp_c2s_ip_fragments,
+                          @v1.$tcp_s2c_ip_fragments AS tcp_s2c_ip_fragments,
+                          @v1.$tcp_c2s_lost_bytes AS tcp_c2s_lost_bytes,
+                          @v1.$tcp_s2c_lost_bytes AS tcp_s2c_lost_bytes,
+                          @v1.$tcp_c2s_o3_pkts AS tcp_c2s_o3_pkts,
+                          @v1.$tcp_s2c_o3_pkts AS tcp_s2c_o3_pkts,
+                          @v1.$tcp_c2s_rtx_pkts AS tcp_c2s_rtx_pkts,
+                          @v1.$tcp_s2c_rtx_pkts AS tcp_s2c_rtx_pkts,
+                          @v1.$tcp_c2s_rtx_bytes AS tcp_c2s_rtx_bytes,
+                          @v1.$tcp_s2c_rtx_bytes AS tcp_s2c_rtx_bytes,
+                          @v1.$tcp_rtt_ms AS tcp_rtt_ms,
+                          @v1.$tcp_client_isn AS tcp_client_isn,
+                          @v1.$tcp_server_isn AS tcp_server_isn,
+                          @v1.$packet_capture_file AS packet_capture_file,
+                          @v1.$in_src_mac AS in_src_mac,
+                          @v1.$out_src_mac AS out_src_mac,
+                          @v1.$in_dest_mac AS in_dest_mac,
+                          @v1.$out_dest_mac AS out_dest_mac,
+                          @v1.$encapsulation AS encapsulation,
+                          @v1.$dup_traffic_flag AS dup_traffic_flag,
+                          @v1.$tunnel_id_list AS tunnel_id_list,
+                          @v1.$tunnel_endpoint_a_desc AS tunnel_endpoint_a_desc,
+                          @v1.$tunnel_endpoint_b_desc AS tunnel_endpoint_b_desc,
                           @v1.$sip_call_id AS sip_call_id,
                           @v1.$sip_originator_description AS sip_originator_description,
                           @v1.$sip_responder_description AS sip_responder_description,
@@ -597,5 +770,989 @@ pipeline:
                           @v1.$rtp_pcap_path AS rtp_pcap_path,
                           @v1.$rtp_originator_dir AS rtp_originator_dir
           - TRUNCATE v1
+  - name: voip-fusion
+    category: CORRELATE
+    cache:
+      - name: sip
+        type: VALUE
+        ttl: 6 minute
+        schema:
+          ## General
+          - name: recv_time
+            data-type: BIGINT NOT NULL
+          - name: log_id
+            data-type: BIGINT NOT NULL
+          - name: decoded_as
+            data-type: STRING NOT NULL
+          - name: session_id
+            data-type: BIGINT NOT NULL
+          - name: start_timestamp_ms
+            data-type: BIGINT NOT NULL
+          # row-time:
+          - name: start_timestamp
+            data-type: TIMESTAMP_LTZ(3)
+          - name: end_timestamp_ms
+            data-type: BIGINT
+          - name: duration_ms
+            data-type: INT
+          - name: tcp_handshake_latency_ms
+            data-type: INT
+          - name: ingestion_time
+            data-type: BIGINT
+          - name: processing_time
+            data-type: BIGINT
+          - name: insert_time
+            data-type: BIGINT
+          - name: device_id
+            data-type: STRING
+          - name: out_link_id
+            data-type: INT
+          - name: in_link_id
+            data-type: INT
+          - name: device_tag
+            data-type: STRING
+          - name: data_center
+            data-type: STRING
+          - name: device_group
+            data-type: STRING
+          - name: sled_ip
+            data-type: STRING
+          - name: address_type
+            data-type: INT
+          - name: direction
+            data-type: STRING
+          - name: vsys_id
+            data-type: INT
+          - name: t_vsys_id
+            data-type: INT
+          - name: flags
+            data-type: BIGINT
+          - name: flags_identify_info
+            data-type: STRING
+          - name: c2s_ttl
+            data-type: INT
+          - name: s2c_ttl
+            data-type: INT
+          ## Treatment
+          - name: security_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: security_action
+            data-type: STRING
+          - name: monitor_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: shaping_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: proxy_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: statistics_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: sc_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: sc_rsp_raw
+            data-type: ARRAY<BIGINT>
+          - name: sc_rsp_decrypted
+            data-type: ARRAY<BIGINT>
+          - name: proxy_action
+            data-type: STRING
+          - name: proxy_pinning_status
+            data-type: INT
+          - name: proxy_intercept_status
+            data-type: INT
+          - name: proxy_passthrough_reason
+            data-type: STRING
+          - name: proxy_client_side_latency_ms
+            data-type: INT
+          - name: proxy_server_side_latency_ms
+            data-type: INT
+          - name: proxy_client_side_version
+            data-type: STRING
+          - name: proxy_server_side_version
+            data-type: STRING
+          - name: proxy_cert_verify
+            data-type: INT
+          - name: proxy_intercept_error
+            data-type: STRING
+          - name: monitor_mirrored_pkts
+            data-type: INT
+          - name: monitor_mirrored_bytes
+            data-type: INT
+          ## Source
+          - name: client_ip
+            data-type: STRING
+          - name: client_port
+            data-type: INT
+          - name: client_os_desc
+            data-type: STRING
+          - name: client_geolocation
+            data-type: STRING
+          - name: client_country
+            data-type: STRING
+          - name: client_super_administrative_area
+            data-type: STRING
+          - name: client_administrative_area
+            data-type: STRING
+          - name: client_sub_administrative_area
+            data-type: STRING
+          - name: client_asn
+            data-type: BIGINT
+          - name: subscriber_id
+            data-type: STRING
+          - name: imei
+            data-type: STRING
+          - name: imsi
+            data-type: STRING
+          - name: phone_number
+            data-type: STRING
+          - name: apn
+            data-type: STRING
+          ## Destination
+          - name: server_ip
+            data-type: STRING
+          - name: server_port
+            data-type: INT
+          - name: server_os_desc
+            data-type: STRING
+          - name: server_geolocation
+            data-type: STRING
+          - name: server_country
+            data-type: STRING
+          - name: server_super_administrative_area
+            data-type: STRING
+          - name: server_administrative_area
+            data-type: STRING
+          - name: server_sub_administrative_area
+            data-type: STRING
+          - name: server_asn
+            data-type: BIGINT
+          - name: server_fqdn
+            data-type: STRING
+          - name: server_domain
+            data-type: STRING
+          - name: fqdn_category_list
+            data-type: ARRAY<INT>
+          ## Application
+          - name: app_transition
+            data-type: STRING
+          - name: app
+            data-type: STRING
+          - name: app_category
+            data-type: STRING
+          - name: app_debug_info
+            data-type: STRING
+          - name: app_content
+            data-type: STRING
+          - name: app_extra_info
+            data-type: STRING
+          ## Protocol
+          - name: ip_protocol
+            data-type: STRING
+          - name: decoded_path
+            data-type: STRING
+          ## Transmission
+          - name: sent_pkts
+            data-type: BIGINT
+          - name: received_pkts
+            data-type: BIGINT
+          - name: sent_bytes
+            data-type: BIGINT
+          - name: received_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_ip_fragments
+            data-type: BIGINT
+          - name: tcp_s2c_ip_fragments
+            data-type: BIGINT
+          - name: tcp_c2s_lost_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_lost_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_o3_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_o3_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_rtt_ms
+            data-type: INT
+          - name: tcp_client_isn
+            data-type: BIGINT
+          - name: tcp_server_isn
+            data-type: BIGINT
+          ## Other
+          - name: packet_capture_file
+            data-type: STRING
+          - name: in_src_mac
+            data-type: STRING
+          - name: out_src_mac
+            data-type: STRING
+          - name: in_dest_mac
+            data-type: STRING
+          - name: out_dest_mac
+            data-type: STRING
+          - name: encapsulation
+            data-type: STRING
+          - name: dup_traffic_flag
+            data-type: INT
+          - name: tunnel_id_list
+            data-type: ARRAY<BIGINT>
+          - name: tunnel_endpoint_a_desc
+            data-type: STRING
+          - name: tunnel_endpoint_b_desc
+            data-type: STRING
+          ## SIP
+          - name: sip_call_id
+            data-type: STRING
+          - name: sip_originator_description
+            data-type: STRING
+          - name: sip_responder_description
+            data-type: STRING
+          - name: sip_user_agent
+            data-type: STRING
+          - name: sip_server
+            data-type: STRING
+          - name: sip_originator_sdp_connect_ip
+            data-type: STRING
+          - name: sip_originator_sdp_media_port
+            data-type: INT
+          - name: sip_originator_sdp_media_type
+            data-type: STRING
+          - name: sip_originator_sdp_content
+            data-type: STRING
+          - name: sip_responder_sdp_connect_ip
+            data-type: STRING
+          - name: sip_responder_sdp_media_port
+            data-type: INT
+          - name: sip_responder_sdp_media_type
+            data-type: STRING
+          - name: sip_responder_sdp_content
+            data-type: STRING
+          - name: sip_duration_s
+            data-type: INT
+          - name: sip_bye
+            data-type: STRING
+          ## RTP
+          - name: rtp_payload_type_c2s
+            data-type: INT
+          - name: rtp_payload_type_s2c
+            data-type: INT
+          - name: rtp_pcap_path
+            data-type: STRING
+          - name: rtp_originator_dir
+            data-type: INT
+      - name: rtp
+        type: LIST
+        schema:
+          ## General
+          - name: recv_time
+            data-type: BIGINT NOT NULL
+          - name: log_id
+            data-type: BIGINT NOT NULL
+          - name: decoded_as
+            data-type: STRING NOT NULL
+          - name: session_id
+            data-type: BIGINT NOT NULL
+          - name: start_timestamp_ms
+            data-type: BIGINT NOT NULL
+          # row-time:
+          - name: start_timestamp
+            data-type: TIMESTAMP_LTZ(3)
+          - name: end_timestamp_ms
+            data-type: BIGINT
+          - name: duration_ms
+            data-type: INT
+          - name: tcp_handshake_latency_ms
+            data-type: INT
+          - name: ingestion_time
+            data-type: BIGINT
+          - name: processing_time
+            data-type: BIGINT
+          - name: insert_time
+            data-type: BIGINT
+          - name: device_id
+            data-type: STRING
+          - name: out_link_id
+            data-type: INT
+          - name: in_link_id
+            data-type: INT
+          - name: device_tag
+            data-type: STRING
+          - name: data_center
+            data-type: STRING
+          - name: device_group
+            data-type: STRING
+          - name: sled_ip
+            data-type: STRING
+          - name: address_type
+            data-type: INT
+          - name: direction
+            data-type: STRING
+          - name: vsys_id
+            data-type: INT
+          - name: t_vsys_id
+            data-type: INT
+          - name: flags
+            data-type: BIGINT
+          - name: flags_identify_info
+            data-type: STRING
+          - name: c2s_ttl
+            data-type: INT
+          - name: s2c_ttl
+            data-type: INT
+          ## Treatment
+          - name: security_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: security_action
+            data-type: STRING
+          - name: monitor_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: shaping_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: proxy_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: statistics_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: sc_rule_list
+            data-type: ARRAY<BIGINT>
+          - name: sc_rsp_raw
+            data-type: ARRAY<BIGINT>
+          - name: sc_rsp_decrypted
+            data-type: ARRAY<BIGINT>
+          - name: proxy_action
+            data-type: STRING
+          - name: proxy_pinning_status
+            data-type: INT
+          - name: proxy_intercept_status
+            data-type: INT
+          - name: proxy_passthrough_reason
+            data-type: STRING
+          - name: proxy_client_side_latency_ms
+            data-type: INT
+          - name: proxy_server_side_latency_ms
+            data-type: INT
+          - name: proxy_client_side_version
+            data-type: STRING
+          - name: proxy_server_side_version
+            data-type: STRING
+          - name: proxy_cert_verify
+            data-type: INT
+          - name: proxy_intercept_error
+            data-type: STRING
+          - name: monitor_mirrored_pkts
+            data-type: INT
+          - name: monitor_mirrored_bytes
+            data-type: INT
+          ## Source
+          - name: client_ip
+            data-type: STRING
+          - name: client_port
+            data-type: INT
+          - name: client_os_desc
+            data-type: STRING
+          - name: client_geolocation
+            data-type: STRING
+          - name: client_country
+            data-type: STRING
+          - name: client_super_administrative_area
+            data-type: STRING
+          - name: client_administrative_area
+            data-type: STRING
+          - name: client_sub_administrative_area
+            data-type: STRING
+          - name: client_asn
+            data-type: BIGINT
+          - name: subscriber_id
+            data-type: STRING
+          - name: imei
+            data-type: STRING
+          - name: imsi
+            data-type: STRING
+          - name: phone_number
+            data-type: STRING
+          - name: apn
+            data-type: STRING
+          ## Destination
+          - name: server_ip
+            data-type: STRING
+          - name: server_port
+            data-type: INT
+          - name: server_os_desc
+            data-type: STRING
+          - name: server_geolocation
+            data-type: STRING
+          - name: server_country
+            data-type: STRING
+          - name: server_super_administrative_area
+            data-type: STRING
+          - name: server_administrative_area
+            data-type: STRING
+          - name: server_sub_administrative_area
+            data-type: STRING
+          - name: server_asn
+            data-type: BIGINT
+          - name: server_fqdn
+            data-type: STRING
+          - name: server_domain
+            data-type: STRING
+          - name: fqdn_category_list
+            data-type: ARRAY<INT>
+          ## Application
+          - name: app_transition
+            data-type: STRING
+          - name: app
+            data-type: STRING
+          - name: app_category
+            data-type: STRING
+          - name: app_debug_info
+            data-type: STRING
+          - name: app_content
+            data-type: STRING
+          - name: app_extra_info
+            data-type: STRING
+          ## Protocol
+          - name: ip_protocol
+            data-type: STRING
+          - name: decoded_path
+            data-type: STRING
+          ## Transmission
+          - name: sent_pkts
+            data-type: BIGINT
+          - name: received_pkts
+            data-type: BIGINT
+          - name: sent_bytes
+            data-type: BIGINT
+          - name: received_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_ip_fragments
+            data-type: BIGINT
+          - name: tcp_s2c_ip_fragments
+            data-type: BIGINT
+          - name: tcp_c2s_lost_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_lost_bytes
+            data-type: BIGINT
+          - name: tcp_c2s_o3_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_o3_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_pkts
+            data-type: BIGINT
+          - name: tcp_c2s_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_s2c_rtx_bytes
+            data-type: BIGINT
+          - name: tcp_rtt_ms
+            data-type: INT
+          - name: tcp_client_isn
+            data-type: BIGINT
+          - name: tcp_server_isn
+            data-type: BIGINT
+          ## Other
+          - name: packet_capture_file
+            data-type: STRING
+          - name: in_src_mac
+            data-type: STRING
+          - name: out_src_mac
+            data-type: STRING
+          - name: in_dest_mac
+            data-type: STRING
+          - name: out_dest_mac
+            data-type: STRING
+          - name: encapsulation
+            data-type: STRING
+          - name: dup_traffic_flag
+            data-type: INT
+          - name: tunnel_id_list
+            data-type: ARRAY<BIGINT>
+          - name: tunnel_endpoint_a_desc
+            data-type: STRING
+          - name: tunnel_endpoint_b_desc
+            data-type: STRING
+          ## SIP
+          - name: sip_call_id
+            data-type: STRING
+          - name: sip_originator_description
+            data-type: STRING
+          - name: sip_responder_description
+            data-type: STRING
+          - name: sip_user_agent
+            data-type: STRING
+          - name: sip_server
+            data-type: STRING
+          - name: sip_originator_sdp_connect_ip
+            data-type: STRING
+          - name: sip_originator_sdp_media_port
+            data-type: INT
+          - name: sip_originator_sdp_media_type
+            data-type: STRING
+          - name: sip_originator_sdp_content
+            data-type: STRING
+          - name: sip_responder_sdp_connect_ip
+            data-type: STRING
+          - name: sip_responder_sdp_media_port
+            data-type: INT
+          - name: sip_responder_sdp_media_type
+            data-type: STRING
+          - name: sip_responder_sdp_content
+            data-type: STRING
+          - name: sip_duration_s
+            data-type: INT
+          - name: sip_bye
+            data-type: STRING
+          ## RTP
+          - name: rtp_payload_type_c2s
+            data-type: INT
+          - name: rtp_payload_type_s2c
+            data-type: INT
+          - name: rtp_pcap_path
+            data-type: STRING
+          - name: rtp_originator_dir
+            data-type: INT
+    where:
+      - on: sip-double-way-records.ok
+        key-by: vsys_id, SORT_ADDRESS( sip_originator_sdp_connect_ip, sip_originator_sdp_media_port, sip_responder_sdp_connect_ip, sip_responder_sdp_media_port )
+        # SIP
+        process:
+          - if: '@sip.isNull'
+            then:
+              - SET sip FROM withColumns(recv_time to rtp_originator_dir)
+          - if: '@sip.isNotNull && @rtp.isNotNull && @rtp.cardinality() > 0'
+            then:
+              - |-
+                FLAT OUTPUT ok FOR i IN rtp FROM 
+                        @i.$recv_time AS recv_time,
+                        @i.$log_id AS log_id,
+                        'VoIP' AS decode_as,
+                        @i.$session_id AS session_id,
+                        @i.$start_timestamp_ms AS start_timestamp_ms,
+                        @i.$end_timestamp_ms AS end_timestamp_ms,
+                        @i.$duration_ms AS duration_ms,
+                        @i.$tcp_handshake_latency_ms AS tcp_handshake_latency_ms,
+                        @i.$ingestion_time AS ingestion_time,
+                        @i.$processing_time AS processing_time,
+                        @i.$insert_time AS insert_time,
+                        @i.$device_id AS device_id,
+                        @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,
+                        @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,
+                        @i.$flags_identify_info AS flags_identify_info,
+                        
+                        @i.$c2s_ttl AS c2s_ttl,
+                        @i.$s2c_ttl AS s2c_ttl,
+                       
+                        @i.$client_ip AS client_ip,
+                        @i.$client_port AS client_port,
+                        @i.$client_os_desc AS client_os_desc,
+                        @i.$client_geolocation AS client_geolocation,
+                        @i.$client_country AS client_country,
+                        @i.$client_super_administrative_area AS client_super_administrative_area,
+                        @i.$client_administrative_area AS client_administrative_area,
+                        @i.$client_sub_administrative_area AS client_sub_administrative_area,
+                        @i.$client_asn AS client_asn,
+                        
+                        @i.$server_ip AS server_ip,
+                        @i.$server_port AS server_port,
+                        @i.$server_os_desc AS server_os_desc,
+                        @i.$server_geolocation AS server_geolocation,
+                        @i.$server_country AS server_country,
+                        @i.$server_super_administrative_area AS server_super_administrative_area,
+                        @i.$server_administrative_area AS server_administrative_area,
+                        @i.$server_sub_administrative_area AS server_sub_administrative_area,
+                        @i.$server_asn AS server_asn,
+                        
+                        @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.$sip_call_id AS sip_call_id,
+                        @i.$sip_originator_description AS sip_originator_description,
+                        @i.$sip_responder_description AS sip_responder_description,
+                        @i.$sip_user_agent AS sip_user_agent,
+                        @i.$sip_server AS sip_server,
+                        @i.$sip_originator_sdp_connect_ip AS sip_originator_sdp_connect_ip,
+                        @i.$sip_originator_sdp_media_port AS sip_originator_sdp_media_port,
+                        @i.$sip_originator_sdp_media_type AS sip_originator_sdp_media_type,
+                        @i.$sip_originator_sdp_content AS sip_originator_sdp_content,
+                        @i.$sip_responder_sdp_connect_ip AS sip_responder_sdp_connect_ip,
+                        @i.$sip_responder_sdp_media_port AS sip_responder_sdp_media_port,
+                        @i.$sip_responder_sdp_media_type AS sip_responder_sdp_media_type,
+                        @i.$sip_responder_sdp_content AS sip_responder_sdp_content,
+                        @i.$sip_duration_s AS sip_duration_s,
+                        @i.$sip_bye AS sip_bye,
+                        @i.$rtp_payload_type_c2s AS rtp_payload_type_c2s,
+                        @i.$rtp_payload_type_s2c AS rtp_payload_type_s2c,
+                        @i.$rtp_pcap_path AS rtp_pcap_path,
+                        ( @i.$client_ip == sip_originator_sdp_connect_ip).?(1, (@i.$client_ip == sip_responder_sdp_connect_ip).?(2, 0) ),
+                        @i.$rtp_originator_dir AS rtp_originator_dir
+              - TRUNCATE rtp
+          - SCHEDULING USING EVENT TIME FOR NOW + 6 * 60 * 1000
+      - on: rtp-records
+        key-by: vsys_id, SORT_ADDRESS( client_ip, client_port, server_ip, server_port )
+        process:
+          - if: '@sip.isNull'
+            then:
+              - APPEND rtp FROM withColumns(recv_time to rtp_originator_dir)
+          - if: '@sip.isNotNull'
+            then:
+              - |-
+                FLAT OUTPUT ok FOR i IN rtp FROM
+                                @i.$recv_time AS recv_time,
+                                @i.$log_id AS log_id,
+                                'VoIP' AS decode_as,
+                                @i.$session_id AS session_id,
+                                @i.$start_timestamp_ms AS start_timestamp_ms,
+                                @i.$end_timestamp_ms AS end_timestamp_ms,
+                                @i.$duration_ms AS duration_ms,
+                                @i.$tcp_handshake_latency_ms AS tcp_handshake_latency_ms,
+                                @i.$ingestion_time AS ingestion_time,
+                                @i.$processing_time AS processing_time,
+                                @i.$insert_time AS insert_time,
+                                @i.$device_id AS device_id,
+                                @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,
+                                @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,
+                                @i.$flags_identify_info AS flags_identify_info,
+                                
+                                @i.$c2s_ttl AS c2s_ttl,
+                                @i.$s2c_ttl AS s2c_ttl,
+                                
+                                @i.$client_ip AS client_ip,
+                                @i.$client_port AS client_port,
+                                @i.$client_os_desc AS client_os_desc,
+                                @i.$client_geolocation AS client_geolocation,
+                                @i.$client_country AS client_country,
+                                @i.$client_super_administrative_area AS client_super_administrative_area,
+                                @i.$client_administrative_area AS client_administrative_area,
+                                @i.$client_sub_administrative_area AS client_sub_administrative_area,
+                                @i.$client_asn AS client_asn,
+                                
+                                @i.$server_ip AS server_ip,
+                                @i.$server_port AS server_port,
+                                @i.$server_os_desc AS server_os_desc,
+                                @i.$server_geolocation AS server_geolocation,
+                                @i.$server_country AS server_country,
+                                @i.$server_super_administrative_area AS server_super_administrative_area,
+                                @i.$server_administrative_area AS server_administrative_area,
+                                @i.$server_sub_administrative_area AS server_sub_administrative_area,
+                                @i.$server_asn AS server_asn,
+                                
+                                @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.$sip_call_id AS sip_call_id,
+                                @i.$sip_originator_description AS sip_originator_description,
+                                @i.$sip_responder_description AS sip_responder_description,
+                                @i.$sip_user_agent AS sip_user_agent,
+                                @i.$sip_server AS sip_server,
+                                @i.$sip_originator_sdp_connect_ip AS sip_originator_sdp_connect_ip,
+                                @i.$sip_originator_sdp_media_port AS sip_originator_sdp_media_port,
+                                @i.$sip_originator_sdp_media_type AS sip_originator_sdp_media_type,
+                                @i.$sip_originator_sdp_content AS sip_originator_sdp_content,
+                                @i.$sip_responder_sdp_connect_ip AS sip_responder_sdp_connect_ip,
+                                @i.$sip_responder_sdp_media_port AS sip_responder_sdp_media_port,
+                                @i.$sip_responder_sdp_media_type AS sip_responder_sdp_media_type,
+                                @i.$sip_responder_sdp_content AS sip_responder_sdp_content,
+                                @i.$sip_duration_s AS sip_duration_s,
+                                @i.$sip_bye AS sip_bye,
+                                @i.$rtp_payload_type_c2s AS rtp_payload_type_c2s,
+                                @i.$rtp_payload_type_s2c AS rtp_payload_type_s2c,
+                                @i.$rtp_pcap_path AS rtp_pcap_path,
+                                ( @i.$client_ip == sip_originator_sdp_connect_ip).?(1, (@i.$client_ip == sip_responder_sdp_connect_ip).?(2, 0) ),
+                                @i.$rtp_originator_dir AS rtp_originator_dir
+          - SCHEDULING USING EVENT TIME FOR NOW + 6 * 60 * 1000
+    schedule:
+      - if: '@rtp.isNotNull && @rtp.cardinality > 0'
+        then:
+          - |- 
+            FLAT OUTPUT fail FOR i IN rtp FROM @i.$recv_time AS recv_time,
+                            @i.$log_id AS log_id,
+                            @i.$decoded_as AS decoded_as,
+                            @i.$session_id AS session_id,
+                            @i.$start_timestamp_ms AS start_timestamp_ms,
+                            @i.$start_timestamp AS start_timestamp,
+                            @i.$end_timestamp_ms AS end_timestamp_ms,
+                            @i.$duration_ms AS duration_ms,
+                            @i.$tcp_handshake_latency_ms AS tcp_handshake_latency_ms,
+                            @i.$ingestion_time AS ingestion_time,
+                            @i.$processing_time AS processing_time,
+                            @i.$insert_time AS insert_time,
+                            @i.$device_id AS device_id,
+                            @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,
+                            @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,
+                            @i.$flags_identify_info AS flags_identify_info,
+                            @i.$c2s_ttl AS c2s_ttl,
+                            @i.$s2c_ttl AS s2c_ttl,
+                            @i.$security_rule_list AS security_rule_list,
+                            @i.$security_action AS security_action,
+                            @i.$monitor_rule_list AS monitor_rule_list,
+                            @i.$shaping_rule_list AS shaping_rule_list,
+                            @i.$proxy_rule_list AS proxy_rule_list,
+                            @i.$statistics_rule_list AS statistics_rule_list,
+                            @i.$sc_rule_list AS sc_rule_list,
+                            @i.$sc_rsp_raw AS sc_rsp_raw,
+                            @i.$sc_rsp_decrypted AS sc_rsp_decrypted,
+                            @i.$proxy_action AS proxy_action,
+                            @i.$proxy_pinning_status AS proxy_pinning_status,
+                            @i.$proxy_intercept_status AS proxy_intercept_status,
+                            @i.$proxy_passthrough_reason AS proxy_passthrough_reason,
+                            @i.$proxy_client_side_latency_ms AS proxy_client_side_latency_ms,
+                            @i.$proxy_server_side_latency_ms AS proxy_server_side_latency_ms,
+                            @i.$proxy_client_side_version AS proxy_client_side_version,
+                            @i.$proxy_server_side_version AS proxy_server_side_version,
+                            @i.$proxy_cert_verify AS proxy_cert_verify,
+                            @i.$proxy_intercept_error AS proxy_intercept_error,
+                            @i.$monitor_mirrored_pkts AS monitor_mirrored_pkts,
+                            @i.$monitor_mirrored_bytes AS monitor_mirrored_bytes,
+                            @i.$client_ip AS client_ip,
+                            @i.$client_port AS client_port,
+                            @i.$client_os_desc AS client_os_desc,
+                            @i.$client_geolocation AS client_geolocation,
+                            @i.$client_country AS client_country,
+                            @i.$client_super_administrative_area AS client_super_administrative_area,
+                            @i.$client_administrative_area AS client_administrative_area,
+                            @i.$client_sub_administrative_area AS client_sub_administrative_area,
+                            @i.$client_asn AS client_asn,
+                            @i.$subscriber_id AS subscriber_id,
+                            @i.$imei AS imei,
+                            @i.$imsi AS imsi,
+                            @i.$phone_number AS phone_number,
+                            @i.$apn AS apn,
+                            @i.$server_ip AS server_ip,
+                            @i.$server_port AS server_port,
+                            @i.$server_os_desc AS server_os_desc,
+                            @i.$server_geolocation AS server_geolocation,
+                            @i.$server_country AS server_country,
+                            @i.$server_super_administrative_area AS server_super_administrative_area,
+                            @i.$server_administrative_area AS server_administrative_area,
+                            @i.$server_sub_administrative_area AS server_sub_administrative_area,
+                            @i.$server_asn AS server_asn,
+                            @i.$server_fqdn AS server_fqdn,
+                            @i.$server_domain AS server_domain,
+                            @i.$fqdn_category_list AS fqdn_category_list,
+                            @i.$app_transition AS app_transition,
+                            @i.$app AS app,
+                            @i.$app_category AS app_category,
+                            @i.$app_debug_info AS app_debug_info,
+                            @i.$app_content AS app_content,
+                            @i.$app_extra_info AS app_extra_info,
+                            @i.$ip_protocol AS ip_protocol,
+                            @i.$decoded_path AS decoded_path,
+                            @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.$tcp_c2s_ip_fragments AS tcp_c2s_ip_fragments,
+                            @i.$tcp_s2c_ip_fragments AS tcp_s2c_ip_fragments,
+                            @i.$tcp_c2s_lost_bytes AS tcp_c2s_lost_bytes,
+                            @i.$tcp_s2c_lost_bytes AS tcp_s2c_lost_bytes,
+                            @i.$tcp_c2s_o3_pkts AS tcp_c2s_o3_pkts,
+                            @i.$tcp_s2c_o3_pkts AS tcp_s2c_o3_pkts,
+                            @i.$tcp_c2s_rtx_pkts AS tcp_c2s_rtx_pkts,
+                            @i.$tcp_s2c_rtx_pkts AS tcp_s2c_rtx_pkts,
+                            @i.$tcp_c2s_rtx_bytes AS tcp_c2s_rtx_bytes,
+                            @i.$tcp_s2c_rtx_bytes AS tcp_s2c_rtx_bytes,
+                            @i.$tcp_rtt_ms AS tcp_rtt_ms,
+                            @i.$tcp_client_isn AS tcp_client_isn,
+                            @i.$tcp_server_isn AS tcp_server_isn,
+                            @i.$packet_capture_file AS packet_capture_file,
+                            @i.$in_src_mac AS in_src_mac,
+                            @i.$out_src_mac AS out_src_mac,
+                            @i.$in_dest_mac AS in_dest_mac,
+                            @i.$out_dest_mac AS out_dest_mac,
+                            @i.$encapsulation AS encapsulation,
+                            @i.$dup_traffic_flag AS dup_traffic_flag,
+                            @i.$tunnel_id_list AS tunnel_id_list,
+                            @i.$tunnel_endpoint_a_desc AS tunnel_endpoint_a_desc,
+                            @i.$tunnel_endpoint_b_desc AS tunnel_endpoint_b_desc,
+                            @i.$sip_call_id AS sip_call_id,
+                            @i.$sip_originator_description AS sip_originator_description,
+                            @i.$sip_responder_description AS sip_responder_description,
+                            @i.$sip_user_agent AS sip_user_agent,
+                            @i.$sip_server AS sip_server,
+                            @i.$sip_originator_sdp_connect_ip AS sip_originator_sdp_connect_ip,
+                            @i.$sip_originator_sdp_media_port AS sip_originator_sdp_media_port,
+                            @i.$sip_originator_sdp_media_type AS sip_originator_sdp_media_type,
+                            @i.$sip_originator_sdp_content AS sip_originator_sdp_content,
+                            @i.$sip_responder_sdp_connect_ip AS sip_responder_sdp_connect_ip,
+                            @i.$sip_responder_sdp_media_port AS sip_responder_sdp_media_port,
+                            @i.$sip_responder_sdp_media_type AS sip_responder_sdp_media_type,
+                            @i.$sip_responder_sdp_content AS sip_responder_sdp_content,
+                            @i.$sip_duration_s AS sip_duration_s,
+                            @i.$sip_bye AS sip_bye,
+                            @i.$rtp_payload_type_c2s AS rtp_payload_type_c2s,
+                            @i.$rtp_payload_type_s2c AS rtp_payload_type_s2c,
+                            @i.$rtp_pcap_path AS rtp_pcap_path,
+                            @i.$rtp_originator_dir AS rtp_originator_dir
+          - TRUNCATE rtp
+      - if: '@sip.isNotNull'
+        then:
+          - |-
+            OUTPUT fail FROM @sip.$recv_time AS recv_time,
+                        @sip.$log_id AS log_id,
+                        @sip.$decoded_as AS decoded_as,
+                        @sip.$session_id AS session_id,
+                        @sip.$start_timestamp_ms AS start_timestamp_ms,
+                        @sip.$start_timestamp AS start_timestamp,
+                        @sip.$end_timestamp_ms AS end_timestamp_ms,
+                        @sip.$duration_ms AS duration_ms,
+                        @sip.$tcp_handshake_latency_ms AS tcp_handshake_latency_ms,
+                        @sip.$ingestion_time AS ingestion_time,
+                        @sip.$processing_time AS processing_time,
+                        @sip.$insert_time AS insert_time,
+                        @sip.$device_id AS device_id,
+                        @sip.$out_link_id AS out_link_id,
+                        @sip.$in_link_id AS in_link_id,
+                        @sip.$device_tag AS device_tag,
+                        @sip.$data_center AS data_center,
+                        @sip.$device_group AS device_group,
+                        @sip.$sled_ip AS sled_ip,
+                        @sip.$address_type AS address_type,
+                        @sip.$direction AS direction,
+                        @sip.$vsys_id AS vsys_id,
+                        @sip.$t_vsys_id AS t_vsys_id,
+                        @sip.$flags AS flags,
+                        @sip.$flags_identify_info AS flags_identify_info,
+                        @sip.$c2s_ttl AS c2s_ttl,
+                        @sip.$s2c_ttl AS s2c_ttl,
+                        @sip.$security_rule_list AS security_rule_list,
+                        @sip.$security_action AS security_action,
+                        @sip.$monitor_rule_list AS monitor_rule_list,
+                        @sip.$shaping_rule_list AS shaping_rule_list,
+                        @sip.$proxy_rule_list AS proxy_rule_list,
+                        @sip.$statistics_rule_list AS statistics_rule_list,
+                        @sip.$sc_rule_list AS sc_rule_list,
+                        @sip.$sc_rsp_raw AS sc_rsp_raw,
+                        @sip.$sc_rsp_decrypted AS sc_rsp_decrypted,
+                        @sip.$proxy_action AS proxy_action,
+                        @sip.$proxy_pinning_status AS proxy_pinning_status,
+                        @sip.$proxy_intercept_status AS proxy_intercept_status,
+                        @sip.$proxy_passthrough_reason AS proxy_passthrough_reason,
+                        @sip.$proxy_client_side_latency_ms AS proxy_client_side_latency_ms,
+                        @sip.$proxy_server_side_latency_ms AS proxy_server_side_latency_ms,
+                        @sip.$proxy_client_side_version AS proxy_client_side_version,
+                        @sip.$proxy_server_side_version AS proxy_server_side_version,
+                        @sip.$proxy_cert_verify AS proxy_cert_verify,
+                        @sip.$proxy_intercept_error AS proxy_intercept_error,
+                        @sip.$monitor_mirrored_pkts AS monitor_mirrored_pkts,
+                        @sip.$monitor_mirrored_bytes AS monitor_mirrored_bytes,
+                        @sip.$client_ip AS client_ip,
+                        @sip.$client_port AS client_port,
+                        @sip.$client_os_desc AS client_os_desc,
+                        @sip.$client_geolocation AS client_geolocation,
+                        @sip.$client_country AS client_country,
+                        @sip.$client_super_administrative_area AS client_super_administrative_area,
+                        @sip.$client_administrative_area AS client_administrative_area,
+                        @sip.$client_sub_administrative_area AS client_sub_administrative_area,
+                        @sip.$client_asn AS client_asn,
+                        @sip.$subscriber_id AS subscriber_id,
+                        @sip.$imei AS imei,
+                        @sip.$imsi AS imsi,
+                        @sip.$phone_number AS phone_number,
+                        @sip.$apn AS apn,
+                        @sip.$server_ip AS server_ip,
+                        @sip.$server_port AS server_port,
+                        @sip.$server_os_desc AS server_os_desc,
+                        @sip.$server_geolocation AS server_geolocation,
+                        @sip.$server_country AS server_country,
+                        @sip.$server_super_administrative_area AS server_super_administrative_area,
+                        @sip.$server_administrative_area AS server_administrative_area,
+                        @sip.$server_sub_administrative_area AS server_sub_administrative_area,
+                        @sip.$server_asn AS server_asn,
+                        @sip.$server_fqdn AS server_fqdn,
+                        @sip.$server_domain AS server_domain,
+                        @sip.$fqdn_category_list AS fqdn_category_list,
+                        @sip.$app_transition AS app_transition,
+                        @sip.$app AS app,
+                        @sip.$app_category AS app_category,
+                        @sip.$app_debug_info AS app_debug_info,
+                        @sip.$app_content AS app_content,
+                        @sip.$app_extra_info AS app_extra_info,
+                        @sip.$ip_protocol AS ip_protocol,
+                        @sip.$decoded_path AS decoded_path,
+                        @sip.$sent_pkts AS sent_pkts,
+                        @sip.$received_pkts AS received_pkts,
+                        @sip.$sent_bytes AS sent_bytes,
+                        @sip.$received_bytes AS received_bytes,
+                        @sip.$tcp_c2s_ip_fragments AS tcp_c2s_ip_fragments,
+                        @sip.$tcp_s2c_ip_fragments AS tcp_s2c_ip_fragments,
+                        @sip.$tcp_c2s_lost_bytes AS tcp_c2s_lost_bytes,
+                        @sip.$tcp_s2c_lost_bytes AS tcp_s2c_lost_bytes,
+                        @sip.$tcp_c2s_o3_pkts AS tcp_c2s_o3_pkts,
+                        @sip.$tcp_s2c_o3_pkts AS tcp_s2c_o3_pkts,
+                        @sip.$tcp_c2s_rtx_pkts AS tcp_c2s_rtx_pkts,
+                        @sip.$tcp_s2c_rtx_pkts AS tcp_s2c_rtx_pkts,
+                        @sip.$tcp_c2s_rtx_bytes AS tcp_c2s_rtx_bytes,
+                        @sip.$tcp_s2c_rtx_bytes AS tcp_s2c_rtx_bytes,
+                        @sip.$tcp_rtt_ms AS tcp_rtt_ms,
+                        @sip.$tcp_client_isn AS tcp_client_isn,
+                        @sip.$tcp_server_isn AS tcp_server_isn,
+                        @sip.$packet_capture_file AS packet_capture_file,
+                        @sip.$in_src_mac AS in_src_mac,
+                        @sip.$out_src_mac AS out_src_mac,
+                        @sip.$in_dest_mac AS in_dest_mac,
+                        @sip.$out_dest_mac AS out_dest_mac,
+                        @sip.$encapsulation AS encapsulation,
+                        @sip.$dup_traffic_flag AS dup_traffic_flag,
+                        @sip.$tunnel_id_list AS tunnel_id_list,
+                        @sip.$tunnel_endpoint_a_desc AS tunnel_endpoint_a_desc,
+                        @sip.$tunnel_endpoint_b_desc AS tunnel_endpoint_b_desc,
+                        @sip.$sip_call_id AS sip_call_id,
+                        @sip.$sip_originator_description AS sip_originator_description,
+                        @sip.$sip_responder_description AS sip_responder_description,
+                        @sip.$sip_user_agent AS sip_user_agent,
+                        @sip.$sip_server AS sip_server,
+                        @sip.$sip_originator_sdp_connect_ip AS sip_originator_sdp_connect_ip,
+                        @sip.$sip_originator_sdp_media_port AS sip_originator_sdp_media_port,
+                        @sip.$sip_originator_sdp_media_type AS sip_originator_sdp_media_type,
+                        @sip.$sip_originator_sdp_content AS sip_originator_sdp_content,
+                        @sip.$sip_responder_sdp_connect_ip AS sip_responder_sdp_connect_ip,
+                        @sip.$sip_responder_sdp_media_port AS sip_responder_sdp_media_port,
+                        @sip.$sip_responder_sdp_media_type AS sip_responder_sdp_media_type,
+                        @sip.$sip_responder_sdp_content AS sip_responder_sdp_content,
+                        @sip.$sip_duration_s AS sip_duration_s,
+                        @sip.$sip_bye AS sip_bye,
+                        @sip.$rtp_payload_type_c2s AS rtp_payload_type_c2s,
+                        @sip.$rtp_payload_type_s2c AS rtp_payload_type_s2c,
+                        @sip.$rtp_pcap_path AS rtp_pcap_path,
+                        @sip.$rtp_originator_dir AS rtp_originator_dir
+          - TRUNCATE sip
 
 
diff --git a/src/main/resources/log4j2.properties b/src/main/resources/log4j2.properties
index b0b222b..32c696e 100644
--- a/src/main/resources/log4j2.properties
+++ b/src/main/resources/log4j2.properties
@@ -16,7 +16,7 @@
 # limitations under the License.
 ################################################################################
 
-rootLogger.level = DEBUG
+rootLogger.level = INFO
 rootLogger.appenderRef.console.ref = ConsoleAppender
 
 appender.console.name = ConsoleAppender