def enable_cloudwatch_alarm_notifications()

in src/credentials_rotators/npm/stacks/common_stack.py [0:0]

• 13 lines of code
• 5 McCabe index (conditional complexity)

    def enable_cloudwatch_alarm_notifications(self, rotator_lambda, secret_id):
        """
        Adds a cloudwatch alarm to monitor the error metrics.
        Subscribes the given emails to receive email alerts.
        Args:
            rotator_lambda (Function): The lambda function used for rotating a secret
            secret_id (string): The identifier corresponding to the secret in the secrets_config.json file
        """
        # create an sns topic for the rotator lambda monitoring
        alarm_sns_topic_id = f'{secret_id}_alarm_sns_topic'
        alarm_sns_topic = Topic(self,
                                alarm_sns_topic_id)

        # subscribe the given emails to the sns topic
        subscription_emails = get_alarm_subscriptions(secret_id)
        for email in subscription_emails:
            alarm_sns_topic.add_subscription(EmailSubscription(email))

        # add errors metric alarm to rotator lambda
        # should send and email notification if the errors metric >= threshold every single time(evaluation_periods)
        errors_alarm_id = f'{secret_id}_errors_alarm'
        errors_alarm = rotator_lambda.metric_errors().create_alarm(self,
                                                                   errors_alarm_id,
                                                                   threshold=1,
                                                                   evaluation_periods=1)
        errors_alarm.add_alarm_action(SnsAction(alarm_sns_topic))