コードを1枚貼って、テストイベントを実行するだけで、
全台の確認結果が返ってくる手順です。
[ゴール]
Nameタグが web-* / api-* / batch-* のEC2すべてに確認コマンドを流し、NGの台だけを結果に表示します。確認対象のuserdata（Ubuntu + nginx）は次のとおりです。
---bash----------------
#!/bin/bash
set -e
apt-get update
apt-get install -y nginx
systemctl enable nginx
systemctl start nginx--------------------------前提
- 対象EC2でSSM Agentが動いている
- 対象EC2に
AmazonSSMManagedInstanceCore相当のIAMロールが付いている
手順
1. Lambda関数を作る
- Lambdaコンソール → 関数の作成 → 「一から作成」
- 関数名：
userdata-check（任意） - ランタイム：Python 3.12
- 作成
2. コードを貼る
lambda_function.py を全部消して、次を丸ごと貼り付けます。
-----python---------------
import re
import time
import boto3
ssm = boto3.client("ssm")
ec2 = boto3.client("ec2")
# ===== 設定（ここだけ書き換えればOK）=====
DEFAULT_PATTERNS = ["web-*", "api-*", "batch-*"] # Nameタグのパターン
BATCH_SIZE = 50 # send_commandのInstanceIdsは1回50台まで
MAX_WAIT_SEC = 240 # 結果待ちの上限（秒）
# 確認用スクリプト（上のuserdataに対応）
SCRIPT = r"""
h=$(hostname)
echo "$h,cloud_init,$(cloud-init status 2>&1 | head -1)"
echo "$h,nginx_pkg,$(dpkg -s nginx 2>&1 | grep -E '^(Status|Version)' | tr '\n' ' ')"
echo "$h,nginx_active,$(systemctl is-active nginx 2>&1)"
echo "$h,nginx_enabled,$(systemctl is-enabled nginx 2>&1)"
echo "$h,http_local,$(curl -s -o /dev/null -w '%{http_code}' http://localhost/ 2>&1)"
"""
# NG判定に使う文字列
NG_KEYWORDS = (
"status: error", "status: running",
",inactive", ",failed", ",disabled",
"is not installed", "not-found",
",000", ",500", ",502", ",503", ",504",
)
def find_instances(name_patterns):
"""Nameタグ（ワイルドカード可・複数可）でrunning中のEC2を {id: name} で返す"""
found = {}
paginator = ec2.get_paginator("describe_instances")
for page in paginator.paginate(Filters=[
{"Name": "tag:Name", "Values": name_patterns},
{"Name": "instance-state-name", "Values": ["running"]},
]):
for r in page["Reservations"]:
for i in r["Instances"]:
name = next((t["Value"] for t in i.get("Tags", [])
if t["Key"] == "Name"), "")
found[i["InstanceId"]] = name
return found
def send_batch(instance_ids):
res = ssm.send_command(
InstanceIds=instance_ids,
DocumentName="AWS-RunShellScript",
Parameters={"commands": [SCRIPT]},
MaxConcurrency="20",
MaxErrors="10",
)
return res["Command"]["CommandId"]
def collect(cmd_id, expected):
"""全台が終わるまで待って、結果を返す"""
deadline = time.time() + MAX_WAIT_SEC
invs = []
while time.time() < deadline:
time.sleep(5)
invs = ssm.list_command_invocations(
CommandId=cmd_id, Details=True
)["CommandInvocations"]
done = [i for i in invs
if i["Status"] not in ("Pending", "InProgress", "Delayed")]
if len(invs) >= expected and len(done) == len(invs):
break
return invs
def lambda_handler(event, context):
patterns = event.get("name_patterns", DEFAULT_PATTERNS)
dry_run = event.get("dry_run", False) # trueなら対象一覧だけ返す
numeric_only = event.get("numeric_only", False) # 「名前-数字」だけに絞る
instances = find_instances(patterns)
if numeric_only:
prefixes = [p.rstrip("*") for p in patterns]
instances = {
k: v for k, v in instances.items()
if any(re.fullmatch(re.escape(px) + r"\d+", v) for px in prefixes)
}
if not instances:
return {"error": f"対象が0台です: {patterns}"}
if dry_run:
return {
"count": len(instances),
"targets": dict(sorted(instances.items(), key=lambda x: x[1])),
}
ids = list(instances.keys())
batches = [ids[i:i + BATCH_SIZE] for i in range(0, len(ids), BATCH_SIZE)]
# 先に全バッチを送信してから、まとめて待つ
sent = [(send_batch(b), len(b)) for b in batches]
lines, failed = [], []
for cmd_id, expected in sent:
for inv in collect(cmd_id, expected):
iid = inv["InstanceId"]
name = instances.get(iid, "")
if inv["Status"] != "Success":
failed.append(f"{name},{iid},{inv['Status']}")
continue
for plugin in inv["CommandPlugins"]:
for line in plugin["Output"].strip().splitlines():
rest = line.split(",", 1)[1] if "," in line else line
lines.append(f"{name},{iid},{rest}")
ng_lines = sorted(l for l in lines if any(k in l for k in NG_KEYWORDS))
return {
"target_count": len(ids),
"ng_count": len(ng_lines),
"failed_count": len(failed),
"ng_lines": ng_lines,
"failed_instances": failed,
"all_lines": sorted(lines),
}-------------------------------------
貼ったら Deploy を押します。
3. タイムアウトを延ばす
設定 → 一般設定 → 編集 で、タイムアウトを 5分0秒 にします。デフォルトの3秒だと、SSMの完了を待てずに終わります。
4. 権限を付ける
設定 → アクセス権限 から実行ロールを開き、「許可を追加 → インラインポリシーを作成」でJSONを選び、次を貼ります。
json
--------------------{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"ec2:DescribeInstances",
"ssm:SendCommand",
"ssm:ListCommandInvocations",
"ssm:GetCommandInvocation"
],
"Resource": "*"
}
]
}--------------------------その場限りの確認用なので Resource: "*" にしています。使い回すなら、対象を絞ってください。
5. テストイベントで実行する
テスト タブでイベントを作り、まず対象の確認から実行します。
json
{ "dry_run": true }対象のEC2一覧が返ります。想定外の名前（web-test など）が混ざっていないか確認してください。混ざっている場合は、次の指定で「名前-数字」だけに絞れます。
json
{ "numeric_only": true }対象が想定どおりなら、本番を実行します。
json
{}結果の見方
json
------------------------------
{
"target_count": 30,
"ng_count": 2,
"failed_count": 0,
"ng_lines": [
"api-3,i-0123abcd,http_local,000",
"api-3,i-0123abcd,nginx_active,inactive"
],
"failed_instances": [],
"all_lines": ["..."]
}------------------------------ng_linesが空なら、全台OKです。failed_instancesは、SSMでコマンドを実行できなかった台です（Agent停止、ロール未付与など）。- 行の形式は
Nameタグ,インスタンスID,確認項目,結果です。 all_linesに全台の全結果が入っています。NG判定は文字列一致なので、迷ったときはここも確認します。
ハマりどころ
- タイムアウト：3秒のままだと、結果が返る前に終わります。
- ワイルドカードが広すぎる：
web-*はweb-testにも一致します。最初にdry_runで確認します。 - 出力が2,500文字で切れる：API経由で取れる出力は1台あたり2,500文字までです。確認スクリプトは短く保ちます。
- 「成功」と「問題なし」は別：SSMのステータスが成功でも、出力の中にNGが入っていることがあります。
- 作成直後の台：SSM Agentの起動前は、対象に出ないか
failed_instancesに入ります。数分待って再実行します。
補足
- 確認コマンドは読み取りだけなので、実行しても対象EC2の状態は変わりません。
- 別のuserdataを確認したいときは、
SCRIPTの中身とNG_KEYWORDSを書き換えるだけで流用できます。 - 台数が多く5分で終わらない場合は、タイムアウトを最大15分まで延ばします。
