在我之前的文章 “Logstash：使用 aggregate filter 处理 N:N 关系”，我详述了如何使用 aggregation filter 来把关系数据库中的数据进行聚合并把结果写入到 Elasticsearch。在我们的实际使用中，我们可能面临更多的是时序数据。一旦数据被写入到 Elasticsearch 中，在分析数据时，我们可以充分 Elasticsearch 的强大功能进行聚合和搜索。如果我们是使用 Logstash 来采集数据的，那么我们有没有想到使用 aggregation filter 来对数据进行聚合，然后再写入到 Elasticsearch 中。那么在实际的搜索时，我们仅仅只需要进行搜索就可以了。

我们知道在使用 Elasticsearch 进行聚合时，我们经常会使用 date histogram 来进行聚合，并针对其中的一些 term 进行分类，比如：

GET kibana_sample_data_logs/_search
{"size": 0,"aggs": {"per_day": {"date_histogram": {"field": "@timestamp","calendar_interval": "day"},"aggs": {"per_host": {"terms": {"field": "host.keyword","size": 10},"aggs": {"total_per_host_per_day": {"terms": {"field": "host.keyword","size": 10}}}}}}}
}

由于聚合是需要花费运算时间的，如果我们在摄入数据的时候就把这个聚合做好，那么以后我们就直接进行搜索即可。

在今天的展示中，我将在 Logstash 中实现上面的聚合功能。你可以在如下的地址下载：

git clone https://github.com/liu-xiao-guo/logstash_aggregation

展示

在某些情况下，日志量可能会非常高，这使得收集、转换和存储每个单独的事件以进行分析变得昂贵且资源密集。在某些用例中，收集传入事件的聚合视图而不是每个单个事件就足够了。此场景将研究在将给定用户的网络活动摄入到 Elasticsearch 之前，如何在一个时间窗口内汇总网络活动：

准备数据

在今天的展示中，我们将使用如下的数据集来进行：

sample.log

{"event.start": "01:05:2021 05:11:10","destination.ip": "93.115.29.34","destination.host": "test.creditcard.com","source.ip": "10.189.90.63","source.host": "u120823domain.corp.com","client.bytes": "177","server.bytes": "4919","http.request.time": "14","http.response.time": "113","user.name": "u120823","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:11","destination.ip": "161.97.175.85","destination.host": "scam.biz","source.ip": "10.189.121.67","source.host": "u881001domain.corp.com","client.bytes": "250","server.bytes": "1949","http.request.time": "60","http.response.time": "614","user.name": "u881001","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:12","destination.ip": "209.145.54.119","destination.host": "visa.spam.org","source.ip": "10.189.121.90","source.host": "u992001domain.corp.com","client.bytes": "849","server.bytes": "3941","http.request.time": "33","http.response.time": "898","user.name": "u992001","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:13","destination.ip": "93.115.29.34","destination.host": "test.creditcard.com","source.ip": "10.189.90.63","source.host": "u120823domain.corp.com","client.bytes": "609","server.bytes": "678","http.request.time": "90","http.response.time": "95","user.name": "u120823","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:14","destination.ip": "161.97.173.254","destination.host": "mail.campaign.adhoster.biz","source.ip": "10.189.90.65","source.host": "u120392domain.corp.com","client.bytes": "98","server.bytes": "3270","http.request.time": "70","http.response.time": "280","user.name": "u120392","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:15","destination.ip": "144.91.92.188","destination.host": "xyz.test.com","source.ip": "10.189.121.90","source.host": "u992001domain.corp.com","client.bytes": "623","server.bytes": "528","http.request.time": "93","http.response.time": "841","user.name": "u992001","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:16","destination.ip": "64.227.105.9","destination.host": "goooooogol.com","source.ip": "10.189.90.64","source.host": "u210820domain.corp.com","client.bytes": "780","server.bytes": "4097","http.request.time": "25","http.response.time": "550","user.name": "u210820","event.outcome": "blocked","event.type": "firewall","event.category": "spyware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:17","destination.ip": "161.97.173.251","destination.host": "phish.scam.com","source.ip": "10.189.198.142","source.host": "u809323domain.corp.com","client.bytes": "118","server.bytes": "3345","http.request.time": "68","http.response.time": "533","user.name": "u809323","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:18","destination.ip": "161.97.173.251","destination.host": "phish.scam.com","source.ip": "10.189.121.129","source.host": "u884219domain.corp.com","client.bytes": "711","server.bytes": "675","http.request.time": "20","http.response.time": "688","user.name": "u884219","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:19","destination.ip": "185.130.44.108","destination.host": "ads.phish.org","source.ip": "10.189.198.90","source.host": "u112349domain.corp.com","client.bytes": "96","server.bytes": "2363","http.request.time": "65","http.response.time": "564","user.name": "u112349","event.outcome": "blocked","event.type": "firewall","event.category": "command_and_control","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:20","destination.ip": "144.91.92.192","destination.host": "hello.test.xyz","source.ip": "10.189.90.62","source.host": "u329012domain.corp.com","client.bytes": "638","server.bytes": "736","http.request.time": "13","http.response.time": "774","user.name": "u329012","event.outcome": "blocked","event.type": "firewall","event.category": "spyware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:21","destination.ip": "161.97.172.254","destination.host": "hello.word.biz","source.ip": "10.189.90.62","source.host": "u329012domain.corp.com","client.bytes": "717","server.bytes": "2787","http.request.time": "9","http.response.time": "648","user.name": "u329012","event.outcome": "allowed","event.type": "firewall","event.category": "unclassified","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:22","destination.ip": "185.130.44.108","destination.host": "ads.phish.org","source.ip": "10.189.198.10","source.host": "u140192domain.corp.com","client.bytes": "267","server.bytes": "1904","http.request.time": "3","http.response.time": "744","user.name": "u140192","event.outcome": "blocked","event.type": "firewall","event.category": "command_and_control","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:23","destination.ip": "144.91.92.188","destination.host": "xyz.test.com","source.ip": "10.189.90.64","source.host": "u210820domain.corp.com","client.bytes": "772","server.bytes": "4210","http.request.time": "65","http.response.time": "506","user.name": "u210820","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:24","destination.ip": "144.91.92.188","destination.host": "xyz.test.com","source.ip": "10.189.198.10","source.host": "u140192domain.corp.com","client.bytes": "444","server.bytes": "2963","http.request.time": "93","http.response.time": "618","user.name": "u140192","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:25","destination.ip": "64.227.105.9","destination.host": "goooooogol.com","source.ip": "10.189.90.64","source.host": "u210820domain.corp.com","client.bytes": "195","server.bytes": "4724","http.request.time": "64","http.response.time": "160","user.name": "u210820","event.outcome": "blocked","event.type": "firewall","event.category": "spyware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:26","destination.ip": "161.97.173.254","destination.host": "mail.campaign.adhoster.biz","source.ip": "10.189.90.64","source.host": "u210820domain.corp.com","client.bytes": "22","server.bytes": "370","http.request.time": "83","http.response.time": "482","user.name": "u210820","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:27","destination.ip": "185.130.44.108","destination.host": "ads.phish.org","source.ip": "10.189.121.90","source.host": "u992001domain.corp.com","client.bytes": "806","server.bytes": "452","http.request.time": "31","http.response.time": "850","user.name": "u992001","event.outcome": "blocked","event.type": "firewall","event.category": "command_and_control","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:28","destination.ip": "161.97.173.254","destination.host": "mail.campaign.adhoster.biz","source.ip": "10.189.121.129","source.host": "u884219domain.corp.com","client.bytes": "803","server.bytes": "1669","http.request.time": "57","http.response.time": "250","user.name": "u884219","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:29","destination.ip": "185.130.44.108","destination.host": "ads.phish.org","source.ip": "10.189.198.10","source.host": "u140192domain.corp.com","client.bytes": "348","server.bytes": "3544","http.request.time": "81","http.response.time": "573","user.name": "u140192","event.outcome": "blocked","event.type": "firewall","event.category": "command_and_control","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:30","destination.ip": "144.91.92.192","destination.host": "hello.test.xyz","source.ip": "10.189.90.65","source.host": "u120392domain.corp.com","client.bytes": "432","server.bytes": "2655","http.request.time": "64","http.response.time": "859","user.name": "u120392","event.outcome": "blocked","event.type": "firewall","event.category": "spyware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:31","destination.ip": "161.97.172.254","destination.host": "hello.word.biz","source.ip": "10.189.198.142","source.host": "u809323domain.corp.com","client.bytes": "525","server.bytes": "3243","http.request.time": "100","http.response.time": "331","user.name": "u809323","event.outcome": "allowed","event.type": "firewall","event.category": "unclassified","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:32","destination.ip": "144.91.92.188","destination.host": "xyz.test.com","source.ip": "10.189.90.65","source.host": "u120392domain.corp.com","client.bytes": "187","server.bytes": "113","http.request.time": "51","http.response.time": "70","user.name": "u120392","event.outcome": "blocked","event.type": "firewall","event.category": "adware","event.action": "threat_filter"}
{"event.start": "01:05:2021 05:11:33","destination.ip": "161.97.173.251","destination.host": "phish.scam.com","source.ip": "10.189.90.63","source.host": "u120823domain.corp.com","client.bytes": "240","server.bytes": "357","http.request.time": "88","http.response.time": "627","user.name": "u120823","event.outcome": "blocked","event.type": "firewall","event.category": "phish","event.action": "threat_filter"}

这是一个防火墙的数据。我们把上面的数据写入到一个本地的 sample.log 的文件中。这是一个 JSON 格式的日志数据。接下来，我们将使用 Logstash 来摄入这个数据。

运用 aggregation filter 来聚合数据

我们来创建如下的一个配置文件：

logstash.conf

input {stdin {codec => "json"}
}filter {mutate {convert => {"client.bytes" => "integer""server.bytes" => "integer"}}aggregate {task_id => "%{user.name}"code => "map['total_client_bytes'] ||= 0;map['total_client_bytes'] += event.get('client.bytes');map['total_server_bytes'] ||= 0;map['total_server_bytes'] += event.get('server.bytes');"push_map_as_event_on_timeout => truetimeout_task_id_field => "user.name"timeout => 1timeout_code => "event.set('client_bytes_summary', event.get('total_client_bytes') > 1)"}mutate {remove_field => ["@version"]}if [event.action] == "threat_filter" {drop {}}
}output {stdout {}
}

在上面，我们使用 stdin 作为输入。因为 sample.log 的格式是 JSON 的，所以我们设置 codec 为 json。

我们想了解单个用户通过防火墙消耗的带宽。 client.bytes 和 server.bytes 字段被转换为整数值以进行聚合：

    mutate {convert => {"client.bytes" => "integer""server.bytes" => "integer"}}

聚合过滤器插件用于汇总事件中的数据。过滤器按 task_id 参数定义的 user.name 值对事件进行分组。该代码块包含 Ruby 代码，用于计算 client.bytes 和 server.bytes 值的总和。

timeout 字段控制发生聚合的时间窗口。在此实例中设置为 5 秒，但如果需要更积极的聚合，可以将其设置为更大的时间间隔。在 timeout 窗口结束时，将字段的数据总和作为事件推出：

    aggregate {task_id => "%{user.name}"code => "map['total_client_bytes'] ||= 0;map['total_client_bytes'] += event.get('client.bytes');map['total_server_bytes'] ||= 0;map['total_server_bytes'] += event.get('server.bytes');"push_map_as_event_on_timeout => truetimeout_task_id_field => "user.name"timeout => 5timeout_code => "event.set('client_bytes_summary', event.get('total_client_bytes') > 1)"}

鉴于我们只对聚合事件感兴趣，可以使用以下条件以及 drop filter 插件来摆脱原始事件：

if [event.action] == "threat_filter" {drop {}
}

我们使用如下的命令来运行 Logstash：

cat sample.log | ./bin/logstash -f logstash.conf

由于一些原因，你可能在最新的版本中不能完成这个功能。请参阅 issue。

Logstash：使用 aggregation filter 把事件流聚合为一个事件相关推荐

【事件流】浅谈事件冒泡事件捕获------【巷子】
首先在扯淡的时候我们需要先了解一个东西,这个东西就是事件流.1.什么是事件流?解释:当一个HTML元素触发一个事件处理函数的时候,该事件会在该元素节点到根节点之间传播,传播路径所经过的节点都会接受到该 ...
「事件流处理架构」事件流处理的八个趋势
经过二十多年的研究和开发,事件流处理(ESP)软件平台已不再局限于在小生境应用或实验中使用.它们已经成为许多业务环境中实时分析的基本工具. 其动机来自需要分析的流数据量激增,特别是: 物联网传感器数据 ...
C#一个事件中调用另一个事件
为ImageButton1注册事件 protected void Page_Load(object sender, EventArgs e) { ImageButton1.Click += n ...
logstash过滤器插件filter详解及实例
原创作者:峰哥ge 原创地址: https://www.cnblogs.com/FengGeBlog/p/10305318.html logstash过滤器插件filter grok正则捕获 grok ...
事件流--事件冒泡现象及阻止
事件冒泡现象 <!DOCTYPE html> <html lang="en"> <head><meta charset="UTF ...
后处理程序文件大小的变量_【每日一题】(17题)面试官问：JS中事件流，事件处理程序，事件对象的理解？...
关注「松宝写代码」,精选好文,每日一题作者:saucxs | songEagle 2020,实「鼠」不易 2021,「牛」转乾坤风劲潮涌当扬帆,任重道远须奋蹄! 一.前言 2020.12.23 立 ...
html流动模型,javascript的事件流模型都有什么？
事件流:当你在页面触发一个点击事件后,页面上不仅仅有一个元素响应该事件而是多个元素响应同一个事件,因为元素是在容器中的.事件发生的顺序就是事件流,不同的浏览器对事件流的处理不同. JavaScript ...
js 事件流的事件冒泡和事件捕获与阻止事件传播
为了方便引入事件流的概念,我们先来说说什么是事件. 事件就是用户或浏览器自身执行的某种动作.换句话说,我们在浏览网页或者 APP 时,通常会在设备上产生很多交互性的操作,例如点击.选择.滚动屏幕.键盘 ...
js事件流、事件代理等
菜菜的自己发现真的啥都不太懂 - 虽然啥都会一点,但是却什么都不精通. 偶然看见js事件流和事件代理等 - 黑人问号脸,虽然在实际写代码的时候经常用到. 做个总结: 都是学的别人的. 事件代理:通过将 ...

Logstash：使用 aggregation filter 把事件流聚合为一个事件

展示

准备数据

运用 aggregation filter 来聚合数据

Logstash：使用 aggregation filter 把事件流聚合为一个事件相关推荐

最新文章

热门文章