在我之前的文章 “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 => true
timeout_task_id_field => "user.name"
timeout => 1
timeout_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 => true
timeout_task_id_field => "user.name"
timeout => 5
timeout_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。
