优化程序处理逻辑，以Metrics结构输出结果。（TSG-14799）

3 files changed, 341 insertions, 11 deletions

diff --git a/src/test/java/com/zdjizhi/AppProtocol.java b/src/test/java/com/zdjizhi/AppProtocol.java
new file mode 100644
index 0000000..a665b85
--- /dev/null
+++ b/src/test/java/com/zdjizhi/AppProtocol.java
@@ -0,0 +1,287 @@
+package com.zdjizhi;
+
+import com.alibaba.fastjson.annotation.JSONField;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi.common.pojo
+ * @Description:
+ * @date 2023/4/2116:06
+ */
+public class AppProtocol {
+    private Long timestamp;
+    private int vsys_id;
+    private String device_id;
+    private String device_group;
+    private String data_center;
+    private String protocol_stack_id;
+    private String app_name;
+    private Long sessions;
+    private Long in_bytes;
+    private Long out_bytes;
+    private Long in_pkts;
+    private Long out_pkts;
+    private Long c2s_pkts;
+    private Long s2c_pkts;
+    private Long c2s_bytes;
+    private Long s2c_bytes;
+    private Long c2s_fragments;
+    private Long s2c_fragments;
+    private Long c2s_tcp_lost_bytes;
+    private Long s2c_tcp_lost_bytes;
+    private Long c2s_tcp_ooorder_pkts;
+    private Long s2c_tcp_ooorder_pkts;
+    private Long c2s_tcp_retransmitted_pkts;
+    private Long s2c_tcp_retransmitted_pkts;
+    private Long c2s_tcp_retransmitted_bytes;
+    private Long s2c_tcp_retransmitted_bytes;
+    private String client_ip_sketch;
+
+    public AppProtocol(Long timestamp, int vsys_id, String device_id, String device_group, String data_center, String protocol_stack_id, String app_name, Long sessions, Long in_bytes, Long out_bytes, Long in_pkts, Long out_pkts, Long c2s_pkts, Long s2c_pkts, Long c2s_bytes, Long s2c_bytes, Long c2s_fragments, Long s2c_fragments, Long c2s_tcp_lost_bytes, Long s2c_tcp_lost_bytes, Long c2s_tcp_ooorder_pkts, Long s2c_tcp_ooorder_pkts, Long c2s_tcp_retransmitted_pkts, Long s2c_tcp_retransmitted_pkts, Long c2s_tcp_retransmitted_bytes, Long s2c_tcp_retransmitted_bytes, String client_ip_sketch) {
+        this.timestamp = timestamp;
+        this.vsys_id = vsys_id;
+        this.device_id = device_id;
+        this.device_group = device_group;
+        this.data_center = data_center;
+        this.protocol_stack_id = protocol_stack_id;
+        this.app_name = app_name;
+        this.sessions = sessions;
+        this.in_bytes = in_bytes;
+        this.out_bytes = out_bytes;
+        this.in_pkts = in_pkts;
+        this.out_pkts = out_pkts;
+        this.c2s_pkts = c2s_pkts;
+        this.s2c_pkts = s2c_pkts;
+        this.c2s_bytes = c2s_bytes;
+        this.s2c_bytes = s2c_bytes;
+        this.c2s_fragments = c2s_fragments;
+        this.s2c_fragments = s2c_fragments;
+        this.c2s_tcp_lost_bytes = c2s_tcp_lost_bytes;
+        this.s2c_tcp_lost_bytes = s2c_tcp_lost_bytes;
+        this.c2s_tcp_ooorder_pkts = c2s_tcp_ooorder_pkts;
+        this.s2c_tcp_ooorder_pkts = s2c_tcp_ooorder_pkts;
+        this.c2s_tcp_retransmitted_pkts = c2s_tcp_retransmitted_pkts;
+        this.s2c_tcp_retransmitted_pkts = s2c_tcp_retransmitted_pkts;
+        this.c2s_tcp_retransmitted_bytes = c2s_tcp_retransmitted_bytes;
+        this.s2c_tcp_retransmitted_bytes = s2c_tcp_retransmitted_bytes;
+        this.client_ip_sketch = client_ip_sketch;
+    }
+
+    public Long getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(Long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public int getVsys_id() {
+        return vsys_id;
+    }
+
+    public void setVsys_id(int vsys_id) {
+        this.vsys_id = vsys_id;
+    }
+
+    public String getDevice_id() {
+        return device_id;
+    }
+
+    public void setDevice_id(String device_id) {
+        this.device_id = device_id;
+    }
+
+    public String getDevice_group() {
+        return device_group;
+    }
+
+    public void setDevice_group(String device_group) {
+        this.device_group = device_group;
+    }
+
+    public String getData_center() {
+        return data_center;
+    }
+
+    public void setData_center(String data_center) {
+        this.data_center = data_center;
+    }
+
+    public String getProtocol_stack_id() {
+        return protocol_stack_id;
+    }
+
+    @JSONField(name = "protocol_label")
+    public void setProtocol_stack_id(String protocol_stack_id) {
+        this.protocol_stack_id = protocol_stack_id;
+    }
+
+    public String getApp_name() {
+        return app_name;
+    }
+
+    @JSONField(name = "app_full_path")
+    public void setApp_name(String app_name) {
+        this.app_name = app_name;
+    }
+
+    public Long getSessions() {
+        return sessions;
+    }
+
+    public void setSessions(Long sessions) {
+        this.sessions = sessions;
+    }
+
+    public Long getIn_bytes() {
+        return in_bytes;
+    }
+
+    public void setIn_bytes(Long in_bytes) {
+        this.in_bytes = in_bytes;
+    }
+
+    public Long getOut_bytes() {
+        return out_bytes;
+    }
+
+    public void setOut_bytes(Long out_bytes) {
+        this.out_bytes = out_bytes;
+    }
+
+    public Long getIn_pkts() {
+        return in_pkts;
+    }
+
+    public void setIn_pkts(Long in_pkts) {
+        this.in_pkts = in_pkts;
+    }
+
+    public Long getOut_pkts() {
+        return out_pkts;
+    }
+
+    public void setOut_pkts(Long out_pkts) {
+        this.out_pkts = out_pkts;
+    }
+
+    public Long getC2s_pkts() {
+        return c2s_pkts;
+    }
+
+    public void setC2s_pkts(Long c2s_pkts) {
+        this.c2s_pkts = c2s_pkts;
+    }
+
+    public Long getS2c_pkts() {
+        return s2c_pkts;
+    }
+
+    public void setS2c_pkts(Long s2c_pkts) {
+        this.s2c_pkts = s2c_pkts;
+    }
+
+    public Long getC2s_bytes() {
+        return c2s_bytes;
+    }
+
+    public void setC2s_bytes(Long c2s_bytes) {
+        this.c2s_bytes = c2s_bytes;
+    }
+
+    public Long getS2c_bytes() {
+        return s2c_bytes;
+    }
+
+    public void setS2c_bytes(Long s2c_bytes) {
+        this.s2c_bytes = s2c_bytes;
+    }
+
+    public Long getC2s_fragments() {
+        return c2s_fragments;
+    }
+
+    public void setC2s_fragments(Long c2s_fragments) {
+        this.c2s_fragments = c2s_fragments;
+    }
+
+    public Long getS2c_fragments() {
+        return s2c_fragments;
+    }
+
+    public void setS2c_fragments(Long s2c_fragments) {
+        this.s2c_fragments = s2c_fragments;
+    }
+
+    public Long getC2s_tcp_lost_bytes() {
+        return c2s_tcp_lost_bytes;
+    }
+
+    public void setC2s_tcp_lost_bytes(Long c2s_tcp_lost_bytes) {
+        this.c2s_tcp_lost_bytes = c2s_tcp_lost_bytes;
+    }
+
+    public Long getS2c_tcp_lost_bytes() {
+        return s2c_tcp_lost_bytes;
+    }
+
+    public void setS2c_tcp_lost_bytes(Long s2c_tcp_lost_bytes) {
+        this.s2c_tcp_lost_bytes = s2c_tcp_lost_bytes;
+    }
+
+    public Long getC2s_tcp_ooorder_pkts() {
+        return c2s_tcp_ooorder_pkts;
+    }
+
+    public void setC2s_tcp_ooorder_pkts(Long c2s_tcp_ooorder_pkts) {
+        this.c2s_tcp_ooorder_pkts = c2s_tcp_ooorder_pkts;
+    }
+
+    public Long getS2c_tcp_ooorder_pkts() {
+        return s2c_tcp_ooorder_pkts;
+    }
+
+    public void setS2c_tcp_ooorder_pkts(Long s2c_tcp_ooorder_pkts) {
+        this.s2c_tcp_ooorder_pkts = s2c_tcp_ooorder_pkts;
+    }
+
+    public Long getC2s_tcp_retransmitted_pkts() {
+        return c2s_tcp_retransmitted_pkts;
+    }
+
+    public void setC2s_tcp_retransmitted_pkts(Long c2s_tcp_retransmitted_pkts) {
+        this.c2s_tcp_retransmitted_pkts = c2s_tcp_retransmitted_pkts;
+    }
+
+    public Long getS2c_tcp_retransmitted_pkts() {
+        return s2c_tcp_retransmitted_pkts;
+    }
+
+    public void setS2c_tcp_retransmitted_pkts(Long s2c_tcp_retransmitted_pkts) {
+        this.s2c_tcp_retransmitted_pkts = s2c_tcp_retransmitted_pkts;
+    }
+
+    public Long getC2s_tcp_retransmitted_bytes() {
+        return c2s_tcp_retransmitted_bytes;
+    }
+
+    public void setC2s_tcp_retransmitted_bytes(Long c2s_tcp_retransmitted_bytes) {
+        this.c2s_tcp_retransmitted_bytes = c2s_tcp_retransmitted_bytes;
+    }
+
+    public Long getS2c_tcp_retransmitted_bytes() {
+        return s2c_tcp_retransmitted_bytes;
+    }
+
+    public void setS2c_tcp_retransmitted_bytes(Long s2c_tcp_retransmitted_bytes) {
+        this.s2c_tcp_retransmitted_bytes = s2c_tcp_retransmitted_bytes;
+    }
+
+    public String getClient_ip_sketch() {
+        return client_ip_sketch;
+    }
+
+    public void setClient_ip_sketch(String client_ip_sketch) {
+        this.client_ip_sketch = client_ip_sketch;
+    }
+}
diff --git a/src/test/java/com/zdjizhi/DatasketchesTest.java b/src/test/java/com/zdjizhi/DatasketchesTest.java
index 24585fd..09cb291 100644
--- a/src/test/java/com/zdjizhi/DatasketchesTest.java
+++ b/src/test/java/com/zdjizhi/DatasketchesTest.java
@@ -4,11 +4,13 @@ import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson2.*;
 import com.zdjizhi.utils.JsonMapper;
 import org.apache.datasketches.hll.HllSketch;
+import org.apache.datasketches.hll.TgtHllType;
 import org.apache.datasketches.hll.Union;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.Test;
 
+import java.lang.instrument.Instrumentation;
 import java.util.*;
 
 /**
@@ -161,16 +163,47 @@ public class DatasketchesTest {
         System.out.println("Hutool Byte(Base64):" + JSONObject.toJSONString(dataMap));
         System.out.println(JSONUtil.toJsonStr(dataMap));
 
-
 //        sendMessage(JSONObject.toJSONString(dataMap));
     }
 
+    @Test
+    public void HllSketchStorageTest() {
+        TgtHllType hllType = TgtHllType.HLL_4;
+//        TgtHllType hllType = TgtHllType.HLL_6;
+//        TgtHllType hllType = TgtHllType.HLL_8;
+
+        HllSketch sketch4 = new HllSketch(4,hllType);
+        HllSketch sketch8 = new HllSketch(8,hllType);
+        HllSketch sketch12 = new HllSketch(12,hllType);
+        HllSketch sketch16 = new HllSketch(16,hllType);
+        HllSketch sketch21 = new HllSketch(21,hllType);
+
+        HashSet<String> IPSet = new HashSet<>();
+
+        for (int i = 0; i < 500000; i++) {
+            String ip = makeIPv4Random();
+            IPSet.add(ip);
+            sketch4.update(ip);
+            sketch8.update(ip);
+            sketch12.update(ip);
+            sketch16.update(ip);
+            sketch21.update(ip);
+        }
+        System.out.println(IPSet.size());
+        System.out.println(sketch4.toString());
+        System.out.println(sketch8.toString());
+        System.out.println(sketch12.toString());
+        System.out.println(sketch16.toString());
+        System.out.println(sketch21.toString());
+
+    }
+
 
     //随机生成ip
     private static String makeIPv4Random() {
         int v4_1 = new Random().nextInt(255) + 1;
-        int v4_2 = new Random().nextInt(255);
-        int v4_3 = new Random().nextInt(255);
+        int v4_2 = new Random().nextInt(100);
+        int v4_3 = new Random().nextInt(100);
         int v4_4 = new Random().nextInt(255);
         return v4_1 + "." + v4_2 + "." + v4_3 + "." + v4_4;
     }
diff --git a/src/test/java/com/zdjizhi/FastJsonTest.java b/src/test/java/com/zdjizhi/FastJsonTest.java
index 125a7cd..7d50fc1 100644
--- a/src/test/java/com/zdjizhi/FastJsonTest.java
+++ b/src/test/java/com/zdjizhi/FastJsonTest.java
@@ -1,15 +1,8 @@
 package com.zdjizhi;
 
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONObject;
-import com.alibaba.fastjson2.JSONPath;
-import com.alibaba.fastjson2.JSONReader;
-import com.zdjizhi.common.pojo.AppProtocol;
-import com.zdjizhi.utils.StringUtil;
+import com.alibaba.fastjson2.*;
 import org.junit.Test;
 
-import java.util.ArrayList;
-
 /**
  * @author qidaijie
  * @Package com.zdjizhi
@@ -37,6 +30,22 @@ public class FastJsonTest {
         }
 
         System.out.println(JSONPath.contains(value, dataTypeExpr));
+
+        String message = "{\"common_schema_type\":\"HTTP\",\"common_sessions\":1,\"http_request_line\":\"GET sampleFile.html HTTP/1.1\",\"http_host\":\"www.texaslotto.com\",\"http_url\":\"www.texaslotto.com/sampleFile.html\",\"http_user_agent\":\"xPTS/2.0\",\"http_response_line\":\"HTTP/1.1 200 OK\",\"http_isn\":1953597368,\"http_proxy_flag\":0,\"http_version\":\"http1\",\"http_response_latency_ms\":1,\"http_session_duration_ms\":2,\"http_response_content_type\":\"text/html\",\"http_sequence\":80,\"common_protocol_label\":\"ETHERNET.IPv4.UDP.GTP.IPv4.TCP\",\"common_c2s_byte_diff\":17200,\"common_c2s_pkt_diff\":120,\"common_s2c_byte_diff\":16490,\"common_s2c_pkt_diff\":81,\"common_c2s_ipfrag_num\":0,\"common_s2c_ipfrag_num\":0,\"common_first_ttl\":64,\"common_c2s_tcp_unorder_num\":0,\"common_s2c_tcp_unorder_num\":0,\"common_c2s_tcp_lostlen\":0,\"common_s2c_tcp_lostlen\":0,\"common_c2s_pkt_retrans\":0,\"common_s2c_pkt_retrans\":0,\"common_c2s_byte_retrans\":0,\"common_s2c_byte_retrans\":0,\"common_flags\":24720,\"common_flags_identify_info\":\"{\\\"Server is Local\\\":1,\\\"Inbound\\\":201,\\\"C2S\\\":1,\\\"S2C\\\":2}\",\"common_direction\":73,\"common_app_full_path\":\"http\",\"common_app_label\":\"http\",\"common_tcp_client_isn\":1953597368,\"common_tcp_server_isn\":1950649408,\"common_server_ip\":\"192.50.199.25\",\"common_client_ip\":\"192.50.146.197\",\"common_server_port\":80,\"common_client_port\":22533,\"common_stream_dir\":3,\"common_address_type\":4,\"common_address_list\":\"IPv4_TCP<22533-80-192.50.146.197-192.50.199.25>|GTP<111001144-851056526>|IPv4_UDP<2152-2152-192.50.235.220-192.50.135.83>|MAC<000c299b2fa4-000c2915b4f4>\",\"common_start_time\":1680475247,\"common_end_time\":1680475247,\"common_con_duration_ms\":23,\"common_s2c_pkt_num\":81,\"common_s2c_byte_num\":16490,\"common_c2s_pkt_num\":120,\"common_c2s_byte_num\":17200,\"common_establish_latency_ms\":2,\"common_client_location\":\"日本.Unknown.Unknown\",\"common_server_location\":\"日本.Unknown.Unknown\",\"common_service_category\":[6223,6219,5093,5089],\"common_apn\":\"cmiott.owflr.mcto60g.com\",\"common_imsi\":\"460045157091460\",\"common_imei\":\"8626070583005833\",\"common_phone_number\":\"861440152028973\",\"common_tunnel_endpoint_a_desc\":\"test_50_gtp\",\"common_tunnel_endpoint_b_desc\":\"test_50_gtp\",\"common_tunnels\":[{\"tunnels_schema_type\":\"GTP\",\"gtp_a2b_teid\":111001144,\"gtp_b2a_teid\":851056526,\"gtp_endpoint_a_ip\":\"192.50.235.220\",\"gtp_endpoint_b_ip\":\"192.50.135.83\",\"gtp_endpoint_a_port\":2152,\"gtp_endpoint_b_port\":2152},{\"tunnels_schema_type\":\"MULTIPATH_ETHERNET\",\"c2s_source_mac\":\"00:0c:29:9b:2f:a4\",\"c2s_destination_mac\":\"00:0c:29:15:b4:f4\",\"s2c_source_mac\":\"00:0c:29:15:b4:f4\",\"s2c_destination_mac\":\"00:0c:29:9b:2f:a4\"}],\"common_stream_trace_id\":\"578829229323951427\",\"common_l4_protocol\":\"IPv4_TCP\",\"common_sled_ip\":\"192.168.40.161\",\"common_device_id\":\"unknown\",\"common_device_tag\":\"{\\\"tags\\\":[{\\\"tag\\\":\\\"device_group\\\",\\\"value\\\":\\\"group-xxg-7400\\\"},{\\\"tag\\\":\\\"data_center\\\",\\\"value\\\":\\\"center-xxg-7400\\\"}]}\",\"common_t_vsys_id\":1,\"common_policy_id\":0,\"common_service\":2,\"common_action\":0,\"common_vsys_id\":1}";
+        JSONObject originalLog = JSON.parseObject(message);
+
+        String value1 = originalLog.getString("common_device_tag");
+        System.out.println(JSONPath.eval(value1, "$.tags[?(@.tag=='data_center')][0].value"));
+        System.out.println(JSONPath.of("$.tags[?(@.tag=='data_center')].value").extract(JSONReader.of(value1)));
+
+        JSONArray jsonArray = originalLog.getJSONObject("common_device_tag").getJSONArray("tags");
+        for (Object array : jsonArray) {
+            JSONObject object = JSON.parseObject(array.toString());
+            if ("device_group".equals(object.getString("tag"))) {
+                System.out.println(object.getString("value"));
+            }
+        }
+
     }
 
     @Test
@@ -60,4 +69,5 @@ public class FastJsonTest {
         System.out.println(appProtocol.getApp_name());
         System.out.println(appProtocol.getProtocol_stack_id());
     }
+
 }