gfwleak/galaxy-tsg-olap-app-protocol-stat-traffic-merge
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化程序处理逻辑，以Metrics结构输出结果。（TSG-14799）

Browse Source
This commit is contained in:
qidaijie
2023-05-19 14:03:40 +08:00
parent 2f437b5d72
commit 5445972400
19 changed files with 295 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/com/zdjizhi/common/config/GlobalConfig.java
View File

@@ -24,6 +24,7 @@ public class GlobalConfig {
* System
*/
public static final Integer SOURCE_PARALLELISM = GlobalConfigLoad.getIntProperty(0, "source.parallelism");
public static final String MEASUREMENT_NAME = GlobalConfigLoad.getStringProperty(1, "measurement.name");
public static final Integer PARSE_PARALLELISM = GlobalConfigLoad.getIntProperty(0, "parse.parallelism");
public static final Integer WINDOW_PARALLELISM = GlobalConfigLoad.getIntProperty(0, "window.parallelism");
public static final Integer COUNT_WINDOW_TIME = GlobalConfigLoad.getIntProperty(0, "count.window.time");

18
src/main/java/com/zdjizhi/common/config/GlobalConfigLoad.java
View File

@@ -13,26 +13,26 @@ import java.util.Properties;
public final class GlobalConfigLoad {
private static Properties propKafka = new Properties();
private static Properties propDefault = new Properties();
private static Properties propService = new Properties();
public static String getStringProperty(Integer type, String key) {
static String getStringProperty(Integer type, String key) {
if (type == 0) {
return propService.getProperty(key);
} else if (type == 1) {
return propKafka.getProperty(key);
return propDefault.getProperty(key);
} else {
return null;
}
}
public static Integer getIntProperty(Integer type, String key) {
static Integer getIntProperty(Integer type, String key) {
if (type == 0) {
return Integer.parseInt(propService.getProperty(key));
} else if (type == 1) {
return Integer.parseInt(propKafka.getProperty(key));
return Integer.parseInt(propDefault.getProperty(key));
} else {
return null;
}
@@ -42,7 +42,7 @@ public final class GlobalConfigLoad {
if (type == 0) {
return Long.parseLong(propService.getProperty(key));
} else if (type == 1) {
return Long.parseLong(propKafka.getProperty(key));
return Long.parseLong(propDefault.getProperty(key));
} else {
return null;
}
@@ -52,7 +52,7 @@ public final class GlobalConfigLoad {
if (type == 0) {
return StringUtil.equals(propService.getProperty(key).toLowerCase().trim().toUpperCase(Locale.ENGLISH), "true");
} else if (type == 1) {
return StringUtil.equals(propKafka.getProperty(key).toLowerCase().trim().toUpperCase(Locale.ENGLISH), "true");
return StringUtil.equals(propDefault.getProperty(key).toLowerCase().trim().toUpperCase(Locale.ENGLISH), "true");
} else {
return null;
}
@@ -61,9 +61,9 @@ public final class GlobalConfigLoad {
static {
try {
propService.load(GlobalConfigLoad.class.getClassLoader().getResourceAsStream("service_flow_config.properties"));
propKafka.load(GlobalConfigLoad.class.getClassLoader().getResourceAsStream("default_config.properties"));
propDefault.load(GlobalConfigLoad.class.getClassLoader().getResourceAsStream("default_config.properties"));
} catch (IOException | RuntimeException e) {
propKafka = null;
propDefault = null;
propService = null;
}
}

257
src/main/java/com/zdjizhi/common/pojo/AppProtocol.java
View File

@@ -1,257 +0,0 @@
package com.zdjizhi.common.pojo;
import com.alibaba.fastjson.annotation.JSONField;
/**
* @author qidaijie
* @Package com.zdjizhi.common.pojo
* @Description:
* @date 2023/4/2116:06
*/
public class AppProtocol {
private Long timestamp;
private int vsys_id;
private String device_id;
private String device_group;
private String data_center;
private String protocol_stack_id;
private String app_name;
private Long sessions;
private Long in_bytes;
private Long out_bytes;
private Long in_pkts;
private Long out_pkts;
private Long c2s_pkts;
private Long s2c_pkts;
private Long c2s_bytes;
private Long s2c_bytes;
private Long c2s_fragments;
private Long s2c_fragments;
private Long c2s_tcp_lost_bytes;
private Long s2c_tcp_lost_bytes;
private Long c2s_tcp_ooorder_pkts;
private Long s2c_tcp_ooorder_pkts;
private Long c2s_tcp_retransmitted_pkts;
private Long s2c_tcp_retransmitted_pkts;
private Long c2s_tcp_retransmitted_bytes;
private Long s2c_tcp_retransmitted_bytes;
private String client_ip_sketch;
public Long getTimestamp() {
return timestamp;
}
public void setTimestamp(Long timestamp) {
this.timestamp = timestamp;
}
public int getVsys_id() {
return vsys_id;
}
public void setVsys_id(int vsys_id) {
this.vsys_id = vsys_id;
}
public String getDevice_id() {
return device_id;
}
public void setDevice_id(String device_id) {
this.device_id = device_id;
}
public String getDevice_group() {
return device_group;
}
public void setDevice_group(String device_group) {
this.device_group = device_group;
}
public String getData_center() {
return data_center;
}
public void setData_center(String data_center) {
this.data_center = data_center;
}
public String getProtocol_stack_id() {
return protocol_stack_id;
}
@JSONField(name = "protocol_label")
public void setProtocol_stack_id(String protocol_stack_id) {
this.protocol_stack_id = protocol_stack_id;
}
public String getApp_name() {
return app_name;
}
@JSONField(name = "app_full_path")
public void setApp_name(String app_name) {
this.app_name = app_name;
}
public Long getSessions() {
return sessions;
}
public void setSessions(Long sessions) {
this.sessions = sessions;
}
public Long getIn_bytes() {
return in_bytes;
}
public void setIn_bytes(Long in_bytes) {
this.in_bytes = in_bytes;
}
public Long getOut_bytes() {
return out_bytes;
}
public void setOut_bytes(Long out_bytes) {
this.out_bytes = out_bytes;
}
public Long getIn_pkts() {
return in_pkts;
}
public void setIn_pkts(Long in_pkts) {
this.in_pkts = in_pkts;
}
public Long getOut_pkts() {
return out_pkts;
}
public void setOut_pkts(Long out_pkts) {
this.out_pkts = out_pkts;
}
public Long getC2s_pkts() {
return c2s_pkts;
}
public void setC2s_pkts(Long c2s_pkts) {
this.c2s_pkts = c2s_pkts;
}
public Long getS2c_pkts() {
return s2c_pkts;
}
public void setS2c_pkts(Long s2c_pkts) {
this.s2c_pkts = s2c_pkts;
}
public Long getC2s_bytes() {
return c2s_bytes;
}
public void setC2s_bytes(Long c2s_bytes) {
this.c2s_bytes = c2s_bytes;
}
public Long getS2c_bytes() {
return s2c_bytes;
}
public void setS2c_bytes(Long s2c_bytes) {
this.s2c_bytes = s2c_bytes;
}
public Long getC2s_fragments() {
return c2s_fragments;
}
public void setC2s_fragments(Long c2s_fragments) {
this.c2s_fragments = c2s_fragments;
}
public Long getS2c_fragments() {
return s2c_fragments;
}
public void setS2c_fragments(Long s2c_fragments) {
this.s2c_fragments = s2c_fragments;
}
public Long getC2s_tcp_lost_bytes() {
return c2s_tcp_lost_bytes;
}
public void setC2s_tcp_lost_bytes(Long c2s_tcp_lost_bytes) {
this.c2s_tcp_lost_bytes = c2s_tcp_lost_bytes;
}
public Long getS2c_tcp_lost_bytes() {
return s2c_tcp_lost_bytes;
}
public void setS2c_tcp_lost_bytes(Long s2c_tcp_lost_bytes) {
this.s2c_tcp_lost_bytes = s2c_tcp_lost_bytes;
}
public Long getC2s_tcp_ooorder_pkts() {
return c2s_tcp_ooorder_pkts;
}
public void setC2s_tcp_ooorder_pkts(Long c2s_tcp_ooorder_pkts) {
this.c2s_tcp_ooorder_pkts = c2s_tcp_ooorder_pkts;
}
public Long getS2c_tcp_ooorder_pkts() {
return s2c_tcp_ooorder_pkts;
}
public void setS2c_tcp_ooorder_pkts(Long s2c_tcp_ooorder_pkts) {
this.s2c_tcp_ooorder_pkts = s2c_tcp_ooorder_pkts;
}
public Long getC2s_tcp_retransmitted_pkts() {
return c2s_tcp_retransmitted_pkts;
}
public void setC2s_tcp_retransmitted_pkts(Long c2s_tcp_retransmitted_pkts) {
this.c2s_tcp_retransmitted_pkts = c2s_tcp_retransmitted_pkts;
}
public Long getS2c_tcp_retransmitted_pkts() {
return s2c_tcp_retransmitted_pkts;
}
public void setS2c_tcp_retransmitted_pkts(Long s2c_tcp_retransmitted_pkts) {
this.s2c_tcp_retransmitted_pkts = s2c_tcp_retransmitted_pkts;
}
public Long getC2s_tcp_retransmitted_bytes() {
return c2s_tcp_retransmitted_bytes;
}
public void setC2s_tcp_retransmitted_bytes(Long c2s_tcp_retransmitted_bytes) {
this.c2s_tcp_retransmitted_bytes = c2s_tcp_retransmitted_bytes;
}
public Long getS2c_tcp_retransmitted_bytes() {
return s2c_tcp_retransmitted_bytes;
}
public void setS2c_tcp_retransmitted_bytes(Long s2c_tcp_retransmitted_bytes) {
this.s2c_tcp_retransmitted_bytes = s2c_tcp_retransmitted_bytes;
}
public String getClient_ip_sketch() {
return client_ip_sketch;
}
public void setClient_ip_sketch(String client_ip_sketch) {
this.client_ip_sketch = client_ip_sketch;
}
}

8
src/main/java/com/zdjizhi/common/pojo/Fields.java
View File

@@ -26,9 +26,9 @@ public class Fields {
private Long s2c_tcp_retransmitted_pkts;
private Long c2s_tcp_retransmitted_bytes;
private Long s2c_tcp_retransmitted_bytes;
private byte[] client_ip_sketch;
private String client_ip_sketch;
public Fields(Long sessions, Long in_bytes, Long out_bytes, Long in_pkts, Long out_pkts, Long c2s_pkts, Long s2c_pkts, Long c2s_bytes, Long s2c_bytes, Long c2s_fragments, Long s2c_fragments, Long c2s_tcp_lost_bytes, Long s2c_tcp_lost_bytes, Long c2s_tcp_ooorder_pkts, Long s2c_tcp_ooorder_pkts, Long c2s_tcp_retransmitted_pkts, Long s2c_tcp_retransmitted_pkts, Long c2s_tcp_retransmitted_bytes, Long s2c_tcp_retransmitted_bytes, byte[] client_ip_sketch) {
public Fields(Long sessions, Long in_bytes, Long out_bytes, Long in_pkts, Long out_pkts, Long c2s_pkts, Long s2c_pkts, Long c2s_bytes, Long s2c_bytes, Long c2s_fragments, Long s2c_fragments, Long c2s_tcp_lost_bytes, Long s2c_tcp_lost_bytes, Long c2s_tcp_ooorder_pkts, Long s2c_tcp_ooorder_pkts, Long c2s_tcp_retransmitted_pkts, Long s2c_tcp_retransmitted_pkts, Long c2s_tcp_retransmitted_bytes, Long s2c_tcp_retransmitted_bytes, String client_ip_sketch) {
this.sessions = sessions;
this.in_bytes = in_bytes;
this.out_bytes = out_bytes;
@@ -203,11 +203,11 @@ public class Fields {
this.s2c_tcp_retransmitted_bytes = s2c_tcp_retransmitted_bytes;
}
public byte[] getClient_ip_sketch() {
public String getClient_ip_sketch() {
return client_ip_sketch;
}
public void setClient_ip_sketch(byte[] client_ip_sketch) {
public void setClient_ip_sketch(String client_ip_sketch) {
this.client_ip_sketch = client_ip_sketch;
}
}

54
src/main/java/com/zdjizhi/common/pojo/Metrics.java Normal file
View File

@@ -0,0 +1,54 @@
package com.zdjizhi.common.pojo;
/**
* @author qidaijie
* @Package com.zdjizhi.common.pojo
* @Description:
* @date 2023/5/911:42
*/
public class Metrics {
private String name;
private Tags tags;
private Fields fields;
private long timestamp;
public Metrics(String name, Tags tags, Fields fields, long timestamp) {
this.name = name;
this.tags = tags;
this.fields = fields;
this.timestamp = timestamp;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Tags getTags() {
return tags;
}
public void setTags(Tags tags) {
this.tags = tags;
}
public Fields getFields() {
return fields;
}
public void setFields(Fields fields) {
this.fields = fields;
}
public long getTimestamp() {
return timestamp;
}
public void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
}

35
src/main/java/com/zdjizhi/common/pojo/Tags.java
View File

@@ -1,5 +1,7 @@
package com.zdjizhi.common.pojo;
import com.alibaba.fastjson.annotation.JSONField;
/**
* @author qidaijie
* @Package com.zdjizhi.common.pojo
@@ -11,16 +13,16 @@ public class Tags {
private String device_id;
private String device_group;
private String data_center;
private String protocol_label;
private String app_full_path;
private String protocol_stack_id;
private String app_name;
public Tags(int vsys_id, String device_id, String device_group, String data_center, String protocol_label, String app_full_path) {
public Tags(int vsys_id, String device_id, String device_group, String data_center, String protocol_stack_id, String app_name) {
this.vsys_id = vsys_id;
this.device_id = device_id;
this.device_group = device_group;
this.data_center = data_center;
this.protocol_label = protocol_label;
this.app_full_path = app_full_path;
this.protocol_stack_id = protocol_stack_id;
this.app_name = app_name;
}
public int getVsys_id() {
@@ -55,19 +57,26 @@ public class Tags {
this.data_center = data_center;
}
public String getProtocol_label() {
return protocol_label;
public String getProtocol_stack_id() {
return protocol_stack_id;
}
public void setProtocol_label(String protocol_label) {
this.protocol_label = protocol_label;
@JSONField(name = "protocol_label")
public void setProtocol_stack_id(String protocol_stack_id) {
this.protocol_stack_id = protocol_stack_id;
}
public String getApp_full_path() {
return app_full_path;
public String getApp_name() {
return app_name;
}
public void setApp_full_path(String app_full_path) {
this.app_full_path = app_full_path;
@JSONField(name = "app_full_path")
public void setApp_name(String app_name) {
this.app_name = app_name;
}
@Override
public String toString() {
return vsys_id + device_id + device_group + data_center + protocol_stack_id + app_name;
}
}

8
src/main/java/com/zdjizhi/topology/ApplicationProtocolTopology.java
View File

@@ -3,7 +3,9 @@ package com.zdjizhi.topology;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.zdjizhi.common.config.GlobalConfig;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.common.pojo.Metrics;
import com.zdjizhi.common.pojo.Tags;
import com.zdjizhi.utils.functions.filter.DataTypeFilter;
import com.zdjizhi.utils.functions.keyby.DimensionKeyBy;
import com.zdjizhi.utils.functions.map.MetricsParseMap;
@@ -41,10 +43,10 @@ public class ApplicationProtocolTopology {
.name("appProtocolFilter").setParallelism(GlobalConfig.SOURCE_PARALLELISM);
SingleOutputStreamOperator<Tuple2<String, AppProtocol>> parseDataMap = appProtocolFilter.map(new MetricsParseMap())
SingleOutputStreamOperator<Tuple2<Tags, Fields>> parseDataMap = appProtocolFilter.map(new MetricsParseMap())
.name("ParseDataMap").setParallelism(GlobalConfig.PARSE_PARALLELISM);
SingleOutputStreamOperator<AppProtocol> dispersionCountWindow = parseDataMap.keyBy(new DimensionKeyBy())
SingleOutputStreamOperator<Metrics> dispersionCountWindow = parseDataMap.keyBy(new DimensionKeyBy())
.window(TumblingProcessingTimeWindows.of(Time.seconds(GlobalConfig.COUNT_WINDOW_TIME)))
.reduce(new DispersionCountWindow(), new MergeCountWindow())
.name("DispersionCountWindow")

11
src/main/java/com/zdjizhi/utils/functions/keyby/DimensionKeyBy.java
View File

@@ -1,14 +1,9 @@
package com.zdjizhi.utils.functions.keyby;
import com.alibaba.fastjson2.JSONObject;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.common.pojo.Tags;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import java.util.Map;
/**
* @author qidaijie
@@ -16,11 +11,11 @@ import java.util.Map;
* @Description:
* @date 2021/7/2112:13
*/
public class DimensionKeyBy implements KeySelector<Tuple2<String, AppProtocol>, String> {
public class DimensionKeyBy implements KeySelector<Tuple2<Tags, Fields>, String> {
@Override
public String getKey(Tuple2<String, AppProtocol> value) throws Exception {
public String getKey(Tuple2<Tags, Fields> value) throws Exception {
//以map拼接的key分组
return value.f0;
return value.f0.toString();
}
}

25
src/main/java/com/zdjizhi/utils/functions/map/MetricsParseMap.java
View File

@@ -4,7 +4,8 @@ import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.common.pojo.Tags;
import com.zdjizhi.utils.StringUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -16,30 +17,26 @@ import org.apache.flink.api.java.tuple.Tuple2;
* @Description:
* @date 2021/5/2715:01
*/
public class MetricsParseMap implements MapFunction<String, Tuple2<String, AppProtocol>> {
public class MetricsParseMap implements MapFunction<String, Tuple2<Tags, Fields>> {
private static final Log logger = LogFactory.get();
@Override
@SuppressWarnings("unchecked")
public Tuple2<String, AppProtocol> map(String message) {
public Tuple2<Tags, Fields> map(String message) {
try {
JSONObject originalLog = JSON.parseObject(message);
JSONObject fieldsObject = JSONObject.parseObject(originalLog.getString("fields"));
JSONObject tagsObject = JSONObject.parseObject(originalLog.getString("tags"));
fieldsObject.putAll(tagsObject);
Fields fields = JSONObject.parseObject(originalLog.getString("fields"), Fields.class);
Tags tags = JSONObject.parseObject(originalLog.getString("tags"), Tags.class);
AppProtocol appProtocol = JSON.to(AppProtocol.class, fieldsObject);
String appFullPath = appProtocol.getApp_name();
String appFullPath = tags.getApp_name();
if (StringUtil.isNotBlank(appFullPath)) {
String appName = appFullPath.substring(appFullPath.lastIndexOf(".") + 1);
String protocolLabel = appProtocol.getProtocol_stack_id();
String protocolLabel = tags.getProtocol_stack_id();
appProtocol.setApp_name(appName);
appProtocol.setProtocol_stack_id(protocolLabel.concat(".").concat(appFullPath));
tags.setApp_name(appName);
tags.setProtocol_stack_id(protocolLabel.concat(".").concat(appFullPath));
}
return new Tuple2<>(tagsObject.toJSONString(), appProtocol);
return new Tuple2<>(tags, fields);
} catch (RuntimeException e) {
logger.error("An error occurred in the original log parsing reorganization,error message is:" + e);
return new Tuple2<>(null, null);

37
src/main/java/com/zdjizhi/utils/functions/map/ResultFlatMap.java
View File

@@ -5,9 +5,9 @@ import cn.hutool.log.LogFactory;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONWriter;
import com.zdjizhi.common.config.GlobalConfig;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Metrics;
import com.zdjizhi.common.pojo.Tags;
import com.zdjizhi.utils.StringUtil;
import com.zdjizhi.utils.general.FormatConverterUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
@@ -17,20 +17,16 @@ import org.apache.flink.util.Collector;
* @Description:
* @date 2021/7/2114:52
*/
public class ResultFlatMap implements FlatMapFunction<AppProtocol, String> {
public class ResultFlatMap implements FlatMapFunction<Metrics, String> {
private static final Log logger = LogFactory.get();
@Override
@SuppressWarnings("unchecked")
public void flatMap(AppProtocol appProtocol, Collector<String> out) throws Exception {
public void flatMap(Metrics metrics, Collector<String> out) throws Exception {
try {
JSONObject tags = FormatConverterUtil.getTags(appProtocol);
JSONObject conversion = FormatConverterUtil.structureConversion(appProtocol);
String protocolStackId = tags.getString("protocol_stack_id");
out.collect(FormatConverterUtil.updateTagsData(conversion, tags));
tags.remove("app_name");
Tags tags = metrics.getTags();
String protocolStackId = tags.getProtocol_stack_id();
out.collect(getResultJson(metrics));
tags.setApp_name(null);
StringBuilder stringBuilder = new StringBuilder();
String[] protocolIds = protocolStackId.split(GlobalConfig.PROTOCOL_SPLITTER);
@@ -38,13 +34,14 @@ public class ResultFlatMap implements FlatMapFunction<AppProtocol, String> {
for (int i = 0; i < protocolIdsNum - 1; i++) {
if (StringUtil.isBlank(stringBuilder.toString())) {
stringBuilder.append(protocolIds[i]);
tags.put("protocol_stack_id", stringBuilder.toString());
out.collect(FormatConverterUtil.updateTagsData(conversion, tags));
tags.setProtocol_stack_id(stringBuilder.toString());
metrics.setTags(tags);
out.collect(getResultJson(metrics));
} else {
stringBuilder.append(".").append(protocolIds[i]);
tags.put("protocol_stack_id", stringBuilder.toString());
conversion.put("tags", tags);
out.collect(FormatConverterUtil.updateTagsData(conversion, tags));
tags.setProtocol_stack_id(stringBuilder.toString());
metrics.setTags(tags);
out.collect(getResultJson(metrics));
}
}
} catch (RuntimeException e) {
@@ -52,4 +49,10 @@ public class ResultFlatMap implements FlatMapFunction<AppProtocol, String> {
e.printStackTrace();
}
}
private static String getResultJson(Metrics metrics) {
return JSONObject.toJSONString(metrics
, JSONWriter.Feature.WriteNullStringAsEmpty
, JSONWriter.Feature.WriteNullNumberAsZero);
}
}

15
src/main/java/com/zdjizhi/utils/functions/statistics/DispersionCountWindow.java
View File

@@ -2,7 +2,8 @@ package com.zdjizhi.utils.functions.statistics;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.common.pojo.Tags;
import com.zdjizhi.utils.general.MetricUtil;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -13,18 +14,18 @@ import org.apache.flink.api.java.tuple.Tuple2;
* @Description:
* @date 2023/4/2314:02
*/
public class DispersionCountWindow implements ReduceFunction<Tuple2<String, AppProtocol>> {
public class DispersionCountWindow implements ReduceFunction<Tuple2<Tags, Fields>> {
private static final Log logger = LogFactory.get();
@Override
public Tuple2<String, AppProtocol> reduce(Tuple2<String, AppProtocol> value1, Tuple2<String, AppProtocol> value2) throws Exception {
public Tuple2<Tags, Fields> reduce(Tuple2<Tags, Fields> value1, Tuple2<Tags, Fields> value2) throws Exception {
try {
AppProtocol cacheData = value1.f1;
AppProtocol newData = value2.f1;
Fields cacheData = value1.f1;
Fields newData = value2.f1;
MetricUtil.statisticsMetrics(cacheData, newData);
Fields metricsResult = MetricUtil.statisticsMetrics(cacheData, newData);
return new Tuple2<>(value1.f0, cacheData);
return new Tuple2<>(value1.f0, metricsResult);
} catch (RuntimeException e) {
logger.error("An exception occurred during incremental aggregation! The message is:" + e.getMessage());
return value1;

18
src/main/java/com/zdjizhi/utils/functions/statistics/MergeCountWindow.java
View File

@@ -2,7 +2,10 @@ package com.zdjizhi.utils.functions.statistics;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.config.GlobalConfig;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.common.pojo.Metrics;
import com.zdjizhi.common.pojo.Tags;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -14,17 +17,18 @@ import org.apache.flink.util.Collector;
* @Description:
* @date 2023/4/2314:43
*/
public class MergeCountWindow extends ProcessWindowFunction<Tuple2<String, AppProtocol>, AppProtocol, String, TimeWindow> {
public class MergeCountWindow extends ProcessWindowFunction<Tuple2<Tags, Fields>, Metrics, String, TimeWindow> {
private static final Log logger = LogFactory.get();
@Override
public void process(String windowKey, Context context, Iterable<Tuple2<String, AppProtocol>> input, Collector<AppProtocol> output) throws Exception {
public void process(String windowKey, Context context, Iterable<Tuple2<Tags, Fields>> input, Collector<Metrics> output) throws Exception {
try {
Long endTime = context.window().getEnd() / 1000;
for (Tuple2<String, AppProtocol> tuple : input) {
AppProtocol data = tuple.f1;
data.setTimestamp(endTime);
output.collect(data);
for (Tuple2<Tags, Fields> tuple : input) {
Tags tags = tuple.f0;
Fields fields = tuple.f1;
Metrics metrics = new Metrics(GlobalConfig.MEASUREMENT_NAME, tags, fields, endTime);
output.collect(metrics);
}
} catch (RuntimeException e) {
logger.error("An exception occurred in the process of full data aggregation! The message is:" + e.getMessage());

91
src/main/java/com/zdjizhi/utils/general/FormatConverterUtil.java
View File

@@ -1,91 +0,0 @@
package com.zdjizhi.utils.general;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONWriter;
import com.zdjizhi.common.pojo.AppProtocol;
/**
* @author qidaijie
* @Package com.zdjizhi.utils.general
* @Description:
* @date 2023/5/519:04
*/
public class FormatConverterUtil {
/**
* 生成tags类型数据
*
* @param appProtocol 结果集
* @return tags结果
*/
public static JSONObject getTags(AppProtocol appProtocol) {
JSONObject tags = new JSONObject();
tags.fluentPut("vsys_id", appProtocol.getVsys_id())
.fluentPut("device_id", appProtocol.getDevice_id())
.fluentPut("device_group", appProtocol.getDevice_group())
.fluentPut("data_center", appProtocol.getData_center())
.fluentPut("protocol_stack_id", appProtocol.getProtocol_stack_id())
.fluentPut("app_name", appProtocol.getApp_name());
return tags;
}
/**
* 将数据结构转换为最终的结构
*
* @param appProtocol 结果集
* @return 结果数据
*/
public static JSONObject structureConversion(AppProtocol appProtocol) {
JSONObject metrics = new JSONObject();
JSONObject fields = new JSONObject();
fields.fluentPut("sessions", appProtocol.getSessions())
.fluentPut("in_bytes", appProtocol.getIn_bytes())
.fluentPut("out_bytes", appProtocol.getOut_bytes())
.fluentPut("in_pkts", appProtocol.getIn_pkts())
.fluentPut("out_pkts", appProtocol.getOut_pkts())
.fluentPut("c2s_bytes", appProtocol.getC2s_bytes())
.fluentPut("s2c_bytes", appProtocol.getS2c_bytes())
.fluentPut("c2s_pkts", appProtocol.getC2s_pkts())
.fluentPut("s2c_pkts", appProtocol.getS2c_pkts())
.fluentPut("c2s_fragments", appProtocol.getC2s_fragments())
.fluentPut("s2c_fragments", appProtocol.getS2c_fragments())
.fluentPut("c2s_tcp_lost_bytes", appProtocol.getC2s_tcp_lost_bytes())
.fluentPut("s2c_tcp_lost_bytes", appProtocol.getS2c_tcp_lost_bytes())
.fluentPut("c2s_tcp_ooorder_pkts", appProtocol.getC2s_tcp_ooorder_pkts())
.fluentPut("s2c_tcp_ooorder_pkts", appProtocol.getS2c_tcp_ooorder_pkts())
.fluentPut("c2s_tcp_retransmitted_pkts", appProtocol.getC2s_tcp_retransmitted_bytes())
.fluentPut("s2c_tcp_retransmitted_pkts", appProtocol.getS2c_tcp_retransmitted_bytes())
.fluentPut("c2s_tcp_retransmitted_bytes", appProtocol.getC2s_tcp_retransmitted_pkts())
.fluentPut("s2c_tcp_retransmitted_bytes", appProtocol.getS2c_tcp_retransmitted_pkts())
.fluentPut("client_ip_sketch", appProtocol.getClient_ip_sketch());
metrics.put("timestamp", appProtocol.getTimestamp());
metrics.put("name", "application_protocol_stat");
metrics.fluentPut("timestamp", appProtocol.getTimestamp())
.fluentPut("name", "application_protocol_stat")
.fluentPut("fields", fields);
return metrics;
}
/**
* 更新结果集tags数据（不同协议层级），并输出json
*
* @param conversion 结果集
* @param tags tags结果
* @return 结果json
*/
public static String updateTagsData(JSONObject conversion, JSONObject tags) {
conversion.put("tags", tags);
return JSONObject.toJSONString(conversion
, JSONWriter.Feature.WriteNullStringAsEmpty
, JSONWriter.Feature.WriteNullNumberAsZero);
}
}

61
src/main/java/com/zdjizhi/utils/general/MetricUtil.java
View File

@@ -1,9 +1,8 @@
package com.zdjizhi.utils.general;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.zdjizhi.common.pojo.AppProtocol;
import com.zdjizhi.common.pojo.Fields;
import com.zdjizhi.utils.StringUtil;
import org.apache.datasketches.hll.HllSketch;
import org.apache.datasketches.hll.Union;
@@ -27,35 +26,47 @@ public class MetricUtil {
* @param cacheData 缓存中数据
* @param newData 新数据
*/
public static void statisticsMetrics(AppProtocol cacheData, AppProtocol newData) {
cacheData.setSessions(MetricUtil.longSum(cacheData.getSessions(), newData.getSessions()));
public static Fields statisticsMetrics(Fields cacheData, Fields newData) {
cacheData.setIn_bytes(MetricUtil.longSum(cacheData.getIn_bytes(), newData.getIn_bytes()));
cacheData.setOut_pkts(MetricUtil.longSum(cacheData.getOut_bytes(), newData.getOut_bytes()));
cacheData.setIn_pkts(MetricUtil.longSum(cacheData.getIn_pkts(), newData.getIn_pkts()));
cacheData.setOut_pkts(MetricUtil.longSum(cacheData.getOut_pkts(), newData.getOut_pkts()));
Long sessions = MetricUtil.longSum(cacheData.getSessions(), newData.getSessions());
cacheData.setC2s_bytes(MetricUtil.longSum(cacheData.getC2s_bytes(), newData.getC2s_bytes()));
cacheData.setS2c_bytes(MetricUtil.longSum(cacheData.getS2c_bytes(), newData.getS2c_bytes()));
cacheData.setC2s_pkts(MetricUtil.longSum(cacheData.getC2s_pkts(), newData.getC2s_pkts()));
cacheData.setS2c_pkts(MetricUtil.longSum(cacheData.getS2c_pkts(), newData.getS2c_pkts()));
Long inBytes = MetricUtil.longSum(cacheData.getIn_bytes(), newData.getIn_bytes());
Long outBytes = MetricUtil.longSum(cacheData.getOut_bytes(), newData.getOut_bytes());
Long inPkts = MetricUtil.longSum(cacheData.getIn_pkts(), newData.getIn_pkts());
Long outPkts = MetricUtil.longSum(cacheData.getOut_pkts(), newData.getOut_pkts());
cacheData.setC2s_fragments(MetricUtil.longSum(cacheData.getC2s_fragments(), newData.getC2s_fragments()));
cacheData.setS2c_fragments(MetricUtil.longSum(cacheData.getS2c_fragments(), newData.getS2c_fragments()));
cacheData.setC2s_tcp_lost_bytes(MetricUtil.longSum(cacheData.getC2s_tcp_lost_bytes(), newData.getC2s_tcp_lost_bytes()));
cacheData.setS2c_tcp_lost_bytes(MetricUtil.longSum(cacheData.getS2c_tcp_lost_bytes(), newData.getS2c_tcp_lost_bytes()));
Long c2sBytes = MetricUtil.longSum(cacheData.getC2s_bytes(), newData.getC2s_bytes());
Long s2cBytes = MetricUtil.longSum(cacheData.getS2c_bytes(), newData.getS2c_bytes());
Long c2sPkts = MetricUtil.longSum(cacheData.getC2s_pkts(), newData.getC2s_pkts());
Long s2cPkts = MetricUtil.longSum(cacheData.getS2c_pkts(), newData.getS2c_pkts());
cacheData.setC2s_tcp_ooorder_pkts(MetricUtil.longSum(cacheData.getC2s_tcp_ooorder_pkts(), newData.getC2s_tcp_ooorder_pkts()));
cacheData.setS2c_tcp_ooorder_pkts(MetricUtil.longSum(cacheData.getS2c_tcp_ooorder_pkts(), newData.getS2c_tcp_ooorder_pkts()));
Long c2sFragments = MetricUtil.longSum(cacheData.getC2s_fragments(), newData.getC2s_fragments());
Long s2cFragments = MetricUtil.longSum(cacheData.getS2c_fragments(), newData.getS2c_fragments());
cacheData.setC2s_tcp_retransmitted_bytes(MetricUtil.longSum(cacheData.getC2s_tcp_retransmitted_bytes(), newData.getC2s_tcp_retransmitted_bytes()));
cacheData.setS2c_tcp_retransmitted_bytes(MetricUtil.longSum(cacheData.getS2c_tcp_retransmitted_bytes(), newData.getS2c_tcp_retransmitted_bytes()));
Long c2sTcpLostBytes = MetricUtil.longSum(cacheData.getC2s_tcp_lost_bytes(), newData.getC2s_tcp_lost_bytes());
Long s2cTcpLostBytes = MetricUtil.longSum(cacheData.getS2c_tcp_lost_bytes(), newData.getS2c_tcp_lost_bytes());
cacheData.setC2s_tcp_retransmitted_pkts(MetricUtil.longSum(cacheData.getC2s_tcp_retransmitted_pkts(), newData.getC2s_tcp_retransmitted_pkts()));
cacheData.setS2c_tcp_retransmitted_pkts(MetricUtil.longSum(cacheData.getS2c_tcp_retransmitted_pkts(), newData.getS2c_tcp_retransmitted_pkts()));
Long c2sTcpooorderPkts = MetricUtil.longSum(cacheData.getC2s_tcp_ooorder_pkts(), newData.getC2s_tcp_ooorder_pkts());
Long s2cTcpooorderPkts = MetricUtil.longSum(cacheData.getS2c_tcp_ooorder_pkts(), newData.getS2c_tcp_ooorder_pkts());
cacheData.setClient_ip_sketch(MetricUtil.hllSketchUnion(cacheData.getClient_ip_sketch(), newData.getClient_ip_sketch()));
Long c2sTcpretransmittedPkts = MetricUtil.longSum(cacheData.getC2s_tcp_retransmitted_pkts(), newData.getC2s_tcp_retransmitted_pkts());
Long s2cTcpretransmittedPkts = MetricUtil.longSum(cacheData.getS2c_tcp_retransmitted_pkts(), newData.getS2c_tcp_retransmitted_pkts());
Long c2sTcpretransmittedBytes = MetricUtil.longSum(cacheData.getC2s_tcp_retransmitted_bytes(), newData.getC2s_tcp_retransmitted_bytes());
Long s2cTcpretransmittedBytes = MetricUtil.longSum(cacheData.getS2c_tcp_retransmitted_bytes(), newData.getS2c_tcp_retransmitted_bytes());
String clientIpSketch = MetricUtil.hllSketchUnion(cacheData.getClient_ip_sketch(), newData.getClient_ip_sketch());
return new Fields(sessions,
inBytes, outBytes, inPkts, outPkts,
c2sPkts, s2cPkts, c2sBytes, s2cBytes,
c2sFragments, s2cFragments,
c2sTcpLostBytes, s2cTcpLostBytes,
c2sTcpooorderPkts, s2cTcpooorderPkts,
c2sTcpretransmittedPkts, s2cTcpretransmittedPkts,
c2sTcpretransmittedBytes, s2cTcpretransmittedBytes,
clientIpSketch);
}
/**
@@ -66,9 +77,9 @@ public class MetricUtil {
* @return value1 + value2
*/
private static Long longSum(Long value1, Long value2) {
Long result = 0L;
Long result;
try {
if (value1 > 0 && value2 > 0) {
if (value1 >= 0 && value2 >= 0) {
result = value1 + value2;
} else {
result = value1;