[AWS] CloudWatch 로그 S3 백업 파이프라인 구축

CloudWatch에 적재되는 로그를 S3로 백업하여 CloudWatch Logs 사용량 및 비용 줄이기

1. Lambda 생성

1) IAM Role 생성

• Lambda 실행을 위한 Role 생성
• Lambda 함수의 로그 기록 허용

• 가져올 CloudWatch Log Group 및 백업할 s3에 대한 권한 허용

  Role Policy json (Click)

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Sid": "VisualEditor0",
              "Effect": "Allow",
              "Action": [
                  "logs:CreateLogStream",
                  "logs:CreateLogGroup",
                  "logs:PutLogEvents"
              ],
              "Resource": "arn:aws:logs:*:{Account Num}:log-group:*"
          },
          {
              "Sid": "VisualEditor1",
              "Effect": "Allow",
              "Action": [
                  "logs:CreateExportTask",
                  "s3:PutObject",
                  "logs:DescribeLogStreams",
                  "logs:GetLogEvents",
                  "logs:FilterLogEvents"
              ],
              "Resource": [
                  "{Application Logs Group ARN}:*",
                  "{S3 Bucket ARN}/logs/*"
              ]
          }
      ]
  }
  

2) Lambda 정의

• 런타임
  • Python 3.x 선택
• 추가 설정 > 사용자 지정 실행 역할
  • 기존에 생성한 Role 선택
• 구성 > 환경변수
  • DESTINATION_BUCKET: 저장할 S3 버킷 이름
  • GROUP_NAME: 로그 그룹 이름
  • NDAYS: 저장 기준 (ex. 1일 전이면 1)
  • PREFIX: S3 저장 폴더 이름
• 코드 소스
  • 한국 시간(KST)을 기준으로 전날 로그를 계산하여 백업 진행
  • create_export_task
    • 코드 내부에서 사용하는 create_export_task API는 AWS 정책상 계정 및 리전당 동시에 딱 1개의 백업 작업만 실행
    • 백업할 로그 그룹이 여러 개로 늘어나면, 동시 호출 시 LimitExceededException 에러 발생
      → EventBridge 일정을 다르게 쪼개거나, 이전 task가 완료되었는지 상태를 체크(describe_export_tasks)한 뒤 다음 task를 실행하는 순차 로직을 추가
    코드 보기 (Click)

    import boto3
    import os
    import datetime
    from botocore.exceptions import ClientError
        
    # 로그 그룹 이름
    GROUP_NAME = os.environ['GROUP_NAME']
    # 저장 s3 버킷 이름
    DESTINATION_BUCKET = os.environ['DESTINATION_BUCKET']
    # s3 저장 폴더 이름
    PREFIX = os.environ['PREFIX']
    # 저장 기준 ex. 1일전
    NDAYS = os.environ['NDAYS']
    nDays = int(NDAYS)
        
    def lambda_handler(event, context):
      client = boto3.client('logs')
      
      # 한국 시간 기준으로 로그 저장
      KST = datetime.timezone(datetime.timedelta(hours=9))
      # 현재시간
      currentTime = datetime.datetime.now(KST)
      # 현재 날짜에서 목표 날짜 만큼 빼기(저장 시작 날짜)
      markDay = currentTime.date() - datetime.timedelta(days=nDays)
      # 로그 수집 시작 시간(기준 날짜 0시)
      StartDate = datetime.datetime.combine(markDay,datetime.time.min,tzinfo=KST)
      # 기준날짜 다음날
      nextDay = markDay + datetime.timedelta(days=1)
      # 로그 수집 종료 시간
      EndDate = datetime.datetime.combine(nextDay,datetime.time.min,tzinfo=KST)
      
      # 타임스탬프 변환
      fromDate = int(StartDate.timestamp() * 1000)
      toDate = int(EndDate.timestamp() * 1000)
      
      # S3 저장 경로(ex.2026/01/01)
      BUCKET_PREFIX = os.path.join(PREFIX, markDay.strftime('%Y/%m/%d'))
      
      try:
          response = client.create_export_task(
              logGroupName=GROUP_NAME,
              fromTime=fromDate,
              to=toDate,
              destination=DESTINATION_BUCKET,
              destinationPrefix=BUCKET_PREFIX
          )
          print("💾 Task created: %s" % response['taskId'])
          return response
      except ClientError as e:
          print("⚠️ ClientError : %s" % str(e))
          raise e
      except Exception as e:
          print("❌ Unexpected Error: %s" % str(e))
          raise e
    

    • Deploy 및 테스트
      • 기본 설정으로 테스트 이벤트를 생성하여 정상 작동 여부 확인

---

2. S3 버킷 정책

• Lambda Role 및 CloudWatch Logs 서비스에 대한 S3 업로드 권한 허용

• 위치: S3 버킷 > 권한 > 버킷 정책

  Policy json (click)

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Sid": "AllowLambdaToPutLogs",
              "Effect": "Allow",
              "Principal": {
                  "AWS": "{Lambda Role ARN}"
              },
              "Action": "s3:PutObject",
              "Resource": "{S3 Bucket ARN}/logs/*"
          },
          {
              "Sid": "AllowLambdaGetLogs",
              "Effect": "Allow",
              "Principal": {
                  "AWS": "{Lambda Role ARN}"
              },
              "Action": "s3:GetBucketAcl",
              "Resource": "{S3 Bucket ARN}"
          },
          {
              "Sid": "AllowCloudWatchToPutLogs",
              "Effect": "Allow",
              "Principal": {
                  "Service": "logs.ap-northeast-2.amazonaws.com"
              },
              "Action": "s3:PutObject",
              "Resource": "{S3 Bucket ARN}/logs/*",
              "Condition": {
                  "StringEquals": {
                      "aws:SourceAccount": "{Account Num}"
                  }
              }
          },
          {
              "Sid": "AllowCloudWatchGetBucketAcl",
              "Effect": "Allow",
              "Principal": {
                  "Service": "logs.ap-northeast-2.amazonaws.com"
              },
              "Action": "s3:GetBucketAcl",
              "Resource": "{S3 Bucket ARN}",
              "Condition": {
                  "StringEquals": {
                      "aws:SourceAccount": "{Account Num}"
                  }
              }
          }
      ]
  }
  

---

3. EventBridge Scheduler 설정

• 위치: Amazon EventBridge > Scheduler > 일정 생성
• 설정: 시간대 지정 및 Cron 표현식으로 주기 지정 (ex. 매일 새벽 자동 실행)
• 대상: Lambda Invoke > 생성한 Lambda 함수 선택

---

4. CloudWatch Log Group 보존 설정

• S3로 안전하게 백업되므로, 비용 절감을 위해 기존 로그 그룹의 로그 보존 기간(Retention)을 단기(ex. 3일~7일)로 변경

---

5. 결과

• 정상적으로 배치 작업이 완료되면 아래와 같이 S3 버킷에 날짜별 파티셔닝 포맷으로 로그 파일 적재

---

참고자료

• https://devpongi.tistory.com/46
• https://velog.io/@joshuara7235/AWS-Lambda-로-CloudWatch-로그-S3로-마이그레이션-자동화-하기