AWS Resource Explorer の回数制限に対処する

たまには小ネタで調子を整える回ということで、Resource Explorer についてです。

AWSリソースの検索に非常に便利な機能であるものの、回数制限があるのでイィ感じに対処しましょうという内容です。

目的

今回は ECS に特定のタグがついた Service のリストを取得することにします。

普通に ECS API でやると、こんな感じの流れになるので大変なわけです。

list-clusters でクラスタ一覧を取得
list-services でクラスタ毎のサービス一覧を取得
list-tags-for-resource でサービス毎のタグ一覧を取得
タグ情報から対象のサービスをリスト化

もし describe-services で複数のサービスの情報を一気に取得しつつ、タグ情報も含まれていたりフィルタできれば、まぁギリ許せるかなってくらいなのですが、イベントログは入っているくせにタグが入っていないし、Resource Explorer の方が使い勝手がイィしって感じで選択することになります。

Resource Explorer の準備

Terraform だと、index と view を作っておくだけです。

aws_resourceexplorer2_view | Resources | hashicorp/aws | Terraform

有効にしたら、プログラムを書く前に aws cli で取得したい情報を取れるかを確認しておくと、その後がスムーズになると思います。

aws resource-explorer-2 search \
  --query-string "region:ap-northeast-1 resourcetype:ecs:service tag:env=production"

1 2	aws resource-explorer-2 search \ --query-string "region:ap-northeast-1 resourcetype:ecs:service tag:env=production"

回数上限

便利な代わりといってはアレですが、検索回数が１ヶ月あたり 10000 回に制限されています。

Resource Explorer のクォータ – AWS Resource Explorer

定期処理で実行した場合、
　１分に１回なら月に 44640 回、
　５分に１回なら月に 8928 回、
　60分に１回なら月に 744 回、

となります。仮に５分毎だとしても、１種類の処理で９割を使っちゃうので、複数種類の処理で使うことは難しくなってしまいます。

この対応策として最初の思い浮かぶのは、当然クォータから上限引き上げの申請をすることですが、現在はこの項目の上限増加は対応していません、という返答が返ってきます。

もし上限に達してしまった場合は、その月はもう検索ができないので、来月を待たなくてはいけなくなります。自分の場合は、１日の 10時台にリセットされました。

こういうタイプの詰み方ってあまりないと思うので、何かしら救済措置がほしいところですね。せめて問い合わせ１回目は、処理に対策を施した上で回数をリセットしてくれるとか。頼んでみてはいないので、意外とやってくれるかもですが、無理なら待つか別アカウントで開発準備するとかになります。

対策方針

上限は上げられないし、ECS API でやるのも書くだけとはいえ実行回数がかなり増えるので、あまりやりたくない。となれば、Resource Explorer のままで戦えるようにするしかありません。

何を検索するか次第ではあるのですが、今回の ECS Service のように１回作成されたあとは長期間そのまま存在するようなリソースは、ずっと同じ検索結果になるし、削除されることもサービス終了までほぼないはずです。

そういう性質を前提とした場合、シンプルに検索結果をキャッシュすることで対策が可能です。それこそ /tmp/ に JSON ファイルなどで読み書きするだけで十分です。

もし１つの処理における１回の検索結果を６時間キャッシュした場合、１ヶ月あたり 124回の実行回数になります。もし Multi-AZ でも 2AZ 分ならその倍の 248回で収まります。

回数としては上限に対して十分すぎる余裕があり、この６時間という期間において特に最初の Service 作成時にそこまで不都合がないのであれば、採用できるのではなかろうか。ということで、それっぽい処理の話に入っていきます。

Lambda Python の例

Lambda に定期実行させる処理で、ECS Service のリストを Resource Explorer で取得します。下記コードは適当にクラスメソッドとかをコピペしてるので、大体の参考ということで。

まずはキャッシュの読み書き処理を書いておきます。
Lambda の /tmp はデフォで 512MB 無料なエフェメラルストレージなのと、デプロイしなければ２回目以降の実行でもデータが残っているので、わりと好き勝手に使うことが可能です。

    def readCache(self, name, period):
        file = "%s/%s" % (self.tmp_dir, name)
        if not os.path.exists(file): return False

        now   = int(time.time())
        mtime = int(os.stat(file).st_mtime)
        if now - mtime > period: return False

        f = open(file, "r")
        result = f.read()
        f.close()

        return result

    def writeCache(self, name, content):
        file = "%s/%s" % (self.tmp_dir, name)
        f = open(file, "w")
        f.write(content)
        f.close()
        os.chmod(file, 0o777)

    def jsonDefault(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        raise TypeError (f'Type {obj} is not serializable')

def readCache(self, name, period):

file = "%s/%s" % (self.tmp_dir, name)

if not os.path.exists(file): return False

now = int(time.time())

mtime = int(os.stat(file).st_mtime)

if now - mtime > period: return False

f = open(file, "r")

result = f.read()

f.close()

return result

def writeCache(self, name, content):

file = "%s/%s" % (self.tmp_dir, name)

f = open(file, "w")

f.write(content)

f.close()

os.chmod(file, 0o777)

def jsonDefault(self, obj):

if isinstance(obj, (datetime, date)):

return obj.isoformat()

raise TypeError (f'Type {obj} is not serializable')

それを使いつつの、こちらがメインの検索処理です。

キャッシュが有効期限内で存在してれば使って、なければ検索してキャッシュしておく、という至って普通の内容です。結果に json.dumps できない日付型を含むので、変換だけしてあげています。

    def search(self, query_string, view_arn=None, cache_sec=10800):
        result = []
        params = {
            "QueryString": query_string,
            "MaxResults" : 1000,
        }
        if view_arn:
            params["ViewArn"] = view_arn

        file_hash  = hashlib.md5(f"{query_string}{str(view_arn)}".encode('utf-8')).hexdigest()[0:8]
        cache_name = f"resource_explorer_{file_hash}.json"
        cache_json = self.readCache(cache_name, cache_sec)

        if cache_json and not self.NO_CACHE:
            result = json.loads(cache_json)
        else:
            explorer = self.getResourceExplorer()
            try:
                while True:
                    res       = explorer.search(**params)
                    resources = res["Resources"]
                    if not resources: break

                    result += resources
                    token   = res.get("NextToken")
                    if not token: break
                    params["NextToken"] = token
            except Exception as e:
                print("Error: %s" % e)

            self.writeCache(cache_name, json.dumps(result, default=self.jsonDefault))

        return result

def search(self, query_string, view_arn=None, cache_sec=10800):

result = []

params = {

"QueryString": query_string,

"MaxResults" : 1000,

}

if view_arn:

params["ViewArn"] = view_arn

file_hash = hashlib.md5(f"{query_string}{str(view_arn)}".encode('utf-8')).hexdigest()[0:8]

cache_name = f"resource_explorer_{file_hash}.json"

cache_json = self.readCache(cache_name, cache_sec)

if cache_json and not self.NO_CACHE:

result = json.loads(cache_json)

else:

explorer = self.getResourceExplorer()

try:

while True:

res = explorer.search(**params)

resources = res["Resources"]

if not resources: break

result += resources

token = res.get("NextToken")

if not token: break

params["NextToken"] = token

except Exception as e:

print("Error: %s" % e)

self.writeCache(cache_name, json.dumps(result, default=self.jsonDefault))

return result

これを使ってクラスタ名やサービス名を整理できますよ、と。

１分間隔で実行しても全然問題ないし、コードも綺麗に使いやすくなる上に処理時間も速くなる、という GG です。

        cluster_service_names = {}

        query     = f"region:{self.getRegion()} resourcetype:ecs:service tag:{self.enable_key}={self.enable_value}"
        resources = resource_explorer.search(query, None, 21600)
        if not resources:
            return cluster_service_names

        for info in resources:
            cluster_name = info["Arn"].split('/')[-2]
            service_name = info["Arn"].split('/')[-1]

            if cluster_name not in cluster_service_names:
                cluster_service_names[cluster_name] = []

            cluster_service_names[cluster_name].append(service_name)

cluster_service_names = {}

query = f"region:{self.getRegion()} resourcetype:ecs:service tag:{self.enable_key}={self.enable_value}"

resources = resource_explorer.search(query, None, 21600)

if not resources:

return cluster_service_names

for info in resources:

cluster_name = info["Arn"].split('/')[-2]

service_name = info["Arn"].split('/')[-1]

if cluster_name not in cluster_service_names:

cluster_service_names[cluster_name] = []

cluster_service_names[cluster_name].append(service_name)

感想戦

たいした内容ではないとはいえ、バカ正直な方法で ECS 情報を何度も取得するよりは、この１回で済ませた方が美しいと自分は思うのでこうしました、という感じです。

検索対象の変化頻度が激しい場合で、かつ検索頻度も高めにしなくてはならない場合は、キャッシュの期間を短くするか、キャッシュできないということになるので、Resource Explorer なしにバカ正直に書く必要があるかもしれません。

便利な機能といっても、こんなのを手動で使うなんてことはまずないし、結局は自動化の処理に使うと思うので、

処理を書いてみて使えそうなら処理を簡略化するために使い、その時はアカウント単位で上限に到達しないように気を使う必要はありますよ、という選択肢として知っておくといつか役立つかもしれません:-)