def _add_cleanup_resources_lambda()

in cli/src/pcluster/templates/cluster_stack.py [0:0]

• 58 lines of code
• 3 McCabe index (conditional complexity)

    def _add_cleanup_resources_lambda(self):
        """Create Lambda cleanup resources function and its role."""
        cleanup_resources_lambda_role = None
        if self._condition_create_lambda_iam_role():
            s3_policy_actions = ["s3:DeleteObject", "s3:DeleteObjectVersion", "s3:ListBucket", "s3:ListBucketVersions"]

            cleanup_resources_lambda_role = add_lambda_cfn_role(
                scope=self.stack,
                config=self.config,
                function_id="CleanupResources",
                statements=[
                    iam.PolicyStatement(
                        actions=s3_policy_actions,
                        effect=iam.Effect.ALLOW,
                        resources=[
                            self.stack.format_arn(service="s3", resource=self.bucket.name, region="", account=""),
                            self.stack.format_arn(
                                service="s3",
                                resource=f"{self.bucket.name}/{self.bucket.artifact_directory}/*",
                                region="",
                                account="",
                            ),
                        ],
                        sid="S3BucketPolicy",
                    ),
                    get_cloud_watch_logs_policy_statement(
                        resource=self.stack.format_arn(
                            service="logs",
                            account=self.stack.account,
                            region=self.stack.region,
                            resource=get_lambda_log_group_prefix("CleanupResources-*"),
                        )
                    ),
                ],
                has_vpc_config=self.config.lambda_functions_vpc_config,
            )

        cleanup_resources_lambda = PclusterLambdaConstruct(
            scope=self.stack,
            id="CleanupResourcesFunctionConstruct",
            function_id="CleanupResources",
            bucket=self.bucket,
            config=self.config,
            execution_role=(
                cleanup_resources_lambda_role.attr_arn
                if cleanup_resources_lambda_role
                else self.config.iam.roles.lambda_functions_role
            ),
            handler_func="cleanup_resources",
        ).lambda_func

        CustomResource(
            self.stack,
            "CleanupResourcesS3BucketCustomResource",
            service_token=cleanup_resources_lambda.attr_arn,
            properties={
                "ResourcesS3Bucket": self.bucket.name,
                "ArtifactS3RootDirectory": self.bucket.artifact_directory,
                "Action": "DELETE_S3_ARTIFACTS",
            },
        )

        return cleanup_resources_lambda_role, cleanup_resources_lambda