修复动态获取nacos配置

pom.xml

@@ -148,6 +148,10 @@
                     <artifactId>jdk.tools</artifactId>
                     <groupId>jdk.tools</groupId>
                 </exclusion>
+                <exclusion>
+                    <artifactId>guava</artifactId>
+                    <groupId>com.google.guava</groupId>
+                </exclusion>
             </exclusions>
         </dependency>
 
@@ -222,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,6 +1,7 @@
 package com.zdjizhi.common;
 
 import com.zdjizhi.utils.CommonConfigurations;
+import com.zdjizhi.utils.NacosUtils;
 import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;
 
 /**
@@ -48,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");
@@ -84,7 +85,8 @@ public class CommonConfig {
         // 配置加密解密的密码/salt值
         encryptor.setPassword("galaxy");
         // 对"raw_password"进行加密：S5kR+Y7CI8k7MaecZpde25yK8NKUnd6p
-        String password = "galaxy2019";
+//        String password = "galaxy2019";
+        String password = "nacos";
         String encPwd = encryptor.encrypt(password);
         System.out.println(encPwd);
         // 再进行解密：raw_password

src/main/java/com/zdjizhi/etl/DosDetection.java

@@ -85,9 +85,9 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
     private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value) {
         DosEventLog result = null;
         long sketchSessions = value.getSketch_sessions();
-        if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD) {
-            long diff = sketchSessions - CommonConfig.STATIC_SENSITIVITY_THRESHOLD;
-            result = getDosEventLog(value, CommonConfig.STATIC_SENSITIVITY_THRESHOLD, diff, SENSITIVITY_CONDITION_TYPE, SESSIONS_TAG);
+        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;
@@ -98,7 +98,7 @@ 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) {
+        if (sketchSessions > NacosUtils.getIntProperty("static.sensitivity.threshold")) {
             DosBaselineThreshold dosBaselineThreshold = baselineMap.get(destinationIp).get(attackType);
             Integer base = getBaseValue(dosBaselineThreshold, value);
             long diff = sketchSessions - base;
@@ -132,7 +132,7 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
             double percent = getDiffPercent(diff, base);
             Severity severity = judgeSeverity(percent);
             if (severity != Severity.NORMAL) {
-                if (type == BASELINE_CONDITION_TYPE && percent < CommonConfig.BASELINE_SENSITIVITY_THRESHOLD) {
+                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);
@@ -178,8 +178,8 @@ 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 < CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
-                        base = CommonConfig.STATIC_SENSITIVITY_THRESHOLD;
+                    if (sessionRateBaselineType == OTHER_BASELINE_TYPE && base < NacosUtils.getIntProperty("static.sensitivity.threshold")){
+                        base = NacosUtils.getIntProperty("static.sensitivity.threshold");
                     }
                 }
             }
@@ -261,15 +261,15 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
     }
 
     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;

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

@@ -15,7 +15,7 @@ kafka.input.topic.name=DOS-SKETCH-RECORD
 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=dos-detection-job-220516-1
+kafka.input.group.id=2112080949
 #kafka.input.group.id=dos-detection-job-210813-1
 
 #发送kafka metrics并行度大小
@@ -37,8 +37,8 @@ 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
@@ -74,26 +74,26 @@ destination.ip.partition.num=10000
 data.center.id.num=15
 
 #IP mmdb库路径
-ip.mmdb.path=D:\\data\\dat\\bak\\
+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=10
-
-#基线敏感阈值
-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
+##静态敏感阈值，速率小于此值不报警
+#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.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
@@ -139,4 +139,13 @@ sasl.jaas.config.user=admin
 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();
+            }
+        }
+        */
+    }
+}