本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
管理警報通知
AWS IoT Events使用 Lambda 函數來管理警示通知。您可以使用提供的 Lambda 函數,AWS IoT Events也可以建立新函數。
建立 Lambda 函數
AWS IoT Events提供 Lambda 函數,可讓警示傳送和接收電子郵件和簡訊通知。
請求
當您建立警示的 Lambda 函數時,適用下列需求:
-
如果您的警示傳送電子郵件或簡訊通知,您必須擁有可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區域」AWS Lambda。如需支援的區域清單,請參閱 AWS IoT EventsAmazon Web Services 一般參考. AWS Lambda
部署一個 Lambda 函數
本教學課程使用AWS CloudFormation範本來部署 Lambda 函數。此範本會自動建立 IAM 角色,讓 Lambda 函數與 Amazon SES 和 Amazon SNS 搭配使用。
以下說明如何使用 AWS Command Line Interface (AWS CLI) 建立 CloudFormation 堆疊。
-
在設備的終端中,運行
aws --version
以檢查是否已安裝AWS CLI. 如需詳細資訊,請參閱《AWS Command Line Interface 使用者指南》中的安裝 AWS CLI。 -
運行
aws configure list
以檢查您是否在AWS區域AWS CLI中配置了具有本教程的所有AWS資源。如需詳細資訊,請參閱《使用指南》AWS CLI中的AWS Command Line Interface〈配置〉 -
下載 CloudFormation 模板,通知。。模板。
注意
如果您在下載檔案時遇到困難,也可以在中使用該範本CloudFormation 範本。
-
解壓縮內容並以
notificationLambda.template.yaml
儲存在本機。 -
開啟裝置上的終端機,然後瀏覽至下載
notificationLambda.template.yaml
檔案的目錄。 -
若要建立 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資源的權限。
-
如果您使用由提供的 Lambda 函數AWS IoT Events,請務必選擇 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')