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:[email protected]: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: log
enabled: true
encoding: utf-8
paths:
- /tmp/log
fields_under_root: true
output.kafka:
enabled: true
hosts: ["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 => 1
decorate_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"
}
总结
以上的几种方式大家可以简单理解为,项目从小到大的过程,如果只是一个简单的小项目,比如一个用户量不大的管理系统,或者是资源申请比较紧张,那就不妨使用第一种方案,如果是中小型的项目,比如两三个集群节点,可以考虑使用第二种,对于大型项目,或者开发阶段没法准确预估用户量的情况下,用第三种方案准没错
版权归原作者 秃头禅师 所有, 如有侵权,请联系我们删除。