写点什么

SpringCloud 微服务日志经 kafka 缓冲写入到 ELK

用户头像
Rubble
关注
发布于: 4 小时前
SpringCloud微服务日志经kafka缓冲写入到ELK

准备工作

springcloud 微服务日志写入kafka

kafka日志写入logstash

elaticsearch kibana介绍与安装


有了上面的准备之后即可完整的将日志写入 kafka,经 kafka 缓冲后异步写入 elk 中。

启动 kafka

启动 zookeeper 新版本中将不再依赖外部的 zookeeper


$ bin/zookeeper-server-start.sh config/zookeeper.properties
复制代码


启动 kafka


$ bin/kafka-server-start.sh config/server.properties
复制代码


创建一个 Topic

$ bin/kafka-topics.sh --create --topic logger-channel --bootstrap-server localhost:9092
复制代码

启动 elasticsearch

bin/elasticsearch

启动 kibana

bin/kibana

启动 logstash

编写配置文件 kafka.conf


# Sample Logstash configuration for creating a simple# Beats -> Logstash -> Elasticsearch pipeline.
input { kafka { id => "my_plugin_id" bootstrap_servers => "localhost:9092" topics => ["logger-channel"] auto_offset_reset => "latest" }}
filter { #json json { source => "message" } date { match => ["time", "yyyy-MM-dd HH:mm:ss.SSS"] remove_field => ["time"] }}
output { #stdout {} elasticsearch { hosts => ["http://localhost:9200"] index => "logs-%{+YYYY.MM.dd}" #user => "elastic" #password => "changeme" }}
复制代码


启动服务

./bin/logstash -f ./config/kafka.conf --config.reload.automatic
复制代码

配置 SpringBoot 服务日志文件

logback-spring.xml


<?xml version="1.0" encoding="UTF-8"?><!-- scan:当此属性设置为true时,配置文件如果发生改变,将会被重新加载,默认值为true。 scanPeriod:设置监测配置文件是否有修改的时间间隔,如果没有给出时间单位, 默认单位是毫秒当scan为true时,此属性生效。默认的时间间隔为1分钟。 debug:当此属性设置为true时,将打印出logback内部日志信息,实时查看logback运行状态。 默认值为false。 --><!-- <configuration scan="false" scanPeriod="60 seconds" debug="false"> --><configuration>
<!--设置上下文名称,用于区分不同应用程序的记录。一旦设置不能修改, 可以通过%contextName来打印日志上下文名称 --> <contextName>kafka-log-test</contextName>
<property name="applicationName" value="${spring.application.name:-paw-kelk}"/> <!-- 定义日志的根目录 --> <property name="logDir" value="/Users/rubble/logs/${applicationName}" /> <!-- 定义日志文件名称 --> <property name="logName" value="${applicationName}"/> <!-- profile --> <property name="profileActive" value="${spring.profile.active:-dev}" />
<!-- ConsoleAppender 表示控制台输出 --> <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> <!-- 日志输出格式: %d表示日期时间, %thread表示线程名, %-5level:级别从左显示5个字符宽度, %logger{50} 表示logger名字最长50个字符,否则按照句点分割。 %msg:日志消息, %n是换行符 --> <encoder> <pattern>%d{HH:mm:ss.SSS} %contextName [%thread] %-5level %logger{36} - %msg%n</pattern> </encoder> </appender>
<!-- 异常错误日志记录到文件 --> <appender name="logfile" class="ch.qos.logback.core.rolling.RollingFileAppender"> <!-- <Encoding>UTF-8</Encoding> --> <File>${logDir}/${logName}.log</File> <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"> <FileNamePattern>${logDir}/history/${myspringboottest_log}.%d{yyyy-MM-dd}.rar</FileNamePattern> <maxHistory>30</maxHistory> </rollingPolicy> <encoder> <pattern>%d{HH:mm:ss.SSS} %contextName [%thread] %-5level %logger{36} - %msg%n</pattern> </encoder> </appender>

<appender name="kafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender"> <encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder"> <layout class="net.logstash.logback.layout.LogstashLayout" > <includeContext>false</includeContext> <includeCallerData>true</includeCallerData> <customFields>{"appName":"${applicationName}","env":"${profileActive}"}</customFields> <fieldNames class="net.logstash.logback.fieldnames.ShortenedFieldNames"/><!-- <fieldNames class="net.logstash.logback.fieldnames.LogstashFieldNames"/>--> </layout> <charset>UTF-8</charset> </encoder> <!--kafka topic 需要与配置文件里面的topic一致 否则kafka会沉默并鄙视你--> <topic>logger-channel</topic> <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" /> <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" /> <producerConfig>bootstrap.servers=localhost:9092</producerConfig> <!-- 如果kafka不可用则输出到 appender --> <appender-ref ref="logfile" /> </appender>

<!--异步写入kafka,尽量不占用主程序的资源--> <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender"> <neverBlock>true</neverBlock> <includeCallerData>true</includeCallerData> <discardingThreshold>0</discardingThreshold> <queueSize>2048</queueSize> <appender-ref ref="kafkaAppender" /> </appender>
<root level="INFO"> <appender-ref ref="CONSOLE" /><!-- <appender-ref ref="logfile"/>--> <!-- kafka 日志 --> <appender-ref ref="ASYNC" /> </root>
</configuration>
复制代码


启动 springBoot 服务,打印任意日志。

kibana 查看日志

从菜单找到 stack Management>>index patterns>> create index pattern 创建索引



菜单 discover >> 选择刚创建的索引 即可查看日志了



至此 SpringCloud 微服务日志写入到 kafka-elk 完成。

用户头像

Rubble

关注

还未添加个人签名 2021.06.01 加入

还未添加个人简介

评论

发布
暂无评论
SpringCloud微服务日志经kafka缓冲写入到ELK