soracom_exporterを作った話(SORACOM AirをPrometheusで監視してみた)

TL; DR

• SORACOM APIを叩いて、Airのsim session statusをPrometheusで監視できるようにした
• API叩いた結果をnode_exporterのtext collector経由でPrometheusから取得できるよう実装
• SORACOMの話というかむしろPrometheusのexporterをnode exporter text collectorで作った話だな...

はじめに

SORACOM AirのsimでIoTなsomethingを運用していると当然その監視も必要になる。
すると実装次第だが、IoTなsomething内部のOSやapplicationからhealth情報的なのを送るであろう。
しかしNWが通じてないとそれらの情報は送れず、NW生きてるの?疎通とれるの?という観点での監視も必要になる。

で、問題あったときに原因のレイヤー分けとしてAir simのsessionではどうか?がわかると嬉しい。とても嬉しい。
例えばIoT内application疎通はできないがsession statusがOnlineだと、IoTなsomethingの内部のOSやapplicationの内部に問題がありそうだし、
Offlineならデバイス電気切れてない?とか電波状況とかアンテナ壊れてない?とか外側に問題がある可能性が高まる。

ってなわけて、監視環境としてPrometheusを使ってたりすると、SORACOM APIからAirの情報を拾ってきて、Prometheusにいれて、おいおいslackに飛ばしたりgrafanaで可視化したくなるのでそうした。

実装

• soracom_exporter.py(pythonで作る、後述)
  • SORACOM APIを叩く
  • /hoge/node_exporter/text_collector以下にmetricsをtext保存
  • ここはPrometheus python clientで予め関数が準備されてるので処理は丸投げできる
  • 常駐プロセス起動で毎分↑のmetricsを更新
• node_exporter(事前にいれとく)
  • 起動オプションで↑のtext collectorの読み込みを有効にする(後述)
  • OS metricsと一緒にtext_collector以下のmetricsも合わせてPrometheusに返すようになる
• Prometheus(事前にいれとく)
  • node_exporterをscrape jobいれとく
  • node_exporterをscrapeするとOS metricsと一緒にsoracom_exporter生成のmetricsも取得できる

なので、例えば以下のようなファイル構造。

/hoge
|-- prometheus
|   |-- prometheus(バイナリ本体)
|   |-- prometheus.yml
|   |-- (いろいろ)
|-- node_exporter(バイナリ本体)
|   |-- node_exporter
|   |-- text_collector
|       |-- soracom_exporter_session_status.prom(つど更新される)
|-- soracom_exporter
|   |-- soracom_exporter.py

soracom_exporter.py

• 詳細はコメントで解説
• supervisordなどで常駐起動
• export_session_status_metrics に大枠の流れが書いてある

soracom_exporter.py

import json
import logging
import time
import requests
logging.basicConfig(level=logging.DEBUG, format="%(asctime)-15s %(message)s")
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True

# 常駐プロセスにしてタイマー実行するため
# pip install schedule
# cf. https://schedule.readthedocs.io/en/stable/
import schedule

# Prometheus Python Client
# cf. https://github.com/prometheus/client_python
from prometheus_client import CollectorRegistry, Gauge, write_to_textfile # pip install prometheus_client


class SORACOMExporter():
    def __init__(self):
        # config for api
        self.SORACOM_API_KEY_ID = "keyId-xxx"       # ご自分のへ変更
        self.SORACOM_API_KEY_SECRET = "secret-xxx"  # ご自分のへ変更
        self.SORACOM_URL_AUTH = "https://api.soracom.io/v1/auth"
        self.SORACOM_URL_SUBSCRIBERS = "https://api.soracom.io/v1/subscribers?limit=1000"

    def export_session_status_metrics(self):
        # api key id/secretからtokenを生成 (本当は正しく使い回すべき...)
        self._get_soracom_api_token()

        # apiからairのsim一覧を取得してparseする
        # cf. https://dev.soracom.io/jp/docs/api/#!/Subscriber/listSubscribers
        self.subscribers = self._get_subscribers()

        # Prometheusのmetricsっぽいデータに加工して、ファイルに書き出す
        registry = CollectorRegistry()
        self._build_soracom_session_status_metrics(registry, self.subscribers)
        self._write_metrics(registry)

    def _build_soracom_session_status_metrics(self, registry, subscribers):
        # ここでmetricsの名前やlabel/valueなど構造を定義
        soracom_session_status_gauge = Gauge(
            "soracom_session_status",  # metrics name
            "SORACOM session status",  # metrics description
            ["imsi", "name"],  # labels
            registry=registry
        )

        # APIからとってきてサマったデータを入れる
        for subscriber in subscribers:
            metrics_value = 1.0 if subscriber["session_status"] else 0.0 # Onlineなら1.0、Offlineなら0.0
            soracom_session_status_gauge.labels(
                subscriber["imsi"],
                subscriber["name"]
            ).set(metrics_value)

    def _write_metrics(self, registry):
        # ここらへんはPrometheusのpython clientのREADMEに書いてある通りに準備されてるものをそのまま使ってるだけ
        # cf. https://github.com/prometheus/client_python
        text_collector_output_path = "/hoge/node_exporter/text_collector/soracom_exporter_session_status.prom"
        write_to_textfile(text_collector_output_path, registry)
        logging.info("text metrics was written!:%s" % text_collector_output_path)

    def _get_subscribers(self):
        subscribers_json = self._get_soracom_api_json(self.SORACOM_URL_SUBSCRIBERS)

        # parse subscribers json to extract every subscribers's imsi/tag.Name/sessionStatus
        subscribers = []
        for subscriber_json in subscribers_json:
            subscribers.append({
                "imsi": subscriber_json["imsi"],
                "name": subscriber_json["tags"]["name"] if "name" in subscriber_json["tags"] else "",
                "session_status": subscriber_json["sessionStatus"]["online"] if subscriber_json[
                    "sessionStatus"] else False
            })

        return subscribers

    def _get_api_headers(self):
        api_headers = {
            "X-Soracom-API-Key": self.auth_api_key,
            "X-Soracom-Token": self.auth_token,
            "Accept": "application/json",
        }
        return api_headers

    def _get_soracom_api_token(self):
        try:
            auth_headers = {"Content-Type": "application/json"}
            auth_payload = {"authKeyId": self.SORACOM_API_KEY_ID, "authKey": self.SORACOM_API_KEY_SECRET}
            auth_response = requests.post(
                self.SORACOM_URL_AUTH,
                headers=auth_headers,
                data=json.dumps(auth_payload),
                verify=True,
                timeout=60
            )
            auth_response.raise_for_status()
        except requests.exceptions.RequestException as err:
            logging.warning(err)
        self.auth_token = auth_response.json()["token"]
        self.auth_api_key = auth_response.json()["apiKey"]

    def _get_soracom_api_json(self, soracom_api_url):
        try:
            soracom_response = requests.get(
                soracom_api_url,
                headers=self._get_api_headers(),
                verify=True,
                timeout=60
            )
            soracom_response.raise_for_status()
        except requests.exceptions.RequestException as err:
            logging.warning(err)
        return soracom_response.json()


if __name__ == "__main__":
    se = SORACOMExporter()
    schedule.every(1).minutes.do(se.export_session_status_metrics) # 毎分実行
    # 他のmetricsを取りたくなったら export_hoge_metircsを定義して然るべきintervalで実行する
    while True:
        schedule.run_pending()
        time.sleep(1)

出力されるファイルはこんな感じ

$ cat soracom_exporter_session_status.prom
# HELP soracom_session_status SORACOM session status
# TYPE soracom_session_status gauge
soracom_session_status{imsi="00000000000",name="会社検証用"} 1.0
soracom_session_status{imsi="11111111111",name="自宅検証用"} 0.0
...

node_exporterの起動オプション

• こっちもsupervisordなどで常駐起動

node_exporter -web.listen-address ":9100" -collector.textfile.directory /hoge/node_exporter/text_collector/
# バージョン古いかもなんでsyntax注意

所感

この先

• これから何ができるか?
  • slackへの通知
  • grafanaでの可視化
• 他にどんなend pointを監視できそうか?
  • traffic
  • cost
  • cell idみてぶれあるかとか
  • sim session eventも取れなくはないが、ほぼ変化のないものをpollingするのはあほなのでlimitedのイベントハンドラを申し込むが吉

実装

• なぜPrometheusのcustom exporterでなくnode_exporter text collector実装か?
  • 非同期的にmetrics情報を準備できる
    • SORACOM APIはそんなにbulkで叩けるendpointがないので、SIMが多いと1+N的なAPI Callをしたくなってしまう。するとPrometheusからscrapeされてすぐ返事するにはすごくたくさんAPI Callを一気にしないといけなくなるので、非同期的に処理したくなる。なった。ごめんなさい。
  • SORACOM APIを叩く頻度の調整がしやすい
    • session statusは1分ごととかで知りたいが、前述の通り例えばSIMごとの通信量を取得したいとすると、GET /stats/air/subscribers/{imsi}だと5分に1度情報更新だったりして、それを毎分叩くのはアホである。で、intervalを調整できる実装にしたかった。pushgatewayとかメモリにもっとけばcustom exporterでもできなくはないけど。
  • 小規模なら実装しやすい
    • 小規模でぱぱっとやるならtext collector、複雑なことするならcustom exporterな肌感
• (余談)soracom_exporterでいいのか?
  • 外部API叩く系のPrometheusのexporter的なものにAWS Cloudwatch exporterなどがある。でもexportでなくimportしてるがexporterでいいのか?謎い。node exporterとかは監視対象のnode内に置くのでexporterだろうけど...
  • さらにnode exporterのtext collector用にtextを出力するやつはどう呼ぶといいのかよくわかってない
  • soracom_exporterでググってもまだ出てこなかったから、そんな網羅的でもないのにとりま使って見たかった感はあった。

おわり