日志收集工具-logstash

提示：Windows 环境下 安装部署 logstash 采集日志文件

Logstash 服务器数据处理管道，能够从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的存储库中。Logstash 没有 fileBeats 轻量，但是过滤功能更多。

一、下载

1. 官网下载安装 windows 压缩包：https://www.elastic.co/cn/downloads/logstash
   

二、解压部署

1. 解压文件，在 logstash 目录下新建 logstash.conf 配置文件；
2. 编写配置文件如下：
   logstash.conf 包含3个主要部分： 输入(inputs)，过滤器(filters) 和输出(outputs)
下面配置是将日志从 file 同步到 file 中

   input {
     file {
       path => ["D:/project/elk/logstash/input/*.log"]   	# 替换为自己的日志文件路径，单个文件可以省略[]
       start_position => "beginning" 					    # 从文件开头开始读取  如果你只想读取新增的日志内容，可以将其改为 start_position => “end”，默认是结束位置
       sincedb_path => "NUL"  							    # 禁用 sincedb，避免重复读取，避免 Logstash 记录文件读取位置。如果你希望 Logstash 记住读取位置，可以将其设置为一个有效的文件路径
       type => "systemlog" 								# 标记事件类型。提前预定的各个事件对应类型。
       stat_interval => 3 									# 每隔多久检查一次被监听文件状态（是否有更新），默认是 1 秒
     }
   }
   
   filter {
    
   }
   
   output {
     file {
       path => "D:/project/elk/logstash/output/output.log"
       flush_interval => 0 								# 指定刷新间隔，0 代表实时写入
       codec => line {
         format => "%{message}"
       }
     }
   }
   

3. 运行，在bin目录下，执行命令：.\logstash -f logstash.conf；

三、常用插件

1. input
   file：从文件中读取
   stdin：从标准输入读取
   beats：从filebeat或其他beats的数据
   kafka：从 Kafka 主题中消费数据
   jdbc：从数据库中读取数据
   syslog：接受syslog数据
   http：通过 HTTP 接受数据

2. filter
   grok：解析非结构化日志数据
   mutate：修改字段数据，如重命名、删除或添加替换字段
   date：解析和转换日期字段
   json：解析 Json 格式数据
   geoip：根据 IP 地址添加地理位置信息
   drop：丢弃不符合条件的事件
   clone：复制事件

3. output
   elasticsearch：将数据输出到Elasticsearch
   stdout：将数据输出到标准输出，便于测试
   kafka：数据发送到 Kafka
   file：数据写入文件
   jdbc：将数据写入数据库

四、常用配置

1. input
   path（file 插件）：指定文件路径,多个路径用[]包括
   start_position（file 插件）：从文件的开头（beginning）或结尾（end）开始读取，默认end
   bootstrap_servers（kafka 插件）：Kafka 服务器地址
   topics（kafka 插件）：Kafka 主题名称
2. filter
   match（grok 插件）：定义日志解析规则
   source（json 插件）：指定 JSON 字段
   rename（mutate 插件）：重命名字段
   remove_field（mutate 插件）：删除字段
3. output
   hosts（elasticsearch 插件）：Elasticsearch 地址
   index（elasticsearch 插件）：索引名称
   path（file 插件）：输出文件路径