docker elkログ収集システムの導入(kafka方式)

一、logback+elk、tcp方式送信

環境構築参考前のブログ:https://www.cnblogs.com/alan6/p/11667758.html
tcp方式で問題がある:tcp方式はログ量が比較的大きく、同時量が高い場合、ログが失われる可能性がある.kafkaを使用してログメッセージを保存し、トラフィックカットピークを作成することも考えられます.

二、logback+kafka+elk

1、docker zookeeeper+kafkaをインストールする
ラミラー:docker pull wurstmeister/zookeeperdocker pull wurstmeister/kafka
実行zookeeeper:docker run-d--name zookeeeper--restart always--publish 2181:2181--volume/etc/localtime:/etc/localtime wurstmeister/zookeeeper:latest
実行kafka:docker run-d--name kafka--restart always--publish 9092:9092--link zookeeeper--env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181\--env KAFKA_ADVERTISED_HOST_NAME=kafka所在宿ホストのIP--env KAFKA_ADVERTISED_PORT=9092\--volume/etc/localtime:/etc/localtime\wurstmeister/kafka:latest
2、logbackをkafkaに送信するように設定する
サービス側のpomファイルに依存を追加

<dependency>
      <groupId>com.github.danielwegenergroupId>
      <artifactId>logback-kafka-appenderartifactId>
dependency>

logback-spring.xmlプロファイルにappenderを追加

    <appender name="kafka" class="com.github.danielwegener.logback.kafka.KafkaAppender">
        <encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder">
            <layout class="net.logstash.logback.layout.LogstashLayout" >
                <includeContext>trueincludeContext>
                <includeCallerData>trueincludeCallerData>
                <customFields>{"system":"test"}customFields>
                <fieldNames class="net.logstash.logback.fieldnames.ShortenedFieldNames"/>
            layout>
            <charset>UTF-8charset>
        encoder>
        
        <topic>kafka_elktopic>
        <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" />
        <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
        <producerConfig>bootstrap.servers=192.168.33.128:9092producerConfig>
    appender>

    
    <root level="INFO">
        <appender-ref ref="STDOUT" />
        <appender-ref ref="FILE"/>
        <appender-ref ref="kafka" />
    root>

3、logstashの設定
elkを起動し、コンテナに入ります.
#docker exec -it elk/bin/bash
/etc/logstash/conf.d/ディレクトリに入り、プロファイルlogstashを作成します.conf、編集内容、主にinputとoutput

input {
    kafka {
        bootstrap_servers => ["192.168.33.128:9092"]
        auto_offset_reset => "latest"
        consumer_threads => 5
        decorate_events => true
        group_id => "elk"
        topics => ["elk_kafka"]
        type => "bhy"
    }
}

output {
    stdout {}
    elasticsearch {
          hosts => ["192.168.33.128:9200"]
          index => "kafka-elk-%{+YYYY.MM.dd}"
    }
}

編集/etc/init.d/logstash,修正

LS_USER=root // logstash
LS_GROUP=root // logstash

修正が完了したら終了し、elkコンテナ#docker restart elkを再起動します
4、kibanaの配置
前のブログとの差は少なく、Index Patternはlogstashで設定した「kafka-elk-日付」のhttps://www.cnblogs.com/alan6/p/11667758.html