AWS Resource 자동 태그 설정하기

# AWS 리소스 자동 태그 설정하기

클라우드 플랫폼은 서비스 = 돈 이기때문에 서비스를 생성하고 관리하는 요소가 매우 중요하다.

예를들어 m5.xlarge 인스턴스 3개를 테스트 한답시고 누군가가 생성을 해놓았다.

여러 사용자들은 해당 인스턴스가 어떤 용도로 생성이 된것인지 알수 없기 때문에 리소스 관리자를 제외하고는 관심이 없을 것이다. 그렇게 해당 리소스를 생성한 사람은 2달여간 지방으로 출장을 간다고 생각해보자.

해당 인스턴스는 2달간 아무도 사용하지 않고 적지않은 비용을 고스란히 지불해야하는 일이 발생한다.

적어도 해당 인스턴스에 누가 만들었고 어떤 용도로 생성 했는지 정도의 정보만 입력 되어 있다면 리소스 관리자는 해당 정보를 보고 리소스를 보다 유연하고 경제적으로 관리 할 수 있을 것이다.

# 비용으로 부각된 클라우드 환경

클라우드 환경이 유연하고 확장성이 높으며 다양한 워크로드 환경에 대응이 가능하다는 장점을 가지고 있다는 사실은 많은 사람들이 이미 알고있는 부분이다. 이러한 장점 이면에 비용이라는 커다란 단점이 존재한다. 물론 클라우드를 활용하여 다국적 서비스 및 여러 다양한 워크로드 환경을 효율적으로 운영을 하고 있는 회사도 많이 존재한다. 하지만 회사가 클수록 인원이 많을수록 비약적으로 늘어나는 클라우드 리소스를 타이트하게 관리하기란 여간 쉬운일이 아니다.

# 리소스 태깅을 활용한 리소스 관리 전략

AWS는 리소스 태깅전략을 공식적으로 권장한다. 아래의 URL을 참고.

https://docs.aws.amazon.com/ko_kr/tag-editor/latest/userguide/tagging.html

AWS 리소스에 태그 지정 - AWS 리소스에 태그 지정

물론 위의 URL에서처럼 AWS가 제공하는 방법을 그대로 적용 하지 않아도 된다. 단지 리소스에는 태그가 반드시 존재해야하며 태그 정보는 많으면 많을수록 좋다는 것은 모두가 동의하는 부분이다.

회사나 팀에 상황이나 여건에 맞게 리소스 태깅 전략을 세우고, 해당 전략을 어떻게 구현을 할 것인지는 DevOps팀에서 해야하는 부분이기 때문에 개발자와 운영자 그리고 DevOps 엔지니어간의 상호 협력이 잘 되어야 한다.

(DevOps팀이 없는 경우에는 개발팀이나 운영팀에서 진행한다.)

# 구현한 태깅 전략은 아래와 같다.

태깅 전략을 구현하기 위해서는 여러가지 서비스를 어떤식으로 결합 하는가에 따라서 작업 절차가 달라진다.

구현 프로세스는 아래와 같다.

1. CloudTrail에서 리소스에대한 생성, 삭제, 수정 등 변동이 발생할때 해당 정보를 S3에 저장한다.

2. CloudTrail의 모든 이벤트가 발생 시 EventBridge에서 설정한 API 정보가 있는경우 EventBridge가 실행된다.

3. EventBridge가 실행될때 함께 실행하는 서비스를 지정할 수 있는데 해당 서비스로 AWS Lambda를 선택한다.

4. AWS Lambda는 파이썬으로 코딩된 함수가 Deploy 되어 있어야 한다.

5. 마지막으로 사용자가 EC2를 생성하면 해당 EC2에 태깅이 적용되고, EC2와 함께 생성되는 EBS에도 태깅이 된다.

- 말로만 설명해서 위의 프로세스가 잘 와닿지 않을수도 있다.

# 파이썬 코드를 첨부

- AWS Lambda에서 사용한 파이썬 코드를 첨부한다.

import json
import boto3
import logging
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def aws_create_tag(_aws_region, _instance_id: str, _key_name: str, _tag_value: str):
    try:
        client = boto3.client('ec2', region_name=_aws_region)
        client.create_tags(Resources=[_instance_id, ], Tags=[{'Key': _key_name, 'Value': _tag_value}, ])
        logging.info(f'태깅완료 {_key_name}, {_instance_id}')
    except ClientError:
        logging.info(str(ClientError))
        return False
    return True


# 메인 함수라고 할 수 있다.
# 유저 이름을 찾는다.
def lambda_handler(event, context):
    if 'detail' in event:
        try:
            if 'userIdentity' in event['detail']:
                if event['detail']['userIdentity']['type'] == 'AssumedRole':
                    user_name = str(
                        'UserName: ' + event['detail']['userIdentity']['principalId'].split(':')[1] + ', Role: ' +
                        event['detail']['userIdentity']['sessionContext']['sessionIssuer']['userName'] + ' (role)')
                elif event['detail']['userIdentity']['type'] == 'IAMUser':
                    user_name = event['detail']['userIdentity']['userName']
                elif event['detail']['userIdentity']['type'] == 'Root':
                    user_name = 'root'
                else:
                    logging.info('유저 이름을 정의할 수 없습니다. (unknown iam userIdentity) ')
                    user_name = ''
            else:
                logging.info('유저 이름을 정의할 수 없습니다. (no userIdentity data in cloudtrail')
                user_name = ''
        except Exception as e:
            logging.info('유저 이름을 찾을 수 없습니다., exception: ' + str(e))
            user_name = ''

        # 인스턴스 id를 찾는다.
        try:
            instance_id = [x['instanceId'] for x in event['detail']['responseElements']['instancesSet']['items']]
        except Exception as e:
            instance_id = []
        aws_region = event['detail']['awsRegion']
        client = boto3.client('ec2', region_name=aws_region)
        if instance_id:
            for instance in instance_id:
                # 인스턴스에 태깅을 진행한다.
                instance_api = client.describe_instances(InstanceIds=[instance])
                # 모든 ec2 인스턴스의 태그정보를 가져온다.
                if 'Tags' in instance_api['Reservations'][0]['Instances'][0]:
                    instance_tags = instance_api['Reservations'][0]['Instances'][0]['Tags']
                else:
                    instance_tags = []
                # Name 키 항목에 태깅 여부를 체크 후 태깅을 진행한다.
                if instance_tags:
                    instance_name = [x['Value'] for x in instance_tags if x['Key'] and x['Key'] == 'Name']
                    if instance_name:
                        instance_name = instance_name[0]
                else:
                    instance_name = ''
                # Owner 키 항목에 태깅 여부를 체크 후 태깅을 진행한다.
                if instance_tags:
                    if not any(keys.get('Key') == 'Owner' for keys in instance_tags):
                        logging.info(f'"{instance}에 "Owner" 태그가 존재하지 않으므로 생성한다.')
                        aws_create_tag(aws_region, instance, 'Owner', user_name)
                    else:
                        logging.info(f'{instance}에 "Owner" 태그가 이미 존재한다.')
                else:
                    logging.info(f'{instance}에 "Owner" 태그가 존재하지 않으므로 생성한다.')
                    aws_create_tag(aws_region, instance, 'Owner', user_name)

                # EBS 볼륨에 태싱을 진행한다.
                instance_volumes = [x['Ebs']['VolumeId'] for x in
                                    instance_api['Reservations'][0]['Instances'][0]['BlockDeviceMappings']]
                # DEBS 기존에 태깅 존재 여부를 확인한다.
                for volume in instance_volumes:
                    response = client.describe_volumes(VolumeIds=[volume])
                    volume_tags = [x['Tags'] for x in response['Volumes'] if 'Tags' in x]
                    if volume_tags:
                        if any(keys.get('Key') == 'Owner' and keys.get('Key') == 'AttachedInstance' for keys in
                               volume_tags[0]):
                            logging.info(
                                f'볼륨에 대한 태그가 존재하지 않는다. = {volume}의 인스턴스 = {instance}')
                            continue
                        if not any(keys.get('Key') == 'Owner' for keys in volume_tags[0]):
                            logging.info('"Owner" 태그 정보가 없으므로 생성한다.')
                            aws_create_tag(aws_region, volume, 'Owner', user_name)
                        if not any(keys.get('Key') == 'AttachedInstance' for keys in volume_tags[0]):
                            logging.info('"AttachedInstance" 태그 정보가 없으므로 생성한다.')
                            aws_create_tag(aws_region, volume, 'AttachedInstance',
                                           instance + ' - ' + str(instance_name))
                    else:
                        logging.info(f'{volume}에 태깅이 없다, AttachedInstance 태그 정보를 추가한다.')
                        aws_create_tag(aws_region, volume, 'AttachedInstance', instance + ' - ' + str(instance_name))
                        aws_create_tag(aws_region, volume, 'Owner', user_name)
            return {
                'statusCode': 200,
                'body': json.dumps('작업 완료')
            }
        else:
            return {
                'statusCode': 200,
                'body': json.dumps('데이터가 존재하지 않습니다.')
            }

# 결    론

태깅을 활용하여 리소스를 효율적으로 관리하고, 미사용 리소스 관리를 통해 비용 절감 효과를 볼 수 있다.

 

- 끝 -