elasticsearch做如何进行日志采集

首先elasticsearch应理解为日志的存储介质,由于其查询的便利,尤其结合kibana进行可视化查询,很多项目采用elasticsearch来存储收集的日志,至于如何进行日志采集,下面列举了项目中常用的一些采集手段

ElasticsearchAppender

lockback日志组件,集成 ElasticsearchAppender 插件的方式采集,具体操作如下
1,引入pom依赖

    <dependency><groupId>com.internetitem</groupId><artifactId>logback-elasticsearch-appender</artifactId><version>1.6</version></dependency>

2, lockback.xml中新增appender

  <appender name="ELASTIC" class="com.internetitem.logback.elasticsearch.ElasticsearchAppender"><url>http://elastic:elastic@127.0.0.1:9200/_bulk</url><index>log-%d{yyyy-MM}</index> <connectTimeout>30000</connectTimeout> <errorsToStderr>false</errorsToStderr> <includeCallerData>false</includeCallerData> <logsToStderr>false</logsToStderr> <maxQueueSize>104857600</maxQueueSize> <maxRetries>3</maxRetries> <readTimeout>30000</readTimeout> <sleepTime>250</sleepTime> <rawJsonMessage>false</rawJsonMessage> <includeMdc>false</includeMdc> <maxMessageSize>-1</maxMessageSize> <authentication class="com.internetitem.logback.elasticsearch.config.BasicAuthentication"/> <!-- optional --><properties><property><name>host</name><value>${HOSTNAME}</value><allowEmpty>false</allowEmpty></property><property><name>ip</name><value>%ip</value></property><property><name>@timestamp</name><value>%d{yyyy-MM-dd}</value></property><property><name>level</name><value>%level</value></property><property><name>thread</name><value>%thread</value></property><property><name>stacktrace</name><value>%ex</value></property><property><name>logger</name><value>%logger</value></property><property><name>env</name><value>${active}</value></property></properties><headers><header><name>Content-Type</name><value>application/json</value></header></headers>
</appender>

ps: 此配置包含两部分,一部分为连接信息,主要是账号,密码,超时时间等连接配置,另一部分为日志内容信息,比如主机名,ip,日志level等,这些信息最终会自动同步到es中,建议同步之前先建立好索引(es的特性是可以自动建立索引,可能有些字段类型,分词器未必满足需求),或者使用索引模板,事先定义好mapping

ELK

由于 ElasticsearchAppender 是一个非常轻量的插件,功能很单一,只能简单的进行日志采集,对于一些复杂性的需求,比如对传输过来的日志进行清洗,过滤等需求就显得无能为力,而且与项目的集成度过高,如果项目吞吐量很大,比如每秒上万条日志,会造成日志积压(很好理解,单机总会有瓶颈),严重点会影响应用本身的吞吐量,所以稍微大点的项目一般会采用elk
所谓elk 实际上是 elasticsearch+logstash+kibana, elasticsearch和kibana上面都有提过,是用来存储和可视化查询的,最重要的采集工作是logstash去做的,本文只讲实战,理论东西如有兴趣,朋友们可以留言后续会更新
logstash的标准格式为

input {     从哪个地方读取，输入数据。}filter {    依据grok模式对数据进行分析结构化}output {    将分析好的数据输出存储到哪些地方}

1,新增一个配置conf

input {file {     读取日志文件path => ["/tmp/log"]   日志文件路径type => "web log"    文件类型,这个是日志类型,后面filter 可根据类型进行区分start_position => "beginning"    从最开始取数据}
}filter {grok {match => { "message" => "%{IP:clientip} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}output {# 一般通过模板的方式,达到自动创建,比如每天创建新的索引elasticsearch {hosts => ["http://localhost:9200""]user => "test"password => "test"index => "log-%{+YYYY.MM.dd}"template_overwrite => "false"}

ELK + filebeat + kafka

1,上面的elk logstash是直接部署在应用服务器的,对于中小型项目足够,对于一些大型项目,依然会有些不妥,还是上面说的和项目的耦合度太大,理论上还是会出现单点瓶颈,而且logstash使用Java语言写的,大家都知道jvm是很吃资源的,所以大型项目一般会在elk的基础上,采集端使用 filebeat 进行采集,采集到的日志传输到kafka,logstash对接kafka的topic,后续的操作和之前一样.
1, 配置filebeat

filebeat.inputs:
- type: logenabled: trueencoding: utf-8paths:- /tmp/logfields_under_root: trueoutput.kafka:enabled: truehosts: ["127.0.0.1:9092"]topic: "log"username: "test"password: "test"

2,配置logstash

input就变成了kafka
input {kafka{bootstrap_servers => "127.0.0.1:9092"client_id => "consumer_id"group_id => "consumer_group"auto_offset_reset => "latest" consumer_threads => 1decorate_events => true topics => ["log"]}
}filter {grok {match => { "message" => "%{IP:clientip} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}output {# 一般通过模板的方式,达到自动创建,比如每天创建索引elasticsearch {hosts => ["http://localhost:9200""]user => "test"password => "test"index => "log-%{+YYYY.MM.dd}"template_overwrite => "false"}

总结

以上的几种方式大家可以简单理解为,项目从小到大的过程,如果只是一个简单的小项目,比如一个用户量不大的管理系统,或者是资源申请比较紧张,那就不妨使用第一种方案,如果是中小型的项目,比如两三个集群节点,可以考虑使用第二种,对于大型项目,或者开发阶段没法准确预估用户量的情况下,用第三种方案准没错