■ はじめに
ずいぶん昔に作成したものですが…
以下の処理に伴って、イメージを削除したものの、元の EC2 インスタンスの設定によっては、スナップショットが残ってしまうことがあるのでこちらで対応します。
■ プログラムの流れ
- EC2 インスタンスのイメージの一覧情報を取得
- EBS ボリュームのスナップショットの一覧情報を取得
- 上記二つの一覧情報から、削除されたイメージにも関わらず、残っているスナップショットを抽出
- 抽出したスナップショットを削除
■ コード
コードは、以下。
MaintenanceEC2_Snapshot.py
#!/usr/bin/env python
# -*- coding: utf-8-unix; -*-
"""AWS Lambda Function - Maintenance EC2 Snapshots
"""
from __future__ import print_function
import logging
import boto3
import os
# 「Python 3 で少しだけ便利になった datetime の新機能 - Qiita」
# <https://qiita.com/methane/items/d7ac3c9af5a2c659bc51>
from datetime import datetime, timezone, timedelta
TimeZone = timezone(timedelta(hours=+9), 'JST')
# 「Lambdaの本番業務利用を考える① - ログ出力とエラーハンドリング | ナレコムAWSレシピ」
# <https://recipe.kc-cloud.jp/archives/9968>
logger = logging.getLogger()
logLevelTable={'DEBUG':logging.DEBUG,'INFO':logging.INFO,'WARNING':logging.WARNING,'ERROR':logging.ERROR,'CRITICAL':logging.CRITICAL}
if os.environ.get('logging_level') in logLevelTable :
logger.setLevel(logLevelTable[os.environ['logging_level']])
else:
logger.setLevel(logging.WARNING)
Snapshot_Description_Head = "Created by CreateImage("
logger.info("Snapshot_Description_Head:[%s]", Snapshot_Description_Head)
Snapshot_Description_Head_1 = ") for ami-"
logger.info("Snapshot_Description_Head_1:[%s]", Snapshot_Description_Head_1)
Snapshot_Description_Head_2 = " from vol-"
logger.info("Snapshot_Description_Head_2:[%s]", Snapshot_Description_Head_2)
Snapshot_Description_Head_3 = ") for "
logger.info("Snapshot_Description_Head_3:[%s]", Snapshot_Description_Head_3)
#
StartDateTime = datetime.now(TimeZone)
def AmazonEC2_DescribeDeleteSnapshots(response_images, response_snapshots):
try:
delete_snapshots = []
image_ids = {}
for n, k in enumerate(response_images["Images"]):
id = response_images["Images"][n].get("ImageId")
name = response_images["Images"][n].get("Name")
description = response_images["Images"][n].get("Description")
state = response_images["Images"][n].get("State")
creationdate = response_images["Images"][n].get("CreationDate")
image_ids[id] = {
"ImageId" : id,
"Name" : name,
"Description" : description,
"State" : state,
"CreationDate" : creationdate,
}
for n, k in enumerate(response_snapshots["Snapshots"]):
id = response_snapshots["Snapshots"][n].get("SnapshotId")
name = response_snapshots["Snapshots"][n].get("Name")
description = k.get("Description")
state = response_snapshots["Snapshots"][n].get("State")
progress = response_snapshots["Snapshots"][n].get("Progress")
starttime = response_snapshots["Snapshots"][n].get("StartTime")
ownerid = response_snapshots["Snapshots"][n].get("OwnerId")
volumeid = response_snapshots["Snapshots"][n].get("VolumeId")
volumesize = response_snapshots["Snapshots"][n].get("VolumeSize")
instanceid = ""
imageid = ""
index = description.find(Snapshot_Description_Head)
if index >= 0:
index2 = description.find(Snapshot_Description_Head_1)
if index2 >= 0:
instanceid = description[(index+len(Snapshot_Description_Head)):index2]
index3 = description.find(Snapshot_Description_Head_2)
if index3 >= 0:
imageid = description[(index2+len(Snapshot_Description_Head_3)):index3]
logger.debug(("(%04d/%04d) EBS Snapshot | [ ] "
"ID:[%s] Name:[%s] Description:[%s] State:[%s] Progress:[%s] StartTime:[%s] OwnerId:[%s] VolumeId:[%s] VolumeSize:[%s] InstanceId:[%s] ImageId:[%s]"),
n + 1, len(response_snapshots["Snapshots"]),
id, name, description, state, progress, starttime, ownerid, volumeid, volumesize, instanceid, imageid)
if not instanceid:
continue
if not imageid:
continue
if not state == "completed":
continue
if imageid in image_ids:
continue
delete_snapshots.append({
"SnapshotId" : id,
"Name" : name,
"Description" : description,
"State" : state,
"Progress" : progress,
"StartTime" : starttime,
"OwnerId" : ownerid,
"VolumeId" : volumeid,
"VolumeSize" : volumesize,
"InstanceId" : instanceid,
"ImageId" : imageid,
})
except Exception as e:
logger.exception('Error dosomething: %s', e)
else:
pass
finally:
pass
return delete_snapshots
def lambda_handler(event, context):
try:
global StartDateTime
StartDateTime = datetime.now(TimeZone)
logger.info("StartDateTime:[%s]", StartDateTime)
#
logger.info("event:[%s]", event)
logger.info("context:[%s]", context)
#
DryRun = event.get("DryRun")
if not DryRun:
DryRun = False
logger.info("DryRun:[%s]", DryRun)
if not isinstance(DryRun, bool):
return "[ERROR] Parameter DryRun Format is Bool"
#
ec2_client = boto3.client("ec2")
response_snapshots = ec2_client.describe_snapshots(OwnerIds = ['self'])
response_images = ec2_client.describe_images(Owners=['self'])
# Amazon EBS - Delete Snapshots
delete_snapshots = AmazonEC2_DescribeDeleteSnapshots(response_images, response_snapshots)
for n, k in enumerate(delete_snapshots):
logger.info(("(%04d/%04d) Delete Snapshot | [Do] "
"ID:[%s] Name:[%s] Description:[%s] State:[%s] Progress:[%s] StartTime:[%s] OwnerId:[%s] VolumeId:[%s] VolumeSize:[%s] InstanceId:[%s] ImageId:[%s]"),
n + 1, len(delete_snapshots),
k["SnapshotId"], k["Name"], k["Description"], k["State"], k["Progress"], k["StartTime"], k["OwnerId"], k["VolumeId"], k["VolumeSize"], k["InstanceId"], k["ImageId"])
if DryRun:
logger.info("Delete Snapshot - Dry Run. (DryRun is %s)", DryRun)
continue
response_delete_snapshot = ec2_client.delete_snapshot(
SnapshotId=k["SnapshotId"])
logger.info("Delete Snapshot Response | [%s] | %s", k["SnapshotId"], response_delete_snapshot)
except Exception as e:
logger.exception('Error dosomething: %s', e)
else:
pass
finally:
pass
#
return "normal end"
■ Lambda への引数
- DryRun
- ドライランの設定。これが、
falseになっていないと、削除しない。
- ドライランの設定。これが、
{
"DryRun": false
}
■ Lambda に設定する環境変数
- logging_level :
[ログレベル]
■ ハマりポイント
- 昔のことなので、覚えていません…
🤔🤔🤔
■ まとめ
参考になれば ♪♪♪
👋👋👋