新增敏感阈值，过滤告警信息

修改计算平均值方式，先聚合再平均

src/main/java/com/zdjizhi/common/CommonConfig.java

@@ -39,6 +39,8 @@ public class CommonConfig {
 
     public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");
 
+    public static final int SENSITIVITY_THRESHOLD = CommonConfigurations.getIntProperty("sensitivity.threshold");
+
     public static final double BASELINE_SESSIONS_MINOR_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.minor.threshold");
     public static final double BASELINE_SESSIONS_WARNING_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.warning.threshold");
     public static final double BASELINE_SESSIONS_MAJOR_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.major.threshold");

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

@@ -85,29 +85,30 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
 
     private DosEventLog mergeFinalResult(Tuple2<Severity, DosEventLog> eventLogByBaseline, Tuple2<Severity, DosEventLog> eventLogByStaticThreshold) {
         if (eventLogByBaseline.f0.score > eventLogByStaticThreshold.f0.score) {
-            mergeCondition(eventLogByBaseline.f1, eventLogByStaticThreshold.f1);
             logger.info("merge eventLogByBaseline {} \neventLogByStaticThreshold {}",eventLogByBaseline,eventLogByStaticThreshold);
-            return eventLogByBaseline.f1;
+            return mergeCondition(eventLogByBaseline.f1, eventLogByStaticThreshold.f1);
         } else {
-            mergeCondition(eventLogByStaticThreshold.f1, eventLogByBaseline.f1);
             logger.info("merge eventLogByStaticThreshold {} \neventLogByBaseline {}",eventLogByStaticThreshold,eventLogByBaseline);
-            return eventLogByStaticThreshold.f1;
+            return mergeCondition(eventLogByStaticThreshold.f1, eventLogByBaseline.f1);
         }
     }
 
-    private void mergeCondition(DosEventLog log1, DosEventLog log2) {
+    private DosEventLog mergeCondition(DosEventLog log1, DosEventLog log2) {
         if (log1 != null && log2 != null) {
             String conditions1 = log1.getConditions();
             String conditions2 = log2.getConditions();
             log1.setConditions(conditions1 + " and " + conditions2);
+        }else if (log1 == null && log2 != null){
+            log1 = log2;
         }
+        return log1;
     }
 
     private Tuple2<Severity, DosEventLog> getDosEventLogByBaseline(DosSketchLog value, String destinationIp, String attackType) {
         Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
         Integer base = getBaseValue(floodTypeTup, value);
-        long diff = value.getSketch_sessions() - base;
+        long sketchSessions = value.getSketch_sessions();
-        return getDosEventLog(value, base, diff, "baseline");
+        return sketchSessions > CommonConfig.SENSITIVITY_THRESHOLD ? getDosEventLog(value, base, sketchSessions - base, "baseline"):Tuple2.of(Severity.NORMAL, null);
     }
 
     private Tuple2<Severity, DosEventLog> getDosEventLogByStaticThreshold(DosSketchLog value, Map<String, DosDetectionThreshold> thresholdMap) {

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

@@ -80,7 +80,7 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
                 }
             }
             String sourceIpList = StringUtils.join(sourceIpSet, ",");
-            return Tuple6.of(sessions/cnt,packets/cnt,bytes/cnt,sourceIpList,startTime,duration);
+            return Tuple6.of(sessions/cnt/duration,packets/cnt/duration,bytes/cnt/duration,sourceIpList,startTime,duration);
         }catch (Exception e){
             logger.error("聚合中间结果集失败 {}",e);
         }

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

@@ -65,9 +65,9 @@ public class ParseSketchLog {
                         long sketchBytes = Long.parseLong(obj.get("sketch_bytes").toString());
                         dosSketchLog.setSource_ip(sourceIp);
                         dosSketchLog.setDestination_ip(destinationIp);
-                        dosSketchLog.setSketch_sessions(sketchSessions/sketchDuration);
+                        dosSketchLog.setSketch_sessions(sketchSessions);
-                        dosSketchLog.setSketch_packets(sketchPackets/sketchDuration);
+                        dosSketchLog.setSketch_packets(sketchPackets);
-                        dosSketchLog.setSketch_bytes(sketchBytes*8/sketchDuration);
+                        dosSketchLog.setSketch_bytes(sketchBytes);
                         collector.collect(dosSketchLog);
                         logger.debug("数据解析成功：{}",dosSketchLog.toString());
                     }

src/main/resources/common.properties

@@ -75,6 +75,9 @@ ip.mmdb.path=D:\\data\\dat\\
 #ip.mmdb.path=/home/bigdata/topology/dat/
 #ip.mmdb.path=/home/bigdata/wlh/topology/dos-detection/dat/
 
+#敏感阈值，速率小于此值不报警
+sensitivity.threshold=100
+
 #基于baseline判定dos攻击的上下限
 baseline.sessions.minor.threshold=0.1
 baseline.sessions.warning.threshold=0.5