[Python] CLB 인스턴스 상태 및 등록, 해제하는 파이썬 스크립트

Shell Script 작성 글
https://nyyang.tistory.com/109?category=990664

[Shell Script] CLB 인스턴스 상태 및 등록, 해제하는 쉘 스크립트

위에 Shell Script로 작성했던 글을 Python으로도 작성해보았다.


### 스크립트 목적

- 고객사 중 한 곳이 Legacy하게 배포 작업을 진행하는 곳이 있음 (SFTP 수동 배포)
1) 특정 CLB에 붙어 있는 Instance 중 A존 Detach
2) 배포가 완료되면 A존 Attach
3) 특정 CLB에 붙어 있는 Instance 중 C존 Detach
4) 배포가 완료되면 C존 Attach

- 배포할 때 마다 GUI 웹 콘솔에서 클릭하면서 하기에는 너무 번거로움


### 스크립트 설명

1) --status, --detach, --attach를 시작으로 CLB에 등록되어 있는 EC2 인스턴스들의 상태 확인, EC2 인스턴스들 Detach, EC2 인스턴스들 Attach 하는 스크립트이다.

2) --attach [CLB_NAME] [INSTANCES_MATCHED] [ZONE]
=> INSTANCES_MATCHED는 만약 내가 --attach MY-CLB PRD-WEB ap-northeast-2a라고 명령어를 입력한다면
모든 EC2 인스턴스 내역 중에 PRD-WEB이란 문자가 포함된 EC2만을 가져온다.
=> 그 EC2가 특정 Zone이여야만 한다. (ap-northeast-2a, .. )


### 스크립트 개선 필요 내용

1. 하드코딩한 부분이 있음 (ap-northeast-2a, ap-northeast-2c)
=> 서울 리전 사용하는 특정 고객에 대해서만 스크립트를 수행하기 때문에 하드코딩

2. python clb_deploy.py =>> 이 부분이 좀 더러움. clb_deploy --attach ~~ 이런 식으로 파이썬에서 보기 좋게 바꿔줄 수 있는데 굳이 안해도 될 것 같아서 안함

3. 기타 등등


### 주의 사항

1. IAM User의 Access Key, Secret Key를 ~/.aws/credentials에 넣었음

2. IAM User의 Policy는 EC2, ELB에 대한 권한이 존재해야함

import sys
import boto3
from typing import List
from typing import Dict


def clb_exist(client, keyword: str) -> bool:
    """
    args[0]에 해당하는 CLB가 존재하는지 확인하는 함수입니다.
    """
    try:
        response = client.describe_load_balancers(
            LoadBalancerNames=[
                keyword,
            ]
        )
    except Exception as e:
        print(f"{keyword} 탐색 중 error 발생.\n", e)
        return False
    
    return True
    

def find_instances(client, keyword: str, zone: str) -> Dict[str, str]:
    """
    keyword에 해당하는 문자열을 포함하는 EC2를 Dict로 반환해주는 함수입니다.
    """
    instances_dict = dict()
    
    for reservation in client.describe_instances().get("Reservations"):
        for instance in reservation["Instances"]:
            status = instance.get('State').get('Name')
            if status == "running" or status == "stopped":
                
                # 해당 Zone에 있는지 확인
                instance_zone = instance.get('Placement').get('AvailabilityZone')
                instance_name = "-"
                if instance_zone == zone:
                    for inst in instance["Tags"]:
                        if inst.get('Key') == "Name":
                            instance_name = inst.get("Value")
                            break
                
                # keyword가 instance_name에 포함되어 있으면 instances_dict에 포함한다.
                if keyword in instance_name:
                    instances_dict[instance.get('InstanceId')] = instance_name
    
    return instances_dict

def clb_instance_status(client, keyword: str) -> Dict[str, str]:
    """
    CLB에 등록되어 있는 Instance Dict로 Return 해주는 함수입니다.
    """
        
    clb_instance_status = dict()
    
    for instance in client.describe_instance_health(LoadBalancerName=keyword).get('InstanceStates'):
        instance_id = instance.get('InstanceId')
        instance_status = instance.get('State')
        
        clb_instance_status[instance_id] = instance_status
    # print(f"{keyword} Instances")
    # for key, value in clb_instance_status.items():
    #     print("Instance Id : " + key + " // Status : " + value)
    
    return clb_instance_status
    

def convert_id_to_name(client, instance_id: str) -> str:
    """
    Instance Id를 Instance Name으로 보기 좋게 변경해주는 함수입니다.
    """
    
    instance_info = "-"
    for reservation in client.describe_instances(
        InstanceIds=[instance_id]
        ).get("Reservations"):
        for instance in reservation["Instances"]:
            status = instance.get('State').get('Name')
            if status == "running" or status == "stopped":
                for inst in instance["Tags"]:
                    if inst.get('Key') == "Name":
                        instance_info = inst.get("Value")
            instance_info += f" ({instance.get('Placement').get('AvailabilityZone')})"
    
    return instance_info


def attach_instances(client, clb: str, instance_list: List[str]) -> None:
    """
    CLB에 Instance를 Attach하는 함수입니다.
    """
    for inst_id in instance_list:
        try:
            response = client.register_instances_with_load_balancer(
                LoadBalancerName=clb,
                Instances=[
                    {
                        'InstanceId': inst_id,
                    }
                ]
                )
            print(f"{clb}에 {inst_id} 등록 완료")
        except Exception as e:
            print(f"{clb}에 {inst_id}를 Attach 하는 중 오류 발생.\n" + e)


def detach_instances(client, clb: str , instance_list: List[str]) -> None:
    """
    CLB에 Instance를 Detach하는 함수입니다.
    """
    for inst_id in instance_list:
        try:
            response = client.deregister_instances_from_load_balancer(
                LoadBalancerName=clb,
                Instances=[
                    {
                        'InstanceId': inst_id,
                    }
                ]
                )
            print(f"{clb}에 {inst_id} 해제 완료")
        except Exception as e:
            print(f"{clb}로부터 {inst_id}를 Detach 하는 중 오류 발생.\n" + e)


def command_help() -> None:
    """
    help 함수입니다.
    """
    arg0 = sys.argv[0].split("\\")[1]
    print(f"""Usage : python {arg0} [OPTIONS] [ARGS]

Options:
  -h, --help   : show this help message and exit
  
  -s, --status : show current instances status attathed to classic load balancer
                 python {arg0} --status [CLB_NAME]
                 
  -a, --attach : attach matched instances to classic load balancer
                 python {arg0} --attach [CLB_NAME] [INSTANCES_MATCHED] [ZONE]
                 ex) python {arg0} --attach MYCLB PRD-WEB ap-northeast-2a
                 
  -d, --detach : detach matched instances from classic load balancer
                 python {arg0} --detach [CLB_NAME] [INSTANCES_MATCHED] [ZONE]
                 ex) python {arg0} --detach MY_CLB PRD-WEB ap-northeast-2c
""")

def approve(question: str) -> bool:
    while "the answer is invalid":
        reply = str(input(question+' (y/n): ')).lower().strip()
        if reply[0] == 'y':
            return True
        if reply[0] == 'n':
            return False

def check_zone(zone: str) -> bool:
    zones = ["ap-northeast-2a", "ap-northeast-2c"]
    
    if zone in zones:
        return True
    else:
        return False
    

def main(arguments: List[str]) -> None:
    opt = arguments[0]
    args = arguments[1:]
    
    # Session
    session = boto3.session.Session(profile_name=p_name)
    # boto3 client
    ec2_cli = session.client(service_name='ec2', region_name=r_name)
    elb_cli = session.client(service_name="elb", region_name=r_name)
    
    # 1. Status
    if opt == "--status" or opt == "-s":
        if len(args) != 1:
            raise SystemExit(command_help())
        if clb_exist(elb_cli, args[0]):
            print(f"CLB : {args[0]}에 등록되어 있는 EC2 인스턴스들입니다.\n")
            ec2_instances = clb_instance_status(elb_cli, args[0])
            
            for instance_id, instance_status in ec2_instances.items():
                # EC2 Id를 Name을 변환한 뒤 print
                instance_name = convert_id_to_name(ec2_cli, instance_id)
                
                print(f"Instance : {instance_id} // Statue : {instance_status} // {instance_name}")
            
        else:
            raise SystemExit(command_help())
    
    # 2. Attach
    elif opt == "--attach" or opt == "-a":
        if len(args) != 3:
            raise SystemExit(command_help())
        
        if not check_zone(args[2]):
            print(f"{args[2]}을 ap-northeast-2a 혹은 ap-northeast-2c로 맞춰주세요.")
            raise SystemExit(command_help())
            
        if clb_exist(elb_cli, args[0]):
        
            # EC2 내역 보여주기
            ec2_instances = find_instances(ec2_cli, args[1], args[2])
            
            print(f"키워드 : {args[1]}에 포함되어 있는 EC2 인스턴스들입니다.\n(가용 영역 : {args[2]})")
            for id, name in ec2_instances.items():
                print(f"Instance Id : {id} // Name : {name}")
            
            # Attach 진행할 것인지 확인
            if approve(f"\nCLB({args[0]})에 위 EC2 인스턴스들을 Attach 하시겠습니까?"):
                # Attach 작업 수행
                instance_list = list(ec2_instances)
                attach_instances(elb_cli, args[0], instance_list)
                
            else:
                # Attach 작업 보류
                raise SystemExit("\n스크립트를 종료합니다.")
            
        
    # 3. Detach
    elif opt == "--detach" or opt == "-d":
        if len(args) != 3:
            raise SystemExit(command_help())
        
        if not check_zone(args[2]):
            print(f"{args[2]}을 ap-northeast-2a 혹은 ap-northeast-2c로 맞춰주세요.")
            raise SystemExit(command_help())
        
        if clb_exist(elb_cli, args[0]):
            
            # EC2 내역 보여주기
            ec2_instances = find_instances(ec2_cli, args[1], args[2])
            
            print(f"키워드 : {args[1]}에 포함되는 EC2 인스턴스들입니다.\n(가용 영역 : {args[2]})")
            for id, name in ec2_instances.items():
                print(f"Instance Id : {id} // Name : {name}")
            
            # Detach 진행할 것인지 확인
            if approve(f"\nCLB({args[0]})로부터 위 EC2 인스턴스들을 Detach 하시겠습니까?"):
                # Detach 작업 수행
                instance_list = list(ec2_instances)
                detach_instances(elb_cli, args[0], instance_list)
                
            else:
                # Detach 작업 보류
                raise SystemExit("\n스크립트를 종료합니다.")
    
    # status, attach, detach 아무것도 포함 X
    else:
        command_help()



if __name__ == "__main__":
    p_name = "my_profile_name"	# 수정 필요 (~/.aws/credentials에 등록되어 있는 Alias)
    
    r_name = "ap-northeast-2"
    main(sys.argv[1:])

### 사용

1. --status [CLB_NAME]

2. --help

3. --attach [CLB_NAME] [INSTANCES_MATCHED] [ZONE]

4. --detach [CLB_NAME] [INSTANCES_MATCHED] [ZONE]