我创建了soracom_exporter（一个用于在Prometheus中监视SORACOM Air的工具）的故事

2 年 ago
科, 颖
4 minutes

太长; 没细看

    • SORACOM APIを叩いて、Airのsim session statusをPrometheusで監視できるようにした

 

    • API叩いた結果をnode_exporterのtext collector経由でPrometheusから取得できるよう実装

 

    SORACOMの話というかむしろPrometheusのexporterをnode exporter text collectorで作った話だな…

首先

当使用SORACOM Air的SIM卡运营物联网设备时，自然也需要对其进行监控。这样一来，根据具体的实施方式，物联网设备内部的操作系统和应用程序可能会发送一些健康信息。然而，若网络无法连通，这些信息就无法传送。因此，我们还需要关注网络是否正常工作，是否能够互通。

非常高兴如果在出现问题时，能够通过Air sim会话层分析找到原因。例如，如果IoT内部的应用程序无法通信但会话状态在线，则可能存在IoT设备内部的操作系统或应用程序的问题；而如果会话状态离线，则可能是设备断电了，或者可能是信号问题或天线损坏了，可能是外部出现了问题。

因此，我们使用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 に大枠の流れが書いてある

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注意

根据个人体验判断

接下来

Please note that Chinese is a highly contextual language, and the exact translation may vary depending on the intended meaning or context of the phrase “この先”. The provided translation is a general option.

    • これから何ができるか?

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でググってもまだ出てこなかったから、そんな網羅的でもないのにとりま使って見たかった感はあった。

结束

✖
广告
将在 10 秒后关闭
bannerAds