docker-compose で fluent-plugin-kafka の match + kafka_buffered を試す

[docker\-compose で Kafka の動作確認](http://qiita.com/tily/items/ea212a7fe68d62e34644)ができて、[docker\-fluentd\-image のカスタマイズを試す](http://qiita.com/tily/items/be4883c25dd29a1a8946)のもやったので、組み合わせて docker-compose で fluent-plugin-kafka を軽く試してみた。

## 準備

ディレクトリを作成して移動。

```
$ mkdir custom-fluentd
$ cd custom-fluentd
```

docker-compose.yml を作成する。

```
$ vi docker-compose.yml
```

```yaml
version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 172.17.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    depends_on:
      - zookeeper
  fluentd:
    build: .
    volumes:
      - ./log:/fluentd/log
    depends_on:
      - kafka
```

fluentd のイメージはカスタマイズして fluent-plugin-kafka をインストールするため、Dockerfile を作成する。
(fluent-plugin-kafka だけでなく zookeeper の gem もインストールする必要があった。)

```
$ vi Dockerfile
```

```
FROM fluent/fluentd:latest-onbuild
MAINTAINER YOUR_NAME <...@...>
WORKDIR /home/fluent
ENV PATH /home/fluent/.gem/ruby/2.3.0/bin:$PATH

USER root
RUN apk --no-cache add sudo build-base ruby-dev && \

    sudo -u fluent gem install zookeeper fluent-plugin-kafka && \

    rm -rf /home/fluent/.gem/ruby/2.3.0/cache/*.gem && sudo -u fluent gem sources -c && \
    apk del sudo build-base ruby-dev

EXPOSE 24284

USER fluent
CMD exec fluentd -c /fluentd/etc/$FLUENTD_CONF -p /fluentd/plugins $FLUENTD_OPT
```

fluentd の設定ファイルを作成する。とりあえず受信したログを全部 docker という名前のトピックに送信する設定。

```
$ vi fluent.conf
```

```xml
<source>
  @type  forward
  @id    input1
  @label @mainstream
  port  24224
</source>

<filter **>
  @type stdout
</filter>

<label @mainstream>
  <match **>
    @type kafka_buffered

    brokers kafka:9092
    zookeeper zookeeper:2181
    default_topic docker
  </match>
</label>
```

あと必要なディレクトリを作っておく。(plugins のほうはいらないかも。)

```
$ mkdir log
$ mkdir plugins
```

## コンテナ起動

ドッカコンポーザップディーする。

```
$ docker-compose up -d
```

今回 docker-compose.yml が手抜きなので、zookeeper が立ち上がる前に kafka が立ち上がろうとして落ちてしまったり、kafka が立ち上がる前に fluentd が立ち上がろうとして落ちてしまったりする。
ので、`docker-compose ps` して落ちていたら、手動で `docker-compose start kafka` とか `docker-compose start fluentd` した。

うまく立ち上がると `docker-compose ps` がこんな感じになる。

```
$ docker-compose ps
          Name                         Command               State                         Ports                        
-----------------------------------------------------------------------------------------------------------------------
customfluentd_fluentd_1     /bin/sh -c exec fluentd -c ...   Up      24224/tcp, 24284/tcp, 5140/tcp                     
customfluentd_kafka_1       start-kafka.sh                   Up      0.0.0.0:32797->9092/tcp                            
customfluentd_zookeeper_1   /bin/sh -c /usr/sbin/sshd  ...   Up      0.0.0.0:2181->2181/tcp, 22/tcp, 2888/tcp, 3888/tcp
```

## 動作確認

fluent.conf に指定した docker トピックを kafka に作成しておく。

```
$ docker-compose exec kafka /opt/kafka/bin/kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic docker
Created topic "test".
$ docker-compose exec kafka /opt/kafka/bin/kafka-topics.sh --list --zookeeper zookeeper:2181
docker
```

fluentd の IP を調べて、別コンテナから docker ログを送信する。

```
$ FLUENTD_IP=$(docker inspect -f "{{.NetworkSettings.Networks.customfluentd_default.IPAddress}}" customfluentd_fluentd_1)
$ echo $FLUENTD_IP
172.19.0.4
$ docker run --log-driver=fluentd --log-opt fluentd-address=$FLUENTD_IP:24224 python:alpine echo Hello
Hello
```

kafka-console-consumer.sh を立ち上げてトピックにたまったログを最初から読んでみると、ちゃんとさきほどの Hello というメッセージを受信することができた。

```
$ docker-compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic docker --from-beginning
{"source":"stdout","log":"Hello","container_id":"c207a0ab45f5f67cb8bf1fc04480681a1a8a8b3947aad9025bf725125583c605","container_name":"/tiny_visvesvaraya"}
```

上記の kafka-console-consumer.sh を起動したまま、もう一度上に書いた方法で `echo Hello` してみると、今度はしばらく時間がたってからメッセージを受信することができた。
バッファされているから受信に時間がかかるということなんだと思う。

## まとめ

docker-compose で fluent-plugin-kafka の match + kafka_buffered を手軽に試すことができるようになった。次は source (consume する側) のほうを試してみようと思う。

## 参考資料

* [docker\-compose で Kafka の動作確認 \- Qiita](http://qiita.com/tily/items/ea212a7fe68d62e34644)
* [docker\-fluentd\-image のカスタマイズを試す \- Qiita](http://qiita.com/tily/items/be4883c25dd29a1a8946)
* [fluent/fluent\-plugin\-kafka: Kafka input and output plugin for Fluentd](https://github.com/fluent/fluent-plugin-kafka)