add cloudwatch dashboard

Author: cyphronix

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/app.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import os
 import logging
@@ -57,14 +58,15 @@ def load_kms_key_policies() -> dict:
 def load_cloudwatch_oam_sink_policy() -> dict:
     with open("sra_cloudwatch_oam_sink_policy.json", "r") as file:
         return json.load(file)
-    # ["sra-oam-sink-policy"]["Statement"][0]["Condition"]["ForAnyValue:StringEquals"]["aws:PrincipalOrgID"]
 
 
 def load_sra_cloudwatch_oam_trust_policy() -> dict:
     with open("sra_cloudwatch_oam_trust_policy.json", "r") as file:
         return json.load(file)
-    # ["Statement"][0]["Principal"]["AWS"]
 
+def load_sra_cloudwatch_dashboard() -> dict:
+    with open("sra_cloudwatch_dashboard.json", "r") as file:
+        return json.load(file)
 
 # Global vars
 RESOURCE_TYPE: str = ""
@@ -84,6 +86,7 @@ def load_sra_cloudwatch_oam_trust_policy() -> dict:
 ACCOUNT: str = boto3.client("sts").get_caller_identity().get("Account")
 REGION: str = os.environ.get("AWS_REGION")
 CFN_RESOURCE_ID: str = "sra-bedrock-org-function"
+ALARM_SNS_KEY_ALIAS = "sra-alarm-sns-key"
 
 # CFN_RESPONSE_DATA definition:
 #   dry_run: bool - type of run
@@ -107,7 +110,7 @@ def load_sra_cloudwatch_oam_trust_policy() -> dict:
 KMS_KEY_POLICIES: dict = load_kms_key_policies()
 CLOUDWATCH_OAM_SINK_POLICY: dict = load_cloudwatch_oam_sink_policy()
 CLOUDWATCH_OAM_TRUST_POLICY: dict = load_sra_cloudwatch_oam_trust_policy()
-ALARM_SNS_KEY_ALIAS = "sra-alarm-sns-key"
+CLOUDWATCH_DASHBOARD: dict = load_sra_cloudwatch_dashboard()
 
 # Instantiate sra class objects
 # todo(liamschn): can these files exist in some central location to be shared with other solutions?
@@ -325,6 +328,27 @@ def build_s3_metric_filter_pattern(bucket_names: list, filter_pattern_template:
         s3_filter = s3_filter.replace('&& ($.requestParameters.bucketName = "<BUCKET_NAME_PLACEHOLDER>")', "")
     return s3_filter
 
+def build_cloudwatch_dashboard(dashboard_template, bedrock_accounts, regions):
+    i = 0
+    for bedrock_account in bedrock_accounts:
+        for region in regions:
+            if i == 0:
+                injection_template = copy.deepcopy(dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][2])
+                sensitive_info_template = copy.deepcopy(dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][3])
+            else:
+                dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"].append(copy.deepcopy(injection_template))
+                dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"].append(copy.deepcopy(sensitive_info_template))
+            dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][2 + i][2]["accountId"] = bedrock_account
+            dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][2 + i][2]["region"] = region
+            dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][3 + i][2]["accountId"] = bedrock_account
+            dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][3 + i][2]["region"] = region
+            i += 2
+    dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][0][2]["accountId"] = sts.MANAGEMENT_ACCOUNT
+    dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][0][2]["region"] = sts.HOME_REGION
+    dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][1][2]["accountId"] = sts.MANAGEMENT_ACCOUNT
+    dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][1][2]["region"] = sts.HOME_REGION
+    dashboard_template["sra-bedrock-org"]["widgets"][0]["properties"]["region"] = sts.HOME_REGION
+    return dashboard_template
 
 def create_event(event, context):
     global DRY_RUN_DATA
@@ -704,7 +728,7 @@ def create_event(event, context):
                             "OAMCrossAccountRolePolicyAttach"
                         ] = f"DRY_RUN: Attach {policy_arn} policy to {cloudwatch.CROSS_ACCOUNT_ROLE_NAME} IAM role"
 
-            # 5d) OAM link in bedrock account
+            # 5e) OAM link in bedrock account
             cloudwatch.CWOAM_CLIENT = sts.assume_role(bedrock_account, sts.CONFIGURATION_ROLE, "oam", bedrock_region)
             search_oam_link = cloudwatch.find_oam_link(oam_sink_arn)
             if search_oam_link[0] is False:
@@ -721,6 +745,41 @@ def create_event(event, context):
             else:
                 LOGGER.info("CloudWatch observability access manager link found")
 
+    # 6) Cloudwatch dashboard in security account
+    cloudwatch_dashboard = build_cloudwatch_dashboard(CLOUDWATCH_DASHBOARD, central_observability_params["bedrock_accounts"], central_observability_params["regions"])
+    cloudwatch.CLOUDWATCH_CLIENT = sts.assume_role(SECURITY_ACCOUNT, sts.CONFIGURATION_ROLE, "cloudwatch", sts.HOME_REGION)
+    # sra-bedrock-filter-prompt-injection-metric template ["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][2]
+    # sra-bedrock-filter-sensitive-info-metric template ["sra-bedrock-org"]["widgets"][0]["properties"]["metrics"][3]
+
+    search_dashboard = cloudwatch.find_dashboard(SOLUTION_NAME)
+    if search_dashboard[0] is False:
+        if DRY_RUN is False:
+            LOGGER.info("CloudWatch observability dashboard not found, creating...")
+            cloudwatch.create_dashboard(cloudwatch.SOLUTION_NAME, cloudwatch_dashboard)
+            LIVE_RUN_DATA["CloudWatchDashboardCreate"] = "Created CloudWatch observability dashboard"
+            CFN_RESPONSE_DATA["deployment_info"]["action_count"] += 1
+            CFN_RESPONSE_DATA["deployment_info"]["resources_deployed"] += 1
+            LOGGER.info("Created CloudWatch observability dashboard")
+        else:
+            LOGGER.info("DRY_RUN: CloudWatch observability dashboard not found, creating...")
+            DRY_RUN_DATA["CloudWatchDashboardCreate"] = "DRY_RUN: Create CloudWatch observability dashboard"
+    else:
+        LOGGER.info(f"Cloudwatch dashboard already exists: {search_dashboard[1]}")
+        # check_dashboard = cloudwatch.compare_dashboard(search_dashboard[1], cloudwatch_dashboard)
+        # if check_dashboard is False:
+        #     if DRY_RUN is False:
+        #         LOGGER.info("CloudWatch observability dashboard needs updating...")
+        #         cloudwatch.create_dashboard(cloudwatch.SOLUTION_NAME, cloudwatch_dashboard)
+        #         LIVE_RUN_DATA["OAMDashboardUpdate"] = "Updated CloudWatch observability dashboard"
+        #         CFN_RESPONSE_DATA["deployment_info"]["action_count"] += 1
+        #         CFN_RESPONSE_DATA["deployment_info"]["configuration_changes"] += 1
+        #         LOGGER.info("Updated CloudWatch observability dashboard")
+        #     else:
+        #         LOGGER.info("DRY_RUN: CloudWatch observability dashboard needs updating...")
+        #         DRY_RUN_DATA["OAMDashboardUpdate"] = "DRY_RUN: Update CloudWatch observability dashboard"
+        # else:
+        #     LOGGER.info("CloudWatch observability dashboard is correct")
+
     # End
     # TODO(liamschn): Consider the 256 KB limit for any cloudwatch log message
     if DRY_RUN is False:

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/sra_cloudwatch.py

@@ -407,3 +407,69 @@ def delete_oam_link(self, link_arn: str) -> None:
         except ClientError as e:
             self.LOGGER.info(self.UNEXPECTED)
             raise ValueError(f"Unexpected error executing Lambda function. {e}") from None
+
+    def find_dashboard(self, dashboard_name: str) -> tuple[bool, str]:
+        """Find the CloudWatch dashboard for SRA in the organization.
+
+        Args:
+            dashboard_name (str): name of the dashboard
+
+        Returns:
+            tuple[bool, str]: True if the dashboard is found, False if not, and the dashboard ARN
+        """
+        try:
+            response = self.CLOUDWATCH_CLIENT.list_dashboards()
+            for dashboard in response["DashboardEntries"]:
+                if dashboard["DashboardName"] == dashboard_name:
+                    self.LOGGER.info(f"CloudWatch dashboard {dashboard_name} found: {dashboard['DashboardArn']}")
+                    return True, dashboard["DashboardArn"]
+            self.LOGGER.info(f"CloudWatch dashboard {dashboard_name} not found")
+            return False, ""
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "ResourceNotFoundException":
+                self.LOGGER.info(f"CloudWatch dashboard {dashboard_name} not found. Error code: {error.response['Error']['Code']}")
+                return False, ""
+            else:
+                self.LOGGER.info(self.UNEXPECTED)
+                raise ValueError(f"Unexpected error executing Lambda function. {error}") from None
+
+    def create_dashboard(self, dashboard_name: str, dashboard_body: dict) -> str:
+        """Create the CloudWatch dashboard for SRA in the organization.
+
+        Args:
+            dashboard_name (str): name of the dashboard
+            dashboard_body (str): body of the dashboard
+
+        Returns:
+            str: ARN of the created dashboard
+        """
+        try:
+            response = self.CLOUDWATCH_CLIENT.put_dashboard(
+                DashboardName=dashboard_name,
+                DashboardBody=json.dumps(dashboard_body)
+            )
+            self.LOGGER.info(f"CloudWatch dashboard {dashboard_name} created: {response['DashboardArn']}")
+            return response["DashboardArn"]
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "ResourceAlreadyExistsException":
+                self.LOGGER.info(f"CloudWatch dashboard {dashboard_name} already exists")
+                return self.find_dashboard(dashboard_name)[1]
+            else:
+                self.LOGGER.info(self.UNEXPECTED)
+                raise ValueError(f"Unexpected error executing Lambda function. {error}") from None
+
+    def delete_dashboard(self, dashboard_arn: str) -> None:
+        """Delete the CloudWatch dashboard for SRA in the organization.
+
+        Args:
+            dashboard_arn (str): ARN of the dashboard
+
+        Returns:
+            None
+        """
+        try:
+            self.CLOUDWATCH_CLIENT.delete_dashboards(DashboardNames=[dashboard_arn])
+            self.LOGGER.info(f"CloudWatch dashboard {dashboard_arn} deleted")
+        except ClientError as e:
+            self.LOGGER.info(self.UNEXPECTED)
+            raise ValueError(f"Unexpected error executing Lambda function. {e}") from None

aws_sra_examples/solutions/genai/bedrock_org/lambda/src/sra_cloudwatch_dashboard.json

@@ -0,0 +1,43 @@
+{
+  "sra-bedrock-org": {
+    "widgets": [
+      {
+        "height": 9,
+        "width": 24,
+        "y": 0,
+        "x": 0,
+        "type": "metric",
+        "properties": {
+          "metrics": [
+            [
+              ".",
+              "sra-bedrock-filter-bucket-changes-metric",
+              { "accountId": "<MANAGEMENT_ACCOUNT>", "region": "<HOME_REGION>" }
+            ],
+            [
+              ".",
+              "sra-bedrock-filter-service-changes-metric",
+              { "accountId": "<MANAGEMENT_ACCOUNT>", "region": "<HOME_REGION>" }
+            ],
+            [
+              "sra-bedrock",
+              "sra-bedrock-filter-prompt-injection-metric",
+              { "accountId": "<BEDROCK_ACCOUNT>", "region": "<REGION>" }
+            ],
+            [
+              ".",
+              "sra-bedrock-filter-sensitive-info-metric",
+              { "accountId": "<BEDROCK_ACCOUNT>", "region": "<REGION>" }
+            ]
+          ],
+          "view": "timeSeries",
+          "stacked": false,
+          "region": "<HOME_REGION>",
+          "title": "SRA Bedrock Generative AI Metrics and Alarms",
+          "period": 1,
+          "stat": "Sum"
+        }
+      }
+    ]
+  }
+}