建设网站实训心得,安阳网络推广代理,收费资源下载wordpress主题,加盟网站有哪些一、背景
高并发接口中#xff0c;为了提高接口的高性能#xff0c;在需要保存审计及操作记录的时候#xff0c;往往有以下常见方案#xff1a;
保存到redis数据库异步保存到mysql/mongodb/es等数据库logger打印业务日志#xff0c;采集与展示则交由elk模块
对于第一种…一、背景
高并发接口中为了提高接口的高性能在需要保存审计及操作记录的时候往往有以下常见方案
保存到redis数据库异步保存到mysql/mongodb/es等数据库logger打印业务日志采集与展示则交由elk模块
对于第一种方案接口的高性能依赖于redis的性能第二种方案的关键在于异步可以是基于事件驱动机制常见的CQRS设计就是例子 本文则是介绍第三种方案不同的是我们的数据展示是在业务管理后台而非kibana。
另外生产环境我们的应用程序是部署在k8s容器。
二、设计方案 1、服务打印日志持久化到nfs2、filebeat先挂载nfs再配置采集日志3、kafka作为数据采集的削峰填谷的角色4、Logstash读取kafka的数据解析并存储到指定es数据库5、管理后台连接并读取es数据库展示数据
三、打印业务日志
使用logback使用不同的Logger对象区分普通的jvm日志把业务日志输出到指定的日志文件。
logback-spring.xml
?xml version1.0 encodingUTF-8?
!--该日志将日志级别不同的log信息保存到不同的文件中 --
configurationinclude resourceorg/springframework/boot/logging/logback/defaults.xml/springProperty scopecontext namespringAppNamesourcespring.application.name/springProperty scopecontext namelog_dir sourcelogging.login.path defaultValue../logs/!-- 日志输出位置 --property nameLOG_FILE value${log_dir}/${springAppName}-login/!-- 文件的日志输出样式 --property nameFILE_LOG_PATTERNvalue%d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}} ${LOG_LEVEL_PATTERN:-%5p} ${PID:- } --- [%t] %-40.40logger{39} : %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}/!-- 为logstash输出的JSON格式的Appender --appender namelogstashclassch.qos.logback.core.rolling.RollingFileAppenderfile${LOG_FILE}.log/filerollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy!--日志文件输出的文件名 --fileNamePattern${LOG_FILE}.log.%d{yyyy-MM-dd}.%i.gz/fileNamePatterntimeBasedFileNamingAndTriggeringPolicyclassch.qos.logback.core.rolling.SizeAndTimeBasedFNATPmaxFileSize20MB/maxFileSize/timeBasedFileNamingAndTriggeringPolicy!--日志文件保留天数 --MaxHistory15/MaxHistory/rollingPolicy!-- 日志输出编码 --encoderpattern${FILE_LOG_PATTERN}/patterncharsetutf8/charset/encoder/appenderlogger namecom.xxx.event.handler.LoginLogEventHandler levelINFO additivityfalseappender-ref reflogstash//logger
/configuration普通的jvm日志选择输出在console控制台然后将流重定向到指定的日志文件里。
console
!-- 控制台的日志输出样式 --property nameCONSOLE_LOG_PATTERNvalue%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}/!-- 控制台输出 --appender nameconsole classch.qos.logback.core.ConsoleAppenderfilter classch.qos.logback.classic.filter.ThresholdFilterlevelINFO/level/filter!-- 日志输出编码 --encoderpattern${CONSOLE_LOG_PATTERN}/patterncharsetutf8/charset/encoder/appender!-- 日志输出级别 --root levelINFOappender-ref refconsole//root对于业务应用程序来说所做的事情只需要打印一条日志即可其高性能可想而知。
四、flebeat采集日志
申请一台机器专门采集日志而非在每个Pod容器里使用sidecar边车模式部署filebeat容器。
当然这台机器必须要先挂载nfs。nfs是购买的阿里云“文件存储NAS”服务 下面是filebeat.yml示例
# cat /etc/filebeat/filebeat.yml
filebeat.prospectors:
- input_type: logpaths:- /login_output_service/auth-service.logmultiline.pattern: ^[0-9]{4}-[0-9]{2}-[0-9]{2}multiline.negate: truemultiline.match: afterdocument_type: login_out_logsoutput.kafka:enabled: truehosts: [xxx.alikafka.aliyuncs.com:9092,yyy.alikafka.aliyuncs.com:9092,zzz.alikafka.aliyuncs.com:9092]topic: login_out_logs五、kafka中间件
当然它作为数据采集的中间件所起的作用是削峰填谷可以说在整个链路中不是必须的。
建议你采用mq作为数据采集的过渡。
这里插一句我们可以通过kafka的消息内容反查出filebeat所在的机器。
原本的思路是想通过文件存储NAS查找ECS的挂载详情可惜没找到入口。也就是说无法得到哪些ECS挂载了NFS只能选择通过Kafka 忘记了filebeat是部署在哪个机器好在我们知道kafka因为购买的是阿里云服务。 进入“消息队列Kafka版” 查看消息体内容 {“timestamp”:“2024-04-01T20:26:24.956Z”,“beat”:{“hostname”:“iZbp12202mumv2j30jv6l8Z”,“name”:“iZbp12202mumv2j30jv6l8Z”,“version”:“5.6.9”},“input_type”:“log”,“message”:“2024-04-02 04:25:18.486 INFO [auth-service,] 8 — [AuthService_EventPool_757886445] c.x.s.a.i.e.h.LoginLogEventHandler : LoginLogEventHandler - writeLoginLog - 登录登出日志 [id1774895838313730083\tdate20240402042518\tosVersion 13]”,“offset”:10016,“source”:“/opt/server/auth-service.log”,“type”:“login_out_logs”} 从这里可以看到filebeat机器hostname是iZbp12202mumv2j30jv6l8Z注意需要转换实际上是“i-bp12202mumv2j30jv6l8”。
另外从source字段可以知道日志的源文件在哪。
根据hostname查找ecs所在地
进入“云服务器ECS” 通过这两步我们就通过kafka消息中间件反查得到filebeat所在机器。关于filebeat在上一步已介绍下面将介绍logstash的配置。
六、logstash存储业务日志
Logstash主要是用来解析业务日志先读取kafka的数据然后存储到es数据库。
另外logstash支持多个配置文件建议你分开配置并加载。
# 默认配置
logstash -f /usr/share/logstash/pipeline/logstash.conf# 修改为通配符*.conf
logstash -f /usr/share/logstash/pipeline/*.confauth.conf 详情
# cat /usr/local/logstash-6.8.23/config/auth.confinput {kafka {bootstrap_servers xxx.alikafka.aliyuncs.com:9092,yyy.alikafka.aliyuncs.com:9092,zzz.alikafka.aliyuncs.com:9092codec jsontype _doctopics login_out_logs}
}filter {grok {match {message%{JAVA_DATE:logdate}\s%{XHJVM_VALUE:logType}\s.*登录登出日志\s\[id(%{FIELD_VALUE:id}?\s)date(%{FIELD_VALUE:date}?\s)osVersion(%{FIELD_VALUE:osVersion}?)\]}}# 增加采集时间戳去掉一些非必要的字段date {match [logdate, yyyy-MM-dd HH:mm:ss.SSS]target timestampremove_field [ logdate, logType, host, version, timestamp, path, message, beat, source, input_type, offset]}}output{elasticsearch{# es的地址hosts [10.xxx.xxx.126:9200]index auth_login_out_logs_writecodec line { format %{message}}document_type _doc# es的用户名user {username}# es的密码password {password}document_id %{id}}
}还好无意中找到了logstash所在机器要不然要找出logstash还真是麻烦。。。
因为这里把很多“无用”的字段remove掉了所以在es数据库中也无法得到采集的链路详情。
七、总结
本文试着通过另外一个解决方案来保证高并发接口的高性能顺便介绍了如何找回filebeat所在机器的过程。同时提醒我们在采集数据的时候或者程序开发的时候特别是当你的链路很长的时候把每个环节的主体信息记录下来是多么的重要。
