Logstash可以收集日志数据并对数据进行处理后传递给ES。如果需要使用它采集日志的话需要在所有需要收集日志的服务器上安装Logstash，由于Logstash偏重量级，现在通常不会这样做，而是交给各种beat去完成这个工作，然后最后交给Logstash处理后再传递给ES。如果数据量太大可以在agent与server端之前部署消息队列，一般使用Redis或者Kafka。

一、Logstash安装步骤

1、在ELK官网www.elastic.co下载Logstash 6.3，其他版本安装方式基本一致。Logstash比较重量级的原因是它也需要依赖JAVA环境，所以依然要提前配置JDK。如果是使用的ES7版本，那么已经内置了OpenJDK可以用，只需要配置好JAVA环境变量即可

tar zxf logstash-6.3.tar.gz
mv logstash-6.3 /usr/local/
ln -s logstash-6.3.logstash

二、如何启动Logstash

Logstash有2个配置文件，一个是config/logstash.yml，该配置文件可以修改启动进程数、日志信息、持久化、持久化队列设置等；另一个是jvm.options，这个和Elasticsearch是一样的，主要是优化内存。2个配置文件不做任何修改也是可以启动Logstash的，启动脚本放在bin目录中，下面先使用命令行模式启动（需要注意的是不能用root用户来运行）：

bin/logstash -e 'input{ stdin{} } output{ stdout{} }'  #-e代表使用命令行模式，采用标准输入，标准输出
Settings: Default pipeline workers: 1
Pipeline main started
hello world  #这里手动输入的hello world，作为标准输入
2016-05-19T01:59:03.797Z 0.0.0.0 hello world  #标准输出的结果显示在屏幕
nihao
2016-05-19T02:00:20.013Z 0.0.0.0 nihao

还可以使用codec风格指定输出格式（codec是一种解码编码的工具）

./logstash -e 'input{ stdin{} } output{ stdout{codec => rubydebug} }'  # -e选项是直接用命令行模式，输入采用标准输入，输出采用的codec风格
Settings: Default pipeline workers: 1
Pipeline main started
hello world
{
       "message" => "hello world",  #输入的内容
      "@version" => "1",  #版本号
    "@timestamp" => "2016-05-19T02:09:43.921Z",  #自动生成时间戳
          "host" => "0.0.0.0"  #数据是由哪个节点发过来的

三、自定义Logstash配置

在生产环境中需要用到的规则非常复杂，需要自定义规则文件然后让Logstash根据规则进行工作。自定义配置文件时，主要配置的内容包括input（用于指定输入源，包含本地日志、redis、stdin标准输入等多种方式）、filter（进行规则过滤和处理，重头戏）、output（指定输出目标）。下面是一个配置示例：

input{
  stdin {}
}

output{
  stdout{
    codec  => rubydebug
  }
}

启动服务时加上-t选项可以检查配置文件是否正确，-f选项就是配置文件的路径

nohup logstash -t  -f /etc/logstash.conf &

四、自定义Logstash输入与输出数据源

在Logstash中输入的数据可以通过很多种方式获取，比如本地日志文件、filebeat、数据库等。而输出也可以指定到需要的容器中，如Elasticsearch。

1、使用file插件将本地文件作为Logstash输入。下面是一个使用file插件的简单案例，由于output依然是标准输出，所以会有疯狂刷屏的情况出现：

vi etc/logstash.conf                                   
input {
  file {
    path => [ "/var/log/message","/var/log/*.log" ]  #文件路径
    exclude => "/var/log/*.gz"  #排除不需要监听的文件
    type => "system"  #自定义事件的类型，可用于后续的条件判断
    add_field => {"key" => "test"}  #自定义新增字段
    start_position => "beginning"  #从日志文件头部读取，相反还有end
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

file插件使用了一个sincedb文件来记录当前文件读区位置，即使重新启动服务也不会丢失文件的读取位置。默认情况下sincedb文件放在运行Logstash的用户的主目录中，可以用sincedb_path选项自定义存放路径。

2、将Logstash收集到的数据提交给Elasticsearch

这里需要使用一个output插件——elasticsearch。使用方法也很简单，只需要在配置文件里指定插件就可以了，如下：

[root@server ~]# cat /etc/logstash.conf   
input{
  file{  
    path=> "/var/log/audit/audit.log"
    type=> "system"
    start_position=> "beginning"  
  }
}
output{
  elasticsearch{  #输出到es
    hosts=> ["192.168.44.129:9200","192.168.44.130:9200"]
  }
}

文件设置好了后运行logstash的程序，再到Elasticsearch或者Kibana就可以查看到数据了，如果Logstash的配置文件发生了变化只需要重启Logstash即可，Elasticsearch和Kibana里的数据都会自动更新。因为Logstash不会占用端口，加-w选项可以启动多个线程，提高效率，默认是2，例

nohup /logstash/bin/logstash  -f /etc/logstash.conf -w 4 &

五、Logstash多实例的配置

1、复制logstash/config目录

2、修改新目录中的logstash.yml以下配置

node.name: instance2  #实例2的节点名称
path.data: /data/instance2  #实例2的数据路径
path.config: /etc/logstash2.conf  #实例2的配置文件

3、启动服务即可

本站所有文章均为原创，转载请注明出处！小站维护不易，如果对您有所帮助，希望能点击一下站内广告，谢谢！