文档章节

SpringCloud实践分享-日志收集Kafka-ELK

杨子敬的瞎扯时刻
 杨子敬的瞎扯时刻
发布于 06/11 09:53
字数 1216
阅读 1060
收藏 26

微服务应用在容器化后,日志的查询就会变成困难的问题,虽说有portainer这类的容器管理工具,能够方便的查询每个容器中的日志,但容器到达一定数量后,尤其是应用有多个实例时候,查询就成了头疼的问题。所以我们采用了Kafka-Logstash-Elasticsearch-Kibana的方案来处理日志。

首先说说我的日志收集思路:

  1. 应用将日志传入kafka集群。在集群建立相应topic,并传入日志。
  2. logstash在kafka上消费(读取)日志内容,写入elasticsearch。
  3. kibana读elasticsearch,做对应的展示。

这样的好处,1)几乎不用做特别大的修改,只需做一定的配置工作即可完成日志收集;2)日志内容输入kafka几乎没有什么瓶颈,另外kafka的扩展性能很好,也很简单;3)收集的日志几乎是实时的;4)整体的扩展性很好,很容易消除瓶颈,例如elasticsearch分片、扩展都很容易。

应用侧配置

在应用中,我们只需配置log4j的相应配置,将其日志输出到kafka即可。以下为配置示例,配置中包含将日志输入出命令行和kafka部分中。注意,在输出到kafka中,需要有一个appender类kafka.producer.KafkaLog4jAppender一般是没有的,则我在本地自己写上该类(KafkaLog4jAppender.java),并加入相应的文件路径。

log4j.rootLogger=INFO,console,kafka

# 输出到kafka
log4j.appender.kafka=com.yang.config.KafkaLog4jAppender
  #kafka.producer.KafkaLog4jAppender
log4j.appender.kafka.topic=api-admin
log4j.appender.kafka.brokerList=192.0.0.2:9092,192.0.0.3:9092,192.0.0.4:9092 # 这里填写kafka的ip
log4j.appender.kafka.compressionType=none
log4j.appender.kafka.syncSend=true
log4j.appender.kafka.layout=org.apache.log4j.PatternLayout
log4j.appender.kafka.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n

# 输出到Console

log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d (%t) [%p - %l] %m%n

KafkaLog4jAppender.java

public class KafkaLog4jAppender extends AppenderSkeleton {

    private static final String BOOTSTRAP_SERVERS_CONFIG = ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
    private static final String COMPRESSION_TYPE_CONFIG = ProducerConfig.COMPRESSION_TYPE_CONFIG;
    private static final String ACKS_CONFIG = ProducerConfig.ACKS_CONFIG;
    private static final String RETRIES_CONFIG = ProducerConfig.RETRIES_CONFIG;
    private static final String KEY_SERIALIZER_CLASS_CONFIG = ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
    private static final String VALUE_SERIALIZER_CLASS_CONFIG = ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
    private static final String SECURITY_PROTOCOL = CommonClientConfigs.SECURITY_PROTOCOL_CONFIG;
    private static final String SSL_TRUSTSTORE_LOCATION = SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG;
    private static final String SSL_TRUSTSTORE_PASSWORD = SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG;
    private static final String SSL_KEYSTORE_TYPE = SslConfigs.SSL_KEYSTORE_TYPE_CONFIG;
    private static final String SSL_KEYSTORE_LOCATION = SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG;
    private static final String SSL_KEYSTORE_PASSWORD = SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG;

    private String brokerList = null;
    private String topic = null;
    private String compressionType = null;
    private String securityProtocol = null;
    private String sslTruststoreLocation = null;
    private String sslTruststorePassword = null;
    private String sslKeystoreType = null;
    private String sslKeystoreLocation = null;
    private String sslKeystorePassword = null;

    private int retries = 0;
    private int requiredNumAcks = Integer.MAX_VALUE;
    private boolean syncSend = false;
    private Producer<byte[], byte[]> producer = null;

    public Producer<byte[], byte[]> getProducer() {
        return producer;
    }

    public String getBrokerList() {
        return brokerList;
    }

    public void setBrokerList(String brokerList) {
        this.brokerList = brokerList;
    }

    public int getRequiredNumAcks() {
        return requiredNumAcks;
    }

    public void setRequiredNumAcks(int requiredNumAcks) {
        this.requiredNumAcks = requiredNumAcks;
    }

    public int getRetries() {
        return retries;
    }

    public void setRetries(int retries) {
        this.retries = retries;
    }

    public String getCompressionType() {
        return compressionType;
    }

    public void setCompressionType(String compressionType) {
        this.compressionType = compressionType;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public boolean getSyncSend() {
        return syncSend;
    }

    public void setSyncSend(boolean syncSend) {
        this.syncSend = syncSend;
    }

    public String getSslTruststorePassword() {
        return sslTruststorePassword;
    }

    public String getSslTruststoreLocation() {
        return sslTruststoreLocation;
    }

    public String getSecurityProtocol() {
        return securityProtocol;
    }

    public void setSecurityProtocol(String securityProtocol) {
        this.securityProtocol = securityProtocol;
    }

    public void setSslTruststoreLocation(String sslTruststoreLocation) {
        this.sslTruststoreLocation = sslTruststoreLocation;
    }

    public void setSslTruststorePassword(String sslTruststorePassword) {
        this.sslTruststorePassword = sslTruststorePassword;
    }

    public void setSslKeystorePassword(String sslKeystorePassword) {
        this.sslKeystorePassword = sslKeystorePassword;
    }

    public void setSslKeystoreType(String sslKeystoreType) {
        this.sslKeystoreType = sslKeystoreType;
    }

    public void setSslKeystoreLocation(String sslKeystoreLocation) {
        this.sslKeystoreLocation = sslKeystoreLocation;
    }

    public String getSslKeystoreLocation() {
        return sslKeystoreLocation;
    }

    public String getSslKeystoreType() {
        return sslKeystoreType;
    }

    public String getSslKeystorePassword() {
        return sslKeystorePassword;
    }

    @Override
    public void activateOptions() {
        // check for config parameter validity
        Properties props = new Properties();
        if (brokerList != null)
            props.put(BOOTSTRAP_SERVERS_CONFIG, brokerList);
        if (props.isEmpty())
            throw new ConfigException("The bootstrap servers property should be specified");
        if (topic == null)
            throw new ConfigException("Topic must be specified by the Kafka log4j appender");
        if (compressionType != null)
            props.put(COMPRESSION_TYPE_CONFIG, compressionType);
        if (requiredNumAcks != Integer.MAX_VALUE)
            props.put(ACKS_CONFIG, Integer.toString(requiredNumAcks));
        if (retries > 0)
            props.put(RETRIES_CONFIG, retries);
        if (securityProtocol != null && sslTruststoreLocation != null &&
                sslTruststorePassword != null) {
            props.put(SECURITY_PROTOCOL, securityProtocol);
            props.put(SSL_TRUSTSTORE_LOCATION, sslTruststoreLocation);
            props.put(SSL_TRUSTSTORE_PASSWORD, sslTruststorePassword);

            if (sslKeystoreType != null && sslKeystoreLocation != null &&
                    sslKeystorePassword != null) {
                props.put(SSL_KEYSTORE_TYPE, sslKeystoreType);
                props.put(SSL_KEYSTORE_LOCATION, sslKeystoreLocation);
                props.put(SSL_KEYSTORE_PASSWORD, sslKeystorePassword);
            }
        }

        props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        this.producer = getKafkaProducer(props);
        LogLog.debug("Kafka producer connected to " + brokerList);
        LogLog.debug("Logging for topic: " + topic);
    }

    protected Producer<byte[], byte[]> getKafkaProducer(Properties props) {
        return new KafkaProducer<byte[], byte[]>(props);
    }

    @Override
    protected void append(LoggingEvent event) {
        String message = subAppend(event);
        LogLog.debug("[" + new Date(event.getTimeStamp()) + "]" + message);
        Future<RecordMetadata> response = producer.send(new ProducerRecord<byte[], byte[]>(topic, message.getBytes()));
        if (syncSend) {
            try {
                response.get();
            } catch (InterruptedException ex) {
                throw new RuntimeException(ex);
            } catch (ExecutionException ex) {
                throw new RuntimeException(ex);
            }
        }
    }

    private String subAppend(LoggingEvent event) {
        return (this.layout == null) ? event.getRenderedMessage() : this.layout.format(event);
    }

    public void close() {
        if (!this.closed) {
            this.closed = true;
            producer.close();
        }
    }

    public boolean requiresLayout() {
        return true;
    }

}

logstash配置

logstash可能会有快速启动实例的需求,我们就采用docker部署,能够快速启动、扩展等功能。

镜像就直接logstash官方镜像logstash docker镜像,我们选择了一种最简单启动方式做演示,具体还有多种docker部署方法,可以参考以上链接。

docker启动命令,input输入中,指定kafka集群地址和对应topic;输出,则是elasticsearch集群地址、索引,以及elastisearch配置参数。

#启动命令
docker run -it -d logstash -e 'input { kafka { 
	bootstrap_servers => "kafkaIp1:9092,kafkaIp2:9092" 
	topics => ["api-admin"] } } 
output { elasticsearch { 
	hosts => ["elasticsearch1:9200","elasticsearch2:9200","elasticsearch3:9200"] 
	index => "api-admin-%{+YYYY.MM.dd}" 
	flush_size => 20000 
	idle_flush_time => 10 
	template_overwrite => true 
} }'

在marathon中启动在Command选项中加入参数logstash -e 'input {} output {}'即可。另外说一句, 如果在容器编排系统(mesos/marathon、kubernetes)中,可能会有健康检查要求,其端口为9600。

© 著作权归作者所有

共有 人打赏支持
杨子敬的瞎扯时刻
粉丝 10
博文 11
码字总数 8785
作品 0
西城
程序员
私信 提问
elastic-job项目集成springcloud之后启动报错

本人基于springboot + elastic-job实现了一个分布式作业任务,后来因业务需求,项目需集成springcloud,目的是把服务注册到Eureka以方便调用其他服务。不料,在pom.xml文件中导入springcloud...

HiZack
2017/10/20
1K
2
使用ELK构建微服务的日志平台

1 概述 在微服务架构中,会部署众多的应用,其中有基础应用,比如:网关,服务发现等。同时还有大量的业务应用。所以,如何有效的收集它们的日志,并且方便查询,同时提供友好的可视化展示,...

Java大蜗牛
08/18
0
0
springcloud+sleuth+zipkin+kafka+es

前面已经完成了Springcloud+sleuth+zipkin的入门,以及kafka的安装。至于ES这里就不在说明了,网上安装使用资料挺多的,这里仅仅是将其作为持久化工具使用。 环境说明 jdk1.8 server 64位 in...

xiaomin0322
04/23
0
0
ELK技术栈ElasticSearch,Logstash,Kibana

基础部分 工作原理 es增删改内部原理 ELK集群安装部署 文档管理 索引管理 搜索 聚合分析 分词 数据建模 Java api 零停机 index segment merge 乐观锁并发控制 索引别名 相关度评分算法与定制...

马士兵优效学院
08/19
0
0
白话SpringCloud | 第零章:前言

说在前面 大清早醒来,觉得睡不着了。还是起来,写写博客。但最后发现关于的安全相关的还是比较多内容的,也比较专业,怕是一个多小时完不成的,也罢,那就来写写关于前言吧。 说明 《白话S...

oKong
09/04
0
0

没有更多内容

加载失败,请刷新页面

加载更多

GROUP BY GROUPING SETS

GROUPING SETS 子句是 SELECT 语句的 GROUP BY 子句的扩展。通过 GROUPING SETS 子句,您可采用多种方式对结果分组,而不必使用多个 SELECT 语句来实现这一目的。这就意味着,能够减少响应时...

hblt-j
9分钟前
0
0
selenium之表格的定位

真的勇士, 敢于直面惨淡的warning、 敢于正视淋漓的error 目录 被测试网页的HTML代码 1.遍历表格所有单元格 2.定位表格中的某个元素 3.定位表格中的子元素 总结 浏览器网页常常会包含各类表...

程序猿拿Q
14分钟前
0
0
adb命令启动展讯平台工厂模式

adb命令启动展讯工厂模式: adb shell am start com.sprd.engineermode/com.sprd.engineermode.EngineerModeActivity...

东街小霸王
15分钟前
0
0
mtu值怎样设置才网速最快

一、什么是 MTU 值 1 从字面上来说,MTU 是英文 Maximum Transmission Unit 的缩写,即最大传输单元,它的单位是字节。 在我们常用的以太网中,MTU 一般是 1500,而无线路由器默认一般是 14...

Clarence_D
33分钟前
1
0
McAfee阻止邮件发送功能

自己写的邮件发送功能,在本地机器上可以发送,但是放到服务器上面之后就不能发送了,捕获到的异常是“以一个访问权限不允许的方式做了一个访问套接字的尝试”,找了很久,终于发现是被McAfe...

壹峰
38分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部