修复动态获取nacos配置

集成nacos，修复依赖冲突，启动报错
2022-06-28 11:15:49 +08:00 · 2022-06-28 11:14:54 +08:00 · 2022-06-22 15:58:49 +08:00 · 2022-06-20 18:15:24 +08:00 · 2022-06-13 18:00:57 +08:00 · 2022-05-31 18:00:36 +08:00
22 changed files with 993 additions and 271 deletions
--- a/pom.xml
+++ b/pom.xml
@@ -102,6 +102,13 @@

    <dependencies>

+        <!-- https://mvnrepository.com/artifact/org.jasypt/jasypt -->
+        <dependency>
+            <groupId>org.jasypt</groupId>
+            <artifactId>jasypt</artifactId>
+            <version>1.9.3</version>
+        </dependency>
+
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
@@ -141,6 +148,10 @@
                    <artifactId>jdk.tools</artifactId>
                    <groupId>jdk.tools</groupId>
                </exclusion>
+                <exclusion>
+                    <artifactId>guava</artifactId>
+                    <groupId>com.google.guava</groupId>
+                </exclusion>
            </exclusions>
        </dependency>

@@ -149,6 +160,7 @@
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.2.3</version>
+            <!--<scope>provided</scope>-->
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-log4j12</artifactId>
@@ -198,7 +210,7 @@
        <dependency>
            <groupId>com.zdjizhi</groupId>
            <artifactId>galaxy</artifactId>
-            <version>1.0.6</version>
+            <version>1.0.8</version>
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-log4j12</artifactId>
@@ -214,6 +226,51 @@
                </exclusion>
            </exclusions>
        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.nacos</groupId>
+            <artifactId>nacos-client</artifactId>
+            <version>1.2.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-databind</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-codec</groupId>
+                    <artifactId>commons-codec</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/commons-codec/commons-codec -->
+        <dependency>
+            <groupId>commons-codec</groupId>
+            <artifactId>commons-codec</artifactId>
+            <version>1.11</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.9.10</version>
+        </dependency>
+
+        <!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>22.0</version>
+        </dependency>
    </dependencies>


--- a/src/main/java/com/zdjizhi/common/CommonConfig.java
+++ b/src/main/java/com/zdjizhi/common/CommonConfig.java
@@ -1,12 +1,21 @@
 package com.zdjizhi.common;

 import com.zdjizhi.utils.CommonConfigurations;
+import com.zdjizhi.utils.NacosUtils;
+import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;

 /**
- * Created by wk on 2021/1/6.
+ * @author wlh
+ * @date 2021/1/6
 */
 public class CommonConfig {

+    private static StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
+
+    static {
+        encryptor.setPassword("galaxy");
+    }
+
    public static final int STREAM_EXECUTION_ENVIRONMENT_PARALLELISM = CommonConfigurations.getIntProperty("stream.execution.environment.parallelism");
    public static final String STREAM_EXECUTION_JOB_NAME = CommonConfigurations.getStringProperty("stream.execution.job.name");

@@ -27,6 +36,7 @@ public class CommonConfig {

    public static final String HBASE_BASELINE_TABLE_NAME = CommonConfigurations.getStringProperty("hbase.baseline.table.name");
    public static final int HBASE_BASELINE_TOTAL_NUM = CommonConfigurations.getIntProperty("hbase.baseline.total.num");
+    public static final int HBASE_BASELINE_TTL = CommonConfigurations.getIntProperty("hbase.baseline.ttl");

    public static final int FLINK_FIRST_AGG_PARALLELISM = CommonConfigurations.getIntProperty("flink.first.agg.parallelism");
    public static final int FLINK_DETECTION_MAP_PARALLELISM = CommonConfigurations.getIntProperty("flink.detection.map.parallelism");
@@ -39,13 +49,14 @@ public class CommonConfig {

    public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");

-    public static final int SENSITIVITY_THRESHOLD = CommonConfigurations.getIntProperty("sensitivity.threshold");
-
-    public static final double BASELINE_SESSIONS_MINOR_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.minor.threshold");
-    public static final double BASELINE_SESSIONS_WARNING_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.warning.threshold");
-    public static final double BASELINE_SESSIONS_MAJOR_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.major.threshold");
-    public static final double BASELINE_SESSIONS_SEVERE_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.severe.threshold");
-    public static final double BASELINE_SESSIONS_CRITICAL_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.critical.threshold");
+//    public static final int STATIC_SENSITIVITY_THRESHOLD = NacosUtils.getIntProperty("static.sensitivity.threshold");
+//    public static final double BASELINE_SENSITIVITY_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sensitivity.threshold");
+//
+//    public static final double BASELINE_SESSIONS_MINOR_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sessions.minor.threshold");
+//    public static final double BASELINE_SESSIONS_WARNING_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sessions.warning.threshold");
+//    public static final double BASELINE_SESSIONS_MAJOR_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sessions.major.threshold");
+//    public static final double BASELINE_SESSIONS_SEVERE_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sessions.severe.threshold");
+//    public static final double BASELINE_SESSIONS_CRITICAL_THRESHOLD = NacosUtils.getDoubleProperty("baseline.sessions.critical.threshold");

    public static final String BIFANG_SERVER_URI = CommonConfigurations.getStringProperty("bifang.server.uri");
    public static final String BIFANG_SERVER_TOKEN = CommonConfigurations.getStringProperty("bifang.server.token");
@@ -53,6 +64,8 @@ public class CommonConfig {
    public static final String BIFANG_SERVER_LOGIN_PATH = CommonConfigurations.getStringProperty("bifang.server.login.path");
    public static final String BIFANG_SERVER_POLICY_THRESHOLD_PATH = CommonConfigurations.getStringProperty("bifang.server.policy.threshold.path");

+    public static final String BIFANG_SERVER_POLICY_VSYSID_PATH = CommonConfigurations.getStringProperty("bifang.server.policy.vaysid.path");
+
    public static final int HTTP_POOL_MAX_CONNECTION = CommonConfigurations.getIntProperty("http.pool.max.connection");
    public static final int HTTP_POOL_MAX_PER_ROUTE = CommonConfigurations.getIntProperty("http.pool.max.per.route");
    public static final int HTTP_POOL_REQUEST_TIMEOUT = CommonConfigurations.getIntProperty("http.pool.request.timeout");
@@ -63,7 +76,21 @@ public class CommonConfig {
    public static final int BASELINE_THRESHOLD_SCHEDULE_DAYS = CommonConfigurations.getIntProperty("baseline.threshold.schedule.days");

    public static final String SASL_JAAS_CONFIG_USER = CommonConfigurations.getStringProperty("sasl.jaas.config.user");
-    public static final String SASL_JAAS_CONFIG_PASSWORD = CommonConfigurations.getStringProperty("sasl.jaas.config.password");
+    public static final String SASL_JAAS_CONFIG_PASSWORD = encryptor.decrypt(CommonConfigurations.getStringProperty("sasl.jaas.config.password"));

    public static final int SASL_JAAS_CONFIG_FLAG = CommonConfigurations.getIntProperty("sasl.jaas.config.flag");
+
+    public static void main(String[] args) {
+        StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
+        // 配置加密解密的密码/salt值
+        encryptor.setPassword("galaxy");
+        // 对"raw_password"进行加密：S5kR+Y7CI8k7MaecZpde25yK8NKUnd6p
+//        String password = "galaxy2019";
+        String password = "nacos";
+        String encPwd = encryptor.encrypt(password);
+        System.out.println(encPwd);
+        // 再进行解密：raw_password
+        String rawPwd = encryptor.decrypt(encPwd);
+        System.out.println(rawPwd);
+    }
 }
--- a/src/main/java/com/zdjizhi/common/DosBaselineThreshold.java
+++ b/src/main/java/com/zdjizhi/common/DosBaselineThreshold.java
@@ -0,0 +1,63 @@
+package com.zdjizhi.common;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Objects;
+
+public class DosBaselineThreshold implements Serializable {
+    private ArrayList<Integer> session_rate;
+    private Integer session_rate_baseline_type;
+    private Integer session_rate_default_value;
+
+    public ArrayList<Integer> getSession_rate() {
+        return session_rate;
+    }
+
+    public void setSession_rate(ArrayList<Integer> session_rate) {
+        this.session_rate = session_rate;
+    }
+
+    public Integer getSession_rate_baseline_type() {
+        return session_rate_baseline_type;
+    }
+
+    public void setSession_rate_baseline_type(Integer session_rate_baseline_type) {
+        this.session_rate_baseline_type = session_rate_baseline_type;
+    }
+
+    public Integer getSession_rate_default_value() {
+        return session_rate_default_value;
+    }
+
+    public void setSession_rate_default_value(Integer session_rate_default_value) {
+        this.session_rate_default_value = session_rate_default_value;
+    }
+
+    @Override
+    public String toString() {
+        return "DosBaselineThreshold{" +
+                "session_rate=" + session_rate +
+                ", session_rate_baseline_type=" + session_rate_baseline_type +
+                ", session_rate_default_value=" + session_rate_default_value +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosBaselineThreshold)) {
+            return false;
+        }
+        DosBaselineThreshold that = (DosBaselineThreshold) o;
+        return Objects.equals(getSession_rate(), that.getSession_rate()) &&
+                Objects.equals(getSession_rate_baseline_type(), that.getSession_rate_baseline_type()) &&
+                Objects.equals(getSession_rate_default_value(), that.getSession_rate_default_value());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getSession_rate(), getSession_rate_baseline_type(), getSession_rate_default_value());
+    }
+}
--- a/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java
+++ b/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java
@@ -2,7 +2,11 @@ package com.zdjizhi.common;

 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Objects;

+/**
+ * @author wlh
+ */
 public class DosDetectionThreshold implements Serializable {
    private String profileId;
    private String attackType;
@@ -27,6 +31,30 @@ public class DosDetectionThreshold implements Serializable {
                '}';
    }

+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        DosDetectionThreshold threshold = (DosDetectionThreshold) o;
+        return packetsPerSec == threshold.packetsPerSec &&
+                bitsPerSec == threshold.bitsPerSec &&
+                sessionsPerSec == threshold.sessionsPerSec &&
+                isValid == threshold.isValid &&
+                Objects.equals(profileId, threshold.profileId) &&
+                Objects.equals(attackType, threshold.attackType) &&
+                Objects.equals(serverIpList, threshold.serverIpList) &&
+                Objects.equals(serverIpAddr, threshold.serverIpAddr);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(profileId, attackType, serverIpList, serverIpAddr, packetsPerSec, bitsPerSec, sessionsPerSec, isValid);
+    }
+
    public String getProfileId() {
        return profileId;
    }
--- a/src/main/java/com/zdjizhi/common/DosEventLog.java
+++ b/src/main/java/com/zdjizhi/common/DosEventLog.java
@@ -1,6 +1,7 @@
 package com.zdjizhi.common;

 import java.io.Serializable;
+import java.util.Objects;

 public class DosEventLog implements Serializable {

@@ -140,4 +141,33 @@ public class DosEventLog implements Serializable {
                ", bit_rate=" + bit_rate +
                '}';
    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosEventLog)) {
+            return false;
+        }
+        DosEventLog that = (DosEventLog) o;
+        return getLog_id() == that.getLog_id() &&
+                getStart_time() == that.getStart_time() &&
+                getEnd_time() == that.getEnd_time() &&
+                getSession_rate() == that.getSession_rate() &&
+                getPacket_rate() == that.getPacket_rate() &&
+                getBit_rate() == that.getBit_rate() &&
+                Objects.equals(getAttack_type(), that.getAttack_type()) &&
+                Objects.equals(getSeverity(), that.getSeverity()) &&
+                Objects.equals(getConditions(), that.getConditions()) &&
+                Objects.equals(getDestination_ip(), that.getDestination_ip()) &&
+                Objects.equals(getDestination_country(), that.getDestination_country()) &&
+                Objects.equals(getSource_ip_list(), that.getSource_ip_list()) &&
+                Objects.equals(getSource_country_list(), that.getSource_country_list());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getLog_id(), getStart_time(), getEnd_time(), getAttack_type(), getSeverity(), getConditions(), getDestination_ip(), getDestination_country(), getSource_ip_list(), getSource_country_list(), getSession_rate(), getPacket_rate(), getBit_rate());
+    }
 }
--- a/src/main/java/com/zdjizhi/common/DosMetricsLog.java
+++ b/src/main/java/com/zdjizhi/common/DosMetricsLog.java
@@ -1,6 +1,7 @@
 package com.zdjizhi.common;

 import java.io.Serializable;
+import java.util.Objects;

 public class DosMetricsLog implements Serializable {

@@ -80,4 +81,27 @@ public class DosMetricsLog implements Serializable {
                ", partition_num=" + partition_num +
                '}';
    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosMetricsLog)) {
+            return false;
+        }
+        DosMetricsLog that = (DosMetricsLog) o;
+        return getSketch_start_time() == that.getSketch_start_time() &&
+                getSession_rate() == that.getSession_rate() &&
+                getPacket_rate() == that.getPacket_rate() &&
+                getBit_rate() == that.getBit_rate() &&
+                getPartition_num() == that.getPartition_num() &&
+                Objects.equals(getAttack_type(), that.getAttack_type()) &&
+                Objects.equals(getDestination_ip(), that.getDestination_ip());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getSketch_start_time(), getAttack_type(), getDestination_ip(), getSession_rate(), getPacket_rate(), getBit_rate(), getPartition_num());
+    }
 }
--- a/src/main/java/com/zdjizhi/common/DosSketchLog.java
+++ b/src/main/java/com/zdjizhi/common/DosSketchLog.java
@@ -1,6 +1,7 @@
 package com.zdjizhi.common;

 import java.io.Serializable;
+import java.util.Objects;

 public class DosSketchLog implements Serializable {

@@ -31,6 +32,32 @@ public class DosSketchLog implements Serializable {
                '}';
    }

+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosSketchLog)) {
+            return false;
+        }
+        DosSketchLog sketchLog = (DosSketchLog) o;
+        return getSketch_start_time() == sketchLog.getSketch_start_time() &&
+                getSketch_duration() == sketchLog.getSketch_duration() &&
+                getSketch_sessions() == sketchLog.getSketch_sessions() &&
+                getSketch_packets() == sketchLog.getSketch_packets() &&
+                getSketch_bytes() == sketchLog.getSketch_bytes() &&
+                Objects.equals(getCommon_sled_ip(), sketchLog.getCommon_sled_ip()) &&
+                Objects.equals(getCommon_data_center(), sketchLog.getCommon_data_center()) &&
+                Objects.equals(getAttack_type(), sketchLog.getAttack_type()) &&
+                Objects.equals(getSource_ip(), sketchLog.getSource_ip()) &&
+                Objects.equals(getDestination_ip(), sketchLog.getDestination_ip());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getCommon_sled_ip(), getCommon_data_center(), getSketch_start_time(), getSketch_duration(), getAttack_type(), getSource_ip(), getDestination_ip(), getSketch_sessions(), getSketch_packets(), getSketch_bytes());
+    }
+
    public String getCommon_sled_ip() {
        return common_sled_ip;
    }
--- a/src/main/java/com/zdjizhi/common/DosVsysId.java
+++ b/src/main/java/com/zdjizhi/common/DosVsysId.java
@@ -0,0 +1,22 @@
+package com.zdjizhi.common;
+
+import java.util.Objects;
+
+public class DosVsysId {
+    private int vsysId;
+
+    public int getVsysId() {
+        return vsysId;
+    }
+
+    public void setVsysId(int vsysId) {
+        this.vsysId = vsysId;
+    }
+
+    @Override
+    public String toString() {
+        return "DosVsysId{" +
+                "vsysId=" + vsysId +
+                '}';
+    }
+}
--- a/src/main/java/com/zdjizhi/etl/DosDetection.java
+++ b/src/main/java/com/zdjizhi/etl/DosDetection.java
@@ -1,19 +1,13 @@
 package com.zdjizhi.etl;

-import com.zdjizhi.common.CommonConfig;
-import com.zdjizhi.common.DosDetectionThreshold;
-import com.zdjizhi.common.DosEventLog;
-import com.zdjizhi.common.DosSketchLog;
-import com.zdjizhi.utils.DateUtils;
-import com.zdjizhi.utils.HbaseUtils;
-import com.zdjizhi.utils.IpUtils;
-import com.zdjizhi.utils.SnowflakeId;
+import com.zdjizhi.common.*;
+import com.zdjizhi.utils.*;
 import inet.ipaddr.IPAddress;
 import inet.ipaddr.IPAddressString;
 import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.text.StrBuilder;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.shaded.guava18.com.google.common.collect.TreeRangeMap;
 import org.slf4j.Logger;
@@ -32,25 +26,31 @@ import java.util.concurrent.TimeUnit;
 public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {

    private static final Logger logger = LoggerFactory.getLogger(DosDetection.class);
-    private static Map<String, Map<String, Tuple2<ArrayList<Integer>, Integer>>> baselineMap = new HashMap<>();
-    private final static int BASELINE_SIZE = 144;
+    private static Map<String, Map<String, DosBaselineThreshold>> baselineMap = new HashMap<>();
    private final static NumberFormat PERCENT_INSTANCE = NumberFormat.getPercentInstance();
-    private TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> thresholdRangeMap;
+    private HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> thresholdRangeMap;
+
+    private final static int BASELINE_SIZE = 144;
+    private final static int STATIC_CONDITION_TYPE = 1;
+    private final static int BASELINE_CONDITION_TYPE = 2;
+    private final static int SENSITIVITY_CONDITION_TYPE = 3;
+
+    private final static String SESSIONS_TAG = "sessions";
+    private final static String PACKETS_TAG = "packets";
+    private final static String BITS_TAG = "bits";
+
+    private final static int OTHER_BASELINE_TYPE = 3;

    @Override
    public void open(Configuration parameters) {
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(2,
                new BasicThreadFactory.Builder().namingPattern("Dos-Detection-%d").daemon(true).build());
        try {
-            executorService.scheduleAtFixedRate(() -> {
-                //do something
-                thresholdRangeMap = ParseStaticThreshold.createStaticThreshold();
-            }, 0, CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
+            executorService.scheduleAtFixedRate(() -> thresholdRangeMap = ParseStaticThreshold.createStaticThreshold(), 0,
+                    CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);

-            executorService.scheduleAtFixedRate(() -> {
-                //do something
-                baselineMap = HbaseUtils.readFromHbase();
-            }, 0, CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
+            executorService.scheduleAtFixedRate(() -> baselineMap = ParseBaselineThreshold.readFromHbase(), 0,
+                    CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
        } catch (Exception e) {
            logger.error("定时器任务执行失败", e);
        }
@@ -64,16 +64,14 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
            String destinationIp = value.getDestination_ip();
            String attackType = value.getAttack_type();
            IPAddress destinationIpAddress = new IPAddressString(destinationIp).getAddress();
-            Map<String, DosDetectionThreshold> thresholdMap = thresholdRangeMap.get(destinationIpAddress);
+            DosDetectionThreshold threshold = thresholdRangeMap.getOrDefault(attackType, TreeRangeMap.create()).get(destinationIpAddress);
            logger.debug("当前判断IP：{}, 类型: {}", destinationIp, attackType);
-            if (baselineMap != null && baselineMap.containsKey(destinationIp) && thresholdMap == null) {
-                finalResult = getDosEventLogByBaseline(value, destinationIp, attackType).f1;
-            } else if (baselineMap != null && !baselineMap.containsKey(destinationIp) && thresholdMap != null) {
-                finalResult = getDosEventLogByStaticThreshold(value, thresholdMap).f1;
-            } else if (baselineMap != null && baselineMap.containsKey(destinationIp) && thresholdMap != null) {
-                Tuple2<Severity, DosEventLog> eventLogByBaseline = getDosEventLogByBaseline(value, destinationIp, attackType);
-                Tuple2<Severity, DosEventLog> eventLogByStaticThreshold = getDosEventLogByStaticThreshold(value, thresholdMap);
-                finalResult = mergeFinalResult(eventLogByBaseline, eventLogByStaticThreshold);
+            if (threshold == null && baselineMap.containsKey(destinationIp)) {
+                finalResult = getDosEventLogByBaseline(value);
+            } else if (threshold == null && !baselineMap.containsKey(destinationIp)) {
+                finalResult = getDosEventLogBySensitivityThreshold(value);
+            } else if (threshold != null) {
+                finalResult = getDosEventLogByStaticThreshold(value, threshold);
            } else {
                logger.debug("未获取到当前server IP：{} 类型 {} 静态阈值 和 baseline", destinationIp, attackType);
            }
@@ -84,72 +82,77 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
        return finalResult;
    }

-    private DosEventLog mergeFinalResult(Tuple2<Severity, DosEventLog> eventLogByBaseline, Tuple2<Severity, DosEventLog> eventLogByStaticThreshold) {
-        if (eventLogByBaseline.f0.score > eventLogByStaticThreshold.f0.score) {
-            logger.info("merge eventLogByBaseline {} \neventLogByStaticThreshold {}",eventLogByBaseline,eventLogByStaticThreshold);
-            return mergeCondition(eventLogByBaseline.f1, eventLogByStaticThreshold.f1);
-        } else {
-            logger.info("merge eventLogByStaticThreshold {} \neventLogByBaseline {}",eventLogByStaticThreshold,eventLogByBaseline);
-            return mergeCondition(eventLogByStaticThreshold.f1, eventLogByBaseline.f1);
-        }
-    }
-
-    private DosEventLog mergeCondition(DosEventLog log1, DosEventLog log2) {
-        if (log1 != null && log2 != null) {
-            String conditions1 = log1.getConditions();
-            String conditions2 = log2.getConditions();
-            log1.setConditions(conditions1 + " and " + conditions2);
-        }else if (log1 == null && log2 != null){
-            log1 = log2;
-        }
-        return log1;
-    }
-
-    private Tuple2<Severity, DosEventLog> getDosEventLogByBaseline(DosSketchLog value, String destinationIp, String attackType) {
-        Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
-        Integer base = getBaseValue(floodTypeTup, value);
+    private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value) {
+        DosEventLog result = null;
        long sketchSessions = value.getSketch_sessions();
-        return sketchSessions > CommonConfig.SENSITIVITY_THRESHOLD ? getDosEventLog(value, base, sketchSessions - base, "baseline"):Tuple2.of(Severity.NORMAL, null);
-    }
-
-    private Tuple2<Severity, DosEventLog> getDosEventLogByStaticThreshold(DosSketchLog value, Map<String, DosDetectionThreshold> thresholdMap) {
-        Tuple2<Severity, DosEventLog> result = Tuple2.of(Severity.NORMAL, null);
-        String attackType = value.getAttack_type();
-        if (thresholdMap.containsKey(attackType)) {
-            DosDetectionThreshold threshold = thresholdMap.get(attackType);
-            long base = threshold.getSessionsPerSec();
-            long diff = value.getSketch_sessions() - base;
-            result = getDosEventLog(value, base, diff, "static");
+        if (sketchSessions > NacosUtils.getIntProperty("static.sensitivity.threshold")) {
+            long diff = sketchSessions - NacosUtils.getIntProperty("static.sensitivity.threshold");
+            result = getDosEventLog(value, NacosUtils.getIntProperty("static.sensitivity.threshold"), diff, SENSITIVITY_CONDITION_TYPE, SESSIONS_TAG);
+            result.setSeverity(Severity.MAJOR.severity);
        }
        return result;
    }

-    private Tuple2<Severity, DosEventLog> getDosEventLog(DosSketchLog value, long base, long diff, String tag) {
+    private DosEventLog getDosEventLogByBaseline(DosSketchLog value) {
        DosEventLog result = null;
        String destinationIp = value.getDestination_ip();
        String attackType = value.getAttack_type();
-        Severity severity = Severity.NORMAL;
-        if (diff > 0 && base != 0) {
-            double percent = getDiffPercent(diff, base);
-            severity = judgeSeverity(percent);
-            if (severity != Severity.NORMAL) {
-                result = getResult(value, severity, percent, tag);
-                logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,日志详情\n {}", destinationIp,attackType,base,percent,result);
-            } else {
-                logger.debug("当前server IP：{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value.toString());
-            }
+        long sketchSessions = value.getSketch_sessions();
+        if (sketchSessions > NacosUtils.getIntProperty("static.sensitivity.threshold")) {
+            DosBaselineThreshold dosBaselineThreshold = baselineMap.get(destinationIp).get(attackType);
+            Integer base = getBaseValue(dosBaselineThreshold, value);
+            long diff = sketchSessions - base;
+            result = getDosEventLog(value, base, diff, BASELINE_CONDITION_TYPE, SESSIONS_TAG);
        }
-        return Tuple2.of(severity, result);
+        return result;
    }

-    private DosEventLog getResult(DosSketchLog value, Severity severity, double percent, String tag) {
+    private DosEventLog getDosEventLogByStaticThreshold(DosSketchLog value, DosDetectionThreshold threshold) {
+        long base = threshold.getSessionsPerSec();
+        long diff = value.getSketch_sessions() - base;
+        DosEventLog result = getDosEventLog(value, base, diff, STATIC_CONDITION_TYPE, SESSIONS_TAG);
+        if (result == null) {
+            base = threshold.getPacketsPerSec();
+            diff = value.getSketch_packets() - base;
+            result = getDosEventLog(value, base, diff, STATIC_CONDITION_TYPE, PACKETS_TAG);
+            if (result == null) {
+                base = threshold.getBitsPerSec();
+                diff = value.getSketch_bytes() - base;
+                result = getDosEventLog(value, base, diff, STATIC_CONDITION_TYPE, BITS_TAG);
+            }
+        }
+        return result;
+    }
+
+    private DosEventLog getDosEventLog(DosSketchLog value, long base, long diff, int type, String tag) {
+        DosEventLog result = null;
+        String destinationIp = value.getDestination_ip();
+        String attackType = value.getAttack_type();
+        if (diff > 0 && base != 0) {
+            double percent = getDiffPercent(diff, base);
+            Severity severity = judgeSeverity(percent);
+            if (severity != Severity.NORMAL) {
+                if (type == BASELINE_CONDITION_TYPE && percent < NacosUtils.getDoubleProperty("baseline.sensitivity.threshold")) {
+                    logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过基线敏感阈值,日志详情\n{}", destinationIp, attackType, base, percent, value);
+                } else {
+                    result = getResult(value, base, severity, percent+1, type, tag);
+                    logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,基于{}:{}检测,日志详情\n {}", destinationIp,attackType,base,percent,type,tag,result);
+                }
+            } else {
+                logger.debug("当前server IP:{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value);
+            }
+        }
+        return result;
+    }
+
+    private DosEventLog getResult(DosSketchLog value, long base, Severity severity, double percent, int type, String tag) {
        DosEventLog dosEventLog = new DosEventLog();
        dosEventLog.setLog_id(SnowflakeId.generateId());
        dosEventLog.setStart_time(value.getSketch_start_time());
-        dosEventLog.setEnd_time(value.getSketch_start_time() + CommonConfig.FLINK_WINDOW_MAX_TIME);
+        dosEventLog.setEnd_time(value.getSketch_start_time() + value.getSketch_duration());
        dosEventLog.setAttack_type(value.getAttack_type());
        dosEventLog.setSeverity(severity.severity);
-        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent), value.getSketch_sessions(), tag));
+        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent), base, value.getSketch_sessions(), type, tag));
        dosEventLog.setDestination_ip(value.getDestination_ip());
        dosEventLog.setDestination_country(IpUtils.ipLookup.countryLookup(value.getDestination_ip()));
        String ipList = value.getSource_ip();
@@ -161,12 +164,13 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
        return dosEventLog;
    }

-    private Integer getBaseValue(Tuple2<ArrayList<Integer>, Integer> floodTypeTup, DosSketchLog value) {
+    private Integer getBaseValue(DosBaselineThreshold dosBaselineThreshold, DosSketchLog value) {
        Integer base = 0;
        try {
-            if (floodTypeTup != null) {
-                ArrayList<Integer> baselines = floodTypeTup.f0;
-                Integer defaultVaule = floodTypeTup.f1;
+            if (dosBaselineThreshold != null) {
+                ArrayList<Integer> baselines = dosBaselineThreshold.getSession_rate();
+                Integer defaultVaule = dosBaselineThreshold.getSession_rate_default_value();
+                Integer sessionRateBaselineType = dosBaselineThreshold.getSession_rate_baseline_type();
                if (baselines != null && baselines.size() == BASELINE_SIZE) {
                    int timeIndex = getCurrentTimeIndex(value.getSketch_start_time());
                    base = baselines.get(timeIndex);
@@ -174,6 +178,9 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
                        logger.debug("获取到当前IP: {},类型: {} baseline值为0,替换为P95观测值{}", value.getDestination_ip(), value.getAttack_type(), defaultVaule);
                        base = defaultVaule;
                    }
+                    if (sessionRateBaselineType == OTHER_BASELINE_TYPE && base < NacosUtils.getIntProperty("static.sensitivity.threshold")){
+                        base = NacosUtils.getIntProperty("static.sensitivity.threshold");
+                    }
                }
            }
        } catch (Exception e) {
@@ -182,46 +189,87 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
        return base;
    }

-    private String getConditions(String percent, long sessions, String tag) {
-        switch (tag) {
-            case "baseline":
-                return "sessions > " + percent + " of baseline";
-            case "static":
-                return "sessions > " + sessions + " sessions/s";
+    private String getConditions(String percent, long base, long sessions, int type, String tag) {
+        switch (type) {
+            case STATIC_CONDITION_TYPE:
+                return new StrBuilder()
+                        .append("Rate > ")
+                        .append(base).append(" ")
+                        .append(tag).append("/s")
+                        .toString();
+            case BASELINE_CONDITION_TYPE:
+                return new StrBuilder()
+                        .append(tag).append(" > ")
+                        .append(percent).append(" of baseline")
+                        .toString();
+            case SENSITIVITY_CONDITION_TYPE:
+                return new StrBuilder()
+                        .append(sessions).append(" ")
+                        .append(tag).append("/s Unusually high ")
+                        .append(StringUtils.capitalize(tag))
+                        .toString();
            default:
-                return null;
+                throw new IllegalArgumentException("Illegal Argument type:" + type + ", known types = [1,2,3]");
        }
    }

    private String getSourceCountryList(String sourceIpList) {
-        String[] ipArr = sourceIpList.split(",");
-        HashSet<String> countrySet = new HashSet<>();
-        for (String ip : ipArr) {
-            countrySet.add(IpUtils.ipLookup.countryLookup(ip));
+        if (StringUtil.isNotBlank(sourceIpList)) {
+            String countryList;
+            try {
+                String[] ipArr = sourceIpList.split(",");
+                HashSet<String> countrySet = new HashSet<>();
+                for (String ip : ipArr) {
+                    String country = IpUtils.ipLookup.countryLookup(ip);
+                    if (StringUtil.isNotBlank(country)){
+                        countrySet.add(country);
+                    }
+                }
+                countryList = StringUtils.join(countrySet, ", ");
+                return countryList;
+            } catch (Exception e) {
+                logger.error("{} source IP lists 获取国家失败", sourceIpList, e);
+                return StringUtil.EMPTY;
+            }
+        } else {
+            throw new IllegalArgumentException("Illegal Argument sourceIpList = null");
        }
-        return StringUtils.join(countrySet, ",");
    }

    private int getCurrentTimeIndex(long sketchStartTime) {
-        long currentDayTime = DateUtils.getTimeFloor(new Date(sketchStartTime * 1000L), "P1D").getTime()/1000;
-        long indexLong = (sketchStartTime - currentDayTime) / 600;
-        return Integer.parseInt(Long.toString(indexLong));
+        int index = 0;
+        try {
+            long currentDayTime = DateUtils.getTimeFloor(new Date(sketchStartTime * 1000L), "P1D").getTime() / 1000;
+            long indexLong = (sketchStartTime - currentDayTime) / (86400 / BASELINE_SIZE);
+            index = Integer.parseInt(Long.toString(indexLong));
+        } catch (Exception e) {
+            logger.error("获取time index失败", e);
+        }
+        return index;
+    }
+
+    public static void main(String[] args) {
+//        System.out.println(new DosDetection().getSourceCountryList("192.0.2.3,138.199.14.31,255.255.255.255,121.14.89.209," +
+//                "23.200.74.224,161.117.68.253"));
+//        DosDetection dosDetection = new DosDetection();
+//        System.out.println(dosDetection.judgeSeverity(dosDetection.getDiffPercent(499, 1000)));
+
    }

    private Double getDiffPercent(long diff, long base) {
-        return BigDecimal.valueOf((float)diff/base).setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
+        return BigDecimal.valueOf((float) diff / base).setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
    }

    private Severity judgeSeverity(double diffPercent) {
-        if (diffPercent >= CommonConfig.BASELINE_SESSIONS_MINOR_THRESHOLD && diffPercent < CommonConfig.BASELINE_SESSIONS_WARNING_THRESHOLD) {
+        if (diffPercent >= NacosUtils.getDoubleProperty("baseline.sessions.minor.threshold") && diffPercent < NacosUtils.getDoubleProperty("baseline.sessions.warning.threshold")) {
            return Severity.MINOR;
-        } else if (diffPercent >= CommonConfig.BASELINE_SESSIONS_WARNING_THRESHOLD && diffPercent < CommonConfig.BASELINE_SESSIONS_MAJOR_THRESHOLD) {
+        } else if (diffPercent >= NacosUtils.getDoubleProperty("baseline.sessions.warning.threshold") && diffPercent < NacosUtils.getDoubleProperty("baseline.sessions.major.threshold")) {
            return Severity.WARNING;
-        } else if (diffPercent >= CommonConfig.BASELINE_SESSIONS_MAJOR_THRESHOLD && diffPercent < CommonConfig.BASELINE_SESSIONS_SEVERE_THRESHOLD) {
+        } else if (diffPercent >= NacosUtils.getDoubleProperty("baseline.sessions.major.threshold") && diffPercent < NacosUtils.getDoubleProperty("baseline.sessions.severe.threshold")) {
            return Severity.MAJOR;
-        } else if (diffPercent >= CommonConfig.BASELINE_SESSIONS_SEVERE_THRESHOLD && diffPercent < CommonConfig.BASELINE_SESSIONS_CRITICAL_THRESHOLD) {
+        } else if (diffPercent >= NacosUtils.getDoubleProperty("baseline.sessions.severe.threshold") && diffPercent < NacosUtils.getDoubleProperty("baseline.sessions.critical.threshold")) {
            return Severity.SEVERE;
-        } else if (diffPercent >= CommonConfig.BASELINE_SESSIONS_CRITICAL_THRESHOLD) {
+        } else if (diffPercent >= NacosUtils.getDoubleProperty("baseline.sessions.critical.threshold")) {
            return Severity.CRITICAL;
        } else {
            return Severity.NORMAL;
@@ -232,24 +280,22 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
        /**
         * 判断严重程度枚举类型
         */
-        CRITICAL("Critical", 5),
-        SEVERE("Severe", 4),
-        MAJOR("Major", 3),
-        WARNING("Warning", 2),
-        MINOR("Minor", 1),
-        NORMAL("Normal", 0);
+        CRITICAL("Critical"),
+        SEVERE("Severe"),
+        MAJOR("Major"),
+        WARNING("Warning"),
+        MINOR("Minor"),
+        NORMAL("Normal");

        private final String severity;
-        private final int score;

        @Override
        public String toString() {
            return this.severity;
        }

-        Severity(String severity, int score) {
+        Severity(String severity) {
            this.severity = severity;
-            this.score = score;
        }

    }
--- a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
@@ -21,6 +21,9 @@ import static com.zdjizhi.sink.OutputStreamSink.outputTag;
 public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosSketchLog, Tuple2<String,String>, TimeWindow> {

    private static final Logger logger = LoggerFactory.getLogger(EtlProcessFunction.class);
+    private static final String EMPTY_SOURCE_IP_IPV4 = "0.0.0.0";
+    private static final String EMPTY_SOURCE_IP_IPV6 = "::";
+
    @Override
    public void process(Tuple2<String, String> keys,
                        Context context, Iterable<DosSketchLog> elements,
@@ -60,27 +63,30 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
    }

    private Tuple6<Long, Long, Long,String,Long,Long> sketchAggregate(Iterable<DosSketchLog> elements){
-        int cnt = 1;
        long sessions = 0;
        long packets = 0 ;
        long bytes = 0;
-        long startTime = 0;
+        long startTime = System.currentTimeMillis()/1000;
+        long endTime = System.currentTimeMillis()/1000;
        long duration = 0;
        HashSet<String> sourceIpSet = new HashSet<>();
        try {
            for (DosSketchLog newSketchLog : elements){
-                sessions += newSketchLog.getSketch_sessions();
-                packets += newSketchLog.getSketch_packets();
-                bytes += newSketchLog.getSketch_bytes();
-                startTime = newSketchLog.getSketch_start_time();
-                duration = newSketchLog.getSketch_duration();
-                cnt += 1;
-                if (sourceIpSet.size() < CommonConfig.SOURCE_IP_LIST_LIMIT){
-                    sourceIpSet.add(newSketchLog.getSource_ip());
+                String sourceIp = newSketchLog.getSource_ip();
+                if (StringUtils.equals(sourceIp,EMPTY_SOURCE_IP_IPV4) || StringUtils.equals(sourceIp,EMPTY_SOURCE_IP_IPV6)){
+                    sessions += newSketchLog.getSketch_sessions();
+                    packets += newSketchLog.getSketch_packets();
+                    bytes += newSketchLog.getSketch_bytes();
+                    startTime = newSketchLog.getSketch_start_time() > startTime ? startTime : newSketchLog.getSketch_start_time();
+                    endTime = newSketchLog.getSketch_start_time() > endTime ? newSketchLog.getSketch_start_time() : endTime;
+                    duration = endTime - startTime == 0 ? 5 : endTime - startTime;
+                }else {
+                    if (sourceIpSet.size() < CommonConfig.SOURCE_IP_LIST_LIMIT){
+                        sourceIpSet.add(sourceIp);
+                    }
                }
            }
            String sourceIpList = StringUtils.join(sourceIpSet, ",");
-//            return Tuple6.of(sessions/cnt/duration,packets/cnt/duration,bytes/cnt/duration,sourceIpList,startTime,duration);
            return Tuple6.of(sessions/CommonConfig.FLINK_WINDOW_MAX_TIME,packets/CommonConfig.FLINK_WINDOW_MAX_TIME,
                    bytes*8/CommonConfig.FLINK_WINDOW_MAX_TIME,sourceIpList,startTime,duration);
        }catch (Exception e){
--- a/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java
+++ b/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java
@@ -0,0 +1,109 @@
+package com.zdjizhi.etl;
+
+import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.DosBaselineThreshold;
+import com.zdjizhi.utils.DateUtils;
+import com.zdjizhi.utils.HbaseUtils;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+
+public class ParseBaselineThreshold {
+
+    private static final Logger logger = LoggerFactory.getLogger(ParseBaselineThreshold.class);
+    private static ArrayList<String> floodTypeList = new ArrayList<>();
+
+    private static Table table = null;
+    private static Scan scan = null;
+
+    static {
+        floodTypeList.add("TCP SYN Flood");
+        floodTypeList.add("UDP Flood");
+        floodTypeList.add("ICMP Flood");
+        floodTypeList.add("DNS Flood");
+    }
+
+    private static void prepareHbaseEnv() throws IOException {
+        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
+
+        config.set("hbase.zookeeper.quorum", CommonConfig.HBASE_ZOOKEEPER_QUORUM);
+        config.set("hbase.client.retries.number", "3");
+        config.set("hbase.bulkload.retries.number", "3");
+        config.set("zookeeper.recovery.retry", "3");
+        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, CommonConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
+        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, CommonConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
+
+        TableName tableName = TableName.valueOf(CommonConfig.HBASE_BASELINE_TABLE_NAME);
+        Connection conn = ConnectionFactory.createConnection(config);
+        table = conn.getTable(tableName);
+        long currentTimeMillis = System.currentTimeMillis();
+        scan = new Scan()
+                .setAllowPartialResults(true)
+                .setTimeRange(DateUtils.getSomeDate(new Date(currentTimeMillis), Math.negateExact(CommonConfig.HBASE_BASELINE_TTL)).getTime(), currentTimeMillis)
+                .setLimit(CommonConfig.HBASE_BASELINE_TOTAL_NUM);
+        logger.info("连接hbase成功，正在读取baseline数据");
+    }
+
+
+    static Map<String, Map<String, DosBaselineThreshold>> readFromHbase() {
+        Map<String, Map<String, DosBaselineThreshold>> baselineMap = new HashMap<>();
+        try {
+            prepareHbaseEnv();
+            logger.info("开始读取baseline数据");
+            ResultScanner rs = table.getScanner(scan);
+            for (Result result : rs) {
+                Map<String, DosBaselineThreshold> floodTypeMap = new HashMap<>();
+                String rowkey = Bytes.toString(result.getRow());
+                for (String type:floodTypeList){
+                    DosBaselineThreshold baselineThreshold = new DosBaselineThreshold();
+                    ArrayList<Integer> sessionRate = HbaseUtils.getArraylist(result, type, "session_rate");
+                    if (sessionRate != null && !sessionRate.isEmpty()){
+                        Integer defaultValue = HbaseUtils.getIntegerValue(result, type, "session_rate_default_value");
+                        Integer rateBaselineType = HbaseUtils.getIntegerValue(result, type, "session_rate_baseline_type");
+                        baselineThreshold.setSession_rate(sessionRate);
+                        baselineThreshold.setSession_rate_default_value(defaultValue);
+                        baselineThreshold.setSession_rate_baseline_type(rateBaselineType);
+                        floodTypeMap.put(type,baselineThreshold);
+                    }
+                }
+                baselineMap.put(rowkey, floodTypeMap);
+            }
+            logger.info("格式化baseline数据成功，读取IP共：{}", baselineMap.size());
+        } catch (Exception e) {
+            logger.error("读取hbase数据失败", e);
+        }
+        return baselineMap;
+    }
+
+    public static void main(String[] args) {
+        long currentTimeMillis = System.currentTimeMillis();
+        long p200D = DateUtils.getSomeDate(new Date(currentTimeMillis), Math.negateExact(CommonConfig.HBASE_BASELINE_TTL)).getTime();
+        System.out.println(p200D);
+        System.out.println(currentTimeMillis);
+        System.out.println(currentTimeMillis - p200D);
+
+
+        Map<String, Map<String, DosBaselineThreshold>> baselineMap = readFromHbase();
+        Set<String> keySet = baselineMap.keySet();
+        for (String key : keySet) {
+            Map<String, DosBaselineThreshold> stringTuple2Map = baselineMap.get(key);
+            Set<String> strings = stringTuple2Map.keySet();
+            for (String s:strings){
+                DosBaselineThreshold dosBaselineThreshold = stringTuple2Map.get(s);
+                System.out.println(key+"---"+s+"---"+dosBaselineThreshold);
+            }
+        }
+        System.out.println(baselineMap.size());
+    }
+
+
+
+
+}
--- a/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
+++ b/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
@@ -3,6 +3,7 @@ package com.zdjizhi.etl;
 import com.fasterxml.jackson.databind.JavaType;
 import com.zdjizhi.common.CommonConfig;
 import com.zdjizhi.common.DosDetectionThreshold;
+import com.zdjizhi.common.DosVsysId;
 import com.zdjizhi.utils.HttpClientUtils;
 import com.zdjizhi.utils.JsonMapper;
 import inet.ipaddr.IPAddress;
@@ -29,6 +30,7 @@ public class ParseStaticThreshold {
    private static JsonMapper jsonMapperInstance = JsonMapper.getInstance();
    private static JavaType hashmapJsonType = jsonMapperInstance.createCollectionType(HashMap.class, String.class, Object.class);
    private static JavaType thresholdType = jsonMapperInstance.createCollectionType(ArrayList.class, DosDetectionThreshold.class);
+    private static JavaType vsysIDType = jsonMapperInstance.createCollectionType(ArrayList.class, DosVsysId.class);

    static {
        //加载加密登录密码
@@ -42,7 +44,7 @@ public class ParseStaticThreshold {
        String psw = HttpClientUtils.ERROR_MESSAGE;
        try {
            URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
-            HashMap<String, String> parms = new HashMap<>();
+            HashMap<String, Object> parms = new HashMap<>();
            parms.put("password", "admin");
            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_ENCRYPTPWD_PATH, parms);
            String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build());
@@ -75,7 +77,7 @@ public class ParseStaticThreshold {
        try {
            if (!HttpClientUtils.ERROR_MESSAGE.equals(encryptpwd)) {
                URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
-                HashMap<String, String> parms = new HashMap<>();
+                HashMap<String, Object> parms = new HashMap<>();
                parms.put("username", "admin");
                parms.put("password", encryptpwd);
                HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_LOGIN_PATH, parms);
@@ -99,35 +101,95 @@ public class ParseStaticThreshold {
    }

    /**
-     * 获取静态阈值配置列表
+     * 获取vsysId配置列表
     *
-     * @return thresholds
+     * @return vsysIdList
     */
-    private static ArrayList<DosDetectionThreshold> getDosDetectionThreshold() {
-        ArrayList<DosDetectionThreshold> thresholds = null;
+    private static ArrayList<DosVsysId> getVsysId() {
+        ArrayList<DosVsysId> vsysIdList = null;
        try {
            URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
-            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, null);
+            HashMap<String, Object> parms = new HashMap<>();
+            parms.put("pageSize", -1);
+            parms.put("orderBy", "vsysId desc");
+            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_VSYSID_PATH, parms);
            String token = CommonConfig.BIFANG_SERVER_TOKEN;
            if (!HttpClientUtils.ERROR_MESSAGE.equals(token)) {
                BasicHeader authorization = new BasicHeader("Authorization", token);
-                String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization);
+                BasicHeader authorization1 = new BasicHeader("Content-Type", "application/x-www-form-urlencoded");
+                String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization, authorization1);
                if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
                    HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
                    boolean success = (boolean) resposeMap.get("success");
                    String msg = resposeMap.get("msg").toString();
                    if (success) {
                        HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
-                        thresholds = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(data.get("list")), thresholdType);
-                        logger.info("获取到静态阈值配置{}条", thresholds.size());
+                        Object list = data.get("list");
+                        if (list != null) {
+                            vsysIdList = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(list), vsysIDType);
+                            logger.info("获取到vsysId{}条", vsysIdList.size());
+                        } else {
+                            logger.warn("vsysIdList为空");
+                        }
                    } else {
                        logger.error(msg);
                    }
                }
            }
+        } catch (Exception e) {
+            logger.error("获取vsysId失败,请检查bifang服务或登录配置信息 ", e);
+        }
+        return vsysIdList;
+    }
+
+    /**
+     * 根据vsysId获取静态阈值配置列表
+     *
+     * @return thresholds
+     */
+    private static ArrayList<DosDetectionThreshold> getDosDetectionThreshold() {
+        ArrayList<DosDetectionThreshold> thresholds = null;
+//        ArrayList<DosVsysId> vsysId = getVsysId();
+        try {
+//            if (vsysId != null){
+//             for (DosVsysId dosVsysId : vsysId) {
+                     URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
+                     HashMap<String, Object> parms = new HashMap<>();
+                     parms.put("pageSize", -1);
+                     parms.put("orderBy", "profileId asc");
+                     parms.put("isValid", 1);
+//                     parms.put("vsysId", dosVsysId.getVsysId());
+                     parms.put("vsysId", 1);
+                     HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, parms);
+                     String token = CommonConfig.BIFANG_SERVER_TOKEN;
+                     if (!HttpClientUtils.ERROR_MESSAGE.equals(token)) {
+                         BasicHeader authorization = new BasicHeader("Authorization", token);
+                         BasicHeader authorization1 = new BasicHeader("Content-Type", "application/x-www-form-urlencoded");
+                         String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization, authorization1);
+                         if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
+                             HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
+                             boolean success = (boolean) resposeMap.get("success");
+                             String msg = resposeMap.get("msg").toString();
+                             if (success) {
+                                 HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
+                                 Object list = data.get("list");
+                                 if (list != null) {
+                                     thresholds = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(list), thresholdType);
+                                     logger.info("获取到静态阈值配置{}条", thresholds.size());
+                                 } else {
+                                     logger.warn("静态阈值配置为空");
+                                 }
+                             } else {
+                                 logger.error(msg);
+                             }
+                         }
+                     }
+//             }
+//            }
        } catch (Exception e) {
            logger.error("获取静态阈值配置失败,请检查bifang服务或登录配置信息 ", e);
        }
+
        return thresholds;
    }

@@ -136,25 +198,47 @@ public class ParseStaticThreshold {
     *
     * @return threshold RangeMap
     */
-    static TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> createStaticThreshold() {
-        TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> thresholdRangeMap = TreeRangeMap.create();
+    static HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> createStaticThreshold() {
+        HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> thresholdRangeMap = new HashMap<>(4);
        try {
            ArrayList<DosDetectionThreshold> dosDetectionThreshold = getDosDetectionThreshold();
            if (dosDetectionThreshold != null && !dosDetectionThreshold.isEmpty()) {
                for (DosDetectionThreshold threshold : dosDetectionThreshold) {
+                    String attackType = threshold.getAttackType();
+                    TreeRangeMap<IPAddress, DosDetectionThreshold> treeRangeMap = thresholdRangeMap.getOrDefault(attackType, TreeRangeMap.create());
                    ArrayList<String> serverIpList = threshold.getServerIpList();
                    for (String sip : serverIpList) {
                        IPAddressString ipAddressString = new IPAddressString(sip);
                        if (ipAddressString.isIPAddress()) {
                            IPAddress address = ipAddressString.getAddress();
-                            Map<String, DosDetectionThreshold> floodTypeThresholdMap = thresholdRangeMap.get(address);
-                            if (floodTypeThresholdMap == null) {
-                                floodTypeThresholdMap = new HashMap<>();
+                            if (address.isPrefixed()) {
+                                IPAddress lower = address.getLower();
+                                IPAddress upper = address.getUpper();
+                                if (!address.isMultiple()) {
+                                    lower = address.adjustPrefixLength(address.getBitCount());
+                                    upper = address.toMaxHost().withoutPrefixLength();
+                                }
+                                Map.Entry<Range<IPAddress>, DosDetectionThreshold> lowerEntry = treeRangeMap.getEntry(lower);
+                                Map.Entry<Range<IPAddress>, DosDetectionThreshold> upperEntry = treeRangeMap.getEntry(upper);
+                                if (lowerEntry != null && upperEntry == null) {
+                                    Range<IPAddress> lowerEntryKey = lowerEntry.getKey();
+                                    DosDetectionThreshold lowerEntryValue = lowerEntry.getValue();
+                                    treeRangeMap.put(Range.closedOpen(lowerEntryKey.lowerEndpoint(), lower), lowerEntryValue);
+                                    treeRangeMap.put(Range.closed(lower, upper), threshold);
+                                } else if (lowerEntry == null && upperEntry != null) {
+                                    Range<IPAddress> upperEntryKey = upperEntry.getKey();
+                                    DosDetectionThreshold upperEntryValue = upperEntry.getValue();
+                                    treeRangeMap.put(Range.openClosed(upper, upperEntryKey.upperEndpoint()), upperEntryValue);
+                                    treeRangeMap.put(Range.closed(lower, upper), threshold);
+                                } else {
+                                    treeRangeMap.put(Range.closed(lower, upper), threshold);
+                                }
+                            } else {
+                                treeRangeMap.put(Range.closed(address, address), threshold);
                            }
-                            floodTypeThresholdMap.put(threshold.getAttackType(), threshold);
-                            thresholdRangeMap.put(Range.closed(address.getLower(), address.getUpper()), floodTypeThresholdMap);
                        }
                    }
+                    thresholdRangeMap.put(attackType, treeRangeMap);
                }
            }
        } catch (Exception e) {
@@ -164,15 +248,26 @@ public class ParseStaticThreshold {
    }

    public static void main(String[] args) {
-        TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> staticThreshold = createStaticThreshold();
-        Map<Range<IPAddress>, Map<String, DosDetectionThreshold>> rangeMapMap = staticThreshold.asMapOfRanges();
-        for (Range<IPAddress> range : rangeMapMap.keySet()) {
-            Map<String, DosDetectionThreshold> thresholdMap = rangeMapMap.get(range);
-            for (String type : thresholdMap.keySet()) {
-                DosDetectionThreshold threshold = thresholdMap.get(type);
-                System.out.println(range + "---" + type + "---" + threshold);
+        ArrayList<DosDetectionThreshold> dosDetectionThreshold = getDosDetectionThreshold();
+        dosDetectionThreshold.forEach(System.out::println);
+
+
+        System.out.println("------------------------");
+        HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> staticThreshold = createStaticThreshold();
+
+        System.out.println("------------------------");
+
+        for (String type : staticThreshold.keySet()) {
+            Map<Range<IPAddress>, DosDetectionThreshold> asMapOfRanges = staticThreshold.get(type).asMapOfRanges();
+            for (Range<IPAddress> range : asMapOfRanges.keySet()) {
+                DosDetectionThreshold threshold = asMapOfRanges.get(range);
+                System.out.println(type + "---" + range + "---" + threshold);
            }
+            System.out.println("------------------------");
        }
+//        String s = loginBifangServer();
+//        System.out.println(s);
+
    }


--- a/src/main/java/com/zdjizhi/source/DosSketchSource.java
+++ b/src/main/java/com/zdjizhi/source/DosSketchSource.java
@@ -25,6 +25,7 @@ public class DosSketchSource {
            properties.put("sasl.mechanism", "PLAIN");
            properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+CommonConfig.SASL_JAAS_CONFIG_USER+"\" password=\""+CommonConfig.SASL_JAAS_CONFIG_PASSWORD+"\";");
        }
+
        return streamExeEnv.addSource(new FlinkKafkaConsumer<String>(
                CommonConfig.KAFKA_INPUT_TOPIC_NAME,
                new SimpleStringSchema(), properties))
--- a/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java
+++ b/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java
@@ -1,6 +1,8 @@
 package com.zdjizhi.utils;

 import com.zdjizhi.common.CommonConfig;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


@@ -12,6 +14,33 @@ public class FlinkEnvironmentUtils {

    static {
        streamExeEnv.setParallelism(CommonConfig.STREAM_EXECUTION_ENVIRONMENT_PARALLELISM);
+
+        /*
+        // 每 1000ms 开始一次 checkpoint
+        streamExeEnv.enableCheckpointing(CommonConfig.FLINK_WINDOW_MAX_TIME * 1000);
+
+        // 设置模式为精确一次 (这是默认值)
+        streamExeEnv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
+
+        // 确认 checkpoints 之间的时间会进行 500 ms
+        streamExeEnv.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
+
+        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
+        streamExeEnv.getCheckpointConfig().setCheckpointTimeout(60000);
+
+        // 允许两个连续的 checkpoint 错误
+        streamExeEnv.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
+
+        // 同一时间只允许一个 checkpoint 进行
+        streamExeEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
+
+        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
+        streamExeEnv.getCheckpointConfig().enableExternalizedCheckpoints(
+                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+
+        // 开启实验性的 unaligned checkpoints
+        streamExeEnv.getCheckpointConfig().enableUnalignedCheckpoints();
+        */
    }

 }
--- a/src/main/java/com/zdjizhi/utils/HbaseUtils.java
+++ b/src/main/java/com/zdjizhi/utils/HbaseUtils.java
@@ -1,17 +1,10 @@
 package com.zdjizhi.utils;

-import com.zdjizhi.common.CommonConfig;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.*;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.io.ArrayWritable;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.Writable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;

 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
@@ -22,75 +15,8 @@ import java.util.*;
 * @author wlh
 */
 public class HbaseUtils {
-    private static final Logger logger = LoggerFactory.getLogger(HbaseUtils.class);
-    private static Table table = null;
-    private static Scan scan = null;
-    private static ArrayList<String> floodTypeList = new ArrayList<>();

-    static {
-        floodTypeList.add("TCP SYN Flood");
-        floodTypeList.add("UDP Flood");
-        floodTypeList.add("ICMP Flood");
-        floodTypeList.add("DNS Amplification");
-    }
-
-    private static void prepareHbaseEnv() throws IOException {
-        org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
-
-        config.set("hbase.zookeeper.quorum", CommonConfig.HBASE_ZOOKEEPER_QUORUM);
-        config.set("hbase.client.retries.number", "3");
-        config.set("hbase.bulkload.retries.number", "3");
-        config.set("zookeeper.recovery.retry", "3");
-        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, CommonConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
-        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, CommonConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
-
-        TableName tableName = TableName.valueOf(CommonConfig.HBASE_BASELINE_TABLE_NAME);
-        Connection conn = ConnectionFactory.createConnection(config);
-        table = conn.getTable(tableName);
-        scan = new Scan().setAllowPartialResults(true).setLimit(CommonConfig.HBASE_BASELINE_TOTAL_NUM);
-        logger.info("连接hbase成功，正在读取baseline数据");
-    }
-
-    public static void main(String[] args) {
-        Map<String, Map<String, Tuple2<ArrayList<Integer>, Integer>>> baselineMap = readFromHbase();
-        Set<String> keySet = baselineMap.keySet();
-        for (String key : keySet) {
-            Map<String, Tuple2<ArrayList<Integer>, Integer>> stringTuple2Map = baselineMap.get(key);
-            Set<String> strings = stringTuple2Map.keySet();
-            for (String s:strings){
-                Tuple2<ArrayList<Integer>, Integer> arrayListIntegerTuple2 = stringTuple2Map.get(s);
-                System.out.println(key+"---"+s+"---"+arrayListIntegerTuple2.f0+"---"+arrayListIntegerTuple2.f1);
-            }
-        }
-        System.out.println(baselineMap.size());
-    }
-
-    public static Map<String, Map<String, Tuple2<ArrayList<Integer>, Integer>>> readFromHbase() {
-        Map<String, Map<String, Tuple2<ArrayList<Integer>, Integer>>> baselineMap = new HashMap<>();
-        try {
-            prepareHbaseEnv();
-            logger.info("开始读取baseline数据");
-            ResultScanner rs = table.getScanner(scan);
-            for (Result result : rs) {
-                Map<String, Tuple2<ArrayList<Integer>, Integer>> floodTypeMap = new HashMap<>();
-                String rowkey = Bytes.toString(result.getRow());
-                for (String type:floodTypeList){
-                    ArrayList<Integer> sessionRate = getArraylist(result, type, "session_rate");
-                    if (sessionRate != null && !sessionRate.isEmpty()){
-                        Integer defaultValue = getDefaultValue(result, type, "session_rate_default_value");
-                        floodTypeMap.put(type,Tuple2.of(sessionRate, defaultValue));
-                    }
-                }
-                baselineMap.put(rowkey, floodTypeMap);
-            }
-            logger.info("格式化baseline数据成功，读取IP共：{}", baselineMap.size());
-        } catch (Exception e) {
-            logger.error("读取hbase数据失败", e);
-        }
-        return baselineMap;
-    }
-
-    private static Integer getDefaultValue(Result result, String family, String qualifier) {
+    public static Integer getIntegerValue(Result result, String family, String qualifier) {
        byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        if (value != null){
            return Bytes.toInt(value);
@@ -98,7 +24,7 @@ public class HbaseUtils {
        return 1;
    }

-    private static ArrayList<Integer> getArraylist(Result result, String family, String qualifier) throws IOException {
+    public static ArrayList<Integer> getArraylist(Result result, String family, String qualifier) throws IOException {
        if (containsColumn(result, family, qualifier)) {
            ArrayWritable w = new ArrayWritable(IntWritable.class);
            w.readFields(new DataInputStream(new ByteArrayInputStream(result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier)))));
--- a/src/main/java/com/zdjizhi/utils/HttpClientUtils.java
+++ b/src/main/java/com/zdjizhi/utils/HttpClientUtils.java
@@ -253,12 +253,12 @@ public class HttpClientUtils {
     * 拼装url
     * url ,参数map
     */
-    public static void setUrlWithParams(URIBuilder uriBuilder,String path, Map<String, String> params) {
+    public static void setUrlWithParams(URIBuilder uriBuilder,String path, Map<String, Object> params) {
        try {
            uriBuilder.setPath(path);
            if (params != null && !params.isEmpty()){
-                for (Map.Entry<String, String> kv : params.entrySet()) {
-                    uriBuilder.setParameter(kv.getKey(),kv.getValue());
+                for (Map.Entry<String, Object> kv : params.entrySet()) {
+                    uriBuilder.setParameter(kv.getKey(),kv.getValue().toString());
                }
            }
        } catch (Exception e) {
--- a/src/main/java/com/zdjizhi/utils/IpUtils.java
+++ b/src/main/java/com/zdjizhi/utils/IpUtils.java
@@ -7,20 +7,20 @@ public class IpUtils {
    /**
     * IP定位库工具类
     */
-    public static IpLookup ipLookup = new IpLookup.Builder(false)
-            .loadDataFileV4(CommonConfig.IP_MMDB_PATH + "ip_v4.mmdb")
-            .loadDataFileV6(CommonConfig.IP_MMDB_PATH + "ip_v6.mmdb")
-            .loadDataFilePrivateV4(CommonConfig.IP_MMDB_PATH + "ip_private_v4.mmdb")
-            .loadDataFilePrivateV6(CommonConfig.IP_MMDB_PATH + "ip_private_v6.mmdb")
+    public static IpLookupV2 ipLookup = new IpLookupV2.Builder(false)
+            .loadDataFileV4(CommonConfig.IP_MMDB_PATH + "ip_v4_built_in.mmdb")
+            .loadDataFileV6(CommonConfig.IP_MMDB_PATH + "ip_v6_built_in.mmdb")
+            .loadDataFilePrivateV4(CommonConfig.IP_MMDB_PATH + "ip_v4_user_defined.mmdb")
+            .loadDataFilePrivateV6(CommonConfig.IP_MMDB_PATH + "ip_v6_user_defined.mmdb")
            .build();

    public static void main(String[] args) {
        System.out.println(ipLookup.countryLookup("49.7.115.37"));

-//        String ips = "192.168.50.23,192.168.50.45,192.168.56.9,192.168.56.8,192.168.50.58,192.168.56.7,192.168.56.6,192.168.50.40,192.168.50.19,192.168.50.6,192.168.50.4,192.168.56.17,192.168.50.27,192.168.50.26,192.168.50.18,192.168.56.3,192.168.56.10";
-//        for (String ip:ips.split(",")){
-//            System.out.println(ip+"--"+ipLookup.countryLookup(ip));
-//        }
+        String ips = "182.168.50.23,182.168.50.45,182.168.56.9,182.168.56.8,92.168.50.58,19.168.56.7,12.168.56.6,2.168.50.40,1.168.50.19,9.168.50.6,2.168.50.4,192.168.56.17,192.168.50.27,192.168.50.26,192.168.50.18,192.168.56.3,192.168.56.10";
+        for (String ip:ips.split(",")){
+            System.out.println(ip+"--"+ipLookup.countryLookup(ip));
+        }

    }

--- a/src/main/java/com/zdjizhi/utils/KafkaUtils.java
+++ b/src/main/java/com/zdjizhi/utils/KafkaUtils.java
@@ -4,6 +4,7 @@ import com.zdjizhi.common.CommonConfig;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

+import java.util.Optional;
 import java.util.Properties;

 public class KafkaUtils {
@@ -21,11 +22,14 @@ public class KafkaUtils {
    }

    public static FlinkKafkaProducer<String> getKafkaSink(String topic){
-        return new FlinkKafkaProducer<String>(
+        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                topic,
                new SimpleStringSchema(),
-                getKafkaSinkProperty()
+                getKafkaSinkProperty(),
+                Optional.empty()
        );
+        kafkaProducer.setLogFailuresOnly(true);
+        return kafkaProducer;
    }

 }
--- a/src/main/java/com/zdjizhi/utils/NacosUtils.java
+++ b/src/main/java/com/zdjizhi/utils/NacosUtils.java
@@ -0,0 +1,90 @@
+package com.zdjizhi.utils;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.Listener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+public class NacosUtils {
+    private static final Logger logger = LoggerFactory.getLogger(NacosUtils.class);
+    private static Properties nacosProperties = new Properties();
+    private static Properties commonProperties = new Properties();
+
+
+    private static final String NACOS_SERVER_ADDR = CommonConfigurations.getStringProperty("nacos.server.addr");
+    private static final String NACOS_NAMESPACE = CommonConfigurations.getStringProperty("nacos.namespace");
+    private static final String NACOS_USERNAME = CommonConfigurations.getStringProperty("nacos.username");
+    private static final String NACOS_PASSWORD = CommonConfigurations.getStringProperty("nacos.password");
+    private static final String NACOS_DATA_ID = CommonConfigurations.getStringProperty("nacos.data.id");
+    private static final String NACOS_GROUP = CommonConfigurations.getStringProperty("nacos.group");
+    private static final long NACOS_READ_TIMEOUT = CommonConfigurations.getLongProperty("nacos.read.timeout");
+
+    static {
+        createConfigService();
+    }
+
+    private static void getProperties() {
+        nacosProperties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_SERVER_ADDR);
+        nacosProperties.setProperty(PropertyKeyConst.NAMESPACE, NACOS_NAMESPACE);
+        nacosProperties.setProperty(PropertyKeyConst.USERNAME, NACOS_USERNAME);
+        nacosProperties.setProperty(PropertyKeyConst.PASSWORD, NACOS_PASSWORD);
+    }
+
+    private static void createConfigService() {
+        try {
+            getProperties();
+            ConfigService configService = NacosFactory.createConfigService(nacosProperties);
+            String config = configService.getConfig(NACOS_DATA_ID, NACOS_GROUP, NACOS_READ_TIMEOUT);
+            commonProperties.load(new StringReader(config));
+
+            configService.addListener(NACOS_DATA_ID, NACOS_GROUP, new Listener() {
+                @Override
+                public Executor getExecutor() {
+                    return null;
+                }
+
+                @Override
+                public void receiveConfigInfo(String configMsg) {
+                    try {
+                        commonProperties.clear();
+                        commonProperties.load(new StringReader(configMsg));
+                    } catch (IOException e) {
+                        logger.error("监听nacos配置失败", e);
+                    }
+                    System.out.println(configMsg);
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+            logger.error("获取nacos配置失败", e);
+        }
+    }
+
+    public static String getStringProperty(String key) {
+        return commonProperties.getProperty(key);
+    }
+
+    public static Integer getIntProperty(String key) {
+        return Integer.parseInt(commonProperties.getProperty(key));
+    }
+
+    public static Double getDoubleProperty(String key) {
+        return Double.parseDouble(commonProperties.getProperty(key));
+    }
+
+    public static Long getLongProperty(String key) {
+        return Long.parseLong(commonProperties.getProperty(key));
+    }
+
+    public static Boolean getBooleanProperty(String key) {
+        return "true".equals(commonProperties.getProperty(key).toLowerCase().trim());
+    }
+
+}
--- a/src/main/resources/common.properties
+++ b/src/main/resources/common.properties
@@ -11,11 +11,11 @@ kafka.input.parallelism=1
 kafka.input.topic.name=DOS-SKETCH-RECORD

 #输入kafka地址
-kafka.input.bootstrap.servers=192.168.44.12:9092
-#kafka.input.bootstrap.servers=192.168.44.11:9092,192.168.44.14:9092,192.168.44.15:9092
+#kafka.input.bootstrap.servers=192.168.44.12:9094
+kafka.input.bootstrap.servers=192.168.44.11:9094,192.168.44.14:9094,192.168.44.15:9094

 #读取kafka group id
-kafka.input.group.id=2108231709
+kafka.input.group.id=2112080949
 #kafka.input.group.id=dos-detection-job-210813-1

 #发送kafka metrics并行度大小
@@ -30,26 +30,29 @@ kafka.output.event.parallelism=1

 #发送kafka event topic名
 kafka.output.event.topic.name=DOS-EVENT
-#kafka.output.event.topic.name=test
+#kafka.output.event.topic.name=storm-dos-test

 #kafka输出地址
-kafka.output.bootstrap.servers=192.168.44.12:9092
+kafka.output.bootstrap.servers=192.168.44.12:9094
 #kafka.output.bootstrap.servers=192.168.44.11:9092,192.168.44.14:9092,192.168.44.15:9092

 #zookeeper地址
-#hbase.zookeeper.quorum=192.168.44.12:2181
-hbase.zookeeper.quorum=192.168.44.11:2181,192.168.44.14:2181,192.168.44.15:2181
+hbase.zookeeper.quorum=192.168.44.12:2181
+#hbase.zookeeper.quorum=192.168.44.11:2181,192.168.44.14:2181,192.168.44.15:2181

 #hbase客户端处理时间
 hbase.client.operation.timeout=30000
 hbase.client.scanner.timeout.period=30000

 ##hbase baseline表名
-hbase.baseline.table.name=ddos_traffic_baselines
+hbase.baseline.table.name=dos:ddos_traffic_baselines

 #读取baseline限制
 hbase.baseline.total.num=1000000

+#baseline ttl，单位：天
+hbase.baseline.ttl=30
+
 #设置聚合并行度，2个key
 flink.first.agg.parallelism=1

@@ -57,7 +60,7 @@ flink.first.agg.parallelism=1
 flink.detection.map.parallelism=1

 #watermark延迟
-flink.watermark.max.orderness=1
+flink.watermark.max.orderness=10

 #计算窗口大小，默认600s
 flink.window.max.time=10
@@ -75,18 +78,22 @@ ip.mmdb.path=D:\\data\\dat\\
 #ip.mmdb.path=/home/bigdata/topology/dat/
 #ip.mmdb.path=/home/bigdata/wlh/topology/dos-detection/dat/

-#敏感阈值，速率小于此值不报警
-sensitivity.threshold=100
-
-#基于baseline判定dos攻击的上下限
-baseline.sessions.minor.threshold=0.1
-baseline.sessions.warning.threshold=0.5
-baseline.sessions.major.threshold=1
-baseline.sessions.severe.threshold=3
-baseline.sessions.critical.threshold=8
+##静态敏感阈值，速率小于此值不报警
+#static.sensitivity.threshold=500
+#
+##基线敏感阈值
+#baseline.sensitivity.threshold=0.2
+#
+##基于baseline判定dos攻击的上下限
+#baseline.sessions.minor.threshold=0.5
+#baseline.sessions.warning.threshold=1
+#baseline.sessions.major.threshold=2.5
+#baseline.sessions.severe.threshold=5
+#baseline.sessions.critical.threshold=8

 #bifang服务访问地址
-bifang.server.uri=http://192.168.44.3:80
+bifang.server.uri=http://192.168.44.72:80
+#bifang.server.uri=http://192.168.44.3:80

 #访问bifang只读权限token，bifang内置，无需修改
 bifang.server.token=ed04b942-7df4-4e3d-b9a9-a881ca98a867
@@ -97,6 +104,9 @@ bifang.server.encryptpwd.path=/v1/user/encryptpwd
 #登录bifang服务路径信息
 bifang.server.login.path=/v1/user/login

+#获取vaysId路径信息
+bifang.server.policy.vaysid.path=/v1/system/vsys/
+
 #获取静态阈值路径信息
 bifang.server.policy.threshold.path=/v1/policy/profile/DoS/detection/threshold

@@ -120,11 +130,22 @@ http.pool.response.timeout=60000
 static.threshold.schedule.minutes=10

 #获取baseline周期，默认7天
-baseline.threshold.schedule.days=7
+baseline.threshold.schedule.days=1

 #kafka用户认证配置参数
 sasl.jaas.config.user=admin
-sasl.jaas.config.password=galaxy2019
+#sasl.jaas.config.password=galaxy2019
+#sasl.jaas.config.password=ENC(6MleDyA3Z73HSaXiKsDJ2k7Ys8YWLhEJ)
+sasl.jaas.config.password=6MleDyA3Z73HSaXiKsDJ2k7Ys8YWLhEJ

 #是否开启kafka用户认证配置，1：是；0：否
-sasl.jaas.config.flag=1
+sasl.jaas.config.flag=1
+
+#nacos配置
+nacos.server.addr=192.168.40.42:8848
+nacos.namespace=prod
+nacos.username=nacos
+nacos.password=nacos
+nacos.data.id=dos_detection.properties
+nacos.group=Galaxy
+nacos.read.timeout=5000
--- a/src/test/java/com/zdjizhi/common/IpTest.java
+++ b/src/test/java/com/zdjizhi/common/IpTest.java
@@ -41,7 +41,22 @@ public class IpTest {
        IPAddress pv43 = new IPAddressString("fc00::").getAddress();
        IPAddress pv44 = new IPAddressString("fc00::10:1").getAddress();

-        IPAddress pv45 = new IPAddressString("12.56.4.0/32").getAddress();
+        IPAddress pv45 = new IPAddressString("192.168.42.1").getAddress();
+        IPAddress pv46 = new IPAddressString("192.168.42.1/32").getAddress();
+        IPAddress pv47 = new IPAddressString("12.56.4.0").getAddress();
+
+        IPAddress mask = pv45.getNetwork().getNetworkMask(24, false);
+
+        System.out.println(pv45.isMultiple());
+        System.out.println(pv46.isMultiple());
+        System.out.println(pv46.isPrefixed());
+        System.out.println(pv47.isPrefixed());
+        System.out.println(pv45+"---"+pv45.toMaxHost().withoutPrefixLength()+"---"+pv45.adjustPrefixLength(pv45.getBitCount()));
+        System.out.println(pv45+"---mask:"+pv45.mask(mask).toString());
+
+        System.out.println(pv45.adjustPrefixLength(pv45.getBitCount())+"---"+pv45.toMaxHost().withoutPrefixLength());
+
+        /*
        System.out.println(str5.getUpper()+"---"+str5.getLower());

        System.out.println(rangeMap.span().contains(pv4));
@@ -50,6 +65,7 @@ public class IpTest {
        System.out.println(rangeMap.get(pv42));
        System.out.println(rangeMap.get(pv43));
        System.out.println(rangeMap.get(pv44));
+        */

        /*
        System.out.println(str5.toSequentialRange());
--- a/src/test/java/com/zdjizhi/common/NacosTest.java
+++ b/src/test/java/com/zdjizhi/common/NacosTest.java
@@ -0,0 +1,101 @@
+package com.zdjizhi.common;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.Listener;
+import com.alibaba.nacos.api.exception.NacosException;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2022/3/1016:58
+ */
+public class NacosTest {
+
+    /**
+     * <dependency>
+     * <groupId>com.alibaba.nacos</groupId>
+     * <artifactId>nacos-client</artifactId>
+     * <version>1.2.0</version>
+     * </dependency>
+     */
+
+    private static Properties properties = new Properties();
+    /**
+     * config data id = config name
+     */
+    private static final String DATA_ID = "dos_baseline.properties";
+    /**
+     * config group
+     */
+    private static final String GROUP = "Galaxy";
+
+    private void getProperties() {
+        properties.setProperty(PropertyKeyConst.SERVER_ADDR, "192.168.44.12:8848");
+        properties.setProperty(PropertyKeyConst.NAMESPACE, "flink");
+        properties.setProperty(PropertyKeyConst.USERNAME, "nacos");
+        properties.setProperty(PropertyKeyConst.PASSWORD, "nacos");
+    }
+
+
+    @Test
+    public void GetConfigurationTest() {
+        try {
+            getProperties();
+            ConfigService configService = NacosFactory.createConfigService(properties);
+            String content = configService.getConfig(DATA_ID, GROUP, 5000);
+            Properties nacosConfigMap = new Properties();
+            nacosConfigMap.load(new StringReader(content));
+            System.out.println(nacosConfigMap.getProperty("static.sensitivity.threshold"));
+        } catch (NacosException | IOException e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    @Test
+    public void ListenerConfigurationTest() {
+        getProperties();
+        try {
+            //first get config
+            ConfigService configService = NacosFactory.createConfigService(properties);
+            String config = configService.getConfig(DATA_ID, GROUP, 5000);
+//            System.out.println(config);
+
+            //start listenner
+            configService.addListener(DATA_ID, GROUP, new Listener() {
+                @Override
+                public Executor getExecutor() {
+                    return null;
+                }
+
+                @Override
+                public void receiveConfigInfo(String configMsg) {
+                    System.out.println(configMsg);
+                }
+            });
+        } catch (NacosException e) {
+            e.printStackTrace();
+        }
+
+        //keep running,change nacos config,print new config
+        /*
+        while (true) {
+            try {
+                Thread.sleep(5000);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        */
+    }
+}
Author	SHA1	Message	Date
unknown	84a1e6879a	修复动态获取nacos配置	2022-06-28 11:15:49 +08:00
unknown	ab8f6aba81	修复动态获取nacos配置	2022-06-28 11:14:54 +08:00
unknown	94e8fb807a	集成nacos，修复依赖冲突，启动报错	2022-06-22 15:58:49 +08:00
wanglihui	cead1d4d99	Merge branch 'tsg-22.06' of git.mesalab.cn:bigdata/tsg/flink-dos-detection	2022-06-20 18:15:24 +08:00
徐鹏飞	2d98c3b6e6	修改获取静态阈值逻辑，根据默认指定vsysID获取	2022-06-13 18:00:57 +08:00
徐鹏飞	3dc29a07be	新增根据vsysId从bifang获取静态阈值	2022-05-31 18:00:36 +08:00
wanglihui	1fcdb79739	DoS事件日志对Conditions基于速率检测属性值修正	2022-05-17 13:54:43 +08:00
wanglihui	3d974217d9	集成nacos，将部分配置放到nacos中管理。	2022-04-08 10:16:29 +08:00
wanglihui	db17064f73	更新工具库到1.0.8适配新版MMDB定位库	2022-02-11 16:06:24 +08:00
wanglihui	065e5abb09	增加jasypt配置加密方式	2021-12-20 13:50:18 +08:00
wanglihui	75bbdd2962	更新DoS检测程序，新增读取baseline TTL配置。修复DoS检测-Conditions阈值描述语言逻辑问题。	2021-12-08 13:51:09 +08:00
wanglihui	c46a395d9b	修改静态阈值，上下限判定等配置。	2021-11-26 09:46:53 +08:00
wanglihui	cc3f0cf620	解决获取配置为空时空指针异常bug。	2021-11-05 19:01:12 +08:00
wanglihui	0617b1e614	修改获取基线值策略，当基线type=3且基线值小于静态敏感阈值时，将其替换。	2021-10-29 18:43:05 +08:00
wanglihui	0125b031dd	代码格式化，实体类重写equals、hashcode方法。	2021-10-22 18:38:29 +08:00
wanglihui	177e7461cc	优化构建baseline方式	2021-10-21 18:27:48 +08:00
wanglihui	be916531fb	修改构建threshold RangeMap逻辑，基于attack type为key，避免IP冲突问题。	2021-10-20 18:23:12 +08:00
wanglihui	c692112445	增加基于packets与bits作为static条件判断依据。修复static配置IP冲突问题。	2021-10-19 18:39:13 +08:00
wanglihui	b03ab9642d	修复静态条件与基线条件冲突bug	2021-09-28 16:11:52 +08:00
wanglihui	c44250bf73	新增读取DoS Detection Profiles IP冲突检测机制修复DoS event日志end_time大于当前时间bug	2021-09-26 18:41:36 +08:00
wanglihui	77bc6a844e	修复读取配置IP冲突问题	2021-09-23 18:36:27 +08:00
wanglihui	e930fa23ed	修改判定逻辑，增加基线敏感阈值作为判定条件。	2021-09-16 18:47:00 +08:00
wanglihui	8cd4dea19e	获取静态阈值列表使用bifang默认token 修复获取baseline下标方法	2021-09-15 10:08:17 +08:00
wanglihui	62f3c65d66	基于DoS Sketch一元组进行实时检测	2021-09-14 18:46:23 +08:00
wanglihui	8cfb442c44	增加一元组作为基线生成数据源	2021-09-13 14:14:58 +08:00