ES集群地址
10.0.0.91:9200
10.0.0.92:9200
10.0.0.93:9200
编写Logstash实例
root@elk91:~# vim /etc/logstash/conf.d/08-nginx-to-es.conf
input { file { path => "/var/log/nginx/access.log"start_position => "beginning"}
} filter {#匹配nginx日志grok {match => { "message" => "%{HTTPD_COMMONLOG}" }}# 用于提取用户的设备信息useragent {# 指定从哪个字段解析用户设备信息source => "message"#将解析的结果存储到某个特定字段，若不指定，则默认放在顶级字段。target => "dezyan_user_agent"}geoip {source => "clientip"# 7.17.28版本中需要指定本地数据库文件进行解析，否则可能会长达8-10min的时间才能够进行解析。# 指定要解析的本地数据库文件路径，理论上官网说了默认就是"GeoLite2-City"，但是不配置的确有延迟!database => "/usr/share/logstash/data/plugins/filters/geoip/CC/GeoLite2-City.mmdb"# database => "/usr/share/logstash/data/plugins/filters/geoip/CC/GeoLite2-ASN.mmdb"# 若不不配置此类型，则默认值为"City"，当"database"指向的是"GeoLite2-City.mmdb"时，无需配置此项。# default_database_type => "ASN"}# 转换日期字段date {# 匹配日期字段，将其转换为日期格式，将来存储到ES，基于官方的示例对号入座对应的格式即可。# https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-date.html#plugins-filters-date-match# "timestamp" => "23/Oct/2024:16:25:25 +0800"match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]# 将match匹配的日期修改后的值直接覆盖到指定字段，若不定义，则默认覆盖"@timestamp"。target => "dezyan-timestamp"}# 对指定字段进行转换处理mutate {# 将指定字段转换成我们需要转换的类型convert => {"bytes" => "integer"}remove_field => [ "@version","host","message" ]}
}output { stdout { codec => rubydebug }elasticsearch {# 对应的ES集群主机列表hosts => ["10.0.0.91:9200","10.0.0.92:9200","10.0.0.93:9200"]# 对应的ES集群的索引名称index => "dezyan-elk-nginx"}
}root@elk91:~# rm -f /usr/share/logstash/data/plugins/inputs/file/.sincedb*
root@elk91:~# logstash -rf 08-nginx-to-es.conf
