← 목록

AWS Connect 아웃바운드 콜 테스트 및 CloudWatch 연동 자동 전화 알림 구축

AWS Connect StartOutboundVoiceContact API 테스트부터 CloudWatch → SNS → Lambda → Connect 파이프라인으로 CPU 알람 시 자동 전화 알림까지 구축한 과정

2025-06-10AWS ConnectCloudWatchLambdaMonitoringOutbound Call

개요

AlertHub 애플리케이션에 AWS Connect 전화 알림을 통합하기 전, 먼저 AWS Connect의 아웃바운드 콜 API를 독립적으로 테스트하고, CloudWatch 알람과 연동하여 인프라 이벤트 발생 시 자동으로 전화를 거는 파이프라인을 구축했다.

핵심 목표


API를 통한 자동 전화 걸기 테스트

문제 정의

위급 상황 시 야간에 담당자에게 전화가 가도록 설정해야 했다. 30초 또는 1분 동안 전화를 받지 않으면 A에서 B로 순차 발신하는 구조가 필요했다.

진행 과정에서의 제약 사항


구축 과정

1단계: Amazon Connect 인스턴스 설정

Amazon Connect 인스턴스에서 아웃바운드 통화를 위한 콘택트 플로우를 설정한다.

Amazon Connect 인스턴스 생성: 자격 증명 설정

인스턴스 생성 시 발신 통화 허용을 체크하여 아웃바운드 콜을 활성화한다.

텔레포니 설정: 발신 통화 허용

2단계: AWS SDK 설치 및 자격 증명

AWS SDK를 사용하여 StartOutboundVoiceContact API를 호출한다.

# Python SDK (boto3) 설치
pip install boto3

access key, secret key를 등록하여 AWS 자격 증명을 설정한다.

3단계: Connect Flow 연결 설정 및 API 호출

AWS SDK가 AWS 자격 증명을 참조하고, Connect API를 통해 아웃바운드 콜을 실행한다.

import boto3
 
# Amazon Connect 클라이언트 생성
connect = boto3.client('connect')
 
def start_outbound_call():
    params = {
        'ContactFlowId': '[contact-flow-id]',
        'DestinationPhoneNumber': '[destination-phone]',
        'InstanceId': '[instance-id]',
        'QueueId': '[queue-id]',
        'Attributes': {"Name": "MyAttribute"},
        'SourcePhoneNumber': '[source-phone]'
    }
 
    try:
        response = connect.start_outbound_voice_contact(**params)
        print("Call initiated successfully:", response)
    except Exception as e:
        print("Error initiating call:", str(e))
 
if __name__ == "__main__":
    start_outbound_call()

IAM 유저에 AmazonConnectOutboundPolicy 정책을 생성하여 Amazon Connect 접근 권한을 부여한다.

특정 시간대에만 수행되도록 설정

Connect의 운영 시간(Hours of Operation)을 설정하여 근무 시간 외에만 전화가 가도록 제한한다.

운영 시간 설정: 요일별 시간대 지정


Contact Flow 설정

다수의 후보자: 순차 발신 Flow

여러 담당자에게 순차적으로 전화를 걸기 위한 Contact Flow를 구성했다.

다수 후보자용 Contact Flow 운영 시간 설정

전화 번호 전송(Transfer to phone number) 블록에서 60초 Timeout을 설정하여, 1분 동안 전화를 받지 않으면 자동 종료된다.

Transfer to phone number: 60초 Timeout 설정

단일 수신자 Flow

실제 운영에서는 단일 수신자 Flow를 사용했다.

Contact Flow 구조: Set voice → Check hours of operation → Play prompt

Flow 구성:

  1. Set voice: 음성 설정 (Korean Neural 음성)
  2. Check hours of operation: 근무 시간 확인
  3. Play prompt / Wait: 메시지 재생 및 대기
  4. Success / Error / Timeout 분기 처리

통화 결과 확인: Contact Search API

DisconnectTimestamp 기반 통화 종료 확인

Contact Search API는 기본적으로 완료된 통화의 정보를 제공하므로, 진행 중이거나 완료되지 않은 통화에 대한 직접적인 정보는 다른 방식으로 추적해야 한다.

Contact 객체 내부에 InitiationMethod, Channel, InitiationTimestamp, DisconnectTimestamp, LastUpdateTimestamp 등의 정보가 포함된다.

import boto3
import time
 
connect = boto3.client('connect')
 
def start_outbound_call():
    params = {
        'ContactFlowId': '[contact-flow-id]',
        'DestinationPhoneNumber': '[destination-phone]',
        'InstanceId': '[instance-id]',
        'QueueId': '[queue-id]',
        'Attributes': {"Name": "MyAttribute"},
        'SourcePhoneNumber': '[source-phone]'
    }
 
    try:
        response = connect.start_outbound_voice_contact(**params)
        contact_id = response['ContactId']
        print("Call initiated successfully with Contact ID:", contact_id)
 
        # 통화 상태 확인
        call_status = check_call_status(contact_id, '[instance-id]')
        print("Call status:", call_status)
    except Exception as e:
        print("Error initiating call:", str(e))
 
def check_call_status(contact_id, instance_id):
    for i in range(11):  # 11번 반복 (6초 간격)
        time.sleep(5)
        try:
            response = connect.describe_contact(
                InstanceId=instance_id,
                ContactId=contact_id
            )
            if 'DisconnectTimestamp' in response['Contact']:
                disconnect_time = response['Contact']['DisconnectTimestamp']
                print(f"Call disconnected at: {disconnect_time}")
                return "Call ended"
        except Exception as e:
            print("Error checking call status:", str(e))
 
    return "Call in progress or not connected"
 
if __name__ == "__main__":
    start_outbound_call()

60초 동안 전화가 가도록 설정했으므로 10번 반복해서 6초 간격으로 상태를 확인한다.

테스트 결과

시나리오출력
전화 성공 (수신 후 종료)Call disconnected at: {시간} → Call status: Call ended
전화 실패 (미수신)Call status: Call in progress or not connected

전화 성공 시 로그: Contact ID 생성 및 Call ended 출력

Amazon Connect 콘솔: Contact Details

콘솔에서 통화 상세 정보를 확인할 수 있다. Channel: Voice, Initiation method: API, Disconnect reason 등이 기록된다.

Contact Details: 통화 상세 정보

Contact Search: 통화 이력 목록


CloudWatch 연동: 자동 전화 알림 파이프라인

CloudWatch 알람이 발생하면 SNS를 통해 Lambda를 트리거하고, Lambda가 Connect API를 호출하여 자동으로 전화를 거는 파이프라인을 구축했다.

CloudWatch Alarm → SNS → Lambda → Amazon Connect → Phone

테스트 환경 구성

운영 환경과 동일한 조건에서 테스트하기 위해 별도의 인프라를 구성했다.

항목설정
VPC[Company]-VPC
서브넷[Company]-PRD-MonitoringTest-Public-Subnet (10.x.x.0/28)
보안 그룹[Company]-PRD-MonitoringTest-SG
인스턴스[Company]_PRD_MonitoringTest
라우팅0.0.0.0/0 → Internet Gateway

CloudWatch 알람 생성

인스턴스의 CPUUtilization 메트릭을 기준으로 알람을 설정한다.

CloudWatch 알람 조건 설정: CPU 90% 초과 시 경보

Lambda 함수 IAM 설정

Lambda 함수가 Connect API를 호출할 수 있도록 IAM 역할을 구성한다.

LambdaRole에 부여한 정책:

정책유형설명
AWSLambdaBasicExecutionRoleAWS 관리형Lambda 기본 실행 권한
AmazonConnect_FullAccessAWS 관리형Connect API 호출 권한
Lambda_Invoke_Policy고객 관리형CloudWatch 이벤트에 의한 Lambda 호출 허용

Lambda IAM 역할: 3개 정책 연결

Lambda_Invoke_Policy는 해당 리전, 계정 내의 특정 Lambda 함수에만 InvokeFunction 권한을 부여하는 커스텀 정책이다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "lambda:InvokeFunction",
            "Resource": "arn:aws:lambda:[region]:[account-id]:function:CloudWatchToAWSConnect"
        }
    ]
}

Lambda 함수에 LambdaRole 적용

Lambda 함수 코드

CloudWatch 알람 상태가 ALARM일 때만 Connect 아웃바운드 콜을 실행한다. mention 속성에 알람 설명을 전달하여 TTS로 음성 안내한다.

import boto3
 
def lambda_handler(event, context):
    instance_id = "[instance-id]"
    caller_phone_number = "[source-phone]"
    destination_phone_number = "[destination-phone]"
    contact_flow_id = "[contact-flow-id]"
 
    print(event)
 
    if event['detail']['state']['value'] == 'ALARM':
        connect = boto3.client('connect')
        response = connect.start_outbound_voice_contact(
            DestinationPhoneNumber=destination_phone_number,
            InstanceId=instance_id,
            ContactFlowId=contact_flow_id,
            SourcePhoneNumber=caller_phone_number,
            Attributes={
                'mention': event['detail']['configuration']['description']
            }
        )
        print(f"Call response: {response}")
 
    return {
        'statusCode': 200,
        'body': 'Call initiated successfully'
    }

테스트 실행

EC2 인스턴스에 CloudWatch Agent를 설치하고 CPU 부하를 발생시켜 테스트한다.

# CloudWatch Agent 설치 및 설정
sudo yum install -y amazon-cloudwatch-agent
sudo vim /opt/aws/amazon-cloudwatch-agent/bin/config.json
 
# Agent 시작
sudo /opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent-ctl -a start
 
# CPU 부하 발생 (stress 테스트)
sudo stress --cpu 4 --timeout 1000

CloudWatch Agent 설정에서 30초 간격으로 CPU 메트릭을 수집하도록 구성한다.

{
    "agent": {
        "metrics_collection_interval": 30,
        "logfile": "/var/log/amazon-cloudwatch-agent.log"
    },
    "metrics": {
        "append_dimensions": {
            "InstanceId": "[instance-id]"
        },
        "metrics_collected": {
            "cpu": {
                "measurement": [
                    "cpu_usage_idle",
                    "cpu_usage_user",
                    "cpu_usage_system"
                ],
                "metrics_collection_interval": 30
            }
        }
    }
}

비용 분석

음성 통화 요금

항목요금 (미국 기준)
인바운드 음성 통화 (DID 번호)$0.018 / 분
인바운드 음성 통화 (Toll-Free 번호)$0.012 / 분
아웃바운드 음성 통화 (미국)$0.0065 / 분
아웃바운드 음성 통화 (한국)$0.065 / 분

전화번호 사용 요금

항목요금 (미국 기준)
DID 번호 (일반 전화번호)$0.10 / 번호 / 일
Toll-Free 번호 (무료 전화번호)$0.12 / 번호 / 일

DID의 경우 고객(전화를 거는 쪽)이 통화 비용을 부담하고, Toll-Free의 경우 기업(전화 받는 쪽)이 부담한다.

SMS 전송 요금

항목요금 (미국 기준)
SMS 전송 (미국)$0.00645 / 건
SMS 전송 (한국)$0.047 / 건

기타 비용

항목비용
Connect 인스턴스 생성무료
Flow 생성무료 (통화 발생 시 과금)
IVR 사용일반 통화 요금에 포함
Lambda 호출호출 횟수와 실행 시간에 따라 청구
S3 스토리지통화 기록 저장 시 발생

발신자 번호 제약

한국의 경우 Amazon Connect에서 발신자 번호를 발급하지 않아 미국 Toll-Free 번호를 사용했다.


성과


배운 점

새 채널을 붙일 때는 되는지를 보기 전에 얼마가 드는지와 누가 받는지를 먼저 봐야 한다는 걸 이 테스트에서 배웠다. 전화는 가장 확실한 알림이지만 건당 비용이 붙고, 한국에서는 발신 번호부터 제약이 있었다.

통화가 연결됐는지를 어떻게 확인할지도 문제였다. 발신에 성공했다는 응답은 사람이 받았다는 뜻이 아니다. DisconnectTimestamp로 수신 여부를 판단하게 된 이유다. 알림은 보냈다가 아니라 닿았다까지 확인해야 완성된다.

이 검증 결과는 AlertHub 멀티채널 알림 시스템 구축에서 실제 서비스에 통합됐고, 전화를 어느 심각도에 쓸지는 CloudWatch 알람 기반 통합 모니터링의 채널 분리 기준을 따랐다.


이어 읽기