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にて可視化を行います。

　本投稿では、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)を受信していることも確認できます。

４．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を参照してください。

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

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

４－３．主要パラメータ

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

bootstrap_servers

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

client_id

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

security_protocol

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

topic_id

producerが送信するtopic名称

value_serializer

Messageに使用するSerializerクラス。メッセージをエンコードするクラスを指定する。

buffer_memory

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

acks

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

batch_size

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

compression_type

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

key_serializer

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

linger_ms

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

max_request_size

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

metadata_fetch_timeout_ms

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

metadata_max_age_ms

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

receive_buffer_bytes

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

reconnect_backoff_ms

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

request_timeout_ms

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

retries

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

retry_backoff_ms

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

send_buffer_bytes

TCPデータ送信する際に使用する最大バッファサイズ。

５．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

• 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）である。