(ARC310) Python WEBアプリからEMF形式でメトリクスを送信してみた

re:Invent 2023のセッションARC310 Detecting and mitigating gray failuresで紹介されていたEMFによるメトリクスの作成、複合アラームによる部分的な障害検出をサンプルアプリで試してみました。

AWS re:Invent 2023

#AWS

佐々木一洋

2023.12.26

はじめに

ARC310 Detecting and mitigating gray failuresではEMF(Embedding Metrics Format)で送信したメトリクスをヘルスチェックや障害検知に応用するテクニックが紹介されていました。今回の一連の記事では紹介されていた手法に従ってサンプルアプリのメトリクスを取得して障害を検知してみます。この記事では手始めにFlask アプリケーションからEMFでメトリクスを送信します。

サンプルアプリ

アーキテクチャ

アプリケーションはALBおよびフロントエンドとバックエンドのAPIから構成されています。フロントエンドAPIは同じA/ZにプロビジョニングされたバックエンドAPIに依存しています。

APIのメトリクス

フロントエンドAPIがバックエンドAPIを呼び出す度にAPI呼び出しの成否をメトリクスBackendHealthとして送信します。バックエンドAPIは処理の所要時間をBackendLatencyとして送信します。どちらのメトリクスでも障害範囲を特定するためにA/Z名とコンテナIDをディメンションに設定しています。

API	メトリクス	ディメンジョン	単位	値の例
フロントエンド	app/BackendHealth	{DockerId, AvailabilityZone}, {AvailabilityZone}	None	0(成功), 1(失敗)
バックエンド	app/BackendLatency	{DockerId, AvailabilityZone}, {AvailabilityZone}	MilliSecond	500

レイテンシの注入

障害を再現するためにA/Zごとのレイテンシの最大、最小を指定できるようにします。設定はパラメータストアに保存し、バックエンドAPI実行時に参照します。

{
  "ap-northeast-1a": [
    50,
    100
  ],
  "ap-northeast-1c": [
    500,
    800
  ],
  "ap-northeast-1d": [
    500,
    800
  ],
  "local": [
    50,
    100
  ]
}

ソースコード

サンプルアプリケーションのコードはgistで公開しています。

EMFでのメトリクス送信

EMFでのメトリクスは以下のような経路で送信します。

APIからloggerへログとしてEMFのJSONを送信
loggerのハンドラがUDPでサイドカーのCloudWatch Agentに送信
CloudWatch AgentがCloud Watch Logsに送信

CloudWatch AgentはTCP/UDPにログを送信するとログを解析して上でCloudWatch Logsに送信します。今回はECSのサイドカーとしてエージェントを起動する手順を参考にして設定しました。

Python標準モジュールのlogging.handlers.DatagramHandlerはログをPickle形式に変換して送信するため、UDPソケットをラップしたStreamを作成し、StreamHandlerを使ってテキストのままログを送信します。

# UDPソケットにテキストでログを送信するストリーム
class SockStream:
    def __init__(self, host, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.dest = (host, port)

	def write(self, data):
        try:
            self.sock.sendto(data.encode(), self.dest)
        except Exception as e:
            app.logger.warn(e)
    def flush(self):
        pass

# ロガーの設定
cwa = logging.StreamHandler(SockStream("0.0.0.0", 25888))
cwa.setFormatter(Formatter())
cwa.setLevel(logging.DEBUG)
logger.addHandler(cwa)

上記の設定で下記のようにメトリクスを送信します。

# メトリクスをEMF形式に整形してログに出力する
def put_metrics(metrics):
    meta = metadata()
    dimensions = [["AvailabilityZone"], ["AvailabilityZone", "DockerId"]]

    ts = int(time.time() * 1000)
    data = {
        "time": ts,
        "_aws": {
            "LogGroupName": "gray-failure-emf",
            "Timestamp": ts,
            "CloudWatchMetrics": [
                {
                    "Namespace": "app",
                    "Dimensions": dimensions,
                    "Metrics": list(
                        map(lambda m: {"Name": m["Name"], "Unit": m["Unit"]}, metrics)
                    ),
                }
            ],
        },
    }

    for k, v in meta.items():
        data[k] = v

    for m in metrics:
        data[m["Name"]] = m["Value"]
    logger.info(json.dumps(data))