ELK日志分析平台

前段时间,对ELK做了一定的研习,中间踩了许多坑,全都是自己一个人一点点查阅资料解决的,在此做个记录,对logstash方面的记录比较多,其他关于elasticserarch或者kibana以后有了更深入的学习后,再来扩展。

Logstash

  • logstash 如何安装详细参考官网。在centos系统上,一般logstash配置文件目录为:/etc/logstash,
    执行文件目录为:/usr/share/logstash,logstash运行的日志目录:/var/log/messages;

  • logstash的运行依赖JVM,所以需要先安装jdk,安装好jdk后需要配置环境变量;

    在测试阶段,我们使用logstash命令启动程序,只要环境变量配置正确一般是没什么问题的,但是应用阶段我们一般是以服务形式来启动程序,如systemctl start logstash,这时可能会遇到报错的情况,找不到JAVA_HOME,使用以下命令可以解决:ln -s /usr/local/jdk1.8.0_181/bin/java /bin/java,意为:为/bin/java 建立一个软连接,需要的源文件在/usr/local/jdk1.8.0_181/bin/java 中,因为以服务方式启动logstash时,默认去/bin/java路径中寻找java源文件(JAVA_HOME供java命令使用,系统并不关心)。

    输入 which java可以看到返回/bin/java,表示系统认为的java执行路径为/bin/java;

    输入echo &PATH可以查看系统对你输入的命令进行的搜索路径。

  • 服务方式启动常用命令:

    1
    2
    3
    4
    5
    systemctl status logstash
    systemctl restart logstash
    systemctl stop logstash
    systemctl enable logstash 开机自启动
    systemctl disable logstash 停止开机自启动
  • 命令行运行示例:logstash -e 'input { stdin { } } output { stdout {} }'

    -e 标志表示直接从命令行指定配置

    注:windows下要将命令中的单引号变成双引号

  • 配置文件运行示例:logstash -f my_logstash.conf

    centos系统下,./logstash -f my_logstash.conf,”./“表示当前目录;

    -f 标志表示从文件指定配置,配置文件的编写语法为ruby。

    my_logstash.conf:

    1
    2
    3
    4
    5
    6
    7
    8
    9
      input {
    stdin {}
    }
    filter {}
    output {
    stdout {
    codec => rubydebug{}
    }
    }

    logstash的配置文件分为input、filter和output三部分。

    • input输入部分,支持命令行、文件、TCP/UDP、syslog输入和http_poller抓取等;

    • filter过滤部分,有很多插件可以使用,如grok、mutate、date等;

    • output输出部分,可以输出到elasticsearch、TCP/UDP、HDFS和命令行等。此外,也可以保存成文件,报警发送到Nagios。

logstash -f /etc/logstash.d/ or logstash -f /ect/logstash.d &

可以运行/etc/logstash.d目录下所有的文本文件,然后在自己内存里拼接成一个完整的大配置文件,再去执行。

需要注意的有以下几点

  1. 命令行指定配置会覆盖文件配置;
  2. 若要导入数据进Elasticsearch,需要用filter/date插件来修改默认的@timestamp字段值;
  3. logstash每次对某一文件处理后便不会再从头处理,Logstash会保存其位置,并仅在添加新行时处理它们。如需再次处理,需删除.sincedb文件,一般该文件在安装目录/data/plugins/inputs/file文件夹下;
  4. logstash列出目录下所有文件时是字母排序的。而logstash配置段的filter和output都是顺序执行的,所以顺序非常重要。采用多文件管理的用户,推荐采用数字编号方式命名配置文件,同时在配置中严谨采用if判断限定不同日志的动作;

logstash-input

  • 从文件输入:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    input {
    file{
    path => "D:/ELK/new_logs/*.log"
    start_position => "beginning"
    codec => plain{
    charset => "GBK"
    }
    }
    }

    path表示读取的文件路径,可以使用*通配符来读取目录下所有的文件;

    start_position表示从开始位置读取(仅在第一次读取时有效);

    codec表示读取文件使用的编码格式;

    此外还有一些参数,如:

    discover_interval =>15,每隔15秒读取一次文件

    exclude => [“fileName1”,”fileNmae2”],排除哪些文件,也就是不去读取那些文件

    ……

  • 从redis读取:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    input {
    redis {
    type => "systemlog"
    host => "192.168.1.31"
    password => '123456'
    port => "6379"
    db => "6"
    data_type => "list"
    key => "systemlog"
    }
    }

    input–>redis默认编码是json,所以当从redis读取过来的不是json格式的数据时,会有tags:_jsonparsefailure出现,添加codec => plain即可。

    input–>redis设置threads => n(n>1)时,data_type 要设置为list类型,这样当开启多个logstash进程时,才能起到多进程的作用,否则只会重复输出。

    小技巧:当使用redis做消息队列中转日志数据时,读取日志文件输出到redis时,先不要对文件做任何处理,当再次从redis中读取输出到elasticsearch时,再处理。

logstash-grok插件

  1. 标准正则:(?<field_name>the pattern here)

    注:若想匹配某些特殊字符,前面要加反斜杠”\”,如{、(匹配中括号和小括号。

  2. 预先定义一个grok表达式,然后使用%{}引用之

    USERNAME [a-zA-Z0-9._-]+

    %{USERNAME:user_name:data_type}

    注:data_type目前只支持int和float。

  3. 将grok表达式预先统一写入到一个地方,然后用filter/grok的patterns_dir选项来指明

    patterns_dir => [“/path/to/your/own/patterns”]

  4. 多行匹配

    Grok正则和普通正则一样,默认不支持匹配回车换行,若想匹配之在表达式开头加(?m)标记即可。

  5. 多项匹配(用|隔开也可以,但是可读性不强)

    1
    2
    3
    4
    5
    match => [
    “message”, “正则表达式1”,
    “message”, “正则表达式2”,
    “message”, “正则表达式3”,
    ]

注:当正则匹配失败,系统会添加 tags字段,其值为_grokparsefailure。

若想过滤掉匹配失败的事件,可以在输出时加上condition语句:

1
2
3
if "_grokparsefailure" not in [tags] {
elasticsearch {…}
}

logstash-json插件

1
2
3
4
5
json {
source => "data"
target => "json_data"
remove_field => ["data"]
}

source指向具有json格式的字段,target指向新建的字段名,remove_field删除字段名。

注:当具有json格式的字段中其key和value是使用单引号括起来的时,json解析不成功;同样地,json解析不成功时,添加 tags字段,其值为_jsonparsefailure

解决办法:将单引号转变成双引号,使用mutate插件:

1
2
3
mutate{
gsub => ["data", "'", '"']
}

logstash-date插件

将日志中的时间提取出来存到@timestamp中:

1
2
3
4
date {
match => ["logdate", "yyyy-MM-dd HH:mm:ss,SSS"]
target => "@timestamp"
}

logdate为从日志中提取出来的时间字段,具体格式要参考官方文档。

logstash-条件语句

1
2
3
4
5
6
7
if expression {

} else if expression {

} else {

}

相等:==, !=, <, >, <=, >=

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

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

布尔操作:and(与), or(或), nand(非与), xor(非或)

注:使用条件表达式匹配正则时,正则表达式需要用/正则表达式/括起来。

logstash-output -> elasticsearch

1
2
3
4
5
6
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "requests.log"
}
}

注:index索引字段不能有大写字母。

Elasticsearch

关于elasticsearch(以下简称es)的几个坑:

  1. es同一个索引下:相同的字段名不能有不同的格式;

    例如:在某一个索引下”name”字段已有的数据为字符串,然后又要推一条”name”为json格式的数据,就会报错,无法正常写入;已有数据release_time: “2018-08-08”,再推数据release_time: “2018-1-3”会报错,必须改成release_time: “2018-01-03”。

  2. es 规定 root 用户不能启动 es,所以需要创建一个用户来启动 es;

    1
    2
    useradd es
    sudo -u es ./elasticsearch
  3. es配置network.host后出现启动不成功的问题:

    elasticsearch.yml文件中:

    设置:cluster.initial_master_nodes: [“node-1”]

    在最后加上这两句,要不然,外面浏览器就访问不了

    http.cors.enabled: true
    http.cors.allow-origin: “*”

Kibana

kibana启动命令: systemctl start kibana.service(每个命令后面多个service)

Filebeat

非服务形式启动filebeat: ./filebeat -e -c /etc/filebeat/modules.d/filebeat.yml

filebeat读取文件的记录文件存放地址:/usr/share/filebeat/bin/data/registry