Go to Qiita Advent Calendar 2024 Top
LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@hayayu0(林)

複数のAWS APIを組み合わせてこういうのが欲しいんだよ集

Posted at

どういうことか

1つのAPIでは欲しい情報として足りない場合に、複数のAPIを実行して結果をマージさせたデータが欲しいことがあります。
そういった関数をいくつか作ったので、事例としてまとめました。
AWS API単体で取得できてほしいよ という強い想いで、このようなタイトルの付け方になってます。

組み合わせ実例集

個人的によく使っているものです。
いくつか紹介します。言語はPython(boto3)です。
ReadOnlyAccess の IAMロールがアタッチされている AWS Lambda から呼び出すことが可能です。
ローカルで実行する場合は、boto3 API の事前設定（IAMキー設定、リージョン指定、プロファイル指定、等）などは実施済みとします。

EC2ごとのEBSの容量

これ、標準APIで欲しいと思いませんかぁぁぁ！
残念ながら無いので組み合わせます。

  • ec2 : describe_instances 1回
  • ec2 : describe_volumes 1回
import boto3

ec2 = boto3.client('ec2')

instances = []
for reserv in ec2.describe_instances()['Reservations']:
  instances.extend(reserv['Instances'])

volumes = ec2.describe_volumes()['Volumes']

for instance in instances:
  instance['EbsSizeList'] = []
  for blockdevice in instance.get('BlockDeviceMappings', [{}]):
    if blockdevice.get('Ebs', {}).get('VolumeId'):
      filtered = list(filter(lambda x: x['VolumeId']==blockdevice['Ebs']['VolumeId'], volumes))
      blockdevice['Ebs'] = filtered[0]
      instance['EbsSizeList'].append(filtered[0]['Size'])

まず、EC2インスタンスの一覧は describe_instances API で取得しますが、結果が ['Reservations'][*]['Instances'][*] の形式となっているものをフラット化しています。
EC2インスタンスごとに ['BlockDeviceMappings'][*]['Ebs'] があり、この配下にあるデータを、より情報量の多い describe_volumes APIで取得した結果で置き換えています。

また、元のデータに 'EbsSizeList' キーを追加して、EBSの数だけ容量(GiB)の数値がリストに含まれるようにしていて、簡単に合計容量が出せるようにしています。

EC2ごとのEBSの合計容量と内訳のGiBを知りたいだけの場合は、

for instance in instances:
    print(instance['InstanceId'], sum(instance['EbsSizeList']), instance['EbsSizeList'])

を実行することで、

i-0xxxxxxxxxxxxxxxx 20 [20]
i-0xxxxxxxxxxxxxxxx 225 [25, 100, 100]
i-0xxxxxxxxxxxxxxxx 80 [50, 30]

のように取得できます。

この調子で紹介が続きます。

ELB(ALB・NLB)のタグ

EC2の describe_instances や RDSの describe_db_instances は、一覧の中にタグ情報が含まれていますが、他のサービスでは取れないものが多いです。ELBもそうです。
標準APIに入ってて欲しいです、タグ情報。

  • elbv2 : describe_load_balancers 1回
  • elbv2 : describe_tags N回(ELBの数)
import boto3

elbv2 = boto3.client('elbv2')

elbs = []
for pg in elbv2.get_paginator('describe_load_balancers').paginate():
    elbs.extend(pg['LoadBalancers'])

for elb in elbs:
    key = 'Tags'
    elb.update({
        key : elbv2.describe_tags(ResourceArns=[elb['LoadBalancerArn']])['TagDescriptions'][0][key]
    })

ELBのリストに 'Tags' としてタグ情報がマージされます。

対象の数が多い場合を考慮して、ページネーションでデータが欠落しないよう対応します。

参考　boto3 で pagination が発生する API を良い感じに処理する

ELB(ALB・NLB)ごとのリスナー

複数のポート番号でリッスンしているELBが混在している場合は、あると便利な組み合わせです。

  • elbv2 : describe_load_balancers 1回
  • elbv2 : describe_listeners N回(ELBの数)
import boto3

elbv2 = boto3.client('elbv2')

elbs = []
for pg in elbv2.get_paginator('describe_load_balancers').paginate():
    elbs.extend(pg['LoadBalancers'])

for elb in elbs:
    key = 'Listeners'
    elb.update({
        key : elbv2.describe_listeners(LoadBalancerArn=elb['LoadBalancerArn'])[key]
    })

ELBのリストに 'Listeners' としてリスナー情報がマージされます。

さらに acm:list_certificates と組み合わせることで、ELBと証明書のデータの組み合わせデータも生成することが可能です。

専有ホストごとのEC2の割り当て量

EC2で専有ホスト(Dedicated Host)はあまり使う機会が無いかもしれませんが、専有ホストを使っていると取りたい情報です。
AWS API単体では、稼働している EC2の割り当て量しか取れないのです。

  • ec2 : describe_instances 1回
  • ec2 : describe_hosts 1回
  • ec2 : describe_instance_types 1回
import boto3

ec2 = boto3.client('ec2')

hosts = ec2.describe_hosts()['Hosts']
instance_reservs = ec2.describe_instances(Filters=[{"Name":"tenancy","Values":["host"]}])['Reservations']
instances = []
for reserv in instance_reservs:
    instances.extend(reserv['Instances'])

instance_types = []
for x in ec2.get_paginator('describe_instance_types').paginate():
    instance_types.extend(x['InstanceTypes'])

for host in hosts:
    sumcpu=0
    instances_on_host = list(filter(lambda x: x['Placement']['HostId']==host['HostId'], instances))
    host['AssignedInstances'] = []
    for instance in instances_on_host:
        filtered = filter(lambda x: x['InstanceType']==instance['InstanceType'], instance_types)
        sumcpu += list(filtered)[0]['VCpuInfo']['DefaultVCpus']
        host['AssignedInstances'].append(instance)

    host['AssignedVCpus'] = sumcpu

'AssignedInstances' キーに対象インスタンスのリストがマージされます。
'AssignedVCpus' には、対象インスタンスのvCPU数の合計値が入るようにしています。この数値と'AvailableCapacity' の 'AvailableVCpus' の比率を取ることで、後どれだけ余裕があるかもわかります。

IAMユーザーごとのIAMグループ

IAMユーザーは複数のIAMグループに含むことができます。よって、IAMユーザーごとにどのIAMグループに入っているかを出力します。

  • iam : list_users 1回
  • iam : list_groups_for_user N回
import boto3

iam = boto3.client('iam')

users = []
for pg in iam.get_paginator('list_users').paginate():
    users.extend(pg['Users'])

for u in users:
    key = 'Groups'
    u.update({
        key : iam.list_groups_for_user(UserName=u['UserName'])[key]
    })

これによって、result にはIAMユーザーのリストが 'Groups' 付きで入ります。

通常の list_users APIよりも階層構造を減らしてシンプルにしています。
各ユーザーには list_groups_for_user API で得られた結果の 'Groups' を、そのキー名のままマージしています。

IAMユーザーごとのIAMポリシー

  • iam : list_users 1回
  • iam : list_attached_user_policies N回
import boto3

iam = boto3.client('iam')

# IAMユーザー
users = []
for pg in iam.get_paginator('list_users').paginate():
    users.extend(pg['Users'])

for u in users:
    key = 'AttachedPolicies'
    u.update({
        key : iam.list_attached_user_policies(UserName=u['UserName'])[key]
    })

IAMユーザーごとのIAMグループとほぼ似たようなコードになります。
ユーザーのリストにマージする際のIAMポリシーのキー名は 'AttachedPolicies' にしています。

IAMグループごとのIAMポリシー

  • iam : list_groups 1回
  • iam : list_attached_group_policies N回
import boto3

iam = boto3.client('iam')

groups = []
for pg in iam.get_paginator('list_groups').paginate():
    groups.extend(pg['Groups'])

for g in groups:
    key = 'AttachedPolicies'
    g.update({
        key : iam.list_attached_group_policies(GroupName=g['GroupName'])[key]
    })

IAMユーザーグループのリストにポリシーである 'AttachedPolicies' をマージします。

事例集はここまでです。

マルチスレッドによる高速化

APIの実行回数が多くなる場合、以下のようにマルチスレッドで実行すると高速化する可能性があります。(必ずしも高速化するとは限りません）
上記に示した IAMユーザーとIAMグループ のケースで ThreadPoolExecutorを使用した例です。

import boto3
from concurrent.futures import ThreadPoolExecutor

users = []
for pg in iam.get_paginator('list_users').paginate():
    users.extend(pg['Users'])

def task(x):
    session = boto3.Session()
    return session.client('iam').list_groups_for_user(UserName=x['UserName'])

g_user = ThreadPoolExecutor(max_workers=12).map(task, users)

for x,y in zip(users, list(g_user)):
    x.update({'Groups':y['Groups']})

Session() をtask()内で呼んでいる理由は、Sessionはスレッドセーフではないためです。
参考: https://dev.classmethod.jp/articles/python-boto3-session-thread-unsafe/

元々sessionを使っていなくても、boto3.client() が内部的に session() を呼んでいるようでもあります。

Lambda関数で呼び出すときの注意点

単一APIであればデータ量も実行時間も、特に気にしなくて良かったものが、マージするとどうしてもデータ量が多くなり、時間もかかります。

Lambda関数で実装する場合は、Lambda関数の実行時間を十分延ばしましょう。

Lambda関数でAPI Gatewayなどを経由してWebブラウザーで戻り値を受け取る場合は、REST API の 29秒の制限や、Lambda関数のレスポンスサイズ 6MBの制限があるため、成り立たない組み合わせもあるかもしれません。
実装には注意が必要です。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?