Logstash - 按关键字过滤系统日志。

Question

我是ELK的新手，我有一个logstash存储多个网络设备产生的系统日志。所有类型的事件都被标记在一个设备上（不幸的是），我只对存储DHCP租约的日志感兴趣。以下是logstash存储的日志样本

host:10.10.2.21 tags:_grokparsefailure type:dhcp @timestamp:May 6, 2020 @ 22:50:35.176 message:<190>May 6 22:50:33 10.7.24.13 Kiwi_Syslog_Server 2020-05-07 02:50:35 MOPS-E200E-CE-1 %%01SYSTATE/6/HEALTH(l): cpu=23 totalmemory=390127084 curmemory=176017096 @version:1 _id:gjYK7XEBLIinXt5dutFF _type:_doc _index:dhcp-2020.05.07-001 _score: -
host:10.10.2.21 tags:_grokparsefailure type:dhcp @timestamp:May 6, 2020 @ 22:50:33.727 message:<186>May 6 22:50:32 10.6.5.2 Kiwi_Syslog_Server May 7 2020 02:50:33 Data_Center-S9306-AS-1 OSPF/2/IFAUTFAIL:OID 1.3.6.1.2.1.14.16.2.6: A packet is received on a non-virtual interface from a router whose authentication key or authentication type conflicts with the local authentication key or authentication type. (IfIpAddress=10.4.0.254, AddressLessIf=0, ProcessId=3, RouterId=10.6.64.11, PacketSrc=10.4.0.44, ConfigErrorType=5, PacketType=1, InstanceName=) @version:1 _id:WzYK7XEBLIinXt5dtMzM _type:_doc _index:dhcp-2020.05.07-001 _score: -
host:10.10.2.21 tags:_grokparsefailure type:dhcp @timestamp:May 6, 2020 @ 22:50:32.478 message:<188>May 6 22:50:31 10.7.32.7 Kiwi_Syslog_Server date=2020-05-06 time=22:50:32 devname=MOTP_GT-FS248D-CE-7 device_id=S248DN3X17000508 log_id=0103034134 type=event subtype=system pri=warning vd=root user="alertd" msg="[First Event] TEMPERATURE_SENSOR_1 (49.0C) cleared warning threshold of (50.0C)." @version:1 _id:4jYK7XEBLIinXt5dsMg6 _type:_doc _index:dhcp-2020.05.07-001 _score: -
message:<134>May 6 22:52:29 10.7.24.23 Kiwi_Syslog_Server May 6 22:52:30 National-Trust-CE-1 PRESSDHCPSERVER assigned 172.18.20.24 to E8:93:09:CC:2A:C6 type:dhcp host:10.10.2.21 @version:1 @timestamp:May 6, 2020 @ 22:52:30.528 _id:jjgM7XEBLIinXt5dfFvH _type:_doc _index:dhcp-2020.05.07-001 _score: -

我只想存储有DHCP信息的最后一行日志。我观察到有DHCP信息的日志中都有 "assigned "这个词，所以我在寻找一种方法让logstash寻找 "assigned "这个词，并且只存储那行日志。

以下是我对logstash的配置

input {
  udp  {
    type => "dhcp"
    port => "518"
 }
}


filter {
 if [type] == "dhcp" {
   grok {
        match => {
            "message" => "assigned"
        }
    }

 }
}

output {
if [type] == "dhcp" {
  elasticsearch { hosts => ["localhost:9200"] 
  index => "dhcp-%{+yyyy.MM.dd}-001"
}
}
}

请你协助我，非常感谢

Answer 1

你好Narendra Rajcoomar。

我不是100％确定我得到了你的问题，所以请纠正我，如果我错了，我会编辑我的答案。

你正在使用提供的 logstash.conf 文件，你在弹性搜索中得到了上面的几行，其中也包括你不感兴趣的事件，你希望ES只保存像最后一行的事件。

如果这样做是正确的，你可能会注意到，除了最后一行，其他事件都有一个 tag 叫做 _grokparsefailure.这是因为它们确实没有包含 "指定 "的关键字，因此它们不是 "匹配"。

只要改变 logstash.conf 输出部分相应地，它将放弃不必要的日志。

input {
    udp  {
        type => "dhcp"
        port => "518"
    }
}

filter {
    if [type] == "dhcp" {
        grok {
            match => {"message" => "assigned"}
        }
    }
}

output {
    if [type] == "dhcp" and "_grokparsefailure" not in [tags] {
        elasticsearch { 
            hosts => ["localhost:9200"] 
            index => "dhcp-%{+yyyy.MM.dd}-001"
        }
    }
}