log weekly august 2024 boto3 ecs task

boto3로 ECS Task 호출하기

2024년 8월 5일 월요일

Today I Learned

날짜

2024년 8월 5일 월요일

내용

ECS 태스크 호출하기

아무리 생각해도 서버 외부에서 호출해야 했다. 람다는 15분 제한 시간에 넘치고, 서버에서 돌리기엔 처음 네이버에 호출하는 순간 CPU 사용률이 80%에 달한다!

다행히 boto3로 ECS Task를 호출하는 방법이 있어 적용했다. 태스크를 직접 요청시에 적용할 수 도 있지만, 미리 정의해놓고 호출해서 원하는 로직만 처리하도록 구현했다. 다만, 일부 파라미터들은 일시적인 환경변수로 지정해줘야 했다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

ecs_client = boto3.client(
    "ecs",
    region_name=region,
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
)

def ecs_task_definition():
    task_setting = dict()
    if DEVELOPMENT:
        task_setting = {
            "cluster": , # 실행할 클러스터의 arn
            "taskDefinition":, # 태스크 정의 이름
            "subnets": [], # VPC 내 서브넷들의 ID
            "securityGroups": , # VPC 내 보안그룹 ID
            "name": , # 컨테이너 이름
        }

    return task_setting

def call_ecs_task(payload: dict):
    """ECS Task 실행"""
    try:
        task_setting = ecs_task_definition()
        response = ecs_client.run_task(
            cluster=task_setting["cluster"],
            taskDefinition=task_setting["taskDefinition"],
            launchType="FARGATE",
            networkConfiguration={
                "awsvpcConfiguration": {
                    "subnets": task_setting["subnets"],
                    "securityGroups": task_setting["securityGroups"],
                    "assignPublicIp": "ENABLED",
                }
            },
            overrides={
                "containerOverrides": [
                    {
                        "name": task_setting["name"],
                        "environment": [
                            {"name": "ACTION", "value": payload["action"]},
                            {"name": "SERVICE_ID", "value": payload["service_id"]},
                            {"name": "DB_USER", "value": get_db_password()["username"]},
                            {
                                "name": "DB_PASSWORD",
                                "value": get_db_password()["password"],
                            },
                            {"name": "DATA", "value": json.dumps(payload["data"])},
                        ],
                    }
                ]
            },
        )
        task_arn = response["tasks"][0]["taskArn"]
        return task_arn
    except Exception as e:
        logging.info("error in call_ecs_task")
        logging.exception(str(e))

기존에 lambda_function.py에 있는 lambda_handler 를 자동으로 호출되었는데, ECS에선 내가 실행할 스크립트를 추가해줘야 했다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# lambda_entrypoint.py
import os
import json
from lambda_function import lambda_handler

def main():
    # 환경 변수에서 데이터를 읽어옵니다.
    event = {
        "action": os.getenv("ACTION"),
        "service_id": os.getenv("SERVICE_ID"),
        "data": json.loads(os.getenv("DATA")),
    }
    context = {}  # Lambda context가 필요하다면 설정

    # lambda_handler 함수 호출
    lambda_handler(event, context)

if __name__ == "__main__":
    main()

그리고 수정한 도커파일

1
2
3
4
5
6
7
8
9
10
11
12

FROM python:3.12-slim

# 애플리케이션 파일 복사
COPY . /app

WORKDIR /app

# 필요한 패키지 설치
RUN pip install --no-cache-dir -r requirements.txt

# 엔트리포인트를 설정하여 컨테이너가 시작될 때 스크립트를 실행하도록 합니다.
ENTRYPOINT ["python", "lambda_entrypoint.py"]

생각보다 쉽게 잘 호출되고 작동해서 당황했다… 왜 돌아가냐..

회고

QA는 해도해도 자꾸 할게 발견된다. 이대로 출시됐다면 끔찍하다 증말..

CATALOG
    weekly