개요
개발하고 있는 서비스에 실제 유저가 유입되기 시작하면서,
DB에 유저가 가입되는 현황을 실시간으로 체크하고 싶은 니즈가 생겼다.
그래서 팀이 자주 쓰는 Slack 채널을 파고,
알림봇을 만들기로 결정.
여러 방법이 있었다.
API 호출 기준으로 알림을 전송하는 방식도 있었지만
그러면 컨트롤러 로직의 책임이 분산되고 설정 파일이 무거워진다는 문제점이 따라왔다.
그래서 API호출 기준이 아닌, DB에 새로운 데이터가 들어오는 순간을 트리거 포인트로 잡기로 했다.
HOW TO?
[1] Trigger 작성하기
-
조건: User 테이블의 data가 새로 insert되는 상황
-
알림에 들어갈 데이터 : 유저명, 유저이메일, 가입시점
-
방법: 추가된 new data를 별도의 로그 테이블에 기록하는 조건부 트리거를 적용
-
유의사항
-구글 도메인으로 가입한 유저를 제외하고 알림을 띄우기
DELIMITER $$
CREATE TRIGGER trg_log_new_owner
AFTER INSERT ON user
FOR EACH ROW
BEGIN
DECLARE userName VARCHAR(255);
DECLARE userEmail VARCHAR(255);
SELECT name, email
INTO userName, userEmail
FROM user
WHERE id = NEW.user_id;
IF userEmail NOT LIKE '%@google.com' THEN
END IF;
END$$
DELIMITER ;
로그 테이블이 필요한 이유?
Slack은 외부 시스템으로
알림 기능에 필요한 슬랙 웹훅을 호출하려면 http요청이 필요한데,
트리거 내부에서 직접적으로 외부 http호출이 불가능하다.
즉, 트리거 발생과 동시에 알림을 보낼 수 X.
신규 유저 데이터를 감지하고 임시로 쌓아두어 별도로 이를 웹 훅으로 전달하는 프로세스가 필요하다.
그래서 트리거 선언 안에
data insert에 따라 로그 테이블에도 데이터를 쌓도록 로직을 추가했다.
INSERT INTO user_creation_log (
name,
joined_at
) VALUES (
userName,
NEW.joined_at
);DB에 INSERT가 일어난 즉시 알림을 보내는 것이 아니라
로그 테이블에 먼저 기록하고 별도의 프로세스가 이를 읽어 Slack으로 전송하기 때문에 비동기 로직에 속한다.
이로써 가입 로직과 알림 로직의 책임이 분리되고
본 데이터의 안정성과 성능을 유지할 수 있다.
[2] notifier 만들기
로그 테이블을 주기적으로 확인하여 Slack 알림을 전송한 후
알림 발송 완료된 데이터를 삭제하도록 배치 프로세스를 작성한다.
신규 사용자 가입 → 가입처리 API/웹 코드 → user_creation_log 테이블 insert
↓
별도 notifier 프로세스
(5초 간격으로 user_creation_log 테이블 확인하여 알림 발송)
↓
Slack 알림 발송 후, user_creation_log 데이터 삭제
import os
import time
import pymysql
import requests
from urllib.parse import urlparse
(중략)
# 슬랙 메시지 전송 함수
def send_to_slack(name, email, joined_at):
message = {
"text": (
"👥 새로운 유저가 가입했습니다!\n"
f"*이름:* {name}\n"
f"*이메일:* {email or '이메일 없음'}\n"
f"*가입 시간:* {joined_at.strftime('%Y-%m-%d %H:%M:%S')}"
)
}
response = requests.post(SLACK_WEBHOOK_URL, json=message)
return response.status_code == 200
def run():
while True:
try:
conn = pymysql.connect(
host=DB_HOST)
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM user_creation_log ORDER BY joined_at ASC")
logs = cursor.fetchall()
for log in logs:
name = log['name']
email = log.get('email')
joined_at = log.get('joined_at')
success = send_to_slack(name, email, joined_at)
if success:
cursor.execute("DELETE FROM user_creation_log WHERE id = %s", (log['id'],))
conn.commit()
conn.close()
except Exception as e:
print("오류 발생:", e)Polling주기를 5초로 설정해두어 실시간 알림 전송이 가능하게 해두었다.
도커로 서비스 컨테이너와 다같이 띄우고 잘 작동하는지 확인한다.
서비스 보안 상, 이미지는 첨부할 수 없지만
슬랙 알림이 잘 뜨는 것을 확인하였다.
이제 유저가 가입했는지 DB를 계속 조회하지 않아도
SNS알림으로 쉽게 파악할 수 있게 되었다.
