Logstash 聚合过滤器，将信息添加到下一行答案

【问题标题】：Logstash Aggregate filter, add information to next linesLogstash 聚合过滤器，将信息添加到下一行
【发布时间】：2016-11-15 12:43:11
【问题描述】：

我正在编写一个 logstash 2.4.0 配置来查看 HTTP 日志。我们希望将在 Header 字段中传递的 PORT 包含在下面的 Line 字段中。没有定义特定的结束事件。虽然我也尝试过添加结束事件。

我当前使用的输入日志文件是：

HEADER 9200
LINE 1 2016-10-05 08:39:00 Some log data
LINE 2 2016-10-05 08:40:00 Some other log data
FOOTER
HEADER 9300
LINE 4 2016-11-05 08:39:00 Some log data in another log
LINE 5 2016-11-05 08:40:00 Some other log data in another log
FOOTER

我想要这样的输出：输出中当前缺少 Server_port 字段

{"message" => "HEADER 9200",
 "@version" => "1",
 "@timestamp" => "2016-11-15T11:17:18.425Z",
 "path" => "test.log",
 "host" => "hostname",
 "type" => "event",
 "env" => "test",
 "port" => 9200,
 "tags" => [[0] "Header"]    }
{"message" => "LINE 1 2016-10-05 08:39:00 Some log data",
 "@version" => "1",
 "@timestamp" => "2016-11-15T11:17:20.186Z",
 "path" => "test.log",
 "host" => "hostname",
 "type" => "event",
 "env" => "test",
 "logMessage" => "1 2016-10-05 08:39:00 Some log data",
 "Server_port" => 9200,
 "tags" => [[0] "Line"]}
{"message" => "LINE 2 2016-10-05 08:40:00 Some other log data",
 "@version" => "1",<
 "@timestamp" => "2016-11-15T11:17:20.192Z",
 "path" => "test.log",
 "host" => "hostname",
 "type" => "event",
 "env" => "test",
 "logMessage" => "2 2016-10-05 08:40:00 Some other log data",
 "Server_port" => 9200,
 "tags" => [[0] "Line"]}
{"message" => "FOOTER",
 "@version" => "1",
 "@timestamp" => "2016-11-15T11:17:20.195Z",
 "path" => "test.log",
 "host" => "hostname",
 "type" => "event",
 "env" => "test",
 "tags" => [[0] "Footer"]}

尝试了不同的东西后，我目前使用的配置如下，硬编码的taskid='abcd'用于测试：

input{ file{    path => "test.log"
                start_position => "beginning"
                sincedb_path => "/dev/null"
                ignore_older => 0
                type => "event"
                add_field => { "env" => "test"} }
}
filter{
        grok {
                break_on_match => false
                tag_on_failure => []
                match => {"message" => ["^HEADER%{SPACE}%{INT:port:int}"]}
                add_tag => ["Header"]
                }
        grok {
                break_on_match => false
                tag_on_failure => []
                match => {"message" => "^LINE%{SPACE}%{GREEDYDATA:logMessage}"}
                add_tag => ["Line"]
                }
        grok {
                break_on_match => false
                tag_on_failure => []
                match => {"message" => "^FOOTER"}
                add_tag => ["Footer"]
                }    
       if "Header" in [tags]{
                aggregate{
                        task_id => "abcd"
                        code => "map['server_port'] ||= 0; map['server_port']=event['port']"
                        push_map_as_event_on_timeout => true
                        push_previous_map_as_event => true
                        map_action => "create"
                }
        }
        elseif "Line" in [tags]{
                aggregate{
                        task_id => "abcd"
                        code => "event.set('server_port',map['server_port'])"                                                   
                        map_action => "update"
                }
        }
        else if "Footer" in [tags]{
                aggregate{
                        task_id => "abcd"
                        code => "event.set('server_port',map['server_port'])"                                                   
                        map_action => "update"
                        end_of_task => true
                        timeout => 120
                }
        }
}
output {
  stdout { codec => rubydebug }
}

虽然此配置运行没有错误，但它并未创建 server_port 字段。我哪里错了？

【问题讨论】：

标签： events logstash aggregate multiline elastic-stack

【解决方案1】：

在摆弄了更多之后，我有了一个工作测试用例。我已将配置更改如下：

grok {
                break_on_match => false
                tag_on_failure => []
                match => {
                   "message" => ["^HEADER%{SPACE}%{INT:taskid:int}%{SPACE}%{INT:port:int}"]
                }
                add_tag => ["Header"]
                }

和

 if "Header" in [tags]{
            aggregate{
                    task_id => "%{taskid}"
                    code => "map['port']=event.get('port')"
                    map_action => "create"
            }
    }
    elseif "Line" in [tags]{
            aggregate{
                    task_id =>"%{taskid}"
                    code => "event.set('port',map['port'])"
                    map_action => "update"
            }
    }
    else if "Footer" in [tags]{
            aggregate{
                    task_id => "%{taskid}"
                    code => "event.set('port',map['port'])"
                    map_action => "update"
                    end_of_task => true
                    timeout => 120
            }
    }

并在日志中添加了一个任务 id 字段：

HEADER 123 9200
LINE 123 2016-10-05 08:39:00 Some log data

【讨论】：