基于Filebeat、Logstash和Elasticsearch实现微服务日志采集与存储
基于Filebeat、Logstash和Elasticsearch实现微服务日志采集与存储
程序猿杜小头
发布于 2022-12-01 21:34:33
发布于 2022-12-01 21:34:33
基于Filebeat、Logstash和Elasticsearch实现微服务日志采集与存储
1 技术栈
Filebeat-7.9.1Logstash-7.9.1Elasticsearch-7.9.1Logback-1.2.3
2 日志标准化
日志标准化是指所有微服务日志组件的配置均基于一个模板，模板即Logback日志组件的配置文件logback-spring.xml。在该配置文件中你可以定义日志的输出格式、日志的翻滚策略和基于日志级别分离的日志输出策略等。下面基于以下特性给出参考配置模板：
INFO与ERROR级别日志分别记录在info.log和error.log文件中。- 日志输出格式：%d{yyyy-MM-dd HH:mm:ss} %level {system_name} {module_name} %thread %logger{60}:%method:%line >>> %m%n。
info.log和error.log日志文件默认每天压缩一次并生成gz压缩文件，如果info.log或error.log文件体积超过100MB，则触发二次压缩。gz压缩文件仅保留最近30天但总体积不能超过6GB。- 所有日志文件路径为：/apps/logs/{system_name}/{module_name}/。
<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
<!-- 系统名 -->
<property name="system_name" value="transformers"/>
<!-- 模块名，即服务名称 -->
<property name="module_name" value="optimus_prime"/>
<!-- 日志文件存放位置 -->
<Property name="log_path" value="/apps/logs/${system_name}/${module_name}"/>
<!-- 日志压缩文件存放位置 -->
<Property name="archive_path" value="${log_path}/archives"/>
<!-- info级别日志文件名 -->
<Property name="info_log_name" value="${module_name}-info"/>
<!-- error级别日志文件名 -->
<Property name="error_log_name" value="${module_name}-error"/>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss} %level ${system_name} ${module_name} %thread %logger{60}:%method:%line >>> %m%n</pattern>
</encoder>
</appender>
<!-- info级日志记录器 -->
<appender name="INFO_APPENDER" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 如果日志文件已经存在，则追加记录；设置为false会清空已经存在的日志内容 -->
<append>true</append>
<file>${log_path}/${info_log_name}.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<!-- 一般，每天仅压缩一次 -->
<fileNamePattern>${archive_path}/%d{yyyy-MM, aux}/${info_log_name}.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
<!-- info.log文件最大体积100MB，超过该阈值则触发二次压缩 -->
<maxFileSize>100MB</maxFileSize>
<!-- 压缩文件仅保留最近30天历史 -->
<maxHistory>30</maxHistory>
<!-- 压缩文件累积体量最大6GB，超过则删除存活时间最久的压缩文件开始 -->
<totalSizeCap>6GB</totalSizeCap>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>INFO</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss} %level ${system_name} ${module_name} %thread %logger{60}:%method:%line >>> %m%n</pattern>
</encoder>
</appender>
<!-- error级日志记录器 -->
<appender name="ERROR_APPENDER" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 如果日志文件已经存在，则追加记录；设置为false会清空已经存在的日志内容 -->
<append>true</append>
<file>${log_path}/${error_log_name}.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<!-- 一般，每天仅压缩一次 -->
<fileNamePattern>${archive_path}/%d{yyyy-MM, aux}/${error_log_name}.%d{yyyy-MM-dd}.%i.log.gz</fileNamePattern>
<!-- error.log文件最大体积100MB，超过该阈值则触发二次压缩 -->
<maxFileSize>100MB</maxFileSize>
<!-- 压缩文件仅保留最近30天历史 -->
<maxHistory>30</maxHistory>
<!-- 压缩文件累积体量最大6GB，超过则删除存活时间最久的压缩文件开始 -->
<totalSizeCap>6GB</totalSizeCap>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>ERROR</level>
</filter>
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss} %level ${system_name} ${module_name} %thread %logger{60}:%method:%line >>> %m%n</pattern>
</encoder>
</appender>
<logger name="org.mybatis" level="ERROR" additivity="true"/>
<logger name="org.springframework" level="ERROR" additivity="true"/>
<logger name="springfox.documentation" level="ERROR" additivity="true"/>
<logger name="org.apache.http" level="ERROR" additivity="true"/>
<root level="INFO">
<appender-ref ref="STDOUT"/>
<appender-ref ref="INFO_APPENDER"/>
<appender-ref ref="ERROR_APPENDER"/>
</root>
</configuration>
3 日志采集
日志采集使用Filebeat来实现，只需要在每个服务实例宿主机器上部署一个Filebeat实例即可，Filebeat会到指定路径下的日志文件中采集日志数据。对于INFO级别的日志采集，大家应该都没什么困惑，因为它们始终是一行信息；但是对于ERROR级别的日志该如何采集呢？毕竟ERROR级别日志是一个多行堆栈信息。幸运的是，Filebeat具备multiline采集能力，可以将多行堆栈信息聚合为单行日志事件。multiline有四个配置项：
multiline.type，多行日志聚合类型，有两种类型：pattern和count。multiline.pattern，多行匹配正则表达式，搭配multiline.negate与multiline.match选项，那么就可以将那些与该正则表达式匹配的行视为一个新的多行日志事件或者是上一多行日志事件的延续。multiline.negate，该配置决定了正则表达式的匹配方向，默认值为false，即正向(即匹配)。multiline.match，该配置声明了匹配或不匹配的连续多行日志在一个多行日志事件中的位置，其值为：after|before。
下面基于以下思路给出Filebeat参考配置：
以
yyyy-MM-dd模式开头的单行日志被看作是一个新多行日志事件；而不以yyyy-MM-dd模式开头的单行日志则被看作是上一多行日志事件的延续。
# ============================== Filebeat inputs ===============================
filebeat.inputs:
- type: log
enabled: true
paths:
- /apps/logs/${system_name}/*/*-info.log
- type: log
enabled: true
paths:
- /apps/logs/${system_name}/*/*-error.log
multiline.type: pattern
multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
# ------------------------------ Logstash Output -------------------------------
output.logstash:
hosts: ["LOGSTASH_HOST:5044"]
如果配置了多个
Logstash Output，那么Filebeat会采用轮训的方式随机向Logstash传输日志数据。
4 日志预处理与转发
Logstash负责接收Filebeat采集到的日志数据，然后利用Grok表达式对原生日志数据进行Key Value映射操作，最后将数据发送到Elasticsearch中去。
下面基于以下特性给出Logstash参考配置：
- 每天生成一个索引，而不是所有日志数据都堆积在一个索引中，这样不利于索引生命周期管理。
- 索引名称格式：
elk-YYYY.MM.dd
input {
beats {
port => 5044
client_inactivity_timeout => 1800
}
}
filter {
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:logLevel} %{WORD:systemName} %{DATA:moduleName} %{DATA:threadName} %{JAVACLASS:className}:%{JAVAMETHOD:methodName}:%{NUMBER:lineNum} >>> %{GREEDYDATA:message}" }
overwrite => [ "message" ]
remove_field => [ "log", "host", "ecs", "tags", "agent", "input", "container" ]
}
if "_grokparsefailure" in [tags] {
drop {}
}
}
output {
elasticsearch {
hosts => ["ES_HOST:9200"]
user => "ES_USERNAME"
password => "ES_PASSWORD"
index => "elk-%{+YYYY.MM.dd}"
}
}
5 日志存储
Elasticsearch主要用于存储日志数据以及暴露Restful API供上层业务方查询日志数据。但是在Elasticsearch正式接入日志数据之前，你必须做一些初始化操作。
5.1 索引生命周期管理
对于微服务产生的日志是不需要永久保存的，建议保存最近7天数据即可。由于本文日志索引生成策略为按天生成，那么只需要删除那些存活时间超过7天的日志索引即可。在Index Lifecycle Management特性(Elasticsearch 6.6后引入)的加持下，我们可以很容易地实现自动保存最近7天数据。
5.1.1 创建索引生命周期管理策略
PUT /_ilm/policy/sdwan-log-ilm-policy
{
"policy": {
"phases": {
"delete": {
"min_age": "7d",
"actions": {
"delete": {}
}
}
}
}
}
5.1.2 检查索引生命周期状态
我们可以通过索引生命周期EXPLAIN API查看索引创建时间、已存活时间等信息。
GET /elk-*/_ilm/explain
5.2 创建索引模板
PUT /_index_template/sdwan-log-index-template
{
"index_patterns": [
"elk-*"
],
"template": {
"settings": {
"lifecycle": {
"name": "sdwan-log-ilm-policy",
"rollover_alias": "sdwan-log-ilm-policy-alias"
},
"number_of_shards": "1",
"max_result_window": "1000000",
"number_of_replicas": "1",
"analysis": {
"analyzer": {
"sdwan_analyzer": {
"type": "custom",
"tokenizer": "standard",
"char_filter": [
"sdwan_char_filter"
]
}
},
"char_filter": {
"sdwan_char_filter": {
"type": "mapping",
"mappings": [
"a => -a-",
"b => -b-",
"c => -c-",
"d => -d-",
"e => -e-",
"f => -f-",
"g => -g-",
"h => -h-",
"i => -i-",
"j => -j-",
"k => -k-",
"l => -l-",
"m => -m-",
"n => -n-",
"o => -o-",
"p => -p-",
"q => -q-",
"r => -r-",
"s => -s-",
"t => -t-",
"u => -u-",
"v => -v-",
"w => -w-",
"x => -x-",
"y => -y-",
"z => -z-"
]
}
}
}
},
"mappings": {
"properties": {
"@timestamp": {
"type": "date"
},
"@version": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"className": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"lineNum": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"logLevel": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"message": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"methodName": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
},
"analyzer": "sdwan_analyzer"
},
"moduleName": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
},
"analyzer": "sdwan_analyzer"
},
"systemName": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
},
"analyzer": "sdwan_analyzer"
},
"threadName": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"timestamp": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
}
}
}
}
6 搜索日志数据
GET /elk-2021.02.03/_search
{
"from": 0,
"size": 1,
"timeout": "10s",
"_source": {
"exclude": [
"@version",
"@timestamp"
]
},
"track_total_hits": true,
"query": {
"bool": {
"must": [
{
"match_phrase": {
"moduleName": "admin"
}
}
]
}
},
"sort": [
{
"timestamp.keyword": {
"order": "desc"
}
}
]
}
搜索结果如下，其中_source字段中的内容就是微服务所产生的原生日志数据。
{
"took": 2,
"timed_out": false,
"_shards": {
"total": 1,
"successful": 1,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 18,
"relation": "eq"
},
"max_score": null,
"hits": [
{
"_index": "elk-2021.02.03",
"_type": "_doc",
"_id": "byCRZ3cBbZJ5iJayNjn2",
"_score": null,
"_source": {
"systemName": "ccn",
"logLevel": "INFO",
"moduleName": "ccn-admin",
"lineNum": "251",
"methodName": "report",
"className": "c.c.cmss.cnfusion.ccn.pm.scheduler.CapacityReportScheduler",
"message": "capacity report, resp body: {\"code\":200,\"data\":[],\"msg\":\"\"}",
"threadName": "scheduling-1",
"timestamp": "2021-02-03 19:05:00"
}
}
]
}
}
于是，一条INFO级别的日志就保存到了Elasticsearch中了。
2021-02-03 19:05:00 INFO ccn ccn-admin scheduling-1 c.c.cmss.cnfusion.ccn.pm.scheduler.CapacityReportScheduler:report:251 >>> capacity report, resp body: {"code":200,"data":[],"msg":""}
这里就不展示ERROR级别的数据了，message字段内容对应的是完整的异常堆栈信息，太长了，显示很不友好。
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2021-03-04，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
Elasticsearch Service
腾讯云 Elasticsearch Service（ES）是云端全托管海量数据检索分析服务，拥有高性能自研内核，集成X-Pack。ES 支持通过自治索引、存算分离、集群巡检等特性轻松管理集群，也支持免运维、自动弹性、按需使用的 Serverless 模式。使用 ES 您可以高效构建信息检索、日志分析、运维监控等服务，它独特的向量检索还可助您构建基于语义、图像的AI深度应用。