기존 프로젝트를 진행하며 하나의 EC2 내에 Docker를 통한 Airflow 환경을 구축하였습니다.
저의 목적은 Airflow Celery Executor를 사용해 여러 개의 Worker node를 만들어 task를 분산처리하는 것입니다.
그러나, t3-medium 옵션의 EC2 인스턴스 하나로는 메모리 사용량이 감당이 되지않기에,
Worker node를 분리하여 별도의 EC2 인스턴스로 생성하고자 합니다.
또한, CPU 사용량이 90퍼에 도달하면 새로운 EC2 인스턴스(worker node)를 자동으로 생성하고자 합니다.
이를 위해 AWS의 Auto Scaling Group을 생성해줄 것입니다.
Auto Scaling이란,
CPU, 메모리, 디스크, 네트워크 트래픽과 같은 시스템 자원들의 Metric 값을 모니터링하여 서버 사이즈를 자동으로 조절 하는 서비스를 말합니다.
Auto Scaling Group이란?
EC2 인스턴스를 조정 및 관리 목적의 논리 단위로 취급될 수 있도록 구성한 그룹으로,
EC2 인스턴스의 최소 및 최대 인스턴스 수를 지정하여 이 범위안에서 Scale in/out이 일어나게 할 수 있습니다.
시작 템플릿(launch_template) 생성
AWS의 EC2 시작 템플릿으로 가서 EC2 인스턴스 생성할 때와 유사하게
시작 템플릿을 생성해줍니다.
( Auto scaling을 해줘야하므로 지침에 체크를 해줍니다. )
인스턴스 유형은 t3.small로 선택해주었습니다.
일반적인 EC2 인스턴스 생성과 차이가 있는 부분은,
시작 템플릿에서 서브넷을 선택하지 않아야한다는 것과
"종료 시 삭제"에 '예'를 선택해주어야한다는 것입니다.
또한, Private 서브넷에 넣어 줄 예정이므로 퍼블릭 IP는 비활성화하겠습니다.
이렇게 시작 템플릿을 완성했으면 이를 가지고 Auto Scaling 그룹을 생성해줍니다.
Auto Scaling Group 생성
만든 템플릿을 선택하고 Auto Scaling 그룹 생성을 눌러줍니다.
여기서 VPC와 서브넷을 선택해주겠습니다.
여기서 ec2-rds-4라는 보안그룹은 result_backend에 해당하는 RDS에 연결할 수 있도록 아웃바운드 규칙에 해당 RDS에 대한 포트번호를 등록해준 것입니다.
저의 경우에 Airflow worker node가 최소 1개에서 많아야 3개면 충분하므로,
최소 1개, 최대 용량 3으로 설정해주겠습니다.
또한, CPU가 90퍼에 도달했을 때 자동으로 scale out이 이루어지도록
Automatic scaling 설정을 해주겠습니다.
마지막으로 고급 세부 정보에서 메인 EC2 인스턴스와 동일하게 IAM 역할을 부여해주겠습니다.
- S3 Endpoint Gateway를 위한 AmazonS3FullAccess
- SSM 사용을 위한 AmazonSSMManagedInstanceCore
- EC2에서 RDS로 접근하기 위한 AmazonRDSFullAccess
그런데 AutoScalingGroup을 생성하고 나니 어떤 의문점이 들었습니다.
AutoScalingGroup으로 생성된 EC2 인스턴스들은 종료시 삭제가 되는데 어떻게 Airflow 환경을 재구축하고 worker를 실행하지..?
이러한 의문을 해결해보고자 다양한 시도를 해보았습니다.
초기 시도
-
초기 구상
- AutoScalingGroup에 초기에 존재하는 인스턴스 한 개를 기준으로 삼습니다.
- 이 인스턴스에 Airflow 환경 구성 및 명령어만 입력하면 worker를 동작시킬 수 있도록 세팅을 해둡니다.
- 새로운 인스턴스가 추가될 때마다 Lambda 함수를 통해 기준이 되는 인스턴스의 볼륨을 스냅샷을 따서 해당 인스턴스의 볼륨을 교체해줍니다.
- 그 후 명령어를 통해 Airflow worker를 실행시켜줍니다.
이 방법엔 몇 가지 치명적인 문제점들이 있습니다.
-
스냅샷이 너무 많이 쌓이는 문제
인스턴스가 추가로 생성될 때마다 스냅샷을 따므로 이는 너무 비효율적인 행위였습니다.
-
EC2 인스턴스가 중지되면 삭제되는 문제
설정 상 21시에 EC2가 자동으로 중지가 되는데,
AutoScalingGroup의 인스턴스가 중지가 되면 status가 비정상이라고 판단하여 해당 인스턴스를 아예 종료시켜버립니다.따라서, 이를 막기위한 수단이 필요했고
떠올린 방법은 중지되기전에 대기상태로 만드는 것이였습니다.
위 Lambda 함수를 통해 20시 55분에 해당 인스턴스를 대기 상태로 만드는 것은 가능했지만 또 다음 문제가 발생했습니다.
대기상태에서 시작상태로 바꾸는 것도 생성이라고 판단하여 볼륨을 교체하는 Lambda함수가 Trigger되는 것이였습니다.
그러나 아래 문제가 훨씬 더 치명적이였기 때문에 이 문제를 고치지 않고 이 시도를 그만두었습니다.
- 스냅샷이 다른 인스턴스의 볼륨과 교체가 불가능한 문제
충격적이게도 스냅샷은 시작 템플릿이 동일하더라도
인스턴스가 다르면 교체가 되지않았습니다.따라서, 전제부터가 틀린 명제이므로 이 시도를 이렇게 그만두고 구상을 아예 새로해보았습니다.
두 번째 시도
두 번째 시도는 스냅샷이 아니라 AMI 이미지를 활용해보았습니다.
AMI 이미지를 통해 캡쳐를 해보니 스냅샷과 달리 다른 인스턴스에 대해서도 루트 볼륨 교체를 할 수 있었습니다.
- 두 번째 구상
- EC2 인스턴스(t3.small 기준)에 Airflow 환경 구성 및 명령어만 입력하면
worker를 동작시킬 수 있도록 세팅 후 AMI 이미지를 추출
- 새로운 인스턴스가 생성될 때마다 lambda 함수를 호출하여 해당 AMI 이미지로 루트 볼륨을 교체
- S3에 있는 DAG 파일들로 갱신 후 Airflow worker 실행
- EC2 인스턴스(t3.small 기준)에 Airflow 환경 구성 및 명령어만 입력하면
이렇게 AutoScalingGroup을 세팅해주니 문제없이 동작하는 것을 확인할 수 있었습니다.
그러나 인스턴스가 생성될 때마다 매번 루트 볼륨을 교체해주는 것은 상당히 비효율적인 행위입니다.
이를 어떻게 해결하는게 좋을지 고민을 하다가 제가 놓친 부분을 발견하게 되었습니다.
마지막 시도 및 해결 완료
제가 놓쳤던 부분은..
시작 템플릿을 구성할 때, AMI 이미지를 지정할 수 있었다는 것입니다..!!!
그래서 최종적으로 아래와 같이 수정해주었습니다.
-
Airflow Worker 세팅이 완료된 EC2 인스턴스에서 AMI 이미지를 추출
-
해당 AMI를 기반으로 시작 템플릿을 만들어 Auto Scaling Group에 등록
-
CloudWatch에서 CPU 점유율을 감시하다가 수행해야 할 Task가 많아져 CPU 점유율이 40퍼 이상이 되면 알람을 보냄
-
Auto Scaling Group에서 새로운 인스턴스를 생성
-
인스턴스가 새로 생성될 때 Amazon SNS에 Topic을 전송
-
SNS에 Topic이 도착하면 Lambda가 Trigger됨
-
Lambda 함수가 S3의 DAGs를 새로 생성되는 인스턴스에 업데이트 시킨 후 Docker 기반 Airflow worker를 실행시킴
-
수행해야 할 Task가 사라져 CPU 점유율이 5퍼 미만이 되면 해당 인스턴스를 종료
Auto Scaling Group 추가 구축
위 내용들을 기반으로 실제 구현을 해보겠습니다.
먼저, AMI 이미지를 생성해두었다고 가정하겠습니다.
AMI 이미지를 집어넣은 새로운 시작 템플릿을 생성해줍니다.
그 후에 ASG에서 해당 시작 템플릿으로 변경해줍니다.
이제 CloudWatch 경보를 생성해줍니다.
( CPU 사용량을 기준으로 인스턴스를 scale-in, out 하기 위해서 )
이제 CPU 사용률이 40퍼가 넘어가면 scale-out하기 위해 경보가 울리도록 해줄 겁니다.
왜 40퍼로 결정했냐면?
t3.small 기준, task 3개를 동시에 수행하게되면 메모리를 1.8GB 정도를 차지하게 됩니다.
이 정도 수치가 t3.small 기준 최대치입니다.
( 안정성을 위해 EC2에 Swap Memory 설정을 해주었습니다. )
이때의 CPU 사용률이 40퍼 초반이기에 이 수치로 결정했습니다.
반대로 CPU 사용률이 5퍼 미만으로 떨어지면 수행해야할 Task가 없다고 가정하여 인스턴스를 scale-in 해주기 위해 경보를 만들어줄겁니다.
이제 AutoScalingGroup의 동적 크기 조정 정책으로 가서
이 경보들을 등록해줍니다.
다음으로, 인스턴스가 생성될 때 Lambda 함수를 Trigger 시켜줄 SNS 주제를 생성해줄 것입니다.
Amazon SNS에서 주제를 생성해주고,
다시 AutoScalingGroup으로 돌아와 활동 알림을 생성하여 주제를 등록해줄 겁니다.
이제 이 SNS 주제로 Trigger 되는 lambda 함수를 만들어주겠습니다.
lambda 함수의 목적은 S3의 DAG파일들을 EC2 인스턴스의 dags폴더에 배포시키고 worker를 실행시켜주는 것입니다.
from __future__ import print_function
import json, boto3
import time
import logging
from botocore.exceptions import ClientError
"""
( 이전 계획 )
# AutoScaleGroup으로부터 미리 생성해둔 시작 템플릿을 기반으로 인스턴스가 생성되면 SNS 알림이 발생
# 이 SNS 알림으로 Trigger되는 Lambda 함수로,
# 미리 생성해둔 AMI 이미지를 알림의 주체가 되는 인스턴스의 루트 볼륨과 Replace
# 여기서 AMI 이미지는 Worker가 바로 동작할 수 있는 인스턴스를 저장한 이미지
#-------------------------------------------------------------------------------
def lambda_handler(event, context):
print("Received event: " + json.dumps(event, indent=2))
# AutoScaling 이벤트 알림에서 EC2 Instace ID를 추출
message = event['Records'][0]['Sns']['Message']
autoscalingInfo = json.loads(message)
ec2InstanceId = autoscalingInfo['EC2InstanceId']
# 루트 볼륨을 미리 생성해둔 AMI 이미지로 교체
ec2 = boto3.client('ec2', region_name=region)
try :
response_ami = ec2.create_replace_root_volume_task(
InstanceId=ec2InstanceId,
ImageId=amiImageId,
DeleteReplacedRootVolume=True
)
except Exception as e:
logging.error(f"Error replacing root volume to EC2 instance: {str(e)}")
return {'statusCode': 500, 'body': f'Failed_replace : {str(e)}'}
# 교체가 진행되는 5분을 기다림
time.sleep(300)
print(response_ami)
# dags를 갱신하는 lambda 함수를 호출
lambdaCli = boto3.client('lambda')
try:
response_lambda = lambdaCli.invoke(
FunctionName='ariel-1-deploy-dags'
)
print(response_lambda['Payload'].read())
except Exception as e:
logging.error(f"Error updating dags to EC2 instance: {str(e)}")
return {'statusCode': 500, 'body': f'Failed_update : {str(e)}'}
return {
'statusCode': 200,
'body': 'Auto scaling process completed.'
}
-> 처음부터 시작템플릿에 AMI를 설정할 수 있다는 것을 알게되어 계획을 수정
#-------------------------------------------------------------------------------
"""
# AutoScaleGroup으로부터 미리 생성해둔 시작 템플릿을 기반으로 인스턴스가 생성되면 SNS 알림이 발생
# ( 이때, 시작 템플릿에는 docker-compose의 형태로 worker를 바로 실행시킬 수 있는 세팅이 완료되어있음 )
# 이 SNS 알림으로 Trigger되는 Lambda 함수로,
# AWS CLI와 SSM을 통해 인스턴스에 접근하여 S3에 있는 dags를 배포 후 worker를 실행하는 명령을 내림
# 필요한 IAM Policy
"""
- AmazonEC2FullAccess
- AWSLambdaBasicExecutionRole
- AmazonSSMFullAccess
"""
ssm = boto3.client('ssm')
AWS_S3_BUCKET_NAME = 'team-ariel-1-dags'
EC2_PATH = '/home/ubuntu/airflow/dags'
def lambda_handler(event, context):
print("Received event: " + json.dumps(event, indent=2))
# AutoScaling 이벤트 알림에서 EC2 Instace ID를 추출
message = event['Records'][0]['Sns']['Message']
autoscalingInfo = json.loads(message)
ec2InstanceId = autoscalingInfo['EC2InstanceId']
# 인스턴스가 실행되는데 넉넉하게 5분 대기
time.sleep(300)
try :
response = ssm.send_command(
InstanceIds = [ec2InstanceId],
# shell 명령 사용
DocumentName="AWS-RunShellScript",
Parameters={
'commands': [
# S3에 적재된 dags 갱신 & celery worker 실행
f'aws s3 sync --delete s3://{AWS_S3_BUCKET_NAME} {EC2_PATH} && docker compose -f /home/ubuntu/airflow/docker-compose-worker.yaml up -d'
]
}
)
print(response)
except ClientError as e:
print(f"[ERROR] failed to start worker : {str(e)}")
return {'statusCode': 500, 'body': f'failed to start worker : {str(e)}'}
return {
'statusCode': 200,
'body': 'Auto scaling process completed.'
}여기까지 Airflow Celery Executor를 위한 AutoScalingGroup을 만들어보았습니다.
참고
