Zenn
⚠

【Boto3】waiter の待機処理について考える

2025/03/13に公開
AWS
Python
Boto3
tech

概要

EC2.Client.get_waiter() の waiter.wait() を検証中に気づいたのですが、
Boto3においてwaiterなどの機能はメイン機能に比べ開発が遅いっぽいです。
2025/03/13時点で、EC2.Client.describe_instance_status で取得できるAttachedEbsStatus用の待機処理が用意されていませんでした。

便利ですが、自作の待機処理を実装したほうが良い場合もありそうです。

本書では、waiter.wait()を利用した待機処理と自作の待機処理の2つを検証していこうと思います。

前提

以下のバージョンで検証しました。

$ lsb_release -d
Description:    Ubuntu22.04.3

$ pip list installed
Package            Version
------------------ -----------
Python             3.12.4
pip                24.0
boto3              1.37.7

Waiter 検証

検証用コード

EC2を起動後、インスタンスの状態がrunningかつステータスチェックがokになるまで待機します。
※2025/03/13時点で、EC2.Client.describe_instance_status で取得できるAttachedEbsStatus用のwaiterが用意されていませんでした。

verify_ec2_waiter.py
import boto3

import ly_logging # ログ出力用カスタムライブラリ

logger = ly_logging.logger('DEBUG')

class EC2Client:
  def __init__(self, instance_ids: list):
    self.client = boto3.client('ec2')
    self.instance_ids = instance_ids
    self.instance_state_waiter = self.client.get_waiter('instance_running')
    self.instance_status_waiter = self.client.get_waiter('instance_status_ok')
    self.system_status_waiter = self.client.get_waiter('system_status_ok')
  def waiter_wait_all_status_ok(self):
    """ 全てのステータスが起動状態になるまで待機 """
    logger.info('wait処理を開始します。')
    self.instance_state_waiter.wait(InstanceIds = self.instance_ids)
    self.instance_status_waiter.wait(InstanceIds = self.instance_ids)
    self.system_status_waiter.wait(InstanceIds = self.instance_ids)
    logger.info('Wait処理が完了しました。')
  def start_instances(self):
    """ インスタンスを起動 """
    response = self.client.start_instances(InstanceIds = self.instance_ids)
    return response

def verify_waiter(instance_ids: list):
  """ 
  Waiter の動作確認用
  
  Parameters
  ----------
  instance_ids : list of instance ids
  """
  ec2_client = EC2Client(instance_ids)
  ec2_client.start_instances() # EC2起動
  ec2_client.waiter_wait_all_status_ok() # 起動するまで待機

if __name__ == '__main__':
  params = {'instance_ids': ['i-xxxxx']}
  verify_waiter(**params)

検証結果

マネジメントコンソールをリアルタイムで目視していましたが、
インスタンス状態が「実行中」かつステータスチェック「3/3 のチェックに合格しました」となったタイミングで処理が終了しました。

$ python verify_waiters.py
2025/03/13 16:48:11     [INFO]  wait処理を開始します。
2025/03/13 16:50:13     [INFO]  Wait処理が完了しました。

自作の待機処理で検証

検証用コード

Waiter検証時のコードに追記します。
Waiterの代わりにEC2.Client.describe_instance_statusで各種ステータスを取得しています。

verify_waiters.py
import boto3
import time

import ly_logging # ログ出力用カスタムライブラリ

logger = ly_logging.logger('DEBUG')

class EC2Client:
  def __init__(self, instance_ids: list):
    self.client = boto3.client('ec2')
    self.instance_ids = instance_ids
    self.instance_state_waiter = self.client.get_waiter('instance_running')
    self.instance_status_waiter = self.client.get_waiter('instance_status_ok')
    self.system_status_waiter = self.client.get_waiter('system_status_ok')
  def waiter_wait_all_status_ok(self):
    """ 全てのステータスが起動状態になるまで待機 """
    logger.info('wait処理を開始します。')
    self.instance_state_waiter.wait(InstanceIds = self.instance_ids)
    self.instance_status_waiter.wait(InstanceIds = self.instance_ids)
    self.system_status_waiter.wait(InstanceIds = self.instance_ids)
    logger.info('Wait処理が完了しました。')
  def custom_wait_all_status_ok(self):
    """ 全てのステータスが起動状態になるまで待機 """
    logger.info('自作待機処理を開始します。')
    for _ in range(10): # 待機時間の上限は5分とする
      time.sleep(30)
      response = self.describe_instance_status()
      instance_statuses = response['InstanceStatuses']
      if all(
        ist['InstanceState']['Name'] == 'running'
        and ist['InstanceStatus']['Status'] == 'ok'
        and ist['SystemStatus']['Status'] == 'ok'
        and ist['AttachedEbsStatus']['Status'] == 'ok'
        for ist in instance_statuses
      ):
        logger.info('自作待機処理が完了しました。')
        break
    else:
      logger.error('時間内にインスタンスが起動しませんでした。')
  def describe_instance_status(self):
    """ インスタンスの状態を取得 """
    response = self.client.describe_instance_status(InstanceIds = self.instance_ids)
    return response
  def start_instances(self):
    """ インスタンスを起動 """
    response = self.client.start_instances(InstanceIds = self.instance_ids)
    return response

def verify_waiter(instance_ids: list):
  """ 
  Waiter の動作確認用
  
  Parameters
  ----------
  instance_ids : list of instance ids
  """
  ec2_client = EC2Client(instance_ids)
  ec2_client.start_instances() # EC2起動
  ec2_client.waiter_wait_all_status_ok() # 起動するまで待機

def verify_custom_wait(instance_ids: list):
  """ 
  自作の待機処理の動作確認用
  
  Parameters
  ----------
  instance_ids : list of instance ids
  """
  ec2_client = EC2Client(instance_ids)
  ec2_client.start_instances() # EC2起動
  ec2_client.custom_wait_all_status_ok() # 起動するまで待機


if __name__ == '__main__':
  params = {'instance_ids': ['i-xxxxx']}
  # verify_waiter(**params)
  verify_custom_wait(**params)

検証結果

マネジメントコンソールをリアルタイムで目視していましたが、
インスタンス状態が「実行中」かつステータスチェック「3/3 のチェックに合格しました」となったタイミングで処理が終了しました。

$ python verify_waiters.py
2025/03/13 16:52:08     [INFO]  自作待機処理を開始します。
2025/03/13 16:54:39     [INFO]  自作待機処理が完了しました。

これでステータスチェック3/3全てを確認し起動確認できました。

さいごに

Boto3内の機能でも最新機能に対応していない機能があるようです。
比較的新しめの機能を利用する際にはその辺を考慮して実装する必要がありますね。

Discussion