Created
November 12, 2020 03:52
-
-
Save michaw/449bdaf73a71db4f060b5f95b0038586 to your computer and use it in GitHub Desktop.
AWS Lambda - cloudwatch-alarm-to-slack-python blueprint extended
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
''' | |
Follow these steps to configure the webhook in Slack: | |
1. Navigate to https://<your-team-domain>.slack.com/services/new | |
2. Search for and select "Incoming WebHooks". | |
3. Choose the default channel where messages will be sent and click "Add Incoming WebHooks Integration". | |
4. Copy the webhook URL from the setup instructions and use it in the next section. | |
To encrypt your secrets use the following steps: | |
1. Create or use an existing KMS Key - http://docs.aws.amazon.com/kms/latest/developerguide/create-keys.html | |
2. Expand "Encryption configuration" and click the "Enable helpers for encryption in transit" checkbox | |
3. Paste <SLACK_CHANNEL> into the slackChannel environment variable | |
Note: The Slack channel does not contain private info, so do NOT click encrypt | |
4. Paste <SLACK_HOOK_URL> into the kmsEncryptedHookUrl environment variable and click "Encrypt" | |
Note: You must exclude the protocol from the URL (e.g. "hooks.slack.com/services/abc123"). | |
5. Give your function's role permission for the `kms:Decrypt` action using the provided policy template | |
To Customise: | |
1. You can edit the blocks as you see fit. | |
2. Modify the state_emoji values with any emojis in your Slack instance (this includes custom emojis) | |
''' | |
import boto3 | |
import json | |
import logging | |
import os | |
from base64 import b64decode | |
from urllib.request import Request, urlopen | |
from urllib.error import URLError, HTTPError | |
def _finditem(obj, key): | |
if key in obj: return obj[key] | |
for k, v in obj.items(): | |
if isinstance(v,dict): | |
item = _finditem(v, key) | |
if item is not None: | |
return item | |
elif isinstance(v, list): | |
for i in v: | |
item = _finditem(i, key) | |
if item is not None: | |
return item | |
# The base-64 encoded, encrypted key (CiphertextBlob) stored in the kmsEncryptedHookUrl environment variable | |
ENCRYPTED_HOOK_URL = os.environ['kmsEncryptedHookUrl'] | |
# The Slack channel to send a message to stored in the slackChannel environment variable | |
SLACK_CHANNEL = os.environ['slackChannel'] | |
HOOK_URL = "https://" + boto3.client('kms').decrypt( | |
CiphertextBlob=b64decode(ENCRYPTED_HOOK_URL), | |
EncryptionContext={'LambdaFunctionName': os.environ['AWS_LAMBDA_FUNCTION_NAME']} | |
)['Plaintext'].decode('utf-8') | |
logger = logging.getLogger() | |
logger.setLevel(logging.INFO) | |
def lambda_handler(event, context): | |
logger.info("Event: " + str(event)) | |
message = json.loads(event['Records'][0]['Sns']['Message']) | |
logger.info("Message: " + str(message)) | |
alarm_name = message['AlarmName'] | |
alarm_title = message['AlarmDescription'] | |
old_state = message['OldStateValue'] | |
new_state = message['NewStateValue'] | |
reason = message['NewStateReason'] | |
metric_name = _finditem(message, 'MetricName') | |
namespace = _finditem(message, 'Namespace') | |
dimensions = _finditem(message, 'Dimensions') | |
state_emoji = { | |
'ALARM': 'alert', | |
'OK': 'heavy_check_mark' | |
} | |
slack_message = { | |
'channel': SLACK_CHANNEL, | |
'icon_emoji': ":aws:", | |
'blocks': [ | |
{ | |
'type': "section", | |
"text": { | |
"type": "mrkdwn", | |
"text": ":%s: *%s* :%s:" % (state_emoji[new_state], alarm_title, state_emoji[new_state]) | |
} | |
}, | |
{ | |
"type": "section", | |
"fields": [ | |
{ | |
"type": "mrkdwn", | |
"text": "*New State:*\n*%s*" % (new_state) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Old State:*\n%s" % (old_state) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Service :*\n%s" % (namespace) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Region:*\n%s" % (message['Region']) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Metric:*\n%s" % (metric_name) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Threshold:*\n%s" % (_finditem(message, 'Threshold')) | |
}, | |
{ | |
"type": "mrkdwn", | |
"text": "*Timestamp:*\n%s" % (message['StateChangeTime']) | |
} | |
] | |
}, | |
{ | |
"type": "divider" | |
}, | |
{ | |
'type': "section", | |
"text": { | |
"type": "mrkdwn", | |
"text": "*Dimensions:*" | |
}, | |
'fields': [] | |
}, | |
{ | |
"type": "divider" | |
}, | |
{ | |
"type": "section", | |
"text": { | |
"type": "mrkdwn", | |
"text": "*Details:*\n%s*" % (reason) | |
} | |
} | |
] | |
} | |
if dimensions is not None: | |
for d in dimensions: | |
slack_message['blocks'][3]['fields'].append({ | |
"type": "mrkdwn", | |
"text": "*%s:*\n%s" % (d['name'],d['value']) | |
}) | |
req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8')) | |
try: | |
response = urlopen(req) | |
response.read() | |
logger.info("Message posted to %s", slack_message['channel']) | |
except HTTPError as e: | |
logger.error("Request failed: %d %s", e.code, e.reason) | |
except URLError as e: | |
logger.error("Server connection (%s) failed: %s", HOOK_URL, e.reason) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment