RDS 감사 로그에서 사용자 CUD 쿼리만 골라 Slack으로 알림 받기 — 구독 필터의 한계와 Lambda 우회

운영 DB(RDS for MySQL)는 감사 로그(audit log)를 켜 두면 모든 쿼리가 CloudWatch Logs로 쌓입니다. 여기서 원하는 건 좁습니다. 쓰기 권한이 있는 사람이 운영 DB에 직접 CUD(INSERT·UPDATE·DELETE 등) 쿼리를 날리면, 그 사실을 Slack으로 바로 알림받는 것입니다.

처음에는 "구독 필터에서 거르고 SNS로 Slack에 보내면 끝"이라고 봤습니다. 실제로 해보니 구독 필터만으로는 조건을 다 표현할 수 없어서, SNS 직결을 포기하고 Lambda를 한 단계 넣었습니다. 아래는 그 과정과 막혔던 지점, 그리고 남은 개선 항목입니다.

처음 그린 구성은 이렇습니다.

RDS → CloudWatch Logs → 구독 필터(Subscription Filter) → SNS → Slack

구독 필터에서 "유저가 날린 CUD 쿼리"만 골라 SNS로 보내고, SNS 구독으로 Slack 알림을 받자는 그림입니다. 결론부터 말하면 이 경로로는 원하는 필터링을 완성하지 못했습니다.

막힌 지점은 두 가지였습니다.

첫째, 로그 포맷을 바꾸기 어렵습니다. CloudWatch에 찍히는 감사 로그는 DB 엔진이 직접 만드는 로그라, 애플리케이션 로그처럼 JSON 등 다른 포맷으로 다시 찍게 하려면 별도 공수와 비용이 듭니다. 이미 쌓이고 있는 구조를 그대로 쓰는 게 목표였으므로, 로그는 평문(plain text) CSV 형태 그대로 다뤄야 했습니다.

둘째, 구독 필터의 필터 패턴으로는 조건을 다 표현할 수 없었습니다. CloudWatch 필터 패턴이 지원하는 범위를 정리하면 이렇습니다.

기능	지원
지원되는 정규식 구문	O
정규식과 일치하는 용어 검색	O
비정형(비전형) 로그 이벤트에서 용어 검색	O
JSON 로그 이벤트에서 용어 검색	X (로그가 평문이라 해당 없음)
공백으로 구분된 로그 이벤트에서 용어 검색	X

정규식 자체는 지원하지만, 정작 필요한 NOT 조건을 쓸 수 없다는 점이 걸림돌이었습니다. "특정 유저의 쿼리 중 SELECT는 빼고 CUD만"처럼 부정으로 좁히는 조건을 구독 필터 단계에서 표현하기 어려웠습니다. SNS 자체에는 메시지를 가공하거나 추가로 거르는 단계가 없으므로, 구독 필터에서 못 거르면 그대로 다 Slack에 흘러갑니다.

2. 2차 기획: 구독 필터 → Lambda, 두 단계로 거르기

그래서 SNS 자리에 Lambda를 넣고, 필터링을 두 단계로 나눴습니다.

RDS → CloudWatch Logs → 구독 필터 → Lambda → Slack

1차(구독 필터): CUD 권한이 있는 사용자가 날린 쿼리만 Lambda로 넘긴다. 정규식으로 표현 가능한 만큼만 여기서 거른다.
2차(Lambda): 넘어온 로그 중 CUD 쿼리만 다시 추려 Slack으로 보낸다. NOT이나 prefix 판별처럼 구독 필터에서 못 하는 조건은 코드로 처리한다.

실제로 쌓이는 로그 한 줄은 이런 모양입니다. (값은 예시로 바꿨습니다.)

20250708 02:17:18,ip-172-0-0-0,db_user_a,10.0.0.0,6391819,1740227473,QUERY,service_prod,'INSERT INTO service_prod.products (id, name, price, is_active) VALUES (10028, \'sample\', 1000, 1)',0

콤마(,)로 구분된 CSV이고, 앞쪽 필드의 순서가 고정돼 있습니다. 3번째 필드가 쿼리를 요청한 계정의 id입니다. 이 순서를 이용해 구독 필터 정규식을 작성했습니다.

%^[^,]+,[^,]+,db_user_a,|^[^,]+,[^,]+,db_user_b,|^[^,]+,[^,]+,db_user_c,|^[^,]+,[^,]+,db_user_d,%

[^,]+,는 "콤마가 아닌 문자들 다음에 콤마 하나"로 한 필드를 건너뛴다는 뜻입니다. 앞 두 필드(시각, 서버 IP)를 건너뛴 뒤 3번째 필드가 CUD 권한자 계정 id와 일치하는 줄만 통과시킵니다. |로 권한자 계정을 나열했습니다. 이 단계에서 "권한 있는 사람의 쿼리"까지 좁히고, 쿼리 종류(CUD인지)는 Lambda에서 가립니다.

3. Lambda 구현

3-1. 이벤트 디코딩

CloudWatch Logs가 Lambda로 보내는 이벤트는 base64로 인코딩된 gzip 데이터입니다. 디코딩 → 압축 해제 → JSON 파싱 순서로 풉니다.

# 0-1. base64 디코딩
compressed_payload = base64.b64decode(event['awslogs']['data'])
# 0-2. gzip 압축 해제
uncompressed_payload = gzip.decompress(compressed_payload)
# 0-3. JSON 파싱
payload = json.loads(uncompressed_payload)

log_events = payload["logEvents"]   # ← 실제 로그 줄들이 여기 들어 있음

3-2. 로그 한 줄을 dataclass로 매핑

로그가 CSV라서 단순히 split(',')만 하면 안 됩니다. 쿼리(SQL) 안에 콤마가 들어 있기 때문입니다. 위 예시의 VALUES (10028, 'sample', 1000, 1)만 봐도 콤마가 여러 개입니다. 콤마로 자르면 SQL이 여러 조각으로 쪼개집니다.

그래서 앞쪽 고정 필드 8개(시각·서버 IP·user·클라이언트 IP·connection id·query id·query type·db name)만 순서대로 떼고, 나머지를 다시 합친 뒤 마지막 콤마를 기준으로 SQL과 status(상태 코드)로 나눕니다. status는 항상 줄의 맨 끝에 오는 한 값이라, "마지막 콤마 뒤"로 잡으면 SQL 안의 콤마와 섞이지 않습니다.

@dataclass
class RDSAuditLog:
    datetime: str
    server_ip: str
    user: str
    client_ip: str
    connection_id: str
    query_id: str
    query_type: str
    db_name: str
    sql: str
    status: str

    @classmethod
    def from_log_message(cls, log_message: str):
        fields = log_message.split(',')

        if len(fields) < 10:
            raise Exception(f"Log message format error: {log_message}")

        # 앞의 고정 필드 8개를 순서대로 뗀다
        datetime, server_ip, user, client_ip = fields[0], fields[1], fields[2], fields[3]
        connection_id, query_id, query_type, db_name = fields[4], fields[5], fields[6], fields[7]

        # 나머지를 다시 합쳐 SQL + status 로 둔다 (SQL 안에 콤마가 있으므로)
        sql_and_status = ",".join(fields[8:])
        if sql_and_status.rfind(",") == -1:
            raise Exception(f"Cannot separate SQL and status from: {sql_and_status}")

        # 마지막 콤마 기준: 앞은 SQL, 뒤는 status
        sql_part = sql_and_status[:sql_and_status.rfind(",")].strip().strip("'")
        status = sql_and_status[sql_and_status.rfind(",") + 1:].strip()

        return cls(datetime, server_ip, user, client_ip, connection_id,
                   query_id, query_type, db_name, sql_part, status)

len(fields) < 10 검사는 쿼리에 콤마가 많을수록 필드 수가 늘어난다는 전제 아래, 최소 필드 수(고정 8개 + SQL + status)를 못 채우면 포맷이 깨진 줄로 보고 건너뛰기 위한 것입니다.

3-3. CUD 쿼리만 다시 거르기

구독 필터는 "누가" 날렸는지까지만 좁혔습니다. "무슨" 쿼리인지는 SQL 앞 단어(prefix)로 가립니다. 공백·대소문자를 정규화한 뒤 CUD 계열 키워드로 시작하는지 봅니다.

def sql_startswith_any(self, prefixes) -> bool:
    sql_normalized = self.sql.strip().upper()
    return sql_normalized.startswith(tuple(prefix.upper() for prefix in prefixes))

prefixes = ["TRUNCATE", "INSERT", "UPDATE", "DELETE",
            "DROP", "CREATE", "ALTER", "RENAME", "REPLACE"]

filtered_logs = [log for log in logs if log.sql_startswith_any(prefixes)]

여기까지 통과한 줄만 SELECT 등을 제외한 실제 CUD 쿼리입니다. 구독 필터에서 못 했던 "CUD만"이라는 조건을 코드로 표현한 부분입니다.

3-4. Slack 전송

거른 로그를 시각 | user | sql로 포맷해 Slack Incoming Webhook으로 보냅니다.

if filtered_logs:
    message_lines = [f"- {log.datetime} | {log.user} | {log.sql}" for log in filtered_logs]
    slack_message = "\n".join(message_lines)

    http = urllib3.PoolManager()
    data = {"text": slack_message}
    http.request(
        "POST",
        "https://hooks.slack.com/services/T******/B******/******",  # Webhook URL은 환경변수로 분리 권장
        body=json.dumps(data),
        headers={"Content-Type": "application/json"},
    )
else:
    print("No filtered logs found.")

참고: Webhook URL은 그 자체가 인증 수단이라 코드에 박지 않고 환경변수나 Secrets Manager로 분리하는 편이 안전합니다. 위 예시의 URL은 마스킹한 값입니다.

4. 비용

운영에 올리기 전에 두 항목의 비용을 따져봤습니다.

AWS Lambda: 기존과 동일한 수준으로 봤습니다. 월 100만 건 이하 호출은 프리 티어 범위라 추가 요금이 붙지 않는 구조이고, 실제 한 달 호출 수는 852건(CloudWatch가 집계한 Lambda invocation 횟수)이었습니다.
CloudWatch: "분당 10번씩 쿼리가 발생한다"고 가정해 AWS Pricing Calculator로 추산했을 때 약 2.749 USD가 나왔습니다. Lambda를 만들면 IAM 역할이 기본 설정돼 Lambda 실행 기록이 CloudWatch에도 쌓이므로, 이 로그 적재분도 비용에 포함됩니다.

호출 건수 대비 비용은 크지 않은 편이었습니다.

5. 남은 개선 항목

운영에 붙여 알림은 받고 있지만, 다듬을 부분이 남아 있습니다. 아직 적용 전이라 확정안이 아닌 항목으로 남깁니다.

알림 채널 분리. 어느 Slack 채널로 보낼지 운영/개발 채널 기준을 정해야 합니다.
시각 표기. 로그의 시각이 UTC 기준이라 Slack 메시지에서는 KST로 변환해 보여주는 편이 읽기 편합니다.
SQL 포맷팅. 한 줄로 길게 찍히는 쿼리를 보기 좋게 정리하는 처리가 필요합니다.
성공 여부 표시. 로그 끝의 status 값을 활용해 성공(status = 0)/실패(status ≠ 0)를 함께 표기합니다.
범용 정규식. 지금 구독 필터는 권한자 계정을 |로 일일이 나열합니다. 새 계정이 추가될 때마다 필터를 고쳐야 합니다. 계정명을 공통 접두사로 통일할 수 있다면(app_usr_* 같은 형태) 한 패턴으로 묶을 수 있을지 검토가 필요합니다. (계정 네이밍 규칙과 필터 패턴 동작을 함께 확인해야 하는 항목으로, 아직 검증 전입니다.)
```
%^[^,]+,[^,]+,app_usr_*,[^,]+,[^,]+,[^,]+,QUERY,service_prod,%
```
위험 쿼리 멘션. DROP·DELETE·REPLACE처럼 영향이 큰 쿼리는 알림에 멘션을 붙여 더 눈에 띄게 하는 방안을 고려 중입니다.

마무리

처음에는 "구독 필터 + SNS면 끝"으로 봤지만, 구독 필터의 표현력(특히 NOT 미지원)과 로그가 평문 CSV라는 제약이 겹쳐 그 경로로는 조건을 다 거를 수 없었습니다. "어디까지를 인프라 설정으로 풀고, 어디부터 코드로 풀지"의 경계를 필터 패턴이 표현할 수 있는 범위에 맞춰 나눈 것이 이 작업의 핵심이었습니다. 구독 필터는 정규식으로 표현 가능한 만큼(누가)만 맡고, 나머지 조건(무슨 쿼리인지)은 Lambda로 넘겼습니다.

기존에 쌓이고 있던 로그 구조를 그대로 두고, 그 위에 거르는 단계만 얹어 운영 DB의 직접 쓰기를 가시화했습니다.