[AWS] CloudWatch Alarm을 Webex로 받아보기

 

CloudWatch Alarm을 전송할 수 있는 채널은 여러 가지가 있습니다.

• Email / SNS : AWS SNS에서 Subcscription으로 전송 가능
• Slack / Teams : SNS → Amazon Q Developer 통해서 전송 가능
• 그 외 미지원 채널은 SNS → Lambda를 활용하면 Webex나 Telegram 등 메신저 애플리케이션으로 전송 가능

미지원 채널인 Webex로 알람을 전송하는 흐름은 다음과 같습니다.

 

 

 

구성 흐름은 다음과 같습니다.
- Webex Developer에서 Bot을 생성하고 알람을 수신할 스페이스에 추가합니다.

- 스페이스 ID는 Bot이 참여한 스페이스에 한해서 조회 가능합니다. (아래는 Powershell)

$TOKEN = "BOT ACCESS TOKEN을 입력하세요"

$response = Invoke-RestMethod -Uri "https://webexapis.com/v1/rooms" `
  -Headers @{ Authorization = "Bearer $TOKEN" }

$response.items | Where-Object { $_.title -like "BOT이 초대된 스페이스명을 입력하세요" } | Select-Object id, title

 

- Lambda를 작성합니다. (Lambda 환경변수로 Bot access token과 스페이스 ID를 설정해야합니다.)

- Lambda Role은 Webex API 호출을 통한 알람 전송이기에  AWSLambdaBasicExecutionRole 로 설정해도 동작합니다.

- Lambda의 Resource-based policy은 구독할 SNS Topic으로 지정합니다.

import json
import logging
import os
import traceback
import urllib3
from datetime import datetime, timezone, timedelta

logger = logging.getLogger()
logger.setLevel(logging.INFO)

http = urllib3.PoolManager()

def lambda_handler(event, context):
    """
    CloudWatch 알람을 Webex Bot으로 전송하는 Lambda 함수

    환경 변수:
    - WEBEX_BOT_TOKEN: Webex Bot Access Token
    - WEBEX_ROOM_ID: Webex Space ID
    """

    # 환경 변수에서 Bot Token과 Space ID 가져오기
    bot_token = os.environ.get('WEBEX_BOT_TOKEN')
    room_id = os.environ.get('WEBEX_ROOM_ID')

    if not bot_token or not room_id:
        logger.error("WEBEX_BOT_TOKEN 또는 WEBEX_ROOM_ID 환경 변수가 설정되지 않았습니다.")
        return {
            'statusCode': 500,
            'body': json.dumps('Webex configuration missing')
        }

    # SNS 메시지 파싱
    try:
        records = event.get('Records', [])
        if not records:
            raise ValueError("SNS Records가 이벤트에 없습니다.")

        sns_message = json.loads(records[0]['Sns']['Message'])

        # 알람 정보 추출
        alarm_name = sns_message.get('AlarmName', 'Unknown')
        alarm_description = sns_message.get('AlarmDescription', 'No description')
        new_state = sns_message.get('NewStateValue', 'Unknown')
        old_state = sns_message.get('OldStateValue', 'Unknown')
        reason = sns_message.get('NewStateReason', 'No reason provided')
        region = sns_message.get('Region', 'Unknown')
        timestamp = sns_message.get('StateChangeTime', datetime.utcnow().isoformat())

        # 메트릭 정보
        trigger = sns_message.get('Trigger', {})
        metric_name = trigger.get('MetricName', 'Unknown')
        namespace = trigger.get('Namespace', 'Unknown')
        threshold = trigger.get('Threshold', 'N/A')
        comparison_operator = trigger.get('ComparisonOperator', 'N/A')

        # Dimensions 정보 추출
        dimensions = trigger.get('Dimensions', [])
        dimension_str = ', '.join([f"{d.get('name', 'Unknown')}={d.get('value', 'Unknown')}" for d in dimensions]) if dimensions else 'N/A'

        # 알람 상태 라벨
        state_label = {
            'ALARM':             '🚨 알람 발생',
            'OK':                '✅ 알람 해제',
            'INSUFFICIENT_DATA': '⚠️ 데이터 부족',
        }
        label = state_label.get(new_state, new_state)

        # 비교 연산자 기호 변환
        operator_symbol = {
            'GreaterThanOrEqualToThreshold': '≥',
            'GreaterThanThreshold':          '>',
            'LessThanOrEqualToThreshold':    '≤',
            'LessThanThreshold':             '<',
        }
        op_symbol = operator_symbol.get(comparison_operator, comparison_operator)

        # 타임스탬프 KST 변환
        try:
            kst = timezone(timedelta(hours=9))
            dt_utc = datetime.strptime(timestamp[:19], '%Y-%m-%dT%H:%M:%S').replace(tzinfo=timezone.utc)
            timestamp_kst = dt_utc.astimezone(kst).strftime('%Y-%m-%d %H:%M:%S KST')
        except Exception:
            timestamp_kst = timestamp

        # Webex 메시지 포맷 (Markdown)
        webex_message = f"""**[{label}] {alarm_name}**

---

**상태 변경:** {old_state} → **{new_state}**
**설명:** {alarm_description}

**메트릭:** {namespace} / {metric_name}
**리소스:** {dimension_str}
**임계값:** {op_symbol} {threshold}
**리전:** {region}
**발생 시간:** {timestamp_kst}

---

**상태 변경 사유**
```
{reason}
```"""

        # Webex Messages API 요청 페이로드
        webex_payload = {
            "roomId": room_id,
            "markdown": webex_message
        }

        # Webex Messages API 호출
        encoded_data = json.dumps(webex_payload).encode('utf-8')

        response = http.request(
            'POST',
            'https://webexapis.com/v1/messages',
            body=encoded_data,
            headers={
                'Authorization': f'Bearer {bot_token}',
                'Content-Type': 'application/json'
            },
            timeout=urllib3.Timeout(connect=5.0, read=10.0)
        )

        logger.info(f"Webex API Response Status: {response.status}")
        logger.info(f"Webex API Response Body: {response.data.decode('utf-8')}")

        if response.status in (200, 201):
            return {
                'statusCode': 200,
                'body': json.dumps('Message sent to Webex successfully')
            }
        else:
            logger.error(f"Webex API 오류: {response.status} - {response.data.decode('utf-8')}")
            return {
                'statusCode': response.status,
                'body': json.dumps(f'Failed to send message to Webex: {response.data.decode("utf-8")}')
            }

    except Exception as e:
        logger.error(f"ERROR: {str(e)}")
        logger.error(traceback.format_exc())
        return {
            'statusCode': 500,
            'body': json.dumps(f'Error processing alarm: {str(e)}')
        }

 

 

 

- SNS Topic에서 Lambda를 구독하고, 알람의 Action을 해당 Topic으로 지정합니다.

- 알람이 발생하면 해당 스페이스로 전달됩니다.

 

 

추가로 고려해 볼 만한 점은 다음과 같습니다.

- Lambda 코드에 따라 다르겠지만, 알람 메시지에는 인스턴스 ID, IP 주소 등 민감한 인프라 정보가 포함될 수 있습니다.

Lambda를  Private Subnet에 배치하고 보안그룹으로 아웃바운드를 Webex API(443/tcp)로만 제한하여 불필요한 외부 통신을 차단합니다.

 

- Lambda 환경변수는 콘솔 접근 권한이 있으면 누구나 조회 가능합니다. Bot Access Token, Space ID는 Secrets Manager에서 관리하고, Lambda는 런타임에 GetSecretValue로 조회하는 방식으로 구성할 수 있습니다.