ELK学习总结

最近学习ELK(ElasticSearch, Logstash, Kibana)，总结了一些内容，东西不全，欢迎大家补充，主要是写怎么集成，如果有不正确的地方，也请大家纠正。
安装参考官网：
https://www.elastic.co/guide/en/elastic-stack/current/installing-elastic-stack.html

1.简析

Logstash：收集、解析日志，收集的来源包括file、stdin、syslog、tcp、udp、redis、kafka等，经过filter解析转化，将日志输出到ES、file、stdout、tcp、hdfs、redis、kafka等，https://www.elastic.co/products/logstash

ElasticSearch：查询、分析、存储日志，可以集群部署，后面简称ES。

Kibana：依赖ES对日志进行查询、统计分析，可以制作各种图表，将日志信息展示在web界面。

2.采集数据方式

logstash需要配置input(数据从哪来)、filter(提取重要信息，转化字段类型等)、output(数据输出到哪)；
kibana需要配置ES的ip和port等信息。

logstash配置示例：
#输入
input{
#从文件读取日志信息
file {
#日志文件绝对路径
path=>“D:/ELK/5.4.1/logstash-5.4.1-test/bin/test.log”
#输出到es中存储的类型，用以区分不同应用的日志
type=>“runtimelog”
#监控日志位置，beginning和end两个值可选，如果需要导入历史日志使用beginning
start_position => beginning
#换行的合并成一行
codec=>multiline {
pattern => “^[”
negate => true
what => “previous”
}
}
file {
path=>“D:/tomcat/apache-tomcat-6.0.14-8090/logs/drsp-metadata.log”
type=>“drsp-metadata”
start_position => beginning
}
#udp中读取
udp {
port => 25826
buffer_size => 1452
workers => 3
queue_size => 30000
}
#tcp中读取
tcp {
port => 8888
mode => “server”
ssl_enable => false
}
#从redis中读取
redis {
#redis服务器ip
host => “172.16.193.55”
#redis中存储的key，可以用*模糊匹配
key => “logstash-redis”
data_type => “list”
}
}
#过滤
filter {
if [type]“runtimelog” {
grok {
#通过正则表达式匹配日志格式，获取日志中的关键信息，day、level这些字段会在kibana界面中，用于搜索和创建图表
match => [ “message”,
“%{NOTSPACE:day} %{NOTSPACE:datetime} %{NOTSPACE:thread-id} %{NOTSPACE:level} %{GREEDYDATA:msginfo}”]
}
}
if [type]“drsp-metadata” {
grok {
#当日志格式一致时，可以将正则表达式写入logstash的pattern中，其它应用可以直接引用存储的key，与上面的效果一致
match => [ “message”, “%{JAVALOG}” ]
}
}
}
#输出
output {
#输出到ES中
elasticsearch {
#ES的服务器地址和端口
hosts => [“172.16.193.55:9200”]
#存储的日志索引命名方式，建议如下
index => “logstash-%{type}-%{+YYYY.MM.dd}”
}
#输出到文件中
file {
path => “D:/ELK/logstash-tutorial-output-test.txt”
}
}

2.1 直接读取日志文件

logstash直接读取应用系统输出的日志文件，或者中间件的访问日志文件等。
参考上面的logstash配置示例
1.配置logstash的input，从文件读取的方式。
2.编写日志的正则表达式提取重要信息，用于搜索和创建图表。
例如日志为：
2017-06-13 14:58:23:109 [com.thunisoft.artery.form.FormEngine:346]-[ERROR] null

正则表达式可以写为，JAVALOG即上面filter的match中使用的：
JAVALOG %{NOTSPACE:day} %{NOTSPACE:datetime} [%{NOTSPACE:class}:%{NUMBER:costtime}]-%{NOTSPACE:level} %{GREEDYDATA:msginfo}

并将其存储到存储到logstash的下面目录中：
vendor\bundle\jruby\1.9\gems\logstash-patterns-core-4.1.0\patterns

3.配置过滤器，过滤数据，修改数据类型等
4.配置output，配置ES的访问信息

Kibana中展示的日志信息：
ELK学习总结

以下是使用logback的appender，不需要写正则表达式匹配提取日志信息

2.2 logback+redis+logstash

已经使用logback记录日志的系统可以将logback和logstash对接，先将日志输出到redis，再通过logstash输出到ES，日志流转过程：logback->redis->logstash->ES
1.pom中添加jar包依赖

<dependency>  
    <groupId>com.cwbase</groupId>
    <artifactId>logback-redis-appender</artifactId>
    <version>1.0.0</version>  
</dependency>

2.修改logback.xml，配置appender和appender-ref

<appender name="LOGSTASH" class="com.cwbase.logback.RedisAppender">
    <source>YUSP-PC</source>
	<type>test</type>  <!-- 类型标识 -->
	<host>logstashHost</host>
	<key>logstash-redis</key>   <!-- 这个Key是Redis List的Key，需要和Logstash读取Redis的Key匹配 -->
	<tags>test</tags>
	<mdc>true</mdc>
	<location>true</location>
	<callerStackIndex>0</callerStackIndex>
</appender>
<root level="error">
    ...
	<appender-ref ref="logstash" />
</root>

3.配置logstash的input、filter、output，input参考示例配置成redis输入方式，注意key要与上面配置的key一致，filter可以做一些类型转化，不需要写正则表达式提取重要日志信息，已自动提取，output配置成ES的访问信息。

Kibana中展示的日志信息：
ELK学习总结

2.3 logstash-logback-encoder+logstash

此种方法比上面少了redis
1.pom中添加jar包依赖

<dependency>
   <groupId>net.logstash.logback</groupId>
   <artifactId>logstash-logback-encoder</artifactId>
   <version>4.5.1</version>
</dependency>

2.修改logback.xml，配置appender和appender-ref

<!-- LogstashSocketAppender输出方式为udp，如果使用LogstashTcpSocketAppender，则输出方式为tcp，logstash中input要对应配置成udp或者tcp -->
<appender name="logstash" class="net.logstash.logback.appender.LogstashSocketAppender">
    <host>logstashHost</host>
    <port>logstashPort</port>
</appender>
	
<root level="error">
    ...
	<appender-ref ref="logstash" />
</root>