问题背景:cep模板 对数据源设置分组条件后,告警的数据,和分组条件对不上, 掺杂了,其他的不同组的数据,产生了告警
策略条件:
选择了两个kafka的的topic的数据作为数据源,
对A 数据源 test-topic1, 进行条件过滤, 过滤条件为:login_type = 1
对B 数据源 test-topic2,进行条件过滤,过滤条件为:login_type = 2
分组条件 为 src_ip,hostname两个字段进行分组
进行followby 关联。时间关联的最大时间间隔为 60秒
运行并行度设置为3
通过SourceStream打印的原始数据:
2> {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666859021060,"create_time_desc":"2022-10-27 16:23:41","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type":"1"}
2> {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666859020192,"create_time_desc":"2022-10-27 16:23:40","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type":"1"}
1> {"src_ip":"172.11.11.1","hostname":"hostname2","as":"B","create_time":1666859021231,"create_time_desc":"2022-10-27 16:23:41","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type ":"2"}
经过cep处理后,产了告警
产生告警:{A=[{"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666859021060,"create_time_desc":"2022-10-27 16:23:41","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type":"1"}, {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666859020192,"create_time_desc":"2022-10-27 16:23:40","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type":"1"}], B=[{"src_ip":"172.11.11.1","hostname":"hostname2","as":"B","create_time":1666859021231,"create_time_desc":"2022-10-27 16:23:41","event_type_value":"single","id":"67d32010-1f66-4850-b110-a7087e419c64_0","login_type":"2"}]}
经过src_ip,和hostname分组后, 理论上应该只分组后的相同的 scr_ip,hostname进行事件关联告警
结果其他的分组数据也参和进来关联告警了。
期望的是 login_type = 1 出现至少两次, 接着login_type=2的至少出现1次,且相同的src_ip和hostname
然后结果是下面数据也产生了告警。
{"src_ip":"172.11.11.1","hostname":"hostname1","login_type":1}
{"src_ip":"172.11.11.1","hostname":"hostname1","login_type":1}
{"src_ip":"172.11.11.1","hostname":"hostname1","login_type":2}
怀疑是分组没生效。
然后debug数据源那块的方法kafkaStreamSource() 里面有进行分组,debug后发现确实也进行了keyby
后来找不到其他问题,纠结了下, 怀疑是不是 KeyedSteam.union(KeyedStream)后得到的就不是一个KeyedSteam了。 所以
出现问题的原始代码数据源代码:
//程序具体执行流程
DataStream<JSONObject> sourceStream = SourceProcess.getKafkaStream(env, rule);
DataStream<JSONObject> resultStream = TransformProcess.process(sourceStream, rule);
SinkProcess.sink(resultStream, rule);
public static DataStream<JSONObject> getKafkaStream(StreamExecutionEnvironment env, Rule rule) {
DataStream<JSONObject> inputStream = null;
List<Event> events = rule.getEvents();
if (events.size() > SharingConstant.NUMBER_ZERO) {
for (Event event : events) {
FlinkKafkaConsumer<JSONObject> kafkaConsumer =
new KafkaSourceFunction(rule, event).init();
if (inputStream != null) {
// 多条 stream 合成一条 stream
inputStream = inputStream.union(kafkaStreamSource(env, event, rule, kafkaConsumer));
} else {
// 只有一条 stream
inputStream = kafkaStreamSource(env, event, rule, kafkaConsumer);
}
}
}
return inputStream;
}
private static DataStream<JSONObject> kafkaStreamSource(
StreamExecutionEnvironment env,
Event event,
Rule rule,
FlinkKafkaConsumer<JSONObject> kafkaConsumer) {
DataStream<JSONObject> inputStream = env.addSource(kafkaConsumer);
// 对多个黑白名单查询进行循环
String conditions = event.getConditions();
while (conditions.contains(SharingConstant.ARGS_NAME)) {
// 使用新的redis 数据结构,进行 s.include 过滤
inputStream = AsyncDataStream.orderedWait(inputStream,new RedisNameListFilterSourceFunction(s,rule.getSettings().getRedis()),30,TimeUnit.SECONDS,2000);
conditions = conditions.replace(s, "");
}
// 一般过滤处理
inputStream = AsyncDataStream.orderedWait(inputStream,
new Redis3SourceFunction(event, rule.getSettings().getRedis()), 30, TimeUnit.SECONDS, 2000);
// kafka source 进行 keyBy 处理
return KeyedByStream.keyedBy(inputStream, rule.getGroupBy());
}
public static DataStream<JSONObject> keyedBy(
DataStream<JSONObject> input, Map<String, String> groupBy) {
if (null == groupBy || groupBy.isEmpty() ||"".equals(groupBy.values().toArray()[SharingConstant.NUMBER_ZERO])){
return input;
}
return input.keyBy(
new TwoEventKeySelector(
groupBy.values().toArray()[SharingConstant.NUMBER_ZERO].toString()));
}
public class TwoEventKeySelector implements KeySelector<JSONObject, String> {
private static final long serialVersionUID = 8534968406068735616L;
private final String groupBy;
public TwoEventKeySelector(String groupBy) {
this.groupBy = groupBy;
}
@Override
public String getKey(JSONObject event) {
StringBuilder keys = new StringBuilder();
for (String key : groupBy.split(SharingConstant.DELIMITER_COMMA)) {
keys.append(event.getString(key));
}
return keys.toString();
}
}
问题出现在这里:
// 多条 stream 合成一条 stream
inputStream = inputStream.union(kafkaStreamSource(env, event, rule, kafkaConsumer));
kafkaStreamSource()这个方法返回的是 KeyedStream ,
两个KeyedStream unio合并后, 本来以为返回时KeyedStream,结果确是DataStream类型,
结果导致cep分组不生效,一个告警中出现了其他分组的数据。
解决方法, 就是在cep pattern前 根据是否有分组条件再KeyedBy一次
private static DataStream<JSONObject> patternProcess(DataStream<JSONObject> inputStream, Rule rule) {
PatternGen patternGenerator = new PatternGen(rule.getPatterns(), rule.getWindow().getSize());
Pattern<JSONObject, JSONObject> pattern = patternGenerator.getPattern();
if (!rule.getGroupBy().isEmpty()){
inputStream = KeyedByStream.keyedBy(inputStream, rule.getGroupBy());
}
PatternStream<JSONObject> patternStream = CEP.pattern(inputStream, pattern);
return patternStream.inProcessingTime().select(new RuleSelectFunction(rule.getAlarmInfo(), rule.getSelects()));
输入数据:
{"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860300012,"create_time_desc":"2022-10-27 16:45:00","event_type_value":"single","id":"1288a709-d2b3-41c9-b7b7-e45149084514_0","login_type":"1"}
{"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860299272,"create_time_desc":"2022-10-27 16:44:59","event_type_value":"single","id":"1288a709-d2b3-41c9-b7b7-e45149084514_0","login_type":"1"}
{"src_ip":"172.11.11.1","hostname":"hostname2","as":"B","create_time":1666860300196,"create_time_desc":"2022-10-27 16:45:00","event_type_value":"single","id":"1288a709-d2b3-41c9-b7b7-e45149084514_0","login_type":"2"}
不产生告警,符合预期
再次输入同分组的数据:文章来源:https://www.toymoban.com/news/detail-427346.html
2> {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860369307,"create_time_desc":"2022-10-27 16:46:09","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"1"}
2> {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860368471,"create_time_desc":"2022-10-27 16:46:08","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"1"}
2> {"src_ip":"172.11.11.1","hostname":"hostname1","as":"B","create_time":1666860369478,"create_time_desc":"2022-10-27 16:46:09","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"2"}
产生告警:{A=[{"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860368471,"create_time_desc":"2022-10-27 16:46:08","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"1"}, {"src_ip":"172.11.11.1","hostname":"hostname1","as":"A","create_time":1666860369307,"create_time_desc":"2022-10-27 16:46:09","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"1"}], B=[{"src_ip":"172.11.11.1","hostname":"hostname1","as":"B","create_time":1666860369478,"create_time_desc":"2022-10-27 16:46:09","event_type_value":"single","id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","app_id":"2"}]}
告警输出:{"org_log_id":"61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0,61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0,61004dd6-69ec-4d67-845c-8c15e7cc4bf7_0","event_category_id":1,"event_technique_type":"无","event_description":"1","alarm_first_time":1666860368471,"src_ip":"172.11.11.1","hostname":"hostname1","intelligence_id":"","strategy_category_id":"422596451785379862","intelligence_type":"","id":"cc1cd8cd-a626-4916-bdd3-539ea57e898f","event_nums":3,"event_category_label":"资源开发","severity":"info","create_time":1666860369647,"strategy_category_name":"网络威胁分析","rule_name":"ceptest","risk_score":1,"data_center":"guo-sen","baseline":[],"sop_id":"","event_device_type":"无","rule_id":214,"policy_type":"pattern","strategy_category":"/NetThreatAnalysis","internal_event":"1","event_name":"ceptest","event_model_source":"/RuleEngine/OnLine","alarm_last_time":1666860369478}
产生告警符合预期文章来源地址https://www.toymoban.com/news/detail-427346.html
到了这里,关于flink cep数据源keyby union后 keybe失效的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!