skip filter deploy if log group doesn't exist

Author: cyphronix

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

@@ -1058,6 +1058,11 @@ def deploy_metric_filters_and_alarms(region: str, accounts: list, resource_prope
                     cloudwatch.CWLOGS_CLIENT = sts.assume_role(acct, sts.CONFIGURATION_ROLE, "logs", region)
                     cloudwatch.CLOUDWATCH_CLIENT = sts.assume_role(acct, sts.CONFIGURATION_ROLE, "cloudwatch", region)
                     LOGGER.info(f"Filter deploy parameter is 'true'; deploying {filter_name} CloudWatch metric filter...")
+                    search_log_group, log_group_arn = cloudwatch.find_log_group(filter_params["log_group_name"])
+                    if search_log_group is False:
+                        LOGGER.info(f"Log group {filter_params['log_group_name']} not found! Skipping {filter_name} filter deployment...")
+                        LIVE_RUN_DATA[f"{filter_name}_CloudWatch"] = f"Log group {filter_params['log_group_name']} not found! Skipped {filter_name} filter deployment."
+                        continue
                     deploy_metric_filter(
                         region, acct, filter_params["log_group_name"], filter_name, filter_pattern, f"{filter_name}-metric", "sra-bedrock", "1"
                     )

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

@@ -568,3 +568,30 @@ def delete_dashboard(self, dashboard_name: str) -> None:
         except ClientError as e:
             self.LOGGER.info(self.UNEXPECTED)
             raise ValueError(f"Unexpected error executing Lambda function. {e}") from None
+
+    def find_log_group(self, log_group_name: str) -> tuple[bool, str]:
+        """Find the CloudWatch log group for SRA in the organization.
+
+        Args:
+            log_group_name (str): name of the log group
+
+        Raises:
+            ValueError: unexpected error
+
+        Returns:
+            tuple[bool, str]: True if the log group is found, False if not, and the log group ARN
+        """
+        try:
+            response = self.CWLOGS_CLIENT.describe_log_groups(logGroupNamePrefix=log_group_name)
+            for log_group in response["logGroups"]:
+                if log_group["logGroupName"] == log_group_name:
+                    self.LOGGER.info(f"CloudWatch log group {log_group_name} found: {log_group['arn']}")
+                    return True, log_group["arn"]
+            self.LOGGER.info(f"CloudWatch log group {log_group_name} not found")
+            return False, ""
+        except ClientError as error:
+            if error.response["Error"]["Code"] == "ResourceNotFoundException":
+                self.LOGGER.info(f"CloudWatch log group {log_group_name} not found. Error code: {error.response['Error']['Code']}")
+                return False, ""
+            self.LOGGER.info(self.UNEXPECTED)
+            raise ValueError(f"Unexpected error executing Lambda function. {error}") from None