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

Posted 2022-05-12 Elastic 中国社区官方博客

在我之前的文章 “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。