收集日志之一：logstash方式

部署 logstash：
 Logstash 是一个开源的数据收集引擎，基于 ruby 开发，可以水平伸缩，而且 logstash 整个 ELK当中拥有最多插件的一个组件，
 其可以接收来自不同来源的数据并统一输出到指定的且可以是多个不同目的地
 Logstash 参考文档 ：https://www.elastic.co/guide/en/logstash/current/index.html
 https://mirrors.aliyun.com/elasticstack/yum/elastic-7.x/7.6.2/
 
安装jdk
# yum install jdk-8u121-linux-x64.rpm
2.1.2：安装 logstash：
# yum install logstash-5.3.0.rpm
# chown logstash.logstash /usr/share/logstash/data/queue –R #权限更改为 logstash 用户和组，否则启动的时候日志报错 

配置文件路径 
/etc/logstash/logstash.yml配置文件路径 
/etc/logstash/conf.d/  自定义日志收集路径 
/var/log/logstash/logstash-*  日志文件


参数说明：
 /usr/share/logstash/bin/logstash  --help
Usage: environment [options]
    -p port                          set the port (default is 4567)
    -s server                        specify rack server/handler (default is thin)
    -q                               turn on quiet mode (default is off)
    -x                               turn on the mutex lock (default is off)
    -e env                           设置环境 ，（默认为开发,也是终端测试）
    -o addr                          set the host (default is (env == 'development' ? 'localhost' : '0.0.0.0'))
    -t                                logstash配置文件检查
    -f                                  配置文件

配置文件检查
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/dmesg.conf   -t

以进程的方式运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/dmesg.conf

以服务的方式运行
systemctl  restart logstash


基础组件：input、filter、output
input： 输入组件，负责从哪里读取数据
    有：stdin：标准输入、file：读取文件、jdbc：读取数据库、syslog：读取日志服务、ganglia：监控日志、tcp/udp：网络协议、kakfa：消息队列

filter：负责过滤，可以没有

output：输出组件，负责将采集到的数据写入什么位置
    有：stdout：标准输出、file、es、kafka、webhdfs 、codec=>rubydebug：指定输出格式

2.logstash功能详解与使用

2.logstash功能详解与使用,以下测试要ctrl+C结束才能产生”数据“
1.标准输入和输出测试：
        # /usr/share/logstash/bin/logstash -e 'input { stdin {} } output { stdout { codec => "rubydebug" } }'
        hello
        {
         "@timestamp" => 2017-04-20T02:30:01.600Z,  #当前事件的发生时间，
         "@version" => "1",                         #事件版本号，一个事件就是一个 ruby 对象
        "host" => "linux-host3.exmaple.com",         #标记事件发生在哪里
        "message" => "hello"                        #消息的具体内容
        }
2.通过标准输入收集数据，然后输出到某个文件：
        # /usr/share/logstash/bin/logstash -e 'input { stdin {} } output { file { path => "/tmp/logstash-linux39.txt" } }'

3.将输出改成elasticsearch：
        # /usr/share/logstash/bin/logstash -e 'input { stdin {} } output { elasticsearch { hosts => ["192.168.80.120:9200"]  index => "linux39-%{+YYYY.MM.dd}" } }'

4.将输入改成日志文件：
        start_position  string, one of ["beginning", "end"]     从"/var/log/bootstrap.log中的什么位置开始读取内容到ES中，beginning指所有内容。 end指当前位置开始，之前的内容不读取。
        stat_interval number or string_duration             日志收集的间隔时间，多少秒读取一次/var/log/dmesg文件
        elasticsearch： output输出服务类型
        index：  es服务器上分片名称
        input中的是file文件，运行logstash的用户需要对该文件有读取权限。
        hosts:   es主机IP
        
        # /usr/share/logstash/bin/logstash -e 'input { file { path => "/var/log/dmesg" start_position => "beginning" stat_interval => "3"} } output { elasticsearch { hosts => ["192.168.80.120:9200"]  index => "linux40-%{+YYYY.MM.dd}" } }'

5.脚本处理,在后台自动运行，上面4种是以进程的方法在后台处理，使用脚本处理是以服务方式进行运行 ，(下面讲)kibana 部署及日志收集
# cat /etc/logstash/conf.d/dmesg.conf 
input { 
  file { 
    path => "/var/log/dmesg" 
    start_position => "beginning" 
    stat_interval => "3"
  } 
} 
output { 
  elasticsearch { 
    hosts => ["192.168.80.120:9200"]  
    index => "linux39-%{+YYYY.MM.dd}" 
  } 
}


使用浏览器查看 head / cerebro  插件显示的索引状态：