增加一元组作为基线生成数据源

3 files changed, 20 insertions, 15 deletions

diff --git a/src/main/java/com/zdjizhi/etl/DosDetection.java b/src/main/java/com/zdjizhi/etl/DosDetection.java
index d173399..df141ad 100644
--- a/src/main/java/com/zdjizhi/etl/DosDetection.java
+++ b/src/main/java/com/zdjizhi/etl/DosDetection.java
@@ -108,7 +108,8 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
         Integer base = getBaseValue(floodTypeTup, value);
         long sketchSessions = value.getSketch_sessions();
-        return sketchSessions > CommonConfig.SENSITIVITY_THRESHOLD ? getDosEventLog(value, base, sketchSessions - base, "baseline"):Tuple2.of(Severity.NORMAL, null);
+        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) {
diff --git a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
index aab3e88..df13909 100644
--- a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
@@ -69,15 +69,19 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
         HashSet<String> sourceIpSet = new HashSet<>();
         try {
             for (DosSketchLog newSketchLog : elements){
-                sessions += newSketchLog.getSketch_sessions();
-                packets += newSketchLog.getSketch_packets();
-                bytes += newSketchLog.getSketch_bytes();
-                startTime = newSketchLog.getSketch_start_time();
-                duration = newSketchLog.getSketch_duration();
-                cnt += 1;
-                if (sourceIpSet.size() < CommonConfig.SOURCE_IP_LIST_LIMIT){
-                    sourceIpSet.add(newSketchLog.getSource_ip());
+                String sourceIp = newSketchLog.getSource_ip();
+                if ("0.0.0.0".equals(sourceIp) || "::".equals(sourceIp)){
+                    sessions += newSketchLog.getSketch_sessions();
+                    packets += newSketchLog.getSketch_packets();
+                    bytes += newSketchLog.getSketch_bytes();
+                    startTime = newSketchLog.getSketch_start_time();
+                    duration = newSketchLog.getSketch_duration();
+                }else {
+                    if (sourceIpSet.size() < CommonConfig.SOURCE_IP_LIST_LIMIT){
+                        sourceIpSet.add(sourceIp);
+                    }
                 }
+                cnt += 1;
             }
             String sourceIpList = StringUtils.join(sourceIpSet, ",");
 //            return Tuple6.of(sessions/cnt/duration,packets/cnt/duration,bytes/cnt/duration,sourceIpList,startTime,duration);
diff --git a/src/main/resources/common.properties b/src/main/resources/common.properties
index 4f781b4..dfd3ef4 100644
--- a/src/main/resources/common.properties
+++ b/src/main/resources/common.properties
@@ -22,15 +22,15 @@ kafka.input.group.id=2108231709
 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=test
+#kafka.output.event.topic.name=DOS-EVENT
+kafka.output.event.topic.name=test
 
 #kafka输出地址
 kafka.output.bootstrap.servers=192.168.44.12:9092
@@ -45,7 +45,7 @@ hbase.client.operation.timeout=30000
 hbase.client.scanner.timeout.period=30000
 
 ##hbase baseline表名
-hbase.baseline.table.name=ddos_traffic_baselines
+hbase.baseline.table.name=dos:ddos_traffic_baselines
 
 #读取baseline限制
 hbase.baseline.total.num=1000000
@@ -76,7 +76,7 @@ ip.mmdb.path=D:\\data\\dat\\
 #ip.mmdb.path=/home/bigdata/wlh/topology/dos-detection/dat/
 
 #敏感阈值，速率小于此值不报警
-sensitivity.threshold=100
+sensitivity.threshold=1
 
 #基于baseline判定dos攻击的上下限
 baseline.sessions.minor.threshold=0.1