[AWS] Cloudwatch 로그 슬랙 전송하기 #1

해당 포스팅에선 aws의 서비스 로그들이 어떤 식으로 cloudwatch에 저장되고, 어떤 식으로 가공하는지에 대한 방법 위주로 정리

 

 

개요

개발자 분과 협업중 개발자 분이 ECS에서 실행되는 배치 로그에 대한 성공여부 알림을 받을 수 있는지 문의가 왔다.

 

기존에 향로님이 작성한 RDS slow query to slack 포스팅을 참고해서 slow query에 대한 알람을 슬랙으로 전송하고 있었다. 

이 내용이랑 프로세스는 거의 비슷하니 코드만 조금 수정하면 되지 않을까 싶었다.

 

막상 코드를 보니 cloudwatch의 로그 이벤트는 어떤 형태로 저장되고, 어떤 식으로 로그 데이터를 가져오는지, 가져온 로그 데이터를 어떤 식으로 파싱하는지에 대해 아예 모르겠는 것이다...

 

그럴만한게 node.js에서 파이썬으로 변경만하고 그대로 사용했었던터라 무지한 상태나 다름 없었다.

 

cloudwatch 로그에 대해 자세하게 알아보려고 공식 문서, 여러 자료를 찾아봤는데 향로님 코드를 응용한 글이 대부분이었다.

(내가 아직 검색 능력이나 지식이 부족해서 못찾는 것 같긴 함)

 

결국 직접 람다로 테스트 코드를 작성하고 돌려보고 디버깅하면서(디버깅이라고 표현하는게 맞나) 해결하긴 했다.

이 과정에서 알게 된 내용들을 정리해보려고 한다.

 

 

 

람다 코드 작성

먼저 간단하게 테스트할 람다 함수를 작성해보자

def lambda_handler(event, context):
    return {
        'statusCode': 200,
        'body': event
    }

함수명도 프로젝트에 맞게 변경하고 런타임 세팅만 바꿔주면 되는데, 이 부분은 공식 문서에 잘 나와있으니 따로 정리하진 않겠다.

cloudwatch에서 넘어오는 데이터만 확인하기 위해 간단하게 작성했다. 이 코드가 어떻게 동작하는지 테스트 해보자

 

템플릿을 cloudwatch-logs로 지정해주고 테스트해야 한다.

Cloudwatch log는 Base64로 인코딩 + gzip으로 압축된 형식으로 전송된다. 확인할 땐, 역순으로 압축을 풀고 디코딩해주면 된다.

 

테스트해보면 아래와 같이 cloudwatch에서 어떤 형태로 로그가 전달되는지 확인할 수 있다.

콘솔에 찍힌 로그는 현재 gzip으로 압축된 상태니 gzip 압축 해제하는 사이트에서 압축을 해제해보자

테스트 로그의 data부분을 입력해주면 아래와 같이 cloudwatch 로그가 어떤 json 형태로 전달되는지 확인할 수 있다. (JSON 파싱 사이트)

{
  "messageType": "DATA_MESSAGE",
  "owner": "123456789123",
  "logGroup": "testLogGroup",
  "logStream": "testLogStream",
  "subscriptionFilters": [
    "testFilter"
  ],
  "logEvents": [
    {
      "id": "eventId1",
      "timestamp": 1440442987000,
      "message": "[ERROR] First test message"
    },
    {
      "id": "eventId2",
      "timestamp": 1440442987001,
      "message": "[ERROR] Second test message"
    }
  ]
}

 

매번 테스트 할때마다 압축 해제하는건 번거로우니 아래처럼 데이터를 압축해제하는 함수를 만들어서 사용해도 된다.

import json
import base64
from base64 import b64decode
import gzip

def decompress_json(data):
    compressed_payload = base64.b64decode(data)
    return json.loads(gzip.decompress(data))

def lambda_handler(event, context):
    result = decompress_json(event['awslogs']['data'])
    return result

 

이제 이렇게 넘어오는 데이터를 실제 cloudwatch 로그에 쌓이는 형태로 변경해서 테스트와 람다 함수를 작성하면 된다.

 

 

 

테스트 로그 생성

위에서 생성한 CloudWatch 테스트 데이터를 실제 조회하고 로그 형태로 변경해주어야 한다. (ecs 실행 로그)

현재 배포된 ECS가 정상적으로 완료되면 아래와 같이 로그 마지막 줄에 실행이 완료됐다는 로그가 찍힌다.

2024-01-05 07:28:02 INFO  Main:39 - 매일 실행 - dailyAwsBill 완료

 

이제 이 데이터를 파싱하기 위해 위에서 생성한 테스트 데이터에 넣고 gzip 압축해야 한다.

위 포맷에서 logEvents.message 부분에 넣어주고 gzip 압축 사이트에서 압축하면 테스트 데이터를 얻을 수 있다.

압축한 값을 테스트 코드에 넣어주고 테스트 해보면 아래와 같은 결과를 얻을 수 있다.

{
  "awslogs": {
    "data": "H4sIAAAAAAAACk2PzUoDMRSF932KS9YOZNLU+dlNcVoKVhfTnYhEJ5RA5ocktZTSnRvRteDClQVfwjfq+A4mM53qJuSe7+Tk3O0AABVca7bki03NUQzoIlkkd/M0y5Jpis6coVqXXDnkkyEdnQdhZC8dktVyqqpV7ajh2lz2c08zozgr/uGj0HK9utcPStRGVOVESMOVts4biyx0/k5EVrjtA9NHXpo/27Y9LRK5+4Q7Osv9Nr+LEXY/wwpX0acUU0qiMMAYnxzH/d1zggn1sO/hEeAgJmGMCcyuJtcAcybKeBiBB4evffPxDc3L/uft2c45E3KTrPVYSAnN+9Ph89UVBti52oPdLw4o5XZkAQAA"
  }
}

콘솔에서도 확인할 수 있지만 람다 실행 로그가 저장되는 cloudwatch 로그에서도 확인할 수 있다.

(cloudwatch > 로그 > 로그그룹 > /aws/lambda/{lambda_name}

 

이제 테스트 로그를 슬랙에 전송해보도록 하자. (슬랙 API 설정했다는 가정하에 진행)

import json
import base64
from base64 import b64decode
import gzip
from urllib.request import Request, urlopen

slack_webhook_url = {slack_url}

def decompress_json(data):
    decompress_data = base64.b64decode(data)
    return json.loads(gzip.decompress(decompress_data))

def lambda_handler(event, context):
    result = decompress_json(event['awslogs']['data'])

    # slack message
    slack_message = {
        'text' : "TEST : %s" % result['logEvents'][0]['message']
    }

    req = Request(slack_webhook_url, json.dumps(slack_message).encode('utf-8'))
    
    response = urlopen(req)
    response.read()

위에서 생성된 테스트 로그에서 message 부분에 해당하는 내용만 슬랙에 던져주는 코드이다.

테스트 해보면 아래처럼 슬랙에 알람이 온다.

이제 위 코드에서 각자 필요한 로그를 추출하게 끔 코드만 커스텀해서 사용해주면 된다.

여기까지 성공하면 사실상 거의 끝났다고 봐도 된다. 마지막으로 로그가 쌓이는 cloudwatch와 연동만 해주면 된다.

 

 

 

Cloudwatch & Lambda 연동

cloudwatch에는 구독기능이라는게 있어 람다와 연동이 매우매우 간편하다. 

ECS 로그가 떨어지는 로그그룹으로 이동해서 작업 > 구독필터 > 람다 구독 필터 생성만 해주면 된다.

 

연동할 람다 함수를 선택해주고 (위에서 생성한 람다) , 로그 형식과 식별 가능한 구독 필터 이름을 지정해주고 생성하면 끝

구독 필터는 필요한 경우 사용하면 된다.

 

참고로 정규표현식은 지원하지 않으니 aws 공식 문서를 참고하여 전용 패턴을 써야 한다. (아직도 지원안하는지 모르겠다)

 

이제 로그가 생성되는 로그그룹에 새로운 로그가 생성될 때 마다 슬랙으로 알림이 발송된다 !