docker上でmosquittoで受信したjsonデータをdbとElasticsearchに保存する

![kibana.png](https://qiita-image-store.s3.amazonaws.com/0/133915/d45bddf9-d646-4d12-3be8-b3c2b5f6d736.png)



今回は個人的に使うのでお金かかるAWSIoTであれこれってのはなしで、owncloudとか入れて使ってるconohaのdockerマシン上で全て片付けます。
(手順は書きませんが一応外部に開けるのは特定のハイポートのみにしてます。mosquittoへはSSLで通信させる予定です)

#今回、主に参考にしたとこを下記にまとめます。
[mosquitto](https://mosquitto.org/)
[mosquitto-ssl](https://mosquitto.org/man/mosquitto-tls-7.html)
[paho-mqtt](https://pypi.python.org/pypi/paho-mqtt/1.2)
[pymongo](http://api.mongodb.com/python/current/index.html)
[elasticseach](https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html)
[BulkAPI](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html#docs-bulk)
[Bulk helpers](https://elasticsearch-py.readthedocs.io/en/master/helpers.html#elasticsearch.helpers.bulk)
[how-to-use-bulk-api-to-store-the-keywords-in-es-by-using-python](http://stackoverflow.com/questions/20288770/how-to-use-bulk-api-to-store-the-keywords-in-es-by-using-python)
[elasticsearch-py](https://github.com/elastic/elasticsearch-py)

#ちょっと説明
+ データの流れ　raspberry pi ⇒ mosquitto ⇒ db、elasticseach⇒kibana
+ 今回はmysqlに放り込みましたがmongodbでも似たような手順で出来ます。
+ elasticseachにしてもkibanaにしてもほぼ初めて使うような物なのでとりあえず動く程度の物です。
+ kibana上で使うつもりはなかったのですが、GPS情報を取得して位置情報も送る前提なのでgeo_pointも使用します。
+ kibanaコンテナは(outofmemory？かも)いつの間にか止まっているので、何かしら処置する予定です。

#コンテナ作成
検証しながらなので、docke-composeやスクリプトファイルは使わず、ほとんど手作業です。
簡単なのでコンテナをさくさく立てていきましょう。
##busybox(mysql用)
任意のディレクトリ

```コマンド.
docker pull busybox
docker run -i -t -v /var/lib/mysql --name busybox01 busybox /bin/sh
```

##mysql
任意のディレクトリ

```Dcokerfile.
FROM mysql
EXPOSE 3306
ENV container Docker
ENV MYSQL_ROOT_USER root
ENV MYSQL_ROOT_PASSWORD rootpass
ENV MYSQL_DATABASE log_db
ENV MYSQL_USER test
ENV MYSQL_PASSWORD testpass
```

```コマンド.
docker build -f ./Dockerfile -t mysql:mosql --no-cache=true .
docker run -d --name mosdb01 --volumes-from busybox01 -e MYSQL_ROOT_PASSWORD=rootpass mysql:mosql
```


##Elasticseach
任意のディレクトリ

```コマンド.
docker pull elasticsearch
docker run --name moselasticsearch01 -d elasticsearch
```
##Kibana
任意のディレクトリ

```コマンド.
docker pull kibana
docker run --name moskibana01 --link moselasticsearch01:elasticsearch -d -p 5601:5601 kibana
```

##mosquitto
任意のディレクトリ
このコンテナ上でいろいろやるのでたくさん入れてます。要らないものは削除してください。

```Dockerfile.
FROM       ubuntu
MAINTAINER Ototo
ENV        container docker
RUN        apt-get -y update
RUN        apt-get install -y mosquitto
RUN        apt-get install -y openssl
RUN        apt-get install -y python
RUN        apt-get install -y python-pip
RUN        apt-get install -y mysql-client
RUN        apt-get install -y vim
RUN        pip install --upgrade pip
RUN        apt-get install -y curl
RUN        pip install elasticsearch
RUN        pip install pymongo
RUN        pip install paho-mqtt
RUN        pip install PyMySQL
EXPOSE     1883
ENTRYPOINT service mosquitto start && /bin/bash --login
```

```コマンド.
docker build -f ./Dockerfile -t ubuntu:mosquitto --no-cache=true .
docker run --name mosquitto01 -t -i -d -v /root/mosquitto/mosquitto:/etc/mosquitto --link mosdb01 --link moselasticsearch01 -p 1883:1883  ubuntu:mosquitto /bin/bash
```
コンテナにはいります。

```
docker attach [conteinarID]
```
mysqlにログインします。

```
mysql -h mosdb01 -u root -p
Enter password: rootpass
```

とりあえずテーブルだけ作って出ます。

```
mysql>use log_db

mysql>CREATE TABLE `log_data` ( `id` int(11) NOT NULL AUTO_INCREMENT,
                                `date_time` datetime(1), 
                                `data` json NOT NULL,
                                PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

mysqll>exit
```

コンフィグを編集します。

```
vim /etc/mosquitto/mosquitto.conf
```

パスワードを有効にするため下記を追記

```
allow_anonymous false
password_file /etc/mosquitto/passwd
```

パスワードファイルの作成とユーザ、パスワードの設定

```
mosquitto_passwd -c /etc/mosquitto/passwd mqtt
Password: mqttpass
Reenter password: mqttpass
```

サービスのリスタート(最後にコンテナのリスタートでも良い)

```
service mosquitto restart
```

スクリプトファイルの作成

```
vim /etc/mosquitto/sub.py
```

スクリプト
ここでdbとelasticseachにぶち込んでます。例外処理入れてないので、好みの形で良いので書いた方が無難です。

```py:sub.py
#!/usr/bin/env python
import paho.mqtt.client as mqtt
import pymysql.cursors
from pymongo import MongoClient
from elasticsearch import Elasticsearch
from elasticsearch import helpers
from datetime import datetime


username = "mqtt"
userpass = "mqttpass"
host = "localhost"
topic = "car/log"
eshost = "http://moselasticsearch01:9200"
esindex = "log"
estype = "carlog"


def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe(topic, qos=0)

def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))
    set_mysql(msg.payload)
    set_es(msg.payload)

#insert mysql
def set_mysql(data):
        connection = pymysql.connect(host= 'mosdb01',
                                     user='root',
                                     password='rootpass',
                                     db='log_db',
                                     charset='utf8',
                                     cursorclass=pymysql.cursors.DictCursor)
        with connection.cursor() as cursor:
            sql = "INSERT INTO log_data(date_time,data) VALUES ( %s, %s)"
            r = cursor.execute(sql, ( datetime.now().strftime( '%Y-%m-%d %H:%M:%S' ), data))
            print(r) # -> 1
            connection.commit()

#insert elasticseach
def set_es(data):
        es = Elasticsearch([eshost])
        actions = []
        action = {"_index": esindex,"_type": estype,"_source": data}
        actions.append(action)
        log = helpers.bulk(es, actions)

#main
if __name__ == '__main__':
        client = mqtt.Client()
        client.on_connect = on_connect
        client.on_message = on_message
        client.username_pw_set(username, password=userpass)
        client.connect(host,1883)
        client.loop_forever()

```

サービズの起動はbashログイン時にやらせるやっつけ仕事

```
vim ~/.bashrc
```

下記を追記

```
nohup python /etc/mosquitto/sub.py &
```

elasticseachにインデックスを登録するので下記ファイルを作成


```json:index.json
{
  "mappings" : {
    "carlog" : {
      "properties" : {
        "battery_level" : {
          "type" : "float",
          "index" : "not_analyzed"
        },
        "gps_gga" : {
          "type" : "string",
          "index" : "not_analyzed"
        },
        "gps_gsa" : {
          "type" : "string",
          "index" : "not_analyzed"
        },
        "gps_gsv" : {
          "type" : "string",
          "index" : "not_analyzed"
        },
        "gps_rmc" : {
          "type" : "string",
          "index" : "not_analyzed"
        },
        "location" : {
          "type" : "geo_point"
        },
        "oil_press" : {
          "type" : "float",
          "index" : "not_analyzed"
        },
        "oil_temp" : {
          "type" : "integer",
          "index" : "not_analyzed"
        },
        "timestamp" : {
          "type" : "date",
          "format": "YYYY-MM-dd HH:mm:ss"
        },
        "water_temp" : {
          "type" : "integer",
          "index" : "not_analyzed"
        }
      }
    }
  }
}
```

インデックスの登録

```登録コマンド.
curl -XPUT moselasticsearch01:9200/log --data-binary @index.json
```

インデックスの登録が終わったらコンテナを出たらコンテナの再起動を行ってください。
ここまでで、docker上の作業は完了となります。

もし直にBulkAPIで登録するのであれば、コンテナ上で下記のファイル作成し、コマンドを実行することでで可能です。

```json:log.json
{ "index" : {"_index" : "log","_type" : "carlog"}
{"timestamp":"2016-09-16 19:50:00" ,"battery_level":12.0, "location":{"lon":"144.1","lat":"43.5"}, "water_temp": 90, "oil_temp":80, "oil_press":2.0,  "gps_gsa":"$GPGSA", "gps_rmc":"$GPRMC", "gps_gga":"$GPGGA", "gps_gsv":"$GPGSV"}
```
```コマンド.
curl -XPOST moselasticsearch01:9200/_bulk --data-binary @log.json
```


#raspberry pi
piからコンテナにデータを投げます。
細かいことは抜きにして検証は``pip install paho-mqtt``実行後に下記スクリプトの実行します。

```py:pub.py
#!/usr/bin/env python
import json
import paho.mqtt.client as mqtt
from datetime import datetime

#test_datqa
data = {"timestamp":"" ,"battery_level":0, "location":{"lon":"","lat":""}, "water_temp": 0, "oil_temp":0, "oil_press":0,  "gps_gsa":"", "gps_rmc":"", "gps_gga":"", "gps_gsv":""}

WTMP = 80
OTMP = 90
OPLS = 2.0
BLVL = 12.0
GSA = "$GPGSA"
GGA = "$GPGGA"
RMC ="$GPRMC"
GSV = "$GPGSV"
data["timestamp"]=datetime.now().strftime( '%Y-%m-%d %H:%M:%S' )
data["battery_level"] = BLVL
data["water_temp"] = WTMP
data["oil_temp"] = OTMP
data["oil_press"] = OPLS
data["gps_gsa"] = GSA
data["gps_gga"] = GGA
data["gps_rmc"] = RMC
data["gps_gsv"] = GSV
data["location"]["lon"] = "139.79"
data["location"]["lat"] = "35.67"
json_data = json.dumps(data)

username = "mqtt"
userpass = "mqttpass"
host = "mosquittoサーバのＩＰ"
topic = "car/log"

#main
if __name__ == '__main__':
    client = mqtt.Client()
    client.username_pw_set(username, password=userpass)
    client.connect(host,1883, 60)
    client.publish(topic, payload=json_data, qos=1, retain=False)
    client.disconnect()
```
あとはkibanaにアクセスしてデータが来てるか確認しましょう。
![discover.png](https://qiita-image-store.s3.amazonaws.com/0/133915/0022c49d-f7b6-175e-e20b-6e87d21e7606.png)