AWSロードバランサーとターゲットグループの制御方法

概要

AWSで動的な拡張環境を構築した経験があります。Pythonを使用して、AWSロードバランサーとターゲットグループを制御するためのモジュールを作成しました。EC2の動的な制御に関する詳細は、以下の記事をご参照ください。
AWS EC2インスタンスの自動起動停止の制御手法

ソース

import sys
import os
import traceback
import boto3

def to_exception_detail(exception, includes_trace=False):
    """
    概要: 例外の詳細情報を整形して返す関数
    引数:
        - exception: 例外オブジェクト
        - includes_trace: トレースバック情報を含むかどうかのフラグ
    戻り値: 整形された例外詳細情報の文字列
    """
    exc_type, exc_obj, exc_tb = sys.exc_info()
    fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
    lineno = exc_tb.tb_lineno
    trace = traceback.format_exc() if includes_trace else ''
    return 'at [{0}]({1}:{2}) -> {3}{4}'.format(type(exception).__name__, fname, lineno, exception, trace)

class ELBOp(object):
    """
    概要: AWS Elastic Load Balancer (ELB) と関連するオペレーションを行うクラス
    """

    global elb_instances
    elb_instances = boto3.client('elbv2')

    def register_ec2instances(self, tg_name, ec2_ins_ids):
        """
        概要: 指定されたターゲットグループに EC2 インスタンスを登録するメソッド
        引数:
            - tg_name: ターゲットグループの名前
            - ec2_ins_ids: 登録する EC2 インスタンスのIDリスト
        戻り値: 登録が成功したかどうかのブール値
        例外: 例外が発生した場合、詳細なエラーメッセージを含む例外が発生
        """
        try:
            target_ins_ids = list(
                filter(lambda ec2_ins_id: not self.__existed_in_elb(tg_name, ec2_ins_id), ec2_ins_ids))
            if target_ins_ids is None or len(target_ins_ids) == 0:
                return True
            tg_arn, port = self.__get_tgInfoFromName(tg_name)

            if tg_arn is not None:
                targets = []
                for ec2_id in target_ins_ids:
                    target_info = {}
                    target_info['Id'] = ec2_id
                    target_info['Port'] = int(port)
                    targets.append(target_info)
                response = elb_instances.register_targets(TargetGroupArn=tg_arn, Targets=targets)
                if response is not None and response['ResponseMetadata']['HTTPStatusCode'] == 200:
                    return True

            return False
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def __get_tgInfoFromName(self, tg_name):
        """
        概要: ターゲットグループの名前から情報を取得するメソッド
        引数:
            - tg_name: ターゲットグループの名前
        戻り値: ターゲットグループの ARN とポート番号のタプル
        例外: 例外が発生した場合、詳細なエラーメッセージを含む例外が発生
        """
        try:
            response = elb_instances.describe_target_groups(Names=[tg_name])
            if 'TargetGroups' in response and len(response['TargetGroups']) > 0:
                target_group = response['TargetGroups'][0]
                if 'TargetGroupArn' in target_group and 'Port' in target_group:
                    return target_group['TargetGroupArn'], target_group['Port']
            return None, None
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def deregister_ec2instances(self, tg_name, ec2_ins_ids):
        """
        概要: 指定されたターゲットグループから EC2 インスタンスを解除するメソッド
        引数:
            - tg_name: ターゲットグループの名前
            - ec2_ins_ids: 解除する EC2 インスタンスのIDリスト
        戻り値: 解除が成功したかどうかのブール値
        例外: 例外が発生した場合、詳細なエラーメッセージを含む例外が発生
        """
        try:
            target_ins_ids = list(filter(lambda ec2_ins_id: self.__existed_in_elb(tg_name, ec2_ins_id), ec2_ins_ids))
            if target_ins_ids is None or len(target_ins_ids) == 0:
                return True
            tg_arn, port = self.__get_tgInfoFromName(tg_name)
            if tg_arn is not None:
                targets = []
                for ec2_id in target_ins_ids:
                    target_info = {}
                    target_info['Id'] = ec2_id
                    target_info['Port'] = int(port)
                    targets.append(target_info)
                response = elb_instances.deregister_targets(TargetGroupArn=tg_arn, Targets=targets)
                if response is not None and response['ResponseMetadata']['HTTPStatusCode'] == 200:
                    return True
            return False
        except Exception as e:
            raise Exception(to_exception_detail(e))

    def __existed_in_elb(self, tg_name, ec2_ins_id):
        """
        概要: 指定された EC2 インスタンスがターゲットグループに登録されているかどうかを確認するメソッド
        引数:
            - tg_name: ターゲットグループの名前
            - ec2_ins_id: 確認する EC2 インスタンスのID
        戻り値: EC2 インスタンスが登録されていれば True、そうでなければ False
        例外: 例外が発生した場合、詳細なエラーメッセージを含む例外が発生
        """
        try:
            tg_arn, port = self.__get_tgInfoFromName(tg_name)
            response = elb_instances.describe_target_health(TargetGroupArn=tg_arn)
            if response is not None and 'TargetHealthDescriptions' in response:
                targetHealthDescriptions = response['TargetHealthDescriptions']
                for targetHealthDescription in targetHealthDescriptions:
                    if 'Target' in targetHealthDescription and len(targetHealthDescription['Target']) > 0:
                        target = targetHealthDescription['Target']
                        if target['Id'] == ec2_ins_id:
                            return True
            return False
        except Exception as e:
            raise Exception(to_exception_detail(e))

補足

1. 上記のモジュールを動かすために、awscli 用の環境変数を設定することが必要です。以下のように aws configure コマンドで設定することができます。

# aws configure
AWS Access Key ID [****************ETRQ]: 
AWS Secret Access Key [****************oY/N]:
Default region name [ap-northeast-1]:
Default output format [None]: