Logstash input plugins(Kafka Producer/consumer)

##１．目的

　Kafkaは、ProducerとConsumerのAPIを利用することにより、Kafka Brokderに対して、Topicの生成と受信などの操作を行うことが可能です。ElatticStackのコンポーネントの１つであるLogstashには、これらAPIを実装したプラグインが存在します。
　本投稿では、それらのプラグインを***logstash kafka producer***と***logstash kafka consumer***と呼ぶこととします。本投稿では、あるユースケースを実現するための基本設定と基本パラメータについて解説します。



## ２．基本設定

### ２－１．前提構成

　基本設定では、下記構成（ユースケース）に対するLogstashのconfファイルの設定方法について説明します。
このユースケースでは、syslog clientから受信したsyslogデータを下記経路を通じて、最終的にKibanaにて可視化を行います。

![22222222222.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/482612/ff752acb-3803-7b99-cb5d-82c889cf9153.png)

　本投稿では、Logstashの設定に主眼を置いているため、太字の箇所（logstash kafka producer/logstash kafka consumer）の基本設定について解説し、その他コンポーネントについては解説しません。


### ２－２．logstash kafka producerの基本設定

logstash kafka producerの基本設定は以下の通りです。input pluginにはTCP/UDPプラグインを使用して、output pluginにはKafkaを利用しています。


```/etc/logstash/conf.d
### in Kafka producer
input {
       #TCP input moduleにてport5005をlisten
        tcp {
          host => "0.0.0.0"
          port => "50005"
          type => "syslog"
        }
       #udp input moduleにてport5005をlisten
        udp {
          host => "0.0.0.0"
          port => "50005"
          type => "syslog"
        }
}
output {
  kafka {
    #Kafka broker serverのIPアドレスを記載する。
    bootstrap_servers => "broker1:9092,broker2:9092,broker3:9092"
    topic_id => "syslog"
  }
}
```

TCP/UDP portがListenしているがどうかは下記コマンドで確認することが可能です。

```
### in Kafka producer
ss -nltu | grep 50005
udp  UNCONN 0      0                               0.0.0.0:50005        0.0.0.0:*                                                                               
tcp  LISTEN 0      128                                   *:50005              *:*  
```

TCP/UDP input pluginを使用しなくても、syslog pluginでも設定できます。その場合の設定はこちらです。

```/etc/logstash/conf.d
### in Kafka producer
input {
  syslog {
    #syslogの受信ポートを設定する。
      port => 50006
  }
}

output {
  kafka {
    #Kafka broker serverのIPアドレスを記載する。
    bootstrap_servers => "broker1:9092,broker2:9092,broker3:9092"
    topic_id => "syslog"
  }
}

```



### ２－３．logstash kafka consumerの基本設定


Input plugin(logstash kafka consumer)の基本設定は以下の通りです。

```/etc/logstash/conf.d
input {
    kafka {
        #Kafka broker serverのIPアドレスを記載する。
        bootstrap_servers => "broker1:9092,broker2:9092,broker3:9092"
        #Consumerが取得するtopic名称を記載する。
        topics => ["topic"]
    }
}

output {
　　#elasticsearchのIPアドレスを記載する。
    elasticsearch {
        hosts => ["localhost:9200"]
    }
}
```


### ２－４．動作確認

syslog clientにてloggerコマンドを入力すると、

```
#syslog client
logger default-log_777
```

kafka brokerにてtopicの一覧を確認すると、logstashで指定したtopicが生成されていることが確認できます。

```
#kafka broker
# kafka-topics --zookeeper broker1:2181,broker2:2181,broker3 --list
syslog
```

また、kibanaにてdashboard画面を確認すると、syslog clientにて送出したsyslog message(logger default-log_777)を受信していることも確認できます。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/482612/2d137a9a-b1cf-9470-9184-f1e58b3dfbb3.png)



## ４．logstash kafka producerのパラメータ

### ４－１．kafka producerとは
- ProducerはメッセージをTopicに書き込むためのクライアントライブラリです。ユーザアプリケーションはProducerを使用して、Brokerクラスタ上に構成されたTopicの各Partitionにメッセージを書き込みます。
- Producerは定期的にいずれかのBrokerからメタデータを取得することで、各Brokerのホスト名、接続先ポート、各Topic PartitionのLeader Replicaの場所など把握します。Producerはこの情報を元に、どのメッセージをどのBrokerに送信するかを決定します。


#### ４－２．logstash kafka producerとは

　logstash kafka producerは、***Kafka Client 2.1.0***の仕様に準じたプラグインとなっています。本プラグインでは、Kafkaがサポートしているなかで、以下セキュリティ通信をサポートしています。

- SSL (requires plugin version 3.0.0 or later)
- Kerberos SASL (requires plugin version 5.1.0 or later)

　logstash kafka producerのパラメータ設定について、さらに活用するために本節では主要パラメータについて解説します。セキュリティに関するパラメータについては、本記事の対象外とします。

　デフォルトの設定では、セキュリティ通信は無効となっているが、必要に応じて設定をONにすることが可能です。本記事ではセキュリティ設定について触れません。詳細は、[kafka output plugin](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html)を参照してください。

　logstash kafka producerにて、デフォルトで必須で設定しなければならないパラメータは、***topic_idのみ***です。
　デフォルトのコーデックはplainとなっておるが、ユーザー自身で変更することも可能であります。
　例えば、イベント内のすべてのコンテンツをjson形式で送信したい場合は、outputのコンフィグレーションを
下記の通り設定する必要することで実現が可能となります。

```
    output {
      kafka {
        codec => json
        topic_id => "mytopic"
      }
    }
```

#### ４－３．主要パラメータ

　logstash kafka producerの主要パラメータについて下記で解説します。
　また、構成図とのマッピングを下記に載せます（図の関係上、すべてのパラメータを載せることはできなかったので主観で重要そうなパラメータをpick upした）。

![4444444444444.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/482612/a40f13c5-f6cf-beb4-b340-5f08a7234747.png)


<dl>
  <dt>bootstrap_servers</dt>
  <dd>topicを送信するkafka brokderの接続情報（IPアドレス等）を記載する。producerは、メタデータ（topic,partition,replica）の情報を得るために使用する。producerは受信したメタデータをもとに、kafka brokderに対して実データを送信する。フォーマットの記載の仕方は、host1:port1,host2:port2である。</dd>

  <dt>client_id</dt>
  <dd>client idは、kafka brokerへとrequestを作成する際に通知される。request内に含めることでrequest sourceをトラッキングできることがメリットである（IP/Port以外のトラッキング情報として）。デフォルトでは設定する必要はない。</dd>

  <dt>security_protocol</dt>
  <dd>各kafka producer～kafka broker間で使用するセキュリティプロトコルである。PLAINTEXT, SSL, SASL_PLAINTEXTの選択肢から選択することができる。デフォルト値は、PLAINTEXTである。</dd>

  <dt>topic_id</dt>
  <dd>producerが送信するtopic名称</dd>

  <dt>value_serializer</dt>
  <dd>Messageに使用するSerializerクラス。メッセージをエンコードするクラスを指定する。</dd>
</dd>

  <dt>buffer_memory</dt>
  <dd>producer自身が使用可能な自身の最大メモリサイズ。recordをバッファする為に使用することができ、kafka brokerへの送信前に蓄積するための領域となる。デフォルト値は33,354,432 byte(= 33 Mbyte)となる。

  <dt>acks</dt>
  <dd>ProducerがRequest完了前にleaderに求めるackの数。ackが「0」の場合は、すべてのbroker serverからのackを受け取らずにrequestが完了したと判断する。
ackが「１」の場合は、leaderが自身のlogに書き込みが完了したが、すべてのfollowerからのackを返信されていない状態を指す。ackが「all」の場合は、すべてのin-sysnc replicaからレコードを完了した旨のackを受信した状態でrequestが完了したと判断する。
</dd>

  <dt>batch_size</dt>
  <dd>producerが送出するrecord batchの最大サイズ。このパラメータは、Kafka producerとKafka Serverの両方の性能値に影響を大きく与える。デフォルトのbatch sizeは、16384 byteとなる。この値がkafka brokerが許容する受信サイズを超えることができない。</dd>

  <dt>compression_type</dt>
  <dd>producerがrecord batchを送信する際に利用する圧縮方式。選定する圧縮方式に応じて、record batchのoverheadの値が変化する。圧縮方式としては、none、gzip、snappy、lz4を選択することができる。</dd>

  <dt>key_serializer</dt>
  <dd>デフォルト値は"org.apache.kafka.common.serialization.StringSerializer"となる。producerがmessageのkey値をシリアライズするためのSerializer Classを指定する。</dd>

  <dt>linger_ms</dt>
  <dd>送信待機時間。送信シングルスレッドのループ待機時間。prodecer側で指定した最大のRequestサイズ(max_request_size)に達しなくても、この時間が経過したら送信される。</dd>

  <dt>max_request_size</dt>
  <dd>
request最大サイズ。producerが送信するrecord batchの最大サイズ。この上限値に達したとき、producerはrequestを送出する。デフォルト値は、1.048,576 byte(=約1MB)である。</dd>



  <dt>metadata_fetch_timeout_ms</dt>
  <dd>initial meta data requestのさいに、topicのメタデータをfetchするためのタイムアウト値。デフォルトは60,000 msec(=60 sec)である。</dd>

  <dt>metadata_max_age_ms</dt>
  <dd>メタデータを強制リフレッシュするための最大待ち時間。デフォルト値は300,000 msec(= 5 min)である。</dd>

  <dt>receive_buffer_bytes</dt>
  <dd>
データ読み込み時に指定するTCP receive buffeｒの最大サイズ。デフォルト値は、32,768 byte(=約32 kbyte)である。</dd>

  <dt>reconnect_backoff_ms</dt>
  <dd>Connection failが発生した際に、Kafka brokerにreconnectを実施するための待ち時間。デフォルト値は、10msec。</dd>

  <dt>request_timeout_ms</dt>
  <dd>このパラメータは、kafka producerがrequestのレスポンスを待つために使用する最大待ち時間をコントロールする。もしレスポンスがこのタイムアウト内に到着しない場合は、kafka producerは、必要に応じてrequestを再送信する。
</dd>

  <dt>retries</dt>
  <dd>デフォルトのリトライの挙動は、成功するまでリトライをし続けるというものである。リトライ回数はユーザー自身で設定することが可能であるが、送信データの損失を防ぐために、このパラメータを調整することは推奨されない。もし、このリトライ回数を設定した場合（０以上の整数値を設定する）、kafka producerはその設定値しかリトライを実施しないという挙動となる。
０より小さい値を設定した場合はコンフィグレーションエラーとなる。</dd>

  <dt>retry_backoff_ms</dt>
  <dd>topic partitionへのrequestが失敗した場合にretryし続ける最大待ち時間である。デフォルト値は100であり、最大100回リトライを実施することとなる。</dd>


  <dt>send_buffer_bytes</dt>
  <dd>TCPデータ送信する際に使用する最大バッファサイズ。</dd>


</dl>


## ５．logstash kafka producerのパラメータ

### ５－１．kafka consumerとは
- ConsumerはメッセージをTopicから読み出すためのクライアントライブラリです。ユーザアプリケーションはConsumerを使用して、Brokerクラスタ上に構成されたTopicの各Partitionからメッセージを読み出します。
- ConsumerもProducerと同様に、定期的にいずれかのBrokerからメタデータを取得することで、各Brokerのホスト名、接続先ポート、各Topic PartitionのLeader Replicaの場所など把握します。Consumerはこの情報を元に、どのBrokerからRecordを取得するのか決定します。
- ConsumerがRecordをどこまで読みだしたのかを示すOffsetはConsumer側で管理し、Broker側では排他制御を行いません。そのためConsumer数が増加しても、Broker側の負担は少なく済みます。Consumerはアプリケーションの再起動時に、自身のOffsetから再開できるようOffsetを永続化する必要があります。そのため、Consumerは自身のOffsetをKafka上のOffset用Topicや、任意のデータストアに保存します。
- ユーザアプリケーションは、1つ以上のConsumerでConsumer Groupを構成することで、1TopicのデータをGroup内のConsumerで分散読み出しできます。このようにデータを分散読み出しできるため、KafkaはHadoop/Spark/Stormなどの並列分散処理フレームワークと相性が良いです。また、Topicのデータは読み出されてもすぐには削除されないため、同じTopicのデータを複数のConsumer Groupから読み出すことができます。


### ５－２．logstash kafka consumerとは

　logstash kafka consumerは、group management（group idを用いた論理的なグループ管理）を取り扱うことができる。異なる物理マシン上のLogstashインスタンスにて、同じgroup_idを付与することでデータ受信速度を向上させることができます。デフォルト設定では、Logstashインスタンスは１グループにしか所属していません。各コンフィグレーション上のgroup_idを同一の値に設定する必要があります。
Topicに含まれるmessageは、同一のgroup_idをもつ全てのLogstashインスタンスに送信されます。

 以下に含まれるKafka brokerからのメタデータは、＠metadata fieldに含まれる。メタデータは、decorate_eventオプションが「true」に設定された場合のみ送信されます。デフォルトでは、decorate_eventオプションは、falseとなっています。

- [@metadata][kafka][topic]: Kafkatトピック名称
- [@metadata][kafka][consumer_group]: Consumerグループ
- [@metadata][kafka][partition]: messageが含まれていたpartition情報
- [@metadata][kafka][offset]: messageのrecord offset
- [@metadata][kafka][key]: record keyの情報
- [@metadata][kafka][timestamp]: Recordのタイムスタンプ情報。ただし、この情報は、Kafka brokerの設定に依存するので注意が必要です。デフォルトでは、recordが生成されたタイミングのタイムスタンプを意味します。kafka brokerのより詳細な仕様を理解する為には、log.message.timestamp.typeのパラメータ仕様を確認する必要があります。


### ５－３．主要パラメータ（編集中）

　logstash kafka consumerの主要パラメータについて下記で解説します。
　また、構成図とのマッピングを下記に載せます（図の関係上、すべてのパラメータを載せることはできなかったので主観で重要そうなパラメータをpick upしました）。


- auto_commit_interval_ms
ConsumerがもつoffsetがKafkaにコミットする周期間隔。デフォルト値は、5,000 msec。

- auto_offset_reset
ConsumerがFetchリクエストを出した場合に、initial offsetがなかった場合、あるいはoffsetが範囲外だった場合に、Consumerがもつoffset情報をどのように設定し直すかを、このパラメータで決定することができる。

   - earliest: earliest offsetに自動的にリセットする。
   - latest: latest offsetに自動的にリセットする。
   - none: consumerグループ内に以前のOffsetが無い場合、exception(例外通知)をconsumerに送付する。
   - anything else: グループ内に問い合わせを問い合わせず、exception(例外通知)consumerに送付する。


- bootstrap_servers
Kafka brokerクラスタに対しての初回接続の際に使用されるKafkaインスタンスのリスト。複数のKafka borkerを指定する場合は、「host1:port1,host2:port2」の形式で記載します。すべてのクラスタメンバーシップを検知するためのInitial Connectionのみで使用されるため、この値は必ずしもすべてのKafka brokerの値を入力する必要はないです。Initial Connectionが完了したのち、すべてのKafka brokderと通信することが可能です。しかし、念のため（単一のKafka brokderがDownしているケースを考慮して）複数のKafka brokderを記載したほうが耐障害性を考慮すると無難な選択肢となります。

- check_crcs
record内のCRC32を自動的にチェックする機能である。このチェック機能はオーバーヘッドが存在するため、パフォーマンスを最優先する場合は機能をOFFすることを推奨する。


- client_id
このパラメータを設定した場合、requestを作成するタイミングでKafka serverに対して、client_idをKafka brokerに通知されます。client_idを設定することで、ip/port情報以外の情報でトラッキングすることができる。

- connections_max_idle_ms
このパラメータで指定した時間後、idleコネクションを閉じます。

- consumer_threads
デフォルト値は１である。完全なバランスを保つためには、この値はKafka partitionの数と理想的な設定である。


- decorate_events
デフォルトは、falseである。この設定をtrueにすることで、以下のKafkaのメタデータをlogstashのイベント内に含めることができる。

- enable_auto_commit
デフォルト値はtrueである。このパラメータをtrueに設定すると、kafka brokerに対して定期的にコミットを行う。このコミットのは、プロセス失敗が発生した場合に発生する。

- exclude_internal_topics
internal topic（offsetなど）を取り除くかどうかを設定する。

- fetch_max_bytes
fetchリクエストで取得可能なデータ最大量。デフォルト値は、52,428,800（50 MB）

- fetch_max_wait_ms
fetch_min_byteによって設定した条件を満足しないデータが存在しない場合、fetchリクエストを応答する前にkafka brokerがブロックする。そのブロックを行うまでの最大時間。デフォルトでは、500msec(=5 sec)である。

- fetch_min_bytes
consumerからのfetch requestに対してkafka brokerが応答するための最小データ容量。

- group_id
デフォルト値は、「logstash」である。consumerが所属するグループ識別子である。トピック内のMessageは同一group_idのすべてのlogstashインスタンスに通知される。

- heartbeat_interval_ms
同一グループ内のconsumerに適用されるハートビート。ハートビートはconsumerのセッションの正常動作性を確認することに使用される。また、新しいconsumerがjoinしたりleaveした場合のリバランシングに使用される。この値は、session_timeout_msより必ず小さく設定する必要がある。ただし、session_time_outの値の3分の１より大きくするのを推奨するので注意が必要となる。デフォルト値は、3,000
 msec (= 3 sec)である。

- key_deserializer_class
デフォルト値は、"org.apache.kafka.common.serialization.StringDeserializer"である。record keyのシリアル化をデコードするさいに使用されるJavaクラスである。

- max_partition_fetch_bytes
pertitionごとの最大フェッチバイト数。

- max_poll_interval_ms
consumerグループ管理をおこなうときにpoll起動間の最大待ち時間。

- max_poll_records
１処理で取得可能な最大record数。

- metadata_max_age_ms
kafka brokerのメタデータを強制リフレッシュする最大待ち時間。

- partition_assignment_strategy
デフォルト値は、「org.apache.kafka.clients.consumer.RangeAssignor」である。kafka consumerとpartitonの配布のために使用される。

- poll_timeout_ms
topicから新しいメッセージを受信するための最大待ち時間。この時間間隔で指定topicからデータを受信する。

- receive_buffer_bytes
データ受信時に使用するTCP受信バッファ。

- reconnect_backoff_ms
kafka brokerに李コネクトするための最大待ち時間。

- request_timeout_ms
requestのタイムアウト値。kafka clientが投げたrequestのレスポンスの待ち時間。

- retry_backoff_ms
fetchリクエストが失敗したときにtopic partitionに他しいてトライするまでの待ち時間。

- security_protocol
利用するセキュリティプロトコル。PLAINTEXT,SSL,SASL_PLAINTEXT,SASL_SSLのいづれかを使用する。

- send_buffer_bytes
データ送信時のTCPバッファサイズ。

- session_timeout_ms
consumerがdead状態と認識してリバランシングの操作を実施するためのトリガー値である。

- topics
定期購読するtopicsのリストである。デフォルト値は、["logstash"]である。

- topics_pattern
定期購読するtopicのパターンを記載します。この値を設定すると、topicsの値が無視されるので注意が必要である。

- value_deserializer_class
デフォルト値は、"org.apache.kafka.common.serialization.StringDeserializer"である。recordのvalueをデコードする際に資料されるJavaクラス。


## ４．Kafkaの基礎概念

上記パラメータを理解するための、Kafkaの基礎概念を下記に示す。（気が向いたら編集予定）

### ４－１．Record batch

[from this URL](https://medium.com/swlh/exploit-apache-kafkas-message-format-to-save-storage-and-bandwidth-7e0c533edf26)

- Record: messageの最小単位。keyとvalueのペアで構成される。
- Record batch： recordの集合。producerがtopic partitionに送信する際は、record batchの単位で送信される。それぞれのreord batchは複数のrecordで構成される。record batchはrecordに加えて、後述されるoverheadも含まれる。
- Record batch overhead：messageのバージョンによってoverheadの大きさが変化する。producer側の設定では、overheadのサイズは３つの方法により、増減させることとが可能である。linger time、compression、スキーマ定義（valueとkey）である。

 
![1 4UOYy2WLNt3cQCqMDqCYLA.jpeg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/482612/d46e47aa-8abe-6266-0bb1-60837425ba73.jpeg)