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