The purpose of this blog post is to share my recent experience with AWS budget alerting, using above AWS stack.
We can create an alert to trigger alert message when certain threshold increase at AWS budget.
Then what happen is, that alert message will publish to SNS topic and same message will be consumed by Lambda function.
We can do transform the message and sent to Microsoft Team channel through a Webhook. Please refer this document for more details about this approach.
https://aws.amazon.com/premiumsupport/knowledge-center/sns-lambda-webhooks-chime-slack-teams/
Steps
1. Create AWS Budget Alert
a. Create AWS Budget
https://docs.aws.amazon.com/cost-management/latest/userguide/budgets-create.html
b. Create AWS Alert
2. Create Lambda function
https://docs.aws.amazon.com/lambda/latest/dg/getting-started-create-function.html
Now in our case we are using Python Lambda function. so please refer below section.
https://docs.aws.amazon.com/lambda/latest/dg/lambda-python.html
3. Subscribe Lambda function to SNS topic
https://docs.aws.amazon.com/sns/latest/dg/sns-lambda-as-subscriber.html
4. Analyzing the sample code from below document.
https://aws.amazon.com/premiumsupport/knowledge-center/sns-lambda-webhooks-chime-slack-teams/
#!/usr/bin/python3.6
import urllib3
import json
http = urllib3.PoolManager()
def lambda_handler(event, context):
url = "https://outlook.office.com/webhook/xxxxxxx"
msg = {
"text": event['Records'][0]['Sns']['Message']
}
print('Type of message', type(msg))
encoded_msg = json.dumps(msg).encode('utf-8')
resp = http.request('POST',url, body=encoded_msg)
print("=====After encode======",encoded_msg)
print({
"status_code": resp.status,
"response": resp.data
})
5. Unit Testing
6. Integration Testing
7. Debug with Cloud Watch Logs.
8. Comparison of two logs
As you can see the different of above two logs, the success encoded JSON message, child node wrapped with a double quotes.
9. Improve Lambda Function to Transform Output Message.
Now we have improved the Lambda function to transform the output message to another JSON.
#!/usr/bin/python3.6
import urllib3
import json
import ast
http = urllib3.PoolManager()
def lambda_handler(event, context):
url = "https://outlook.office.com/webhook/xxxxxxx"
alert_dict = event['Records'][0]['Sns']['Message']
if isinstance(alert_dict, str):
print('Type of variable is a string')
alert_dict = ast.literal_eval(alert_dict)
else:
print('Type is variable is not a string')
#Construct new json format
title = "Budget Exceed Alert "
limit = (alert_dict['budget']['budgetLimit']['amount'])
threshold = (alert_dict['action']['actionThreshold']['value'])
print("===========", type(limit))
print("===========", threshold)
actual_amount = (limit * threshold) /100
alert_json = dict(AlertTitle=title,TimeStamp=alert_dict['timeStamp'],BudgetName=alert_dict['budget']['budgetName'],BudgetType=alert_dict['budget']['budgetType'],BudgetAmount=alert_dict['budget']['budgetLimit']['amount'],AlertThreshold=alert_dict['action']['actionThreshold']['value'],ActualAmount=actual_amount,AlertType=alert_dict['action']['notificationType'],Message=alert_dict['message'])
summery = {
"text": json.dumps(alert_json)
}
print("===========", type(summery))
print("=====Before JSON======", summery)
json_object = json.dumps(summery)
encoded_msg = json_object.encode('utf-8')
print("=====After encode======",encoded_msg)
resp = http.request('POST',url, body=encoded_msg)
print({
"status_code": resp.status,
"response": resp.data
})
10. Unit Testing
11. Integration Testing
12. Conclusion
- Now both, unite test and integration test will generate same output JSON.
- The way create JSON is important in Python and it's bit tricky.
summery = {
"text": json.dumps(alert_json)
}
- Did you notice one difference in both testing?
