修复动态获取nacos配置

修复DoS检测-Conditions阈值描述语言逻辑问题。

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>
 
@@ -199,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>
@@ -215,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>
 
 

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,14 +49,14 @@ public class CommonConfig {
 
     public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");
 
-    public static final int STATIC_SENSITIVITY_THRESHOLD = CommonConfigurations.getIntProperty("static.sensitivity.threshold");
-    public static final double BASELINE_SENSITIVITY_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.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");
@@ -54,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");
@@ -64,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);
+    }
 }

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());
+    }
+}

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;
     }

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());
+    }
 }

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());
+    }
 }

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;
     }

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 +
+                '}';
+    }
+}

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,10 +26,20 @@ 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) {
@@ -45,7 +49,7 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
             executorService.scheduleAtFixedRate(() -> thresholdRangeMap = ParseStaticThreshold.createStaticThreshold(), 0,
                     CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
 
-            executorService.scheduleAtFixedRate(() -> baselineMap = HbaseUtils.readFromHbase(), 0,
+            executorService.scheduleAtFixedRate(() -> baselineMap = ParseBaselineThreshold.readFromHbase(), 0,
                     CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
         } catch (Exception e) {
             logger.error("定时器任务执行失败", e);
@@ -60,15 +64,15 @@ 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 ((thresholdMap == null || !thresholdMap.containsKey(attackType)) && baselineMap.containsKey(destinationIp)) {
+            if (threshold == null && baselineMap.containsKey(destinationIp)) {
                 finalResult = getDosEventLogByBaseline(value);
-            }else if ((thresholdMap == null || !thresholdMap.containsKey(attackType)) && !baselineMap.containsKey(destinationIp)){
+            } else if (threshold == null && !baselineMap.containsKey(destinationIp)) {
                 finalResult = getDosEventLogBySensitivityThreshold(value);
-            }else if (thresholdMap != null && thresholdMap.containsKey(attackType)){
-                finalResult = getDosEventLogByStaticThreshold(value, thresholdMap);
-            }else {
+            } else if (threshold != null) {
+                finalResult = getDosEventLogByStaticThreshold(value, threshold);
+            } else {
                 logger.debug("未获取到当前server IP：{} 类型 {} 静态阈值 和 baseline", destinationIp, attackType);
             }
 
@@ -78,11 +82,12 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         return finalResult;
     }
 
-    private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value){
+    private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value) {
         DosEventLog result = null;
         long sketchSessions = value.getSketch_sessions();
-        if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
-            result = getDosEventLog(value, CommonConfig.STATIC_SENSITIVITY_THRESHOLD, sketchSessions - CommonConfig.STATIC_SENSITIVITY_THRESHOLD, "sensitivity");
+        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;
@@ -93,27 +98,33 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         String destinationIp = value.getDestination_ip();
         String attackType = value.getAttack_type();
         long sketchSessions = value.getSketch_sessions();
-        if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
-            Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
-            Integer base = getBaseValue(floodTypeTup, value);
-            result = getDosEventLog(value, base, sketchSessions - base, "baseline");
+        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 result;
     }
 
-    private DosEventLog getDosEventLogByStaticThreshold(DosSketchLog value, Map<String, DosDetectionThreshold> thresholdMap) {
-        DosEventLog result = 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");
+    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, String tag) {
+    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();
@@ -121,11 +132,11 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
             double percent = getDiffPercent(diff, base);
             Severity severity = judgeSeverity(percent);
             if (severity != Severity.NORMAL) {
-                if ("baseline".equals(tag) && percent < CommonConfig.BASELINE_SENSITIVITY_THRESHOLD){
-                    logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过基线敏感阈值,日志详情\n{}",destinationIp,attackType,base,percent,value);
-                }else {
-                    result = getResult(value,base, severity, percent, tag);
-                    logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,基于{}检测,日志详情\n {}", destinationIp,attackType,base,percent,tag,result);
+                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);
@@ -134,14 +145,14 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         return result;
     }
 
-    private DosEventLog getResult(DosSketchLog value,long base, Severity severity, double percent, String tag) {
+    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() + value.getSketch_duration());
         dosEventLog.setAttack_type(value.getAttack_type());
         dosEventLog.setSeverity(severity.severity);
-        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent),base, 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();
@@ -153,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);
@@ -166,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) {
@@ -174,59 +189,87 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         return base;
     }
 
-    private String getConditions(String percent,long base, long sessions, String tag) {
-        switch (tag) {
-            case "baseline":
-                return "sessions > " + percent + " of baseline";
-            case "static":
-                return "sessions > " + base + " sessions/s";
-            case "sensitivity":
-                return sessions+" sessions/s Unusually high Sessions";
+    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) {
-        Date date = new Date(1631548860 * 1000L);
-        System.out.println(date);
-        Date p1D = DateUtils.getTimeFloor(date, "P1D");
-        System.out.println(p1D+" "+p1D.getTime()/1000);
-        System.out.println(new DosDetection().getCurrentTimeIndex(1631548860));
-        System.out.println(10+10*0.2);
-        Map<String, DosDetectionThreshold> thresholdMap = null;
-        System.out.println(thresholdMap.containsKey("a"));
+//        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;
@@ -237,49 +280,24 @@ 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;
         }
 
     }
 
-    @Deprecated
-    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);
-        }
-    }
-
-    @Deprecated
-    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;
-    }
-
 }

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,
@@ -70,7 +73,7 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
         try {
             for (DosSketchLog newSketchLog : elements){
                 String sourceIp = newSketchLog.getSource_ip();
-                if ("0.0.0.0".equals(sourceIp) || "::".equals(sourceIp)){
+                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();

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());
+    }
+
+
+
+
+}

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 {
         //加载加密登录密码
@@ -99,40 +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);
             HashMap<String, Object> parms = new HashMap<>();
-            parms.put("pageSize",-1);
-            parms.put("orderBy","profileId asc");
-            parms.put("isValid",1);
-            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, parms);
+            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);
                 BasicHeader authorization1 = new BasicHeader("Content-Type", "application/x-www-form-urlencoded");
-                String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization,authorization1);
+                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;
     }
 
@@ -141,60 +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 = new HashMap<>();
-                            floodTypeThresholdMap.put(threshold.getAttackType(), threshold);
-                            if (address.isPrefixed()){
+                            if (address.isPrefixed()) {
                                 IPAddress lower = address.getLower();
                                 IPAddress upper = address.getUpper();
-                                if (!address.isMultiple()){
+                                if (!address.isMultiple()) {
                                     lower = address.adjustPrefixLength(address.getBitCount());
                                     upper = address.toMaxHost().withoutPrefixLength();
                                 }
-                                Map.Entry<Range<IPAddress>, Map<String, DosDetectionThreshold>> lowerEntry = thresholdRangeMap.getEntry(lower);
-                                Map.Entry<Range<IPAddress>, Map<String, DosDetectionThreshold>> upperEntry = thresholdRangeMap.getEntry(upper);
-                                if (lowerEntry == null && upperEntry == null){
-                                    thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
-                                }else if (lowerEntry != null && upperEntry == null){
+                                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();
-                                    Map<String, DosDetectionThreshold> lowerEntryValue = lowerEntry.getValue();
-                                    lowerEntryValue.put(threshold.getAttackType(), threshold);
-                                    thresholdRangeMap.put(Range.closedOpen(lowerEntryKey.lowerEndpoint(), lower), lowerEntryValue);
-                                    thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
-                                }else if (lowerEntry == null){
+                                    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();
-                                    Map<String, DosDetectionThreshold> upperEntryValue = upperEntry.getValue();
-                                    upperEntryValue.put(threshold.getAttackType(), threshold);
-                                    thresholdRangeMap.put(Range.openClosed(upper, upperEntryKey.upperEndpoint()), upperEntryValue);
-                                    thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
-                                }
-                            }else {
-                                Map.Entry<Range<IPAddress>, Map<String, DosDetectionThreshold>> entry = thresholdRangeMap.getEntry(address);
-                                if (entry != null){
-                                    Range<IPAddress> entryKey = entry.getKey();
-                                    Map<String, DosDetectionThreshold> entryValue = entry.getValue();
-                                    entryValue.put(threshold.getAttackType(), threshold);
-                                    if (entryKey.lowerEndpoint() == entryKey.upperEndpoint()){
-                                        thresholdRangeMap.put(Range.closed(address, address), entryValue);
-                                    }else {
-                                        thresholdRangeMap.put(Range.closed(address, address), floodTypeThresholdMap);
-                                    }
-                                }else {
-                                    thresholdRangeMap.put(Range.closed(address, address), floodTypeThresholdMap);
+                                    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);
                             }
                         }
                     }
+                    thresholdRangeMap.put(attackType, treeRangeMap);
                 }
             }
         } catch (Exception e) {
@@ -204,24 +248,25 @@ public class ParseStaticThreshold {
     }
 
     public static void main(String[] args) {
-
         ArrayList<DosDetectionThreshold> dosDetectionThreshold = getDosDetectionThreshold();
         dosDetectionThreshold.forEach(System.out::println);
 
 
         System.out.println("------------------------");
-        TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> staticThreshold = createStaticThreshold();
+        HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> staticThreshold = createStaticThreshold();
 
         System.out.println("------------------------");
-        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);
-            }
-        }
 
+        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);
 
     }
 

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();
+        */
     }
 
 }

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 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);
-        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)))));

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));
+        }
 
     }
 

src/main/java/com/zdjizhi/utils/KafkaUtils.java

@@ -22,12 +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(),
                 Optional.empty()
         );
+        kafkaProducer.setLogFailuresOnly(true);
+        return kafkaProducer;
     }
 
 }

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());
+    }
+
+}

src/main/resources/common.properties

@@ -11,34 +11,34 @@ 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.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=2109160928
+kafka.input.group.id=2112080949
 #kafka.input.group.id=dos-detection-job-210813-1
 
 #发送kafka metrics并行度大小
 kafka.output.metric.parallelism=1
 
 #发送kafka metrics topic名
-#kafka.output.metric.topic.name=TRAFFIC-TOP-DESTINATION-IP-METRICS
-kafka.output.metric.topic.name=test
+kafka.output.metric.topic.name=TRAFFIC-TOP-DESTINATION-IP-METRICS
+#kafka.output.metric.topic.name=test
 
 #发送kafka event并行度大小
 kafka.output.event.parallelism=1
 
 #发送kafka event topic名
-#kafka.output.event.topic.name=DOS-EVENT
-kafka.output.event.topic.name=storm-dos-test
+kafka.output.event.topic.name=DOS-EVENT
+#kafka.output.event.topic.name=storm-dos-test
 
 #kafka输出地址
 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
@@ -50,6 +50,9 @@ 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
 
@@ -60,7 +63,7 @@ flink.detection.map.parallelism=1
 flink.watermark.max.orderness=10
 
 #计算窗口大小，默认600s
-flink.window.max.time=600
+flink.window.max.time=10
 
 #dos event结果中distinct source IP限制
 source.ip.list.limit=10000
@@ -75,21 +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/
 
-#静态敏感阈值，速率小于此值不报警
-static.sensitivity.threshold=100
-
-#基线敏感阈值
-baseline.sensitivity.threshold=0.2
-
-#基于baseline判定dos攻击的上下限
-baseline.sessions.minor.threshold=0.2
-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.72:80
+#bifang.server.uri=http://192.168.44.3:80
 
 #访问bifang只读权限token，bifang内置，无需修改
 bifang.server.token=ed04b942-7df4-4e3d-b9a9-a881ca98a867
@@ -100,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
 
@@ -123,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

src/test/java/com/zdjizhi/common/IpTest.java

@@ -41,14 +41,18 @@ public class IpTest {
         IPAddress pv43 = new IPAddressString("fc00::").getAddress();
         IPAddress pv44 = new IPAddressString("fc00::10:1").getAddress();
 
-        IPAddress pv45 = new IPAddressString("192.168.42.1/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());
 

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();
+            }
+        }
+        */
+    }
+}