경보 알림 관리 - AWS IoT Events

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

경보 알림 관리

AWS IoT Events는 Lambda 함수를 사용하여 경보 알림을 관리합니다. AWS IoT Events에서 제공하는 Lambda 함수를 사용하거나 새로 만들 수 있습니다.

Lambda 함수 생성

AWS IoT Events는 경보를 통해 이메일 및 SMS 알림을 보내고 받을 수 있는 Lambda 함수를 제공합니다.

요구 사항

경보용 Lambda 함수를 만들려면 다음 요구 사항이 적용됩니다.

  • 이메일 또는 SMS 알림으로 경보를 보내는 경우 AWS Lambda이(가) Amazon SES 및 Amazon SNS와 연동되도록 허용하는 IAM 역할이 있어야 합니다.

    예제 정책:

    { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ses:GetIdentityVerificationAttributes", "ses:SendEmail", "ses:VerifyEmailIdentity" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "sns:Publish", "sns:OptInPhoneNumber", "sns:CheckIfPhoneNumberIsOptedOut" ], "Resource": "*" }, { "Effect": "Deny", "Action": [ "sns:Publish" ], "Resource": "arn:aws:sns:*:*:*" } ] }
  • AWS IoT Events 및 AWS Lambda 모두에 대해 동일한 AWS 리전을 선택해야 합니다. 지원되는 리전은 Amazon Web Services 일반 참조AWS IoT Events 엔드포인트 및 할당량AWS Lambda 엔드포인트 및 할당량을 참조하세요.

Lambda 함수 배포

이 자습서에서는 AWS CloudFormation 템플릿을 사용하여 Lambda 함수를 배포합니다. 이 템플릿은 Lambda 함수가 Amazon SES 및 Amazon SNS와 연동되도록 허용하는 IAM 역할을 자동으로 생성합니다.

다음은 AWS Command Line Interface(AWS CLI)을(를) 사용하여 CloudFormation 스택을 생성하는 방법을 보여줍니다.

  1. 디바이스의 터미널에서 aws --version을 실행하여 AWS CLI가 설치되었는지 확인합니다. 자세한 내용은 AWS Command Line Interface 사용 설명서에서 AWS CLI 설치를 참조하세요.

  2. aws configure list을 실행하여 이 자습서에 대한 모든 AWS 리소스가 포함된 AWS 리전을 구성했는지 AWS CLI를 확인하세요. 자세한 내용은 AWS Command Line Interface 사용 설명서AWS CLI 구성을 참조하세요

  3. CloudFormation 템플릿을 다운로드하세요, notificationLambda.template.yaml.zip.

    참고

    파일을 다운로드하는 데 문제가 있는 경우 CloudFormation 템플릿에서도 템플릿을 사용할 수 있습니다.

  4. 콘텐츠의 압축을 풀고 notificationLambda.template.yaml(으)로 로컬로 저장합니다.

  5. 디바이스에서 터미널을 열고 notificationLambda.template.yaml 파일을 다운로드한 디렉터리로 이동합니다.

  6. CloudFormation 스택을 생성하려면 다음 명령을 실행합니다.

    aws cloudformation create-stack --stack-name notificationLambda-stack --template-body file://notificationLambda.template.yaml --capabilities CAPABILITY_IAM

이 CloudFormation 템플릿을 수정하여 Lambda 함수와 그 동작을 사용자 지정할 수 있습니다.

참고

AWS Lambda은(는) 함수 오류를 두 번 재시도합니다. 함수가 모든 수신 요청을 처리할 만큼 용량이 충분하지 않은 경우 이벤트는 함수로 전송될 때까지 몇 시간 또는 며칠 동안 대기열에서 대기할 수 있습니다. 성공적으로 처리되지 않은 이벤트를 캡처하도록 함수에 대해 배달 못한 메시지 대기열(DLQ)을 구성할 수 있습니다. 자세한 정보는 AWS Lambda 개발자 안내서비동기 호출을 참조하세요.

CloudFormation 콘솔에서 스택을 생성하거나 구성할 수도 있습니다. 자세한 내용을 알아보려면 AWS CloudFormation 사용 설명서스택 작업을 참조하세요.

사용자 지정 Lambda 함수 생성

Lambda 함수를 생성하거나 AWS IoT Events에서 제공하는 함수를 수정할 수 있습니다.

사용자 지정 Lambda 함수를 만들려면 다음 요구 사항이 적용됩니다.

  • Lambda 함수가 지정된 작업을 수행하고 AWS 리소스에 액세스할 수 있도록 허용하는 권한을 추가합니다.

  • AWS IoT Events에서 제공하는 Lambda 함수를 사용하는 경우 Python 3.7 런타임을 선택해야 합니다.

예제 Lambda 함수:

import boto3 import json import logging import datetime logger = logging.getLogger() logger.setLevel(logging.INFO) ses = boto3.client('ses') sns = boto3.client('sns') def check_value(target): if target: return True return False # Check whether email is verified. Only verified emails are allowed to send emails to or from. def check_email(email): if not check_value(email): return False result = ses.get_identity_verification_attributes(Identities=[email]) attr = result['VerificationAttributes'] if (email not in attr or attr[email]['VerificationStatus'] != 'Success'): logging.info('Verification email for {} sent. You must have all the emails verified before sending email.'.format(email)) ses.verify_email_identity(EmailAddress=email) return False return True # Check whether the phone holder has opted out of receiving SMS messages from your account def check_phone_number(phone_number): try: result = sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number) if (result['isOptedOut']): logger.info('phoneNumber {} is not opt in of receiving SMS messages. Phone number must be opt in first.'.format(phone_number)) return False return True except Exception as e: logging.error('Your phone number {} must be in E.164 format in SSO. Exception thrown: {}'.format(phone_number, e)) return False def check_emails(emails): result = True for email in emails: if not check_email(email): result = False return result def lambda_handler(event, context): logging.info('Received event: ' + json.dumps(event)) nep = json.loads(event.get('notificationEventPayload')) alarm_state = nep['alarmState'] default_msg = 'Alarm ' + alarm_state['stateName'] + '\n' timestamp = datetime.datetime.utcfromtimestamp(float(nep['stateUpdateTime'])/1000).strftime('%Y-%m-%d %H:%M:%S') alarm_msg = "{} {} {} at {} UTC ".format(nep['alarmModelName'], nep.get('keyValue', 'Singleton'), alarm_state['stateName'], timestamp) default_msg += 'Sev: ' + str(nep['severity']) + '\n' if (alarm_state['ruleEvaluation']): property = alarm_state['ruleEvaluation']['simpleRule']['inputProperty'] default_msg += 'Current Value: ' + str(property) + '\n' operator = alarm_state['ruleEvaluation']['simpleRule']['operator'] threshold = alarm_state['ruleEvaluation']['simpleRule']['threshold'] alarm_msg += '({} {} {})'.format(str(property), operator, str(threshold)) default_msg += alarm_msg + '\n' emails = event.get('emailConfigurations', []) logger.info('Start Sending Emails') for email in emails: from_adr = email.get('from') to_adrs = email.get('to', []) cc_adrs = email.get('cc', []) bcc_adrs = email.get('bcc', []) msg = default_msg + '\n' + email.get('additionalMessage', '') subject = email.get('subject', alarm_msg) fa_ver = check_email(from_adr) tas_ver = check_emails(to_adrs) ccas_ver = check_emails(cc_adrs) bccas_ver = check_emails(bcc_adrs) if (fa_ver and tas_ver and ccas_ver and bccas_ver): ses.send_email(Source=from_adr, Destination={'ToAddresses': to_adrs, 'CcAddresses': cc_adrs, 'BccAddresses': bcc_adrs}, Message={'Subject': {'Data': subject}, 'Body': {'Text': {'Data': msg}}}) logger.info('Emails have been sent') logger.info('Start Sending SNS message to SMS') sns_configs = event.get('smsConfigurations', []) for sns_config in sns_configs: sns_msg = default_msg + '\n' + sns_config.get('additionalMessage', '') phone_numbers = sns_config.get('phoneNumbers', []) sender_id = sns_config.get('senderId') for phone_number in phone_numbers: if check_phone_number(phone_number): if check_value(sender_id): sns.publish(PhoneNumber=phone_number, Message=sns_msg, MessageAttributes={'AWS.SNS.SMS.SenderID':{'DataType': 'String','StringValue': sender_id}}) else: sns.publish(PhoneNumber=phone_number, Message=sns_msg) logger.info('SNS messages have been sent')

자세한 정보는 AWS Lambda 개발자 설명서AWS Lambda(이)란 무엇입니까?를 참조하세요.

CloudFormation 템플릿

다음 CloudFormation 템플릿을 사용하여 Lambda 함수를 생성하세요.

AWSTemplateFormatVersion: '2010-09-09' Description: 'Notification Lambda for Alarm Model' Resources: NotificationLambdaRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Statement: - Effect: Allow Principal: Service: lambda.amazonaws.com Action: sts:AssumeRole Path: "/" ManagedPolicyArns: - 'arn:aws:iam::aws:policy/AWSLambdaExecute' Policies: - PolicyName: "NotificationLambda" PolicyDocument: Version: "2012-10-17" Statement: - Effect: "Allow" Action: - "ses:GetIdentityVerificationAttributes" - "ses:SendEmail" - "ses:VerifyEmailIdentity" Resource: "*" - Effect: "Allow" Action: - "sns:Publish" - "sns:OptInPhoneNumber" - "sns:CheckIfPhoneNumberIsOptedOut" Resource: "*" - Effect: "Deny" Action: - "sns:Publish" Resource: "arn:aws:sns:*:*:*" NotificationLambdaFunction: Type: AWS::Lambda::Function Properties: Role: !GetAtt NotificationLambdaRole.Arn Runtime: python3.7 Handler: index.lambda_handler Timeout: 300 MemorySize: 3008 Code: ZipFile: | import boto3 import json import logging import datetime logger = logging.getLogger() logger.setLevel(logging.INFO) ses = boto3.client('ses') sns = boto3.client('sns') def check_value(target): if target: return True return False # Check whether email is verified. Only verified emails are allowed to send emails to or from. def check_email(email): if not check_value(email): return False result = ses.get_identity_verification_attributes(Identities=[email]) attr = result['VerificationAttributes'] if (email not in attr or attr[email]['VerificationStatus'] != 'Success'): logging.info('Verification email for {} sent. You must have all the emails verified before sending email.'.format(email)) ses.verify_email_identity(EmailAddress=email) return False return True # Check whether the phone holder has opted out of receiving SMS messages from your account def check_phone_number(phone_number): try: result = sns.check_if_phone_number_is_opted_out(phoneNumber=phone_number) if (result['isOptedOut']): logger.info('phoneNumber {} is not opt in of receiving SMS messages. Phone number must be opt in first.'.format(phone_number)) return False return True except Exception as e: logging.error('Your phone number {} must be in E.164 format in SSO. Exception thrown: {}'.format(phone_number, e)) return False def check_emails(emails): result = True for email in emails: if not check_email(email): result = False return result def lambda_handler(event, context): logging.info('Received event: ' + json.dumps(event)) nep = json.loads(event.get('notificationEventPayload')) alarm_state = nep['alarmState'] default_msg = 'Alarm ' + alarm_state['stateName'] + '\n' timestamp = datetime.datetime.utcfromtimestamp(float(nep['stateUpdateTime'])/1000).strftime('%Y-%m-%d %H:%M:%S') alarm_msg = "{} {} {} at {} UTC ".format(nep['alarmModelName'], nep.get('keyValue', 'Singleton'), alarm_state['stateName'], timestamp) default_msg += 'Sev: ' + str(nep['severity']) + '\n' if (alarm_state['ruleEvaluation']): property = alarm_state['ruleEvaluation']['simpleRule']['inputProperty'] default_msg += 'Current Value: ' + str(property) + '\n' operator = alarm_state['ruleEvaluation']['simpleRule']['operator'] threshold = alarm_state['ruleEvaluation']['simpleRule']['threshold'] alarm_msg += '({} {} {})'.format(str(property), operator, str(threshold)) default_msg += alarm_msg + '\n' emails = event.get('emailConfigurations', []) logger.info('Start Sending Emails') for email in emails: from_adr = email.get('from') to_adrs = email.get('to', []) cc_adrs = email.get('cc', []) bcc_adrs = email.get('bcc', []) msg = default_msg + '\n' + email.get('additionalMessage', '') subject = email.get('subject', alarm_msg) fa_ver = check_email(from_adr) tas_ver = check_emails(to_adrs) ccas_ver = check_emails(cc_adrs) bccas_ver = check_emails(bcc_adrs) if (fa_ver and tas_ver and ccas_ver and bccas_ver): ses.send_email(Source=from_adr, Destination={'ToAddresses': to_adrs, 'CcAddresses': cc_adrs, 'BccAddresses': bcc_adrs}, Message={'Subject': {'Data': subject}, 'Body': {'Text': {'Data': msg}}}) logger.info('Emails have been sent') logger.info('Start Sending SNS message to SMS') sns_configs = event.get('smsConfigurations', []) for sns_config in sns_configs: sns_msg = default_msg + '\n' + sns_config.get('additionalMessage', '') phone_numbers = sns_config.get('phoneNumbers', []) sender_id = sns_config.get('senderId') for phone_number in phone_numbers: if check_phone_number(phone_number): if check_value(sender_id): sns.publish(PhoneNumber=phone_number, Message=sns_msg, MessageAttributes={'AWS.SNS.SMS.SenderID':{'DataType': 'String','StringValue': sender_id}}) else: sns.publish(PhoneNumber=phone_number, Message=sns_msg) logger.info('SNS messages have been sent')