Step Functionsの「実行経過時間」をDatadogで可視化する

はじめに

Geek+の鈴木です。

AWS Step Functions（以下、SFN）を運用していると、「実行開始からどれくらい経過しているか」をリアルタイムに監視したくなる場面があります。

ただし、SFNが標準で提供する実行時間の情報は、基本的に 実行が完了してから確定する“結果の所要時間” です。
そのため、実行中のSFNに対して「開始から何秒経過したか」を標準メトリクスだけで取得することはできません。
こうした事情から、ある実行で想定外のリトライが続き実行時間が長時間になっていても完了するまで気づきにくく、検知が遅れてユーザー影響が大きくなる場合があります。

そこでこの記事では、この課題を解決するために「実行中のSFNの“実行経過時間”をDatadogに送信して可視化」する仕組みを紹介します。

ターゲット

• DatadogにSFNの「実行経過時間」をメトリクスとして送り、可視化したい方

構成

構成はシンプルです。
メインのSFNのメトリクス取得とDatadogへの送信はLambdaが実行しています。

• 実行経過時間の計算
  • 実行中のSFNのstartDateを取得し、now - startDateで時間を算出
• LambdaからDatadogに送信している内容
  • メトリクス：custom.sfn.elapsed.seconds(gauge)
  • 値： 実行中のものがない場合は0
  • タグ：state_machine:<name>、env:<value> など

工夫したこと

• 最大経過秒だけ送る: 実行中のSFNが複数ある場合「いちばん長く動いている1件」の経過秒だけを送ります。目的が“滞留の早期検知”なので最大値だけ見れば十分で、全実行分を送る方式に比べて送信量(＝Datadog側のコスト)やHTTP 429（Too Many Requests）のリスクを増やしにくくなります。(サービスの成長に合わせて今後ブラッシュアップ)

コードのメインの箇所紹介

Datadogへの送信に必要な変数

# Datadog転送に必要な値
METRIC_NAME = "custom.sfn.elapsed.seconds"    # Datadog に送るメトリクス名
DD_SITE = "<DD_SITE>"                         # 例: datadoghq.com
DD_API_URL = f"https://api.{DD_SITE}/api/v1/series"
DD_API_KEY = "<DD_API_KEY>"

SFNの実行経過時間を算出する部分(サンプル)

# 実行中の中で、いちばん経過しているもの(最大値)を集計する
max_elapsed = 0  # RUNNING(実行中)がない場合は 0 のまま
token = None     # ページング用
while True:
    # 対象ステートマシンでRUNNING(実行中)のものを取得
    kwargs = {"stateMachineArn": sm_arn, "statusFilter": "RUNNING", "maxResults": MAX_RESULTS}
    if token:
        # 次ページがある場合
        kwargs["nextToken"] = token
    resp = sfn.list_executions(**kwargs)
    for ex in resp.get("executions", []):
        # 実行開始時刻からの経過秒を計算
        elapsed = int((now - ex["startDate"]).total_seconds())
        if elapsed > max_elapsed:
            # 最大値を更新
            max_elapsed = elapsed
    token = resp.get("nextToken")
    if not token:
        # 全ページ読み終わり
        break

Datadogへメトリクスを送る部分(サンプル)

# 送信するメトリクス(gauge)を組み立てる
series = [{
    "metric": METRIC_NAME,         # メトリクス名
    "type": "gauge",               # 値を上書きできる時系列
    "points": [[ts, max_elapsed]], # [unix秒, 値]
    "tags": tags,                  # 例: ["state_machine:xxx", "env:prod"]
}]
payload = json.dumps({"series": series}).encode("utf-8")  # Datadog Metrics API の形式
headers = {"Content-Type": "application/json", "DD-API-KEY": DD_API_KEY}  # API Key はヘッダ

req = urllib.request.Request(DD_API_URL, data=payload, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=8):
    pass

参考資料

カスタムメトリクス概要

• カスタムメトリクスの送信方法（DogStatsD、HTTP API、カスタム Agent チェックなど）が詳細に解説されています

カスタムメトリクスの料金

• 料金に関してはこちらの記事が参考になります

効果

実行経過時間が一定のしきい値を超えたらアラートを通知することで、ユーザーから問い合わせが来る前に異常に気づけるようになりました。
結果として、影響が広がる前に原因調査・対応ができています。
また、Datadogで継続的に可視化・監視することで、「サービスの信頼性」を守ることにつながっています。

まとめ

SFNは「完了後の所要時間」は追いやすい一方で、「実行中の長時間化」は標準のデータだけだと検知が遅れがちです。
実行中のSFNの最大経過秒を定期送信するだけでも、滞留の早期検知とアラートの起点を作れます。
今後もプロダクトのオブザーバビリティを高め、サービスの信頼性を守っていきたいと思っています。

ここまで記事を読んでいただきありがとうございます。
最後に、少しだけ弊社の紹介をさせてください。
弊社は「サプライチェーンマネジメントの新しいあり方を、ソフトウェアで実現する」プロダクトを開発・提供しているスタートアップです。
少しでも興味を持っていただけた方は、下記をご覧ください。