AWSにおいて、多数のVPCや他のクラウド、オンプレミス拠点を接続する上で、Transit Gatewayは便利なサービスだと思います。
その一方で、接続するネットワークの数が増加するにつれて、構成管理のコストは高くなっていきます。
この記事では、Transit Gatewayのルートテーブルの設定状況を把握しやすくするために作成したツールを紹介します。
作成したツール
以下の3つの情報を取得し、それらを結合することで、各Transit Gatewayルートテーブルに設定されているルーティングを俯瞰できるようにしました。
- Transit Gatewayルートテーブルのルート情報一覧
- Transit Gatewayルートテーブルの一覧
- Transit Gatewayアタッチメントの一覧
それぞれの情報の取得コードと取得結果のイメージは以下の通りです。
1. Transit Gatewayルートテーブルのルート情報一覧
command = [
"aws", "ec2", "search-transit-gateway-routes",
"--transit-gateway-route-table-id", tgw_route_id,
"--filters", "Name=state,Values=active,blackhole",
"--query", "Routes[].[DestinationCidrBlock,TransitGatewayAttachments[0].ResourceId,TransitGatewayAttachments[0].TransitGatewayAttachmentId,TransitGatewayAttachments[0].ResourceType,Type,State]",
"--output", "text"
]
取得イメージ
DestinationCidrBlock ResourceId Attachment-ID ResourceType Type State
-------------------- -------------------------- -------------------------- ------------ ---------- ---------
0.0.0.0/0 vpc-5050gggg6060hhhh tgw-attach-d1d1d1d1d1d1d1d1 vpc static active
10.0.0.0/8 None None None static blackhole
192.168.1.0/24 vpc-1010aaaa2020bbbb tgw-attach-a1a1a1a1a1a1a1a1 vpc propagated active
192.168.2.0/24 vpc-1010cccc2020dddd tgw-attach-b1b1b1b1b1b1b1b1 vpc propagated active
172.16.0.0/24 vpc-3030eeee4040ffff tgw-attach-c1c1c1c1c1c1c1c1 vpc static active
：
：
2. Transit Gatewayルートテーブルの一覧
command = [
"aws", "ec2", "describe-transit-gateway-route-tables",
"--query", "TransitGatewayRouteTables[].[Tags[?Key=='Name'] | [0].Value,TransitGatewayRouteTableId,State]",
"--output", "text"
]
取得イメージ
Route-Table-Name Route-Table-ID State
---------------------------- --------------------------- ----------
network-rtb-production tgw-rtb-1111eeee2222ffff available
network-rtb-staging tgw-rtb-3333gggg4444hhhh available
network-rtb-shared-firewall tgw-rtb-5555iiii6666jjjj available
：
：
### 3. Transit Gatewayアタッチメントの一覧
```Python
command = [
"aws", "ec2", "describe-transit-gateway-attachments",
"--query", "TransitGatewayAttachments[].[Tags[?Key=='Name'] | [0].Value,TransitGatewayId,TransitGatewayAttachmentId,ResourceType,ResourceId,ResourceOwnerId]",
"--output", "text"
]
取得イメージ
Attachment-Name TransitGateway-ID Attachment-ID ResourceType ResourceId ResourceOwner-ID
----------------------- -------------------------- -------------------------- ------------ ------------------------- ----------------
prd-app-vpc-attach tgw-0000aaaa1111bbbb tgw-attach-a1a1a1a1a1a1a1a1 vpc vpc-1010aaaa2020bbbb 111122223333
prd-db-vpc-attach tgw-0000aaaa1111bbbb tgw-attach-b1b1b1b1b1b1b1b1 vpc vpc-1010cccc2020dddd 111122223333
stg-app-vpc-attach tgw-0000aaaa1111bbbb tgw-attach-c1c1c1c1c1c1c1c1 vpc vpc-3030eeee4040ffff 444455556666
shared-fw-vpc-attach tgw-0000aaaa1111bbbb tgw-attach-d1d1d1d1d1d1d1d1 vpc vpc-5050gggg6060hhhh 000000000000
：
：
情報の結合と最終生成物
それらの情報をPythonスクリプトで、以下のように結合するようにしました。
結合イメージ
：
0.0.0.0/0 tgw-attach-d1d1d1d1d1d1d1d1 shared-fw-vpc-attach 000000000000 1 1 0
10.0.0.0/8 None N/A N/A 1 1 1
172.16.0.0/24 tgw-attach-c1c1c1c1c1c1c1c1 stg-app-vpc-attach 444455556666 1 0 1
192.168.1.0/24 tgw-attach-a1a1a1a1a1a1a1a1 prd-app-vpc-attach 111122223333 0 1 1
192.168.2.0/24 tgw-attach-b1b1b1b1b1b1b1b1 prd-db-vpc-attach 111122223333 0 1 1
：
：
Pythonスクリプト実行イメージ
$ python3 main.py
処理完了。ファイルを出力しました: tgw_info_20250626_143820.txt
このファイルをテキストエディタで開き、スプレッドシートに貼り付けてください。
Pythonスクリプトの実行結果を、あらかじめ用意したGoogleスプレッドシートのテンプレートに貼り付けることで、各Transit Gatewayルートテーブルに設定されているルーティングを俯瞰できるようにしました。
Googleスプレッドシートのテンプレート
Pythonスクリプトの実行結果を貼り付けた後のイメージ
表の縦軸にルート情報（宛先CIDR毎の転送先Transit Gatewayアタッチメント名）が、横軸にTransit Gatewayルートテーブルが表示されるようにしています。
行列が交差するセル内の1または0が設定の有無を表し、1であればルートが存在する、0であればルートが存在しないと把握できるようにしています。
同じ情報をマネジメントコンソールで確認しようとする場合、CIDR毎の転送先Transit GatewayアタッチメントIDやVPCのIDの情報しか得られず、転送先のVPCをイメージしにくいです。
今回作成したツールでは、CIDR毎の転送先Transit Gatewayアタッチメント名が表示されているため、通信がどの環境や役割のVPCに向かうのかを把握しやすくなっています。
以下に、今回作成したツールに使用する本データ（Googleスプレッドシートに貼り付ける情報）を出力するPythonスクリプトのコード全文を記載します。
コード全文
import subprocess
import datetime
import json
# TransitGatewayルートテーブルのルート情報一覧を取得する関数
def get_aws_tgw_routes(tgw_route_id):
command = [
"aws", "ec2", "search-transit-gateway-routes",
"--transit-gateway-route-table-id", tgw_route_id,
"--filters", "Name=state,Values=active,blackhole",
"--query", "Routes[].[DestinationCidrBlock,TransitGatewayAttachments[0].ResourceId,TransitGatewayAttachments[0].TransitGatewayAttachmentId,TransitGatewayAttachments[0].ResourceType,Type,State]",
"--output", "text"
]
result = subprocess.run(command, capture_output=True, text=True)
return result.stdout
# 出力を成形してマージするための関数
def format_and_merge_output(output, tgw_route_id):
lines = output.strip().split('\n')
formatted_lines = [f"{tgw_route_id}\t{line}" for line in lines]
return "\n".join(formatted_lines)
# 利用可能なTransitGatewayルートテーブルIDを取得する関数
def get_available_tgw_route_tables():
command = [
"aws", "ec2", "describe-transit-gateway-route-tables",
"--query", "TransitGatewayRouteTables[].[Tags[?Key=='Name'] | [0].Value,TransitGatewayRouteTableId,State]",
"--output", "text"
]
result = subprocess.run(command, capture_output=True, text=True)
lines = result.stdout.strip().split('\n')
# Nameの昇順に並び替え
sorted_lines = sorted(lines, key=lambda x: x.split()[0])
available_tgw_route_tables = [(line.split()[0], line.split()[1]) for line in sorted_lines if line.split()[2] == 'available']
return available_tgw_route_tables
# TGWアタッチメント一覧を取得する関数
def get_tgw_attachments():
command = [
"aws", "ec2", "describe-transit-gateway-attachments",
"--query", "TransitGatewayAttachments[].[Tags[?Key=='Name'] | [0].Value,TransitGatewayId,TransitGatewayAttachmentId,ResourceType,ResourceId,ResourceOwnerId]",
"--output", "text"
]
result = subprocess.run(command, capture_output=True, text=True)
lines = result.stdout.strip().split('\n')
tgw_attachments = {line.split()[2]: (line.split()[0], line.split()[5]) for line in lines} # TGWアタッチメントIDをキー、NameとOwnerIDを値とする辞書を作成
return tgw_attachments
# 出力を並び替えるための関数
def sort_output(lines):
return sorted(lines, key=lambda x: (x.split('\t')[1], x.split('\t')[3]))
# 出力用生データを作成する関数
def create_raw_data_output(sorted_output, available_tgw_route_tables, tgw_attachments):
tgw_route_table_dict = {tgw_id: name for name, tgw_id in available_tgw_route_tables}
raw_data_output = []
for line in sorted_output:
columns = line.split('\t')
tgw_route_id = columns[0]
tgw_route_name = tgw_route_table_dict.get(tgw_route_id, "N/A")
tgw_attachment_name, tgw_attachment_owner = tgw_attachments.get(columns[3], ("N/A", "N/A")) # TGWアタッチメントIDに対応するNameとOwnerIDを取得
raw_data_output.append(f"{tgw_route_id}\t{tgw_route_name}\t{columns[1]}\t{columns[2]}\t{columns[3]}\t{tgw_attachment_name}\t{tgw_attachment_owner}\t{columns[4]}\t{columns[5]}\t{columns[6]}")
return raw_data_output
# 利用可能なTransitGatewayルートテーブルIDを横軸に持つマトリックスを作成する関数
def create_matrix(lines, tgw_route_ids, tgw_attachments):
matrix = []
for line in lines:
columns = line.split('\t')
tgw_attachment_name, tgw_attachment_owner = tgw_attachments.get(columns[3], ("N/A", "N/A")) # TGWアタッチメントIDに対応するNameとOwnerIDを取得
row = columns[:4] + [tgw_attachment_name, tgw_attachment_owner] + columns[4:] + [1 if columns[0] == tgw_id else 0 for tgw_id in tgw_route_ids]
matrix.append(row)
return matrix
# 重複排除するための関数
def remove_duplicates(matrix):
seen = {}
unique_matrix = []
for row in matrix:
key = (row[1], row[3]) # 2カラム目（CIDR）と4カラム目（tgwアタッチメントID）の組み合わせ
if key in seen:
# 既存の行の該当カラムを1にする
for i in range(9, len(row)):
if row[i] == 1:
seen[key][i] = 1
else:
seen[key] = row
unique_matrix.append(row)
return unique_matrix
# 不要なカラムを削除する関数
def remove_unnecessary_columns(matrix):
columns_to_keep = [1, 3, 4, 5] # CIDR, AttachmentId, AttachmentName, AttachmentOwner
filtered_matrix = []
for row in matrix:
filtered_row = [row[i] for i in columns_to_keep] + row[9:] # 9列目以降はそのまま保持
filtered_matrix.append(filtered_row)
return filtered_matrix
# メイン処理
def main():
# AWSと認証が通っていることを確認する
expected_account_id = "XXXXXXXXXXXX" # ここにTGW管理アカウントのIDを設定
command = ["aws", "sts", "get-caller-identity"]
# result = subprocess.run(command, capture_output=True)
result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
if result.returncode != 0:
print(f"AWSアカウントとの認証がされていません。アカウントID: {expected_account_id} と認証を行ってください。")
return
identity = json.loads(result.stdout)
account_id = identity.get("Account")
if account_id != expected_account_id:
print(f"認証されたAWSアカウントIDが期待されるものではありません。期待されるアカウントID: {expected_account_id}, 実際のアカウントID: {account_id}")
return
# 以下よりメイン処理
available_tgw_route_tables = get_available_tgw_route_tables()
tgw_attachments = get_tgw_attachments()
all_formatted_outputs = []
for _, tgw_route_id in available_tgw_route_tables:
output = get_aws_tgw_routes(tgw_route_id)
formatted_output = format_and_merge_output(output, tgw_route_id)
all_formatted_outputs.append(formatted_output)
# すべての出力をマージ
merged_output = "\n".join(all_formatted_outputs)
# 並び替え
sorted_output = sort_output(merged_output.strip().split('\n'))
# 出力用生データを作成
raw_data_output = create_raw_data_output(sorted_output, available_tgw_route_tables, tgw_attachments)
# マトリックスを作成
matrix = create_matrix(sorted_output, [tgw_id for _, tgw_id in available_tgw_route_tables], tgw_attachments)
# 重複排除
unique_matrix = remove_duplicates(matrix)
# 不要なカラムを削除
filtered_matrix = remove_unnecessary_columns(unique_matrix)
# ヘッダーを定義
header1 = ["Transit Gateway ルートテーブルと対応CIDRのマトリクス", "", "", "Transit Gateway ルートテーブルID"] + [tgw_id for _, tgw_id in available_tgw_route_tables]
header2 = ["1の場合、存在する。0の場合、存在しない。", "", "", "Transit Gateway ルートテーブル名"] + [name for name, _ in available_tgw_route_tables]
header3 = ["ルート -> CIDR", "ルート -> アタッチメントID", "アタッチメント名", "アタッチメントのオーナーID"]
header4 = ["TransitGatewayRouteTableID", "TransitGatewayRouteTableName", "CIDR", "ResourceId", "AttachmentId", "AttachmentName", "AttachmentOwner", "ResourceType", "RouteType", "RouteState"]
# 現在の日時を取得してファイル名を生成
now = datetime.datetime.now()
filename = f"tgw_info_{now.strftime('%Y%m%d_%H%M%S')}.txt"
# ファイルに出力を保存
with open(filename, "w") as f:
f.write("\t".join(header1) + "\n")
f.write("\t".join(header2) + "\n")
f.write("\t".join(header3) + "\n")
for row in filtered_matrix:
f.write("\t".join(map(str, row)) + "\n")
f.write("\n\n\n") # 3行ほど余白を開けたいので空行を挿入
f.write("上の表に関する生データ\n")
f.write("\t".join(header4) + "\n")
for line in raw_data_output:
f.write(f"{line}\n")
print(f"処理完了。ファイルを出力しました: {filename}\nこのファイルをテキストエディタで開き、スプレッドシートに貼り付けてください。")
if __name__ == "__main__":
main()
ツールの活用シナリオ
このツールを作成したことで、以下のような需要に応えやすくなりました。
1. 疎通できない事象の切り分け
「特定のVPCやオンプレミス環境から通信ができない」という問い合わせがあった際に、原因がTransit Gatewayのルーティング設定なのか、そうではないのかを迅速に把握できるようになりました。事象の切り分けポイントが1つ減ることで、原因を把握するまでの速度の向上が期待できるようになりました。
2. 設定変更後の確認
Terraformで設定変更をした際に、マネジメントコンソールにログインして、複数のルートテーブルの設定をチェックせずとも、意図した設定ができていることを確認できるようになりました。
3. 定期的な設定情報の棚卸し
「過去に一時的に追加したまま放置されているルート」や「現在は使われていないはずの不要なルート」の有無を把握しやすくなりました。また、過去の出力結果と比較することで、時期ごとのルート設定の変遷も把握しやすくなりました。
感想
今回のツールの主役は、AWS CLIです。構造化されたデータさえ手に入れば、あとはスクリプトでそれらを結合するだけで、今回のような価値ある情報を自在に生成できます。
初めから情報が統合された形で提供されるのが理想ですが、こういった「隙間」があるからこそ、ツールを作成して業務を効率化するという「やりがい」が生まれるのだと考えています。
もちろん、ツール作成自体が目的となり、業務が疎かになるのは本末転倒です。しかし、ツールを作成すべきポイントを的確に見極め、それがもたらす効果を想像して得られる高揚感（ドーパミン）を原動力にすることは、仕事だけでなく、人生にも応用できるものだと考えています。