![如何使用输入作为“文件”将 IP 地址发送到 logstash?](https://linux22.com/image/799680/%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8%E8%BE%93%E5%85%A5%E4%BD%9C%E4%B8%BA%E2%80%9C%E6%96%87%E4%BB%B6%E2%80%9D%E5%B0%86%20IP%20%E5%9C%B0%E5%9D%80%E5%8F%91%E9%80%81%E5%88%B0%20logstash%EF%BC%9F.png)
我正在使用客户端上的另一个 logstash 作为发送器将一些日志发送到 logstash 中央服务器。输入类型为“文件”。服务器上可以正常接收消息,但它没有反映客户端的 IP 地址。它在字段“@source_host”中发送主机名。我可以做些什么来将 IP 作为字段?也许是一个过滤器?
客户端配置:
input {
file {
format => "plain"
path => "/var/log/app/test1.txt"
type => "start"
}
}
output {
redis {
host => "test.example.com"
data_type => "list"
key => "logstash"
}
}
答案1
您可以使用“dns”过滤器进行反向查找,然后使用它来设置字段。 http://logstash.net/docs/1.2.2/filters/dns
答案2
如果您想要的客户端 IP 是静态的,那么我建议您使用以下内容替换内容 @source_hostmutate
筛选
例如:
filter {
mutate {
replace => ["@source_host","xx.xx.xx.xx"]
}
}
如果您只是想要字段中的 IP(而不是 @source_host),您可以将其添加到输入中:
input {
file {
format => "plain"
path => "/var/log/app/test1.txt"
type => "start"
add_field => ['source_ip','xx.xx.xx.xx']
}
}
否则,如果您确实需要解析非静态客户端主机名,那么@丹·加思韦特的答案是正确的。
答案3
这也许是一个很好的例子,可以让你入门,这是发送到 elasticsearch 的 id 的 logstash 配置https://github.com/StamusNetworks/SELKS/blob/master/docker/containers-data/logstash/conf.d/logstash.conf
input {
file {
path => ["/var/log/suricata/*.json"]
#sincedb_path => ["/var/lib/logstash/"]
sincedb_path => ["/usr/share/logstash/since.db"]
codec => json
type => "SELKS"
}
}
filter {
if [type] == "SELKS" {
date {
match => [ "timestamp", "ISO8601" ]
}
ruby {
code => "
if event.get('[event_type]') == 'fileinfo'
event.set('[fileinfo][type]', event.get('[fileinfo][magic]').to_s.split(',')[0])
end
"
}
ruby {
code => "
if event.get('[event_type]') == 'alert'
sp = event.get('[alert][signature]').to_s.split(' group ')
if (sp.length == 2) and /\A\d+\z/.match(sp[1])
event.set('[alert][signature]', sp[0])
end
end
"
}
metrics {
meter => [ "eve_insert" ]
add_tag => "metric"
flush_interval => 30
}
}
if [http] {
useragent {
source => "[http][http_user_agent]"
target => "[http][user_agent]"
}
}
if [src_ip] {
geoip {
source => "src_ip"
target => "geoip"
#database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
#add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
#add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
}
if [dest_ip] {
geoip {
source => "dest_ip"
target => "geoip"
#database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
#add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
#add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
}
}
output {
if [event_type] and [event_type] != 'stats' {
elasticsearch {
hosts => "elasticsearch"
index => "logstash-%{event_type}-%{+YYYY.MM.dd}"
template_overwrite => true
template => "/usr/share/logstash/config/elasticsearch7-template.json"
}
} else {
elasticsearch {
hosts => "elasticsearch"
index => "logstash-%{+YYYY.MM.dd}"
template_overwrite => true
template => "/usr/share/logstash/config/elasticsearch7-template.json"
}
}
}