[Oracle Cloud] Fluentd を使って、Streaming にメッセージを Produce してみる

# はじめに

Oracle Cloud Infractructure(以下OCI)では、ストリーミングデータをリアルタイムに収集・処理が出来る Streaming というサービスが提供されています。Streaming は Apache Kafka 互換のAPIを持っているため、Kafka Client から接続して、データのProduce, Consume が出来ます。

今回は、Fluentd を使って Streaming にデータを Produce する方法を確認していきます。



# 仮想マシン

CentOS7 を適当に準備します。Fluentd が動けば何でもよいです。



# Auth Token生成

Streaming に、Kafka API で接続するためには、IAM ユーザーで Auth Token が必要です。自分の IAMユーザーの詳細画面で、Generate します。

[![1588474181611.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd5e213cd-9d8d-61ca-51d9-d09a52f3bf37.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=440eb7a39c5e4d75b72a2412070b21ec)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd5e213cd-9d8d-61ca-51d9-d09a52f3bf37.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=440eb7a39c5e4d75b72a2412070b21ec)



適当に説明をいれます。

[![1588474211704.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F826783fd-91d3-695a-fdf6-49292d958f01.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=89faeebd672917820612f9622b01c892)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F826783fd-91d3-695a-fdf6-49292d958f01.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=89faeebd672917820612f9622b01c892)



Token が表示されるので、メモっておきます。(画像のものは現在使えません)

**Tokenの中に、; などの記号が含まれていると、正しく動かない気がします。もしハマって正常に動かない時は、Tokenの入れ替えも検討してみてください。**

[![1588474230416.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fb181dd69-3e36-ce09-c95e-3c6d1dd97f76.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=ce7a9621b385f852e537d5b5dc64abff)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fb181dd69-3e36-ce09-c95e-3c6d1dd97f76.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=ce7a9621b385f852e537d5b5dc64abff)



# Stream の作成

OCI Console で、Analytics > Streaming を選択します。その後、Create Stream を選びます。

[![1588519801919.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fdedbe86d-6667-2f82-6f98-a18acbcc18f5.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=b84426187c4ac475c6b065c78c541a88)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fdedbe86d-6667-2f82-6f98-a18acbcc18f5.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=b84426187c4ac475c6b065c78c541a88)



適当にパラメータを入れて作成します

[![1588519908911.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fa12f4d72-c6bd-b10d-533e-d52d94d7a12d.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=40aabbfc6e00eb15e6fb25ea7af419f0)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fa12f4d72-c6bd-b10d-533e-d52d94d7a12d.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=40aabbfc6e00eb15e6fb25ea7af419f0)



Stream を作ったときに、Stream Pool が何もない場合は、Default の Stream Pool が自動的に作成されます。`teststream01`は、DefaultPool に自動配置されます。Stream Pool は、複数のStreamを管理する概念となっていて、Stream への Endpoint を Public にするのか Private にするのか、またデータの暗号に使う鍵は何を使用するのか、といった事をまとめて管理できます。

今回は自動作成にしたので、Endpoint は Public になっていて、暗号化用の鍵は Oracle 側で管理されているものを使う設定です。

[![1588520017422.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd02f1ff7-aaed-1f31-b943-1ef534b77d7c.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=fd23906affa3a22a3c3b4577674e6b5b)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fd02f1ff7-aaed-1f31-b943-1ef534b77d7c.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=fd23906affa3a22a3c3b4577674e6b5b)

上記の画面にある「View Kafka Connection Settings」を押します。

[![1588521024252.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F9e75dd0e-7ce9-71f9-7962-fb86d32dacdf.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=d49a23af436f278045c15d5ae48fff7b)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2F9e75dd0e-7ce9-71f9-7962-fb86d32dacdf.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=d49a23af436f278045c15d5ae48fff7b)

Copy All を押して、すべての設定値をメモっておきます。

[![1588521056591.png](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fc932646e-6fa7-f74f-8f74-59e40736f822.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=8dae892c965b4ffade83295244d47fb8)](https://qiita-user-contents.imgix.net/https%3A%2F%2Fqiita-image-store.s3.ap-northeast-1.amazonaws.com%2F0%2F104247%2Fc932646e-6fa7-f74f-8f74-59e40736f822.png?ixlib=rb-1.2.2&auto=format&gif-q=60&q=75&s=8dae892c965b4ffade83295244d47fb8)





# Fluentd Install

次のURLを確認して事前設定をします。
https://docs.fluentd.org/installation/before-install



Fluentdをインストールします。

```
$ curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent3.sh | sh
```



Systemd で管理されているので、起動します。

```
sudo systemctl start td-agent.service
sudo systemctl enable td-agent.service
```



# Fluentd Configure

Fluentd を編集して、Streaming へ Consume するように設定します。

```
cat <<'EOF' > /etc/td-agent/td-agent.conf
<source>
 @type forward
</source>

<match debug.**>
  @type copy
  <store>
    @type stdout
  </store>
  <store>
    @type kafka2
    brokers cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092
    use_event_time true
    
    # data type settings
    <format>
      @type json
    </format>
    
    # topic settings
    default_topic teststream01
    
    # producer settings
    required_acks -1
    compression_codec gzip
   
    # sasl
    username poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla
    password 8t[shwUN}I-d+{}8Nx_a
    sasl_over_ssl true
    ssl_ca_certs_from_system true
    

    <buffer>
      @type file
      path /var/log/td-agent/buffer/oci.buffer
      chunk_limit_size 8m
      queue_limit_length 256
      flush_at_shutdown true
      flush_interval 1s
      retry_wait 30s
      retry_max_times 9
    </buffer>
  </store>
</match>
EOF
```



Stream に接続するために大事なポイントを説明します。
brokers には、Stream Pool にある、Bootstrap Server の値を指定します。

```
brokers cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092
```



Produce する Topic名を指定します。OCI 側から見ると、Stream の名前を指定します。

```
default_topic teststream01
```



SASL認証に関する設定です。

- `username` : OCI の Stream Pool にある `SASL CONNECTION STRINGS` から、username を抜き出して指定します。
- `password` :  Auth Token を指定します。
- `sasl_over_ssl` : true
- `ssl_ca_certs_from_system` : trueを指定します。ssl_ca_cert で証明書を読み込ませる方法は、自分のやり方だと出来ませんでした。(自分の設定方法が悪そう)

```
# sasl
username poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla
password 8t[shwUN}I-d+{}8Nx_a
sasl_over_ssl true
ssl_ca_certs_from_system true
```



ここまで設定すれば接続できるようになるので、Fluentd を再起動します。

```
sudo systemctl restart td-agent
```



# 動作確認

次のコマンドを実行すると、Fluentd に Hello メッセージを投げられます。

```
echo '{"I say":"hello"}' | /opt/td-agent/embedded/bin/fluent-cat debug.ok
```



 Streaming の画面で、データを確認できます。Load Messages ボタンを押すと、"hello" というメッセージが確認できます。これで、Fluentd 経由で Streaming へ Produce する手順を確認出来ました。

![1588778328650.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/4e142b37-41de-d5cb-a4ee-bef42f1ecd1a.png)





# 参考URL

https://github.com/fluent/fluent-plugin-kafka