【ELK框架系列】Logstash结合Kafka，根据不同Topic创建ES区分索引

旧城等待， 2023-06-19 05:25 1阅读 0赞

# 热门系列： #

*  ### [【ELK框架系列】SpringBoot+Kafka+ELK集成日志采集可视化框架][ELK_SpringBoot_Kafka_ELK] ###
 *  ###   [程序人生，精彩抢先看][Link 1] ###

--------------------

# **目录** #

1.前言

2.实践过程

3.总结

--------------------

# 1.前言 #

前段时间，自己通过SpringCloud框架，结合kafka集成了ELK框架（博客链接：[https://blog.csdn.net/yy339452689/article/details/103296017][ELK_SpringBoot_Kafka_ELK]）。之前的框架整体还有可优化空间以及需要补充的地方，今天特补充一下ELK中根据不同索引，查看不同服务的日志记录。而此情况下，就涉及到logstash与ES的索引结合，下面直接贴干货。

--------------------

# 2.实践过程 #

首先，还是使用kafka再新创建一个新主题：

kafka-topics.bat --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic otherlog

其次在两个微服务项目的logback-spring.xml文件中的kafka appender分别配置上对应的topic。之前框架中已配置了一个名为testlog的，今天这个项目配置上面创建的名为otherlog的。

<appender name="kafkaAppender"  class="com.github.danielwegener.logback.kafka.KafkaAppender">
            
            
            
            <encoder charset="UTF-8" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
                
                <jsonFactoryDecorator class="com.giveu.newwebeurekaclient11003.encoder.MyJsonFactoryDecorator"/>
                <providers>
                    <timestamp>
                        <timeZone>GMT+8</timeZone>
                    </timestamp>
                    
                    <pattern>
                        <pattern>
                            {
                            "ip":"%ip",
                            "severity": "%level",
                            "service": "${springAppName:-}",
                            "trace": "%X{X-B3-TraceId:-}",
                            "span": "%X{X-B3-SpanId:-}",
                            "parent": "%X{X-B3-ParentSpanId:-}",
                            "exportable": "%X{X-Span-Export:-}",
                            "pid": "${PID:-}",
                            "thread": "%thread",
                            "class": "%logger{40}",
                            "rest": "%message",
                            "stack_trace": "%exception{30}"
                            }
                        </pattern>
                    </pattern>
                </providers>
            </encoder>
            
            <topic>otherlog</topic>
            
            <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />
    
            
            <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
    
            
            
            
            <producerConfig>bootstrap.servers=localhost:9092</producerConfig>
            
            <producerConfig>acks=0</producerConfig>
            
            <producerConfig>linger.ms=1000</producerConfig>
            
            <producerConfig>max.block.ms=0</producerConfig>
            
            <producerConfig>client.id=0</producerConfig>
            
            <appender-ref ref="CONSOLE" />
        </appender>

接下来，修改logstash的配置文件，我这边还是在上个框架的core.conf中直接修改，内容如下：

input {
       kafka {
        bootstrap_servers => "localhost:9092"
        topics => ["testlog"]
        codec => "json"
        auto_offset_reset => "earliest" #从最早的偏移量开始消费
        decorate_events => true    #此属性会将当前topic、offset、group、partition等信息也带到message中
        type => "testlog" #所有插件通用属性,尤其在input里面配置多个数据源时很有用
      }
    
      kafka {
        bootstrap_servers => "localhost:9092"
        topics => ["otherlog"]
        codec => "json"
        auto_offset_reset => "earliest" #从最早的偏移量开始消费
        decorate_events => true    #此属性会将当前topic、offset、group、partition等信息也带到message中
        type => "otherlog" #所有插件通用属性,尤其在input里面配置多个数据源时很有用
      }
    }
    
    output {
        
      if [type] == "testlog" {
            elasticsearch {
                hosts => ["localhost:9200"]
                index => "testlog-%{+YYYY.MM.dd}"
            }
        }
    
    if [type] == "otherlog" {
            elasticsearch {
                hosts => ["localhost:9200"]
                index => "otherlog-%{+YYYY.MM.dd}"
            }
        }
    
        stdout {
            codec => rubydebug {metadata => true}  #logstash控制台输出日志和@metadata信息
        }
    }

最后在kibana中根据上面指定的索引名称testlog-\*以及otherlog-\*分别创建新的索引，再分别启动相应的应用与服务，即可达到根据不同主题创建不同索引来区分日志查看啦。效果如下：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 1][]

--------------------

### 3.总结 ###

其实区分创建索引的方式还有很多种，期间也有尝试过其他方式，但是还没有尝试到什么成果，目前这种是已成功且较为简单的方式，希望能够帮助到各位。

尝试期间，有以下几个阻力点：

①logstash中有一个@metadata全局变量，此变量中还有很多元数据，也可以自己通过filter向其中摄入自己定义的数据来使用。但我使用的logstash5.2.2版本貌似需要logstash-kafka插件才可以实现kafka的broker信息。例如`[@metadata][kafka][topic]`：消息被消费的原始kafka主题，这个变量我这边就尝试了多次，无法获取到。导致创建索引失败，卡壳了很久。官方有文档，下面这个帖子也写的很清楚：[https://segmentfault.com/a/1190000016595992][https_segmentfault.com_a_1190000016595992]。大家可以参考一下。

②logstash有好几个任何版本都通用的基础变量，如：

<table> 
 <thead> 
  <tr> 
   <th>设置</th> 
   <th>输入类型</th> 
   <th>要求</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>add_field</code></td> 
   <td>hash</td> 
   <td>No</td> 
  </tr> 
  <tr> 
   <td><code>codec</code></td> 
   <td>codec</td> 
   <td>No</td> 
  </tr> 
  <tr> 
   <td><code>enable_metric</code></td> 
   <td>boolean</td> 
   <td>No</td> 
  </tr> 
  <tr> 
   <td><code>id</code></td> 
   <td>string</td> 
   <td>No</td> 
  </tr> 
  <tr> 
   <td><code>tags</code></td> 
   <td>array</td> 
   <td>No</td> 
  </tr> 
  <tr> 
   <td><code>type</code></td> 
   <td>string</td> 
   <td> <p>No</p> </td> 
  </tr> 
 </tbody> 
</table>

我这边就是引用到了Type才做到了topic区分索引。

> ### 本博客皆为学习、分享、探讨为本，欢迎各位朋友评论、点赞、收藏、关注，一起加油！ ###

![20201028150451530.gif][]

[ELK_SpringBoot_Kafka_ELK]: https://blog.csdn.net/yy339452689/article/details/103296017
[Link 1]: https://blog.csdn.net/yy339452689/article/details/109314083
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191204170933501.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191204170915581.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5,size_16,color_FFFFFF,t_70
[https_segmentfault.com_a_1190000016595992]: https://segmentfault.com/a/1190000016595992
[20201028150451530.gif]: https://img-blog.csdnimg.cn/20201028150451530.gif