logstash

目录：

1、安装logstash
2、logstash快速入门
3、logstash收集系统日志-file
4、logstash收集java日志-codec
5、logstash收集nginx访问日志-json
6、使用syslog收集系统日志
7、logstash收集tcp日志
8、logstash收集slowlog-grok
9、logstash解耦之消息队列

1、安装logstash    <--返回目录

下载并安装GPG key
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch

添加yum仓库
vim /etc/yum.repos.d/logstash.repo

[logstash-2.1]
name=Logstash repository for 2.1.x packages
baseurl=http://packages.elastic.co/logstash/2.1/centos
gpgcheck=1
gpgkey=http://packages.elastic.co/GPG-KEY-elasticsearch
enabled=1

安装logstash
yum install -y logstash

查看安装文件目录：rpm -ql logstash

2、logstash快速入门    <--返回目录

　　/opt/logstash/bin/logstash -e 'input {stdin{}} output {stdout{}}'

 　　/opt/logstash/bin/logstash -e 'input {stdin{}} output {stdout{ codec => rubydebug }}'

 　　/opt/logstash/bin/logstash -e 'input {stdin{}} output {elasticsearch{ hosts => ["192.168.213.200:9200"] }}'

 　　查看es：

　　可以配置多个输出：/opt/logstash/bin/logstash -e 'input {stdin{}} output {elasticsearch{ hosts => ["192.168.213.200:9200"] } stdout{ codec => rubydebug }}'

3、logstash收集系统日志-file    <--返回目录

3.1、控制台输入数据作为input

　　vim /etc/logstash/conf.d/01-logstash.conf，添加以下内容：

input { stdin{} } 
output {
    elasticsearch{ hosts => ["192.168.213.200:9200"] } 
    stdout{ codec => rubydebug }
}

　　控制台输入指令 /opt/logstash/bin/logstash -f /etc/logstash/conf.d/01-logstash.conf，然后再控制台输入一些文本，

 3.2、文件输入作为input

　　vim /etc/logstash/conf.d/01-logstash.conf，添加以下内容：

input {
    file {
        path => "/var/log/a.log"
        type => "a_type"
        start_position => "beginning"
    }
} 
    
output {
    elasticsearch{ hosts => ["192.168.213.200:9200"] } 
    stdout{ codec => rubydebug }
}

　　vim /var/log/a.log，添加一些测试的文本数据，然后输入指令 /opt/logstash/bin/logstash -f /etc/logstash/conf.d/01-logstash.conf启动logstash。

　　start_position => "beginning" 表示从文件开始读入，如果是"end"则是从文件末尾开始读。重启后并不会重复读取文件：

 　　但是当在文件/var/log/a.log中添加一行文本"444"后，a.log文件又从文件开始读取了。

　　type => "a_type" 自定义的一个类型标识名称（名称自己取）。后面elasticsearch的索引可以与a_type一致。

input {
    file {
        path => "/var/log/a.log"
        type => "a_type"
        start_position => "beginning"
    }
} 
    
output {
    elasticsearch{
        hosts => ["192.168.213.200:9200"] 
        index => "a_type-%{+YYYY.MM.dd}"
    } 
    stdout{ codec => rubydebug }
}

4、logstash收集java日志-codec    <--返回目录

　　vim /etc/logstash/conf.d/02-logstash.conf，添加以下内容：

input {
    file {
        path => "/var/log/a.log"
        type => "a_type"
        start_position => "beginning"
    }
    file {
        path => "/var/log/java.log"
        type => "java_type"
        start_position => "beginning"
        codec => multiline {
            #pattern => "^[" #多行文本合并的正则规则
            pattern => "^2021"
            negate => true
            what => "previous" #合并到前面
        }
    }
} 
    
output {
    if [type] == "a_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "a_type-%{+YYYY.MM.dd}"
        } 
    }
    if [type] == "java_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "java_type-%{+YYYY.MM.dd}"
        } 
    }
    
    stdout{ codec => rubydebug }
}

　　多行合并的效果：

5、logstash收集nginx访问日志-json    <--返回目录

 　　nginx修改访问日志格式为json

　　或者将access_log 配置在server

　　访问日志 access.log

 　　vim /etc/logstash/conf.d/02-logstash.conf，添加以下内容：

input {
    file {
        path => "/var/log/a.log"
        type => "a_type"
        start_position => "beginning"
    }
    file {
        path => "/var/log/java.log"
        type => "java_type"
        start_position => "beginning"
        codec => multiline {
            #pattern => "^[" #多行文本合并的正则规则
            pattern => "^2021"
            negate => true
            what => "previous" #合并到前面
        }
    }
    
    file {
        path => "/var/log/nginx/access_json.log"
        codec => json
        type => "nginx_access_log_type"
        start_position => "beginning"
    }
} 
    
output {
    if [type] == "a_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "a_type-%{+YYYY.MM.dd}"
        } 
    }
    if [type] == "java_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "java_type-%{+YYYY.MM.dd}"
        } 
    }
    
    if [type] == "nginx_access_log_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "nginx_access_log_type-%{+YYYY.MM.dd}"
        } 
    }
    
    stdout{ codec => rubydebug }
}

　　/opt/logstash/bin/logstash -f /etc/logstash/conf.d/02-logstash.conf

6、使用syslog收集系统日志    <--返回目录

　　vim /etc/logstash/conf.d/system_syslog_type.conf，添加以下内容：

input {
    syslog {
        type => "system_syslog_type"
        host => "192.168.213.200"
        port => "514"
    }
    
} 
    
output {
    if [type] == "system_syslog_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "system_syslog_type-%{+YYYY.MM.dd}"
        } 
    }
    
    stdout{ codec => rubydebug }
}

　　启动logstash： /opt/logstash/bin/logstash -f /etc/logstash/conf.d/system_syslog_type.conf

　　查看是否监听514端口：

　　配置 vim /etc/rsyslog.conf

　　对应的操作记录也被收集了：

 　　测试时直接在控制台输入: logger "test syslog..."

7、logstash收集tcp日志    <--返回目录

　　vim /etc/logstash/conf.d/tcp_type.conf，添加以下内容：

input {
    tcp {
        type => "tcp_type"
        host => "192.168.213.200"
        port => "6666"
    }
    
} 
    
output {
    if [type] == "tcp_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "tcp_type-%{+YYYY.MM.dd}"
        } 
    }
    
    stdout{ codec => rubydebug }
}

　　启动logstash: /opt/logstash/bin/logstash -f /etc/logstash/conf.d/tcp_type.conf

　　查看是否正在监听6666端口

　　安装 nc: yum install -y nc

　　通过nc发送

　　查看es

8、logstash收集slowlog-grok    <--返回目录

 　　vim /etc/logstash/conf.d/grok_type.conf，添加以下内容：

input {
    stdin {}
} 
filter {
    grok {
        match => {
            "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
        }
    }
}
output {
    if [type] == "grok_type" {
        elasticsearch{
            hosts => ["192.168.213.200:9200"] 
            index => "grok_type-%{+YYYY.MM.dd}"
        } 
    }
    
    stdout{ codec => rubydebug }
}

　　启动logstash：/opt/logstash/bin/logstash -f /etc/logstash/conf.d/grok_type.conf

　　控制台输入：55.3.244.1 GET /index.html 15824 0.043

　　"%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"的解释： IP是logstash内置定义的正则匹配规则

9、logstash解耦之消息队列    <--返回目录

 

 其他参考：

　　1）ELK学习笔记之Logstash详解

　　2）废物大师兄--Logstash介绍