RaspberryPiで測定したデータをKibanaで可視化する

せっかくRaspberryPiで測定したデータ．LCDに表示するだけではもったいない．

これらをグラフとして可視化しない手はない．

IoTプラットフォームとOSS
Rsyslog + Elasticsearch + Kibana
おわりに

IoTプラットフォームとOSS

既にいくつもの商用サービスが展開しており，これらを使う手がある．

Cloud IoT Core | Google Cloud

広範囲に分散したデバイスの接続、管理、データ取り込みを、セキュリティを維持しながら簡単に行えます。

もちろんOSSを使って自宅サーバやレンタルサーバ上に自分で構築することも出来る．

可視化する前段で，データをパイプ処理する仕組みもいろいろ．

その他にも色々あるので，自身の環境にあったものを適当に選ぶ．

今回は既に構築済みの Rsyslog+Elasticsearch+Kibana を使う．

Rsyslog + Elasticsearch + Kibana

送信フォーマットを決める＠RasPi

送信データのフォーマットを決めて，loggerコマンドで送信する．

フォーマットは気温，湿度，気圧のカンマ区切りとする．

$ logger --server ${RsyslogServerIP} --tcp --port 514 "$(./i2c-dev/htu21d.sh t),$(./i2c-dev/htu21d.sh h),$(./i2c-dev/bmp180.py p)"

適当なスクリプトを書いてcronで定期的に送信するといい．

Rsyslogを設定する＠Server

サーバはCentOS7，Rsyslogは8.2002.0を使う．Rsyslogのインストールは長そうなので割愛する．

Rsyslogの役割は，RasPiからメッセージを受け取って，メッセージをElasticsearchに登録出来るフォーマットに変換して，メッセージElasticsearchに送る．

まずは送信側で決めたフォーマットのパースルールを作る．

受信したメッセージはRsyslogのmmnormalizeを使ってパースする．rulebaseファイルの記述方法はLiblognormのドキュメントを参照する．rulebaseファイルの内容で正しくパース出来るかlognormalizerコマンドで確認する．

$ cat raspi.rulebase
version=2
rule=:%temp:float%,%humi:float%,%press:float%
$ echo "30.73,59.81,99641.00" | lognormalizer -r raspi.rulebase
{ "press": "99641.00", "humi": "59.81", "temp": "30.73" }

続いてデータ入力部分の設定する．既存のrsyslog.confに不足部分を書き足す．

$ cat /etc/rsyslog.conf
global(workDirectory="/var/lib/rsyslog")
module(load="imtcp")                   # TCPで受信して，
module(load="mmnormalize")             # mmnormalizeでパースして，
module(load="omelasticsearch")         # Erasticsearchに出力する
input(type="imtcp" port="514" ruleset="raspi")  # 514/tcpで受信して，ルールセット'raspi'を設定
include(file="/etc/rsyslog.d/*.conf")  # 細かい設定は切り出す

続いてパースから出力までの部分を設定する．パースした結果をテンプレートに埋め込み，Elasticsearchへ出力する．どの変数にパースされたかよくわからない場合はjsonmesgの中身を適当なファイルに出力してみるといい．

template(name="ins_erasticsearch"
         type="list"
         option.json="on") {
            constant(value="{")
            constant(value="\"timereported\":\"")           property(name="timereported" dateFormat="rfc3339")      constant(value="\",")
            constant(value="\"hostname\":\"")               property(name="hostname")                               constant(value="\",")
            constant(value="\"syslogfacility-text\":\"")    property(name="syslogfacility-text")                    constant(value="\",")
            constant(value="\"syslogpriority-text\":\"")    property(name="syslogpriority-text")                    constant(value="\",")
            constant(value="\"syslogtag\":\"")              property(name="syslogtag")                              constant(value="\",")
            constant(value="\"temperature\":\"")            property(name="$!temp")                                 constant(value="\",")
            constant(value="\"humidity\":\"")               property(name="$!humi")                                 constant(value="\",")
            constant(value="\"pressure\":\"")               property(name="$!press")                                constant(value="\"")
            constant(value="}\n")
}

template(name="tmp"
         type="string"
         string="TMP: %jsonmesg%\n"
)

ruleset(name="raspi"){
    action(type="mmnormalize" ruleBase="/etc/rsyslog.d/raspi.rulebase")
    #action(type="omfile" dirCreateMode="0700" FileCreateMode="0644" File="/tmp/raspi.log" Template="tmp")
    #action(type="omfile" dirCreateMode="0700" FileCreateMode="0644" File="/tmp/raspi.log" Template="ins_erasticsearch")

    ## Elasticsearch index definition for dynSearchIndex
    $template raspi-index,"raspi-%timereported:1:10:date-rfc3339%"

    action(type="omelasticsearch"
           server="localhost:9200"
           template="ins_erasticsearch"
           dynSearchIndex="on"
           searchType="_doc"
           searchIndex="raspi-index"
           bulkmode="on"
           queue.type="linkedlist"
           queue.size="5000"
           queue.dequeuebatchsize="300"
           action.resumeretrycount="-1")
    & stop
}