[TOC] Logstash 是一个具有实时流水线功能的开源数据收集引擎。Logstash 可以动态地统一来自不同来源的数据，并将数据规范化到您选择的目的地。为各种高级下游分析和可视化用例清理和普及所有数据。 虽然 Logstash 最初推动了日志收集的创新，但它的功能远远超出了该用例。任何类型的事件都可以通过广泛的输入、过滤器和输出插件进行丰富和转换，许多本地编解码器进一步简化了摄取过程。Logstash 通过利用更大量和更多样化的数据来加速您的洞察力。 ## 下载 ```shell [elk@elk02 ~]$ curl -O https://artifacts.elastic.co/downloads/logstash/logstash-7.17.2-linux-x86_64.tar.gz [elk@elk02 ~]$ sudo tar xf logstash-7.17.2-linux-x86_64.tar.gz -C /opt/ ``` ## 修改权限 ```shell [elk@elk02 ~]$ sudo chown -R elk.elk /opt/logstash-7.17.2/ [elk@elk02 ~]$ mkdir /opt/logstash-7.17.2/{logs,pid} ``` ## 配置项 ### 数据源filebeat ```shell [elk@elk02 logstash-7.17.2]$ cd /opt/logstash-7.17.2/ [elk@elk02 logstash-7.17.2]$ cp config/logstash-sample.conf config/logstash.conf [elk@elk02 logstash-7.17.2]$ cat config/logstash.conf input { beats { port => 5044 } } output { elasticsearch { hosts => ["192.168.31.29:9200", "192.168.31.193:9200", "192.168.31.120:9200"] index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}" #user => "elastic" #password => "changeme" } } ``` ### 数据源kafka ```shell [elk@elk02 logstash-7.17.2]$ cat config/logstash.conf input { kafka { bootstrap_servers => "10.0.0.127:9092,10.0.0.128:9092,10.0.0.129:9092" group_id => "logstash" auto_offset_reset => "latest" consumer_threads => "5" topics => ["messages"] type => "logs" } } output { elasticsearch { hosts => ["http://10.0.0.127:9200", "http://10.0.0.128:9200", "http://10.0.0.129:9200"] index => "logstash-%{+YYYY.MM}" #user => "elastic" #password => "changeme" } } ``` ## 启动 ```shell cd /opt/logstash-7.17.2 nohup ./bin/logstash -f config/logstash.conf &>> logs/logstash-server-`date "+%Y%m%d"`.log & echo $! > pid/logstash.pid ``` ## 停止 ```shell cat /opt/logstash-7.17.2/pid/logstash.pid | xargs -I {} kill {} ``` ## 参考文章 logstash官方文档：https://www.elastic.co/guide/en/logstash/7.17/index.html