Logstash语法常用案例解析（一）

摘要

简述logstash的常用插件，以及简单的使用案例

一：基础运行

建议使用supervisor来管理ELK中的各个组件，方便同一管理

安装 https://www.aolens.cn/?p=809      有讲解

提供一个常用的配置：

[program:logstash]
command=/opt/logstash/bin/logstash -f /opt/logstash/conf/index.conf
numprocs=1 ;开几个进程
dirrectory=/opt/logstash
user=root ;用户
stdout_logfile=/opt/logstash/logs/logstash.log
stdout_logfile_maxbytes=1MB ;每个日志大小
stdout_logfile_backups=10 ;保留10个日志文件
stderr_logfile=/opt/logstash/logs/logstash_err.log
stderr_logfile_maxbytes=1MB
stderr_logfile_backups=10

运行参数：

启动logstash服务（常用supervisor守护进程）

./bin/logstash -f /etc/logstash/conf.d/*  -t             #检查配置文件是否ok

./bin/logstash -f  conf.d/nginx.conf -w 5 -l /var/log/logstash/

二：配置语法

1，区域：（section）

Logstash用｛｝来定义区域。可以在区域中定义多个插件区域，插件区域内可以定义键值对

eg:

input {             # 输入数据
       file {
        path=["/var/log/messages","/var/log/*.log"]
          type="system"
          start_position="beginning"
       }
    }
filter{      # 数据过滤处理
    if[type]=="system"{
    grok{
        match=["message",%{COMBINEDAPACHELOG}]
    }
    }
}
output{      # 数据处理输出
  stdout{
    codec=rubydebug
  }
}

2，数据类型：

    string——普通字符串

            name => "Hello world"

            name => 'It\'s a beautiful day'

    array——数组可以是单个或者多个字符串值。

            path => [ "/var/log/messages", "/var/log/*.log" ]

            path => "/data/mysql/mysql.log"

    hash——键值对，注意多个键值对用空格分隔，而不是逗号。

            match => {

                "field1" => "value1"

                "field2" => "value2"

            ... }

    Codec——用来表示数据编码。用于input和output段。便于数据的处理。

            codec => "json"

    number——必须是有效的数值，浮点数或者整数。

            port => 33

    boolean——布尔值必须是TRUE或者false。

            ssl_enable => true

    bytes——指定字节单位。默认是byte。

            my_bytes => "1113" # 1113 bytes

            my_bytes => "10MiB" # 10485760 bytes

            my_bytes => "100kib" # 102400 bytes  Binary (Ki,Mi,Gi,Ti,Pi,Ei,Zi,Yi) 单位1024

            my_bytes => "180 mb" # 180000000 bytes  SI (k,M,G,T,P,E,Z,Y) 单位基于1000

    password——一个单独的字符串。

            my_password => "password"

    path——代表有效的操作系统路径。

            my_path => "/tmp/logstash"

    3：字段引用

    Logstash配置中要使用字段的值，只需要把字段的名字写在中括号[]里。只要是input输入的值，都可以引用

    eg:

            [geoip][location][-1]

    4，条件判断

    表达式支持的操作符

            ==(等于),!=(不等),<(小于),>(大于),<=(小等),>=(大等)

            =~(匹配正则),!~(不匹配正则)

            in(包含),not in(不包含)

            and(与),or(或),nand(与非),xor(非或)

            ()(复合表达式),!()(取反复合表达式结果)

    三：常用插件

    1，插件管理

            ./bin/plugin -h

            install

            uninstall

            update

            list

            eg:

            bin/plugin install logstash-output-webhdfs

            bin/plugin update logstash-input-tcp

    2，常用插件 input,output,filter,codec

    2.1 Input插件

            stdin：标准输入,常用于测试，

input {
    stdin {
        type = "string"
        tags = ["add"]
        codec="plain"
    }
}

            file：从文件系统中读取文件，类似linux下的tail -f。 最常用

input {
    file {
        path = ["/var/log/*.log","/var/log/message"]     ＃ logstash只支持文件的绝对路径
        type = "system"       ＃ type记录文件类型，定义的变量为全局变量，其他插件都可以调用
        start_position = "beginning"
    }
}

            redis：从redis服务器读取，同时使用redis channel和redis list。

input{
    redis{    
        data_type="list"    
        key="logstash-nginx"    
        host="192.168.1.250"    
        port=6379    
        db=1    
        threads=5
        }} 将源数据写入redis
output{
    redis{    
        host="192.168.1.250"
        port=6379    
        db=1    
        data_type="list"    
        key="logstash-nginx"
        }}

            TCP/UDP： 输入

#nc127.0.0.18888 /var/log/nginx/access.json       # 可以是json文件直接传值
 
#echo'{"name":"liuziping","age":"18"}' |nc127.0.0.18888
 
input {
    tcp {
        port = 8888               # 定义tcp监听端口
        codec="json"               # 规定传入的数据为json格式，k/v结构方便分析
        mode = "server"
    }
}

        syslog：监听在514端口的系统日志信息，并解析成RFC3164格式。

input {
    syslog {
        port = "514"
    }
}

        beats: 通过Filebeat发送事件。

2.2:Output 插件

        stdout：标准输出

output {
    stdout {
        codec = rubydebug
        workers = 2
    }
}

        file ：保存成文件

output {
    file {
        path = "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz"
        message_format = "%{message}"
        gzip = true
    }
}

elasticsearch：保存进elasticsearch ，也是最为重要的

output {
    elasticsearch {
        hosts => ["192.168.0.2:9200"]      # 或者cluster => “ClusterName“
        index => "logstash-%{type}-%{+YYYY.MM.dd}"     #索引名，统一格式，方便kibana导入，会讲统一类型的日志，全部导入 这里的type＝input中的type值
        document_type => "nginx"
        workers => 1                          #启动一个进程
        flush_size => 20000              # 攒够20000 条数据一次性发给ES，默认500条
        idle_flush_time => 10          # 如果10s内没攒够 20000 条也发一次给ES,默认1s
        template_overwrite => true
    }
}

        redis：保存到redis中在input插件中已讲解

        TCP：输出TCP

output { 
    tcp {
     host = "192.168.0.2"
      port = 8888
      codec = json_lines
      } 
    }

        Email:发送邮件

        exec：调用命令执行