提交Live Traffic Chart重构后初版代码。（TSG-14799）

4 files changed, 435 insertions, 0 deletions

diff --git a/src/test/java/com/zdjizhi/ConventionalTest.java b/src/test/java/com/zdjizhi/ConventionalTest.java
new file mode 100644
index 0000000..7900d61
--- /dev/null
+++ b/src/test/java/com/zdjizhi/ConventionalTest.java
@@ -0,0 +1,76 @@
+package com.zdjizhi;
+
+import com.zdjizhi.common.config.GlobalConfig;
+import com.zdjizhi.utils.StringUtil;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2023/1/617:54
+ */
+public class ConventionalTest {
+
+    @Test
+    public void protocolTreeTest() {
+        String groupKey = "ETHERNET.IPv4.TCP.UNCATEGORIZED.qq_r2@4";
+        String protocol = groupKey.substring(0, groupKey.indexOf("@"));
+        System.out.println(protocol);
+        StringBuffer stringBuffer = new StringBuffer();
+        String appName = "qq_r2";
+        String[] protocolIds = protocol.split(GlobalConfig.PROTOCOL_SPLITTER);
+        for (String proto : protocolIds) {
+            if (StringUtil.isBlank(stringBuffer.toString())) {
+                stringBuffer.append(proto);
+                System.out.println(stringBuffer.toString());
+            } else {
+                stringBuffer.append(".").append(proto);
+                if (proto.equals(appName)) {
+                    System.out.println(stringBuffer.toString() + "---" + appName);
+                } else {
+                    System.out.println(stringBuffer.toString());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void SplitTest() {
+        String str = "[.]";
+        String protocol = "ETHERNET.IPv4.TCP.http.test";
+
+        System.out.println(Arrays.toString(protocol.split(str)));
+
+        String str2 = "\\.";
+        System.out.println(Arrays.toString(protocol.split(str2)));
+
+        StringBuilder stringBuilder = new StringBuilder();
+        for (int i = 0; i < protocol.split(str).length - 1; i++) {
+            String value = protocol.split(str)[i];
+            if (StringUtil.isBlank(stringBuilder.toString())) {
+                stringBuilder.append(value);
+                System.out.println(stringBuilder.toString());
+            } else {
+                stringBuilder.append(".").append(value);
+                System.out.println(stringBuilder.toString());
+            }
+        }
+        System.out.println("\n\n\n");
+        protocol = "ETHERNET.IPv4.TCP";
+        String app = "http.test";
+        System.out.println(Arrays.toString(app.split(str2)));
+        System.out.println(app.substring(app.lastIndexOf(".") + 1));
+        System.out.println(protocol.concat(".").concat(app));
+
+        System.out.println("\n\n\n");
+        app = "test";
+        System.out.println(Arrays.toString(app.split(str2)));
+        System.out.println(app.substring(app.lastIndexOf(".") + 1));
+        System.out.println(protocol.concat(".").concat(app));
+
+    }
+}
diff --git a/src/test/java/com/zdjizhi/DatasketchesTest.java b/src/test/java/com/zdjizhi/DatasketchesTest.java
new file mode 100644
index 0000000..24585fd
--- /dev/null
+++ b/src/test/java/com/zdjizhi/DatasketchesTest.java
@@ -0,0 +1,248 @@
+package com.zdjizhi;
+
+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.Union;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.Test;
+
+import java.util.*;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2023/3/217:17
+ */
+public class DatasketchesTest {
+
+    @Test
+    public void HllSketchTest() {
+        HashSet<String> strings = new HashSet<>();
+
+        HllSketch sketch = new HllSketch(12);
+
+        for (int i = 0; i < 50; i++) {
+            String ip = "192.168.1." + i;
+            sketch.update(ip);
+            strings.add(ip);
+        }
+
+        System.out.println(sketch.getEstimate() + "--" + strings.size());
+
+        HashSet<String> randomStrings = new HashSet<>();
+
+        HllSketch randomSketch = new HllSketch(12);
+
+        for (int i = 0; i < 50; i++) {
+            String ip = makeIPv4Random();
+            randomSketch.update(ip);
+            randomStrings.add(ip);
+        }
+
+        System.out.println(randomSketch.getEstimate() + "--" + randomStrings.size());
+    }
+
+    @Test
+    public void HllSketchUnionTest() {
+        HashSet<String> strings = new HashSet<>();
+
+        HllSketch sketch = new HllSketch(12);
+
+        for (int i = 0; i < 50; i++) {
+            String ip = "192.168.1." + i;
+            sketch.update(ip);
+            strings.add(ip);
+        }
+
+        HllSketch sketch2 = new HllSketch(12);
+
+        for (int i = 0; i < 10; i++) {
+            String ip = "192.168.2." + i;
+            sketch2.update(ip);
+            strings.add(ip);
+        }
+
+        Union union = new Union(12);
+
+        union.update(sketch);
+        union.update(sketch2);
+        HllSketch sketch_result = HllSketch.heapify(union.getResult().toCompactByteArray());
+
+        System.out.println(sketch.getEstimate() + "--" + strings.size());
+        System.out.println(sketch2.getEstimate() + "--" + strings.size());
+        System.out.println(sketch_result.getEstimate() + "--" + strings.size());
+    }
+
+    @Test
+    public void HllSketchDruidTest() {
+        HashMap<String, Object> dataMap = new HashMap<>();
+
+        HashSet<String> strings = new HashSet<>();
+
+        HllSketch sketch = new HllSketch(12);
+
+        for (int i = 0; i < 50; i++) {
+            String ip = "192.168.1." + i;
+            sketch.update(ip);
+            strings.add(ip);
+        }
+
+        HllSketch sketch2 = new HllSketch(12);
+
+        for (int i = 0; i < 10; i++) {
+            String ip = "192.168.2." + i;
+            sketch2.update(ip);
+            strings.add(ip);
+        }
+
+        Union union = new Union(12);
+
+        union.update(sketch);
+        union.update(sketch2);
+        HllSketch sketch_result1 = HllSketch.heapify(union.getResult().toCompactByteArray());
+
+        HllSketch sketch3 = new HllSketch(12);
+
+        for (int i = 0; i < 10; i++) {
+            String ip = "192.168.3." + i;
+            sketch3.update(ip);
+            strings.add(ip);
+        }
+
+        Union union2 = new Union(12);
+
+        union2.update(sketch_result1);
+        union2.update(sketch3);
+        HllSketch sketch_result2 = HllSketch.heapify(union2.getResult().toCompactByteArray());
+
+        System.out.println(sketch.getEstimate() + "--" + strings.size());
+        System.out.println(sketch2.getEstimate() + "--" + strings.size());
+        System.out.println(sketch3.getEstimate() + "--" + strings.size());
+        System.out.println(sketch_result1.getEstimate() + "--" + strings.size());
+        System.out.println(sketch_result2.getEstimate() + "--" + strings.size());
+
+        Result result = new Result();
+        result.setC2s_pkt_num(10);
+        result.setS2c_pkt_num(10);
+        result.setC2s_byte_num(10);
+        result.setS2c_byte_num(10);
+        result.setStat_time(1679970031);
+        result.setSchema_type("HLLSketchMergeTest");
+
+        //CompactByte
+        result.setIp_object(sketch_result2.toCompactByteArray());
+//        System.out.println(result.toString());
+        //sendMessage(JsonMapper.toJsonString(result);
+
+
+        //UpdatableByte
+        result.setIp_object(sketch_result2.toUpdatableByteArray());
+//        System.out.println(result.toString());
+        //sendMessage(JsonMapper.toJsonString(result);
+
+        //Hashmap
+        dataMap.put("app_name", "TEST");
+        dataMap.put("protocol_stack_id", "HTTP");
+        dataMap.put("vsys_id", 1);
+        dataMap.put("stat_time", 1681370100);
+        dataMap.put("client_ip_sketch", sketch_result2.toUpdatableByteArray());
+
+        System.out.println("Jackson:" + JsonMapper.toJsonString(dataMap));
+        System.out.println("FastJson2:" + JSONObject.toJSONString(dataMap));
+        System.out.println("Hutool:" + JSONUtil.toJsonStr(dataMap) + "\n\n");
+
+
+        dataMap.put("client_ip_sketch", Base64.getEncoder().encode(sketch_result2.toUpdatableByteArray()));
+        System.out.println("FastJson2 Byte(Base64):" + JSONObject.toJSONString(dataMap));
+        System.out.println("Hutool Byte(Base64):" + JSONObject.toJSONString(dataMap));
+        System.out.println(JSONUtil.toJsonStr(dataMap));
+
+
+//        sendMessage(JSONObject.toJSONString(dataMap));
+    }
+
+
+    //随机生成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_4 = new Random().nextInt(255);
+        return v4_1 + "." + v4_2 + "." + v4_3 + "." + v4_4;
+    }
+
+    private static void sendMessage(Object message) {
+        Properties props = new Properties();
+        //kafka地址
+        props.put("bootstrap.servers", "192.168.44.12:9092");
+        props.put("acks", "all");
+        props.put("retries", 0);
+        props.put("linger.ms", 1);
+        props.put("buffer.memory", 67108864);
+//        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+//        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+        KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<String, Object>(props);
+
+        kafkaProducer.send(new ProducerRecord<String, Object>("TRAFFIC-PROTOCOL-TEST", message));
+
+        kafkaProducer.close();
+    }
+}
+
+class Result {
+
+    private String schema_type;
+    private long c2s_byte_num;
+    private long c2s_pkt_num;
+    private long s2c_byte_num;
+    private long s2c_pkt_num;
+    private long stat_time;
+    private byte[] ip_object;
+
+    public void setSchema_type(String schema_type) {
+        this.schema_type = schema_type;
+    }
+
+    public void setC2s_byte_num(long c2s_byte_num) {
+        this.c2s_byte_num = c2s_byte_num;
+    }
+
+    public void setC2s_pkt_num(long c2s_pkt_num) {
+        this.c2s_pkt_num = c2s_pkt_num;
+    }
+
+    public void setS2c_byte_num(long s2c_byte_num) {
+        this.s2c_byte_num = s2c_byte_num;
+    }
+
+    public void setS2c_pkt_num(long s2c_pkt_num) {
+        this.s2c_pkt_num = s2c_pkt_num;
+    }
+
+    public void setStat_time(long stat_time) {
+        this.stat_time = stat_time;
+    }
+
+    public void setIp_object(byte[] ip_object) {
+        this.ip_object = ip_object;
+    }
+
+    @Override
+    public String toString() {
+        return "Result{" +
+                "schema_type='" + schema_type + '\'' +
+                ", c2s_byte_num=" + c2s_byte_num +
+                ", c2s_pkt_num=" + c2s_pkt_num +
+                ", s2c_byte_num=" + s2c_byte_num +
+                ", s2c_pkt_num=" + s2c_pkt_num +
+                ", stat_time=" + stat_time +
+                ", ip_object=" + Arrays.toString(ip_object) +
+                '}';
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/com/zdjizhi/FastJsonTest.java b/src/test/java/com/zdjizhi/FastJsonTest.java
new file mode 100644
index 0000000..125a7cd
--- /dev/null
+++ b/src/test/java/com/zdjizhi/FastJsonTest.java
@@ -0,0 +1,63 @@
+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 org.junit.Test;
+
+import java.util.ArrayList;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2023/4/2116:20
+ */
+public class FastJsonTest {
+
+    @Test
+    public void jsonPathTest() {
+        String dataTypeExpr = "[?(@.name = 'traffic_application_protocol_stat')]";
+        String value = "{\"fields\":{\"c2s_bytes\":120953742,\"c2s_fragments\":0,\"c2s_pkts\":513665,\"c2s_tcp_lost_bytes\":13000,\"c2s_tcp_ooorder_pkts\":7,\"c2s_tcp_retransmitted_bytes\":89555044,\"c2s_tcp_retransmitted_pkts\":240585,\"in_bytes\":64959358,\"in_pkts\":396214,\"out_bytes\":166012,\"out_pkts\":166012,\"s2c_bytes\":28703159,\"s2c_fragments\":0,\"s2c_pkts\":48561,\"s2c_tcp_lost_bytes\":0,\"s2c_tcp_ooorder_pkts\":377,\"s2c_tcp_retransmitted_bytes\":72122,\"s2c_tcp_retransmitted_pkts\":166,\"sessions\":32148},\"name\":\"traffic_application_protocol_stat\",\"tags\":{\"app_full_path\":\"dns\",\"device_id\":\"9800165603247024\",\"device_group\":\"group-xxg-tsgx\",\"vsys_id\":23,\"data_center\":\"center-xxg-tsgx\",\"protocol_label\":\"ETHERNET.IPv4.UDP\"},\"timestamp\":1682046260}";
+        JSONPath dataTypePath = JSONPath.of(dataTypeExpr);
+
+        JSONReader parser = JSONReader.of(value);
+        Object result = dataTypePath.extract(parser);
+        if (result != null) {
+            System.out.println(result.toString());
+        }
+
+
+        Object eval = JSONPath.eval(value, dataTypeExpr);
+        if (eval != null) {
+            System.out.println(eval.toString());
+        }
+
+        System.out.println(JSONPath.contains(value, dataTypeExpr));
+    }
+
+    @Test
+    public void jsonTest() {
+        String message = "{\"fields\":{\"c2s_bytes\":120953742,\"c2s_fragments\":0,\"c2s_pkts\":513665,\"c2s_tcp_lost_bytes\":13000,\"c2s_tcp_ooorder_pkts\":7,\"c2s_tcp_retransmitted_bytes\":89555044,\"c2s_tcp_retransmitted_pkts\":240585,\"in_bytes\":64959358,\"in_pkts\":396214,\"out_bytes\":166012,\"out_pkts\":166012,\"s2c_bytes\":28703159,\"s2c_fragments\":0,\"s2c_pkts\":48561,\"s2c_tcp_lost_bytes\":0,\"s2c_tcp_ooorder_pkts\":377,\"s2c_tcp_retransmitted_bytes\":72122,\"s2c_tcp_retransmitted_pkts\":166,\"sessions\":32148},\"name\":\"traffic_application_protocol_stat\",\"tags\":{\"app_full_path\":\"dns\",\"device_id\":\"9800165603247024\",\"device_group\":\"group-xxg-tsgx\",\"vsys_id\":23,\"data_center\":\"center-xxg-tsgx\",\"protocol_label\":\"ETHERNET.IPv4.UDP\"},\"timestamp\":1682046260}";
+
+        JSONObject originalLog = JSON.parseObject(message);
+        JSONObject fieldsObject = JSONObject.parseObject(originalLog.getString("fields"));
+        JSONObject tagsObject = JSONObject.parseObject(originalLog.getString("tags"));
+
+        tagsObject.putAll(fieldsObject);
+
+        AppProtocol appProtocol = JSON.to(AppProtocol.class, tagsObject);
+        System.out.println(JSONObject.toJSONString(appProtocol));
+
+        System.out.println(appProtocol.getApp_name());
+        System.out.println(appProtocol.getProtocol_stack_id());
+
+        appProtocol.setApp_name("123");
+        appProtocol.setProtocol_stack_id("abc");
+        System.out.println(appProtocol.getApp_name());
+        System.out.println(appProtocol.getProtocol_stack_id());
+    }
+}
diff --git a/src/test/java/com/zdjizhi/FlagsTest.java b/src/test/java/com/zdjizhi/FlagsTest.java
new file mode 100644
index 0000000..8310ee2
--- /dev/null
+++ b/src/test/java/com/zdjizhi/FlagsTest.java
@@ -0,0 +1,48 @@
+package com.zdjizhi;
+
+import org.junit.Test;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2023/4/1810:22
+ */
+public class FlagsTest {
+    /*
+     * 参考资料：https://juejin.cn/post/6879226834597691405
+     *
+     * 会话标记(实际存储为64位无符号整数），32-bit Field标识会话的网络行为，日志记录值和如下值通过Bitwise AND(&)操作进行查询和转换：
+     * 0x00000001 - (1) Asymmetric
+     * 0x00000002 - (2) Bulky
+     * 0x00000004 - (4) CBR Streaming
+     * 0x00000008 - (8) Client is Local
+     * 0x00000010 - (16) Server is Local
+     * 0x00000020 - (32) Download
+     * 0x00000040 - (64) Interactive
+     * 0x00000080 - (128) Inbound
+     * 0x00000100 - (256) Outbound
+     * 0x00000200 - (512) Pseudo Unidirectional
+     * 0x00000400 - (1024) Streaming
+     * 0x00000800 - (2048) Unidirectional
+     * 0x00001000 - (4096) Random looking
+     * 0x00002000 - (8192) C2S
+     * 0x00004000 - (16384) S2C
+     */
+
+    @Test
+    public void bitwiseAND() {
+        Long common_flags = 8200L;
+        Long clientIsLocal = 8L;
+        Long serverIsLocal = 16L;
+
+        System.out.println("common_flags & clientIsLocal = " + (common_flags & clientIsLocal));
+        System.out.println("common_flags & serverIsLocal = " + (common_flags & serverIsLocal)+"\n\n");
+
+        common_flags = 16400L;
+
+        System.out.println("common_flags & clientIsLocal = " + (common_flags & clientIsLocal));
+        System.out.println("common_flags & serverIsLocal = " + (common_flags & serverIsLocal));
+
+    }
+}