feat: Added support for AWS Health events

functions/events/aws_health_event.json

@@ -0,0 +1,34 @@
+{
+  "version": "0",
+  "id": "121345678-1234-1234-1234-123456789012",
+  "detail-type": "AWS Health Event",
+  "source": "aws.health",
+  "account": "123456789012",
+  "time": "2016-06-05T06:27:57Z",
+  "region": "us-west-2",
+  "resources": [
+    "i-abcd1111"
+  ],
+  "detail": {
+    "eventArn": "arn:aws:health:us-west-2::event/AWS_EC2_INSTANCE_STORE_DRIVE_PERFORMANCE_DEGRADED_90353408594353980",
+    "service": "EC2",
+    "eventTypeCode": "AWS_EC2_INSTANCE_STORE_DRIVE_PERFORMANCE_DEGRADED",
+    "eventTypeCategory": "issue",
+    "startTime": "Sat, 05 Jun 2016 15:10:09 GMT",
+    "eventDescription": [
+      {
+        "language": "en_US",
+        "latestDescription": "A description of the event will be provided here"
+      }
+    ],
+    "affectedEntities": [
+      {
+        "entityValue": "i-abcd1111",
+        "tags": {
+          "stage": "prod",
+          "app": "my-app"
+        }
+      }
+    ]
+  }
+}

functions/notify_slack.py

@@ -191,6 +191,80 @@ def format_guardduty_finding(message: Dict[str, Any], region: str) -> Dict[str,
     }
 
 
+class AwsHealthCategory(Enum):
+    """Maps AWS Health eventTypeCategory to Slack message format color
+
+    eventTypeCategory
+        The category code of the event. The possible values are issue,
+        accountNotification, and scheduledChange.
+    """
+
+    accountNotification = "#777777"
+    scheduledChange = "warning"
+    issue = "danger"
+
+
+def format_aws_health(message: Dict[str, Any], region: str) -> Dict[str, Any]:
+    """
+    Format AWS Health event into Slack message format
+
+    :params message: SNS message body containing AWS Health event
+    :params region: AWS region where the event originated from
+    :returns: formatted Slack message payload
+    """
+
+    aws_health_url = (
+        f"https://phd.aws.amazon.com/phd/home?region={region}#/dashboard/open-issues"
+    )
+    detail = message["detail"]
+    resources = message.get("resources", "<unknown>")
+    service = detail.get("service", "<unknown>")
+
+    return {
+        "color": AwsHealthCategory[detail["eventTypeCategory"]].value,
+        "text": f"New AWS Health Event for {service}",
+        "fallback": f"New AWS Health Event for {service}",
+        "fields": [
+            {"title": "Affected Service", "value": f"`{service}`", "short": True},
+            {
+                "title": "Affected Region",
+                "value": f"`{message.get('region')}`",
+                "short": True,
+            },
+            {
+                "title": "Code",
+                "value": f"`{detail.get('eventTypeCode')}`",
+                "short": False,
+            },
+            {
+                "title": "Event Description",
+                "value": f"`{detail['eventDescription'][0]['latestDescription']}`",
+                "short": False,
+            },
+            {
+                "title": "Affected Resources",
+                "value": f"`{', '.join(resources)}`",
+                "short": False,
+            },
+            {
+                "title": "Start Time",
+                "value": f"`{detail.get('startTime', '<unknown>')}`",
+                "short": True,
+            },
+            {
+                "title": "End Time",
+                "value": f"`{detail.get('endTime', '<unknown>')}`",
+                "short": True,
+            },
+            {
+                "title": "Link to Event",
+                "value": f"{aws_health_url}",
+                "short": False,
+            },
+        ],
+    }
+
+
 def format_default(
     message: Union[str, Dict], subject: Optional[str] = None
 ) -> Dict[str, Any]:
@@ -265,6 +339,10 @@ def get_slack_message_payload(
         )
         attachment = notification
 
+    elif isinstance(message, Dict) and message.get("detail-type") == "AWS Health Event":
+        notification = format_aws_health(message=message, region=message["region"])
+        attachment = notification
+
     elif "attachments" in message or "text" in message:
         payload = {**payload, **message}
 

functions/snapshots/snap_notify_slack_test.py

@@ -4,8 +4,56 @@
 
 from snapshottest import Snapshot
 
+
 snapshots = Snapshot()
 
+snapshots[
+    "test_event_get_slack_message_payload_snapshots event_aws_health_event.json"
+] = [
+    {
+        "attachments": [
+            {
+                "color": "danger",
+                "fallback": "New AWS Health Event for EC2",
+                "fields": [
+                    {"short": True, "title": "Affected Service", "value": "`EC2`"},
+                    {"short": True, "title": "Affected Region", "value": "`us-west-2`"},
+                    {
+                        "short": False,
+                        "title": "Code",
+                        "value": "`AWS_EC2_INSTANCE_STORE_DRIVE_PERFORMANCE_DEGRADED`",
+                    },
+                    {
+                        "short": False,
+                        "title": "Event Description",
+                        "value": "`A description of the event will be provided here`",
+                    },
+                    {
+                        "short": False,
+                        "title": "Affected Resources",
+                        "value": "`i-abcd1111`",
+                    },
+                    {
+                        "short": True,
+                        "title": "Start Time",
+                        "value": "`Sat, 05 Jun 2016 15:10:09 GMT`",
+                    },
+                    {"short": True, "title": "End Time", "value": "`<unknown>`"},
+                    {
+                        "short": False,
+                        "title": "Link to Event",
+                        "value": "https://phd.aws.amazon.com/phd/home?region=us-west-2#/dashboard/open-issues",
+                    },
+                ],
+                "text": "New AWS Health Event for EC2",
+            }
+        ],
+        "channel": "slack_testing_sandbox",
+        "icon_emoji": ":aws:",
+        "username": "notify_slack_test",
+    }
+]
+
 snapshots[
     "test_event_get_slack_message_payload_snapshots event_cloudwatch_alarm.json"
 ] = [