Merge pull request #97 from whdalsrnt/master

Implement Budget Alarm

src/spaceone/cost_analysis/error/budget.py

@@ -46,3 +46,7 @@ class ERROR_PROVIDER_FILTER_IS_EMPTY(ERROR_INVALID_ARGUMENT):
 
 class ERROR_BUDGET_ALREADY_EXIST(ERROR_INVALID_ARGUMENT):
     _message = 'Budget already exist. (data_source_id = {data_source_id}, target = {target})'
+
+
+class ERROR_NOTIFICATION_IS_NOT_SUPPORTED_IN_PROJECT_GROUP(ERROR_INVALID_ARGUMENT):
+    _message = 'Notification is not supported in project group. (target = {target})'

src/spaceone/cost_analysis/manager/__init__.py

@@ -12,3 +12,4 @@
 from spaceone.cost_analysis.manager.secret_manager import SecretManager
 from spaceone.cost_analysis.manager.job_manager import JobManager
 from spaceone.cost_analysis.manager.job_task_manager import JobTaskManager
+from spaceone.cost_analysis.manager.notification_manager import NotificationManager

src/spaceone/cost_analysis/manager/budget_usage_manager.py

@@ -3,9 +3,12 @@
 from dateutil.rrule import rrule, MONTHLY
 
 from spaceone.core.manager import BaseManager
+from spaceone.core import utils
 from spaceone.cost_analysis.manager.identity_manager import IdentityManager
+from spaceone.cost_analysis.manager.notification_manager import NotificationManager
 from spaceone.cost_analysis.manager.cost_manager import CostManager
 from spaceone.cost_analysis.manager.budget_manager import BudgetManager
+from spaceone.cost_analysis.manager.data_source_manager import DataSourceManager
 from spaceone.cost_analysis.model.budget_usage_model import BudgetUsage
 from spaceone.cost_analysis.model.budget_model import Budget
 
@@ -18,6 +21,9 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.budget_mgr: BudgetManager = self.locator.get_manager('BudgetManager')
         self.budget_usage_model: BudgetUsage = self.locator.get_model('BudgetUsage')
+        self.identity_mgr: IdentityManager = self.locator.get_manager('IdentityManager')
+        self.notification_mgr: NotificationManager = self.locator.get_manager('NotificationManager')
+        self.data_source_mgr: DataSourceManager = self.locator.get_manager('DataSourceManager')
 
     def create_budget_usages(self, budget_vo: Budget):
         if budget_vo.time_unit == 'TOTAL':
@@ -84,6 +90,144 @@ def update_budget_usage(self, domain_id, data_source_id):
         budget_vos = self.budget_mgr.filter_budgets(domain_id=domain_id, data_source_id=data_source_id)
         for budget_vo in budget_vos:
             self.update_cost_usage(budget_vo.budget_id, domain_id)
+            self.notify_budget_usage(budget_vo)
+
+    def notify_budget_usage(self, budget_vo: Budget):
+        budget_id = budget_vo.budget_id
+        domain_id = budget_vo.domain_id
+        current_month = datetime.now().strftime('%Y-%m')
+        updated_notifications = []
+        is_changed = False
+        for notification in budget_vo.notifications:
+            if current_month not in notification.notified_months:
+                unit = notification.unit
+                threshold = notification.threshold
+                notification_type = notification.notification_type
+                is_notify = False
+
+                if budget_vo.time_unit == 'MONTHLY':
+                    budget_usage_vos = self.filter_budget_usages(budget_id=budget_id, domain_id=domain_id)
+                    total_budget_usage = sum([budget_usage_vo.cost for budget_usage_vo in budget_usage_vos])
+                    budget_limit = budget_vo.limit
+                else:
+                    budget_usage_vos = self.filter_budget_usages(budget_id=budget_id, domain_id=domain_id,
+                                                                 date=current_month)
+                    total_budget_usage = budget_usage_vos[0].cost
+                    budget_limit = budget_usage_vos[0].limit
+
+                if budget_limit == 0:
+                    _LOGGER.debug(f'[notify_budget_usage] budget_limit is 0: {budget_id}')
+                    continue
+
+                budget_percentage = round(total_budget_usage / budget_limit * 100, 2)
+
+                if unit == 'PERCENT':
+                    if budget_percentage > threshold:
+                        is_notify = True
+                        is_changed = True
+                else:
+                    if total_budget_usage > threshold:
+                        is_notify = True
+                        is_changed = True
+
+                if is_notify:
+                    _LOGGER.debug(f'[notify_budget_usage] notify event: {budget_id} (level: {notification_type})')
+                    self._notify_message(budget_vo, current_month, total_budget_usage, budget_limit,
+                                         budget_percentage, threshold, unit, notification_type)
+
+                    updated_notifications.append({
+                        'threshold': threshold,
+                        'unit': unit,
+                        'notification_type': notification_type,
+                        'notified_months': notification.notified_months + [current_month]
+                    })
+                else:
+                    if unit == 'PERCENT':
+                        _LOGGER.debug(f'[notify_budget_usage] skip notification: {budget_id} '
+                                      f'(usage percent: {budget_percentage}%, threshold: {threshold}%)')
+                    else:
+                        _LOGGER.debug(f'[notify_budget_usage] skip notification: {budget_id} '
+                                      f'(usage cost: {total_budget_usage}, threshold: {threshold})')
+
+                    updated_notifications.append(notification.to_dict())
+
+            else:
+                updated_notifications.append(notification.to_dict())
+
+        if is_changed:
+            budget_vo.update({'notifications': updated_notifications})
+
+    def _notify_message(self, budget_vo: Budget, current_month, total_budget_usage, budget_limit, budget_percentage,
+                        threshold, unit, notification_type):
+        data_source_name = self.data_source_mgr.get_data_source(budget_vo.data_source_id, budget_vo.domain_id).name
+        project_name = self.identity_mgr.get_project_name(budget_vo.project_id, budget_vo.domain_id)
+
+        if unit == 'PERCENT':
+            threshold_str = f'{int(threshold)}%'
+        else:
+            threshold_str = format(int(threshold), ',')
+
+        description = f'Please check the budget usage and increase the budget limit if necessary.\n\n'
+        description += (f'Budget Usage (Currency: {budget_vo.currency}): \n'
+                        f'- Usage Cost: {format(round(total_budget_usage, 2), ",")}\n'
+                        f'- Limit: {format(budget_limit, ",")}\n'
+                        f'- Percentage: {budget_percentage}%\n'
+                        f'- Threshold: > {threshold_str}\n')
+
+        if budget_vo.time_unit == 'MONTHLY':
+            period = f'{current_month} ~ {current_month}'
+        else:
+            period = f'{budget_vo.start} ~ {budget_vo.end}'
+
+        message = {
+            'resource_type': 'identity.Project',
+            'resource_id': budget_vo.project_id,
+            'notification_type': 'WARNING' if notification_type == 'WARNING' else 'ERROR',
+            'topic': 'cost_analysis.Budget',
+            'message': {
+                'title': f'Budget usage exceeded - {budget_vo.name}',
+                'description': description,
+                'tags': [
+                    {
+                        'key': 'Budget ID',
+                        'value': budget_vo.budget_id,
+                        'options': {
+                            'short': True
+                        }
+                    },
+                    {
+                        'key': 'Budget Name',
+                        'value': budget_vo.name,
+                        'options': {
+                            'short': True
+                        }
+                    },
+                    {
+                        'key': 'Data Source',
+                        'value': data_source_name,
+                        'options': {
+                            'short': True
+                        }
+                    },
+                    {
+                        'key': 'Period',
+                        'value': period,
+                        'options': {
+                            'short': True
+                        }
+                    },
+                    {
+                        'key': 'Project',
+                        'value': project_name,
+                    }
+                ],
+                'occurred_at': utils.datetime_to_iso8601(datetime.utcnow()),
+            },
+            'notification_level': 'ALL',
+            'domain_id': budget_vo.domain_id
+        }
+
+        self.notification_mgr.create_notification(message)
 
     def filter_budget_usages(self, **conditions):
         return self.budget_usage_model.filter(**conditions)

src/spaceone/cost_analysis/manager/cost_manager.py

@@ -203,7 +203,7 @@ def _check_date_range(self, query):
                 raise ERROR_INVALID_DATE_RANGE(start=start_str, end=end_str,
                                                reason='Request up to a maximum of 1 month.')
 
-            if start + relativedelta(months=12) < now:
+            if start + relativedelta(months=12) < now.replace(day=1):
                 raise ERROR_INVALID_DATE_RANGE(start=start_str, end=end_str,
                                                reason='For DAILY, you cannot request data older than 1 year.')
 
@@ -212,11 +212,11 @@ def _check_date_range(self, query):
                 raise ERROR_INVALID_DATE_RANGE(start=start_str, end=end_str,
                                                reason='Request up to a maximum of 12 months.')
 
-            if start + relativedelta(months=36) < now:
+            if start + relativedelta(months=36) < now.replace(day=1):
                 raise ERROR_INVALID_DATE_RANGE(start=start_str, end=end_str,
                                                reason='For MONTHLY, you cannot request data older than 3 years.')
         elif granularity == 'YEARLY':
-            if start + relativedelta(years=3) < now:
+            if start + relativedelta(years=3) < now.replace(month=1, day=1):
                 raise ERROR_INVALID_DATE_RANGE(start=start_str, end=end_str,
                                                reason='For YEARLY, you cannot request data older than 3 years.')
 

src/spaceone/cost_analysis/manager/identity_manager.py

@@ -4,7 +4,6 @@
 from spaceone.core.manager import BaseManager
 from spaceone.core.connector.space_connector import SpaceConnector
 from spaceone.core import cache
-from spaceone.cost_analysis.error import *
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -19,6 +18,15 @@ def __init__(self, *args, **kwargs):
     def list_projects(self, query, domain_id):
         return self.identity_connector.dispatch('Project.list', {'query': query, 'domain_id': domain_id})
 
+    @cache.cacheable(key='project-name:{domain_id}:{project_id}', expire=300)
+    def get_project_name(self, project_id, domain_id):
+        try:
+            project_info = self.get_project(project_id, domain_id)
+            return f'{project_info["project_group_info"]["name"]} > {project_info["name"]}'
+        except Exception as e:
+            _LOGGER.error(f'[get_project_name] API Error: {e}')
+            return project_id
+
     def get_project(self, project_id, domain_id):
         return self.identity_connector.dispatch('Project.get', {'project_id': project_id, 'domain_id': domain_id})
 

src/spaceone/cost_analysis/manager/notification_manager.py

@@ -0,0 +1,20 @@
+import logging
+
+from spaceone.core import config
+from spaceone.core.manager import BaseManager
+from spaceone.core.connector.space_connector import SpaceConnector
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class NotificationManager(BaseManager):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.notification_connector: SpaceConnector = self.locator.get_connector('SpaceConnector',
+                                                                                 service='notification',
+                                                                                 token=config.get_global('TOKEN'))
+
+    def create_notification(self, message):
+        _LOGGER.debug(f'Notify message: {message}')
+        return self.notification_connector.dispatch('Notification.create', message)

src/spaceone/cost_analysis/model/budget_model.py

@@ -20,6 +20,10 @@ class Notification(EmbeddedDocument):
     threshold = FloatField(required=True)
     unit = StringField(max_length=20, required=True, choices=('PERCENT', 'ACTUAL_COST'))
     notification_type = StringField(max_length=20, required=True, choices=('CRITICAL', 'WARNING'))
+    notified_months = ListField(StringField(), default=[])
+
+    def to_dict(self):
+        return dict(self.to_mongo())
 
 
 class Budget(MongoModel):
@@ -29,7 +33,7 @@ class Budget(MongoModel):
     planned_limits = ListField(EmbeddedDocumentField(PlannedLimit), default=[])
     currency = StringField()
     provider_filter = EmbeddedDocumentField(ProviderFilter, required=True)
-    time_unit = StringField(max_length=20, choices=('TOTAL', 'MONTHLY', 'YEARLY'))
+    time_unit = StringField(max_length=20, choices=('TOTAL', 'MONTHLY'))
     start = StringField(required=True, max_length=7)
     end = StringField(required=True, max_length=7)
     notifications = ListField(EmbeddedDocumentField(Notification), default=[])

src/spaceone/cost_analysis/service/budget_service.py

@@ -96,7 +96,7 @@ def create(self, params):
                 params['limit'] += planned_limit.get('limit', 0)
 
         # Check Notifications
-        self._check_notifications(notifications)
+        self._check_notifications(notifications, project_id, project_group_id)
 
         # Check Duplicated Budget
         budget_vos = self.budget_mgr.filter_budgets(
@@ -114,6 +114,7 @@ def create(self, params):
         budget_usage_mgr: BudgetUsageManager = self.locator.get_manager('BudgetUsageManager')
         budget_usage_mgr.create_budget_usages(budget_vo)
         budget_usage_mgr.update_cost_usage(budget_vo.budget_id, budget_vo.domain_id)
+        budget_usage_mgr.notify_budget_usage(budget_vo)
 
         return budget_vo
 
@@ -154,6 +155,9 @@ def update(self, params):
             for budget_usage_vo in budget_usage_vos:
                 budget_usage_mgr.update_budget_usage_by_vo({'name': params['name']}, budget_usage_vo)
 
+        budget_usage_mgr.update_cost_usage(budget_vo.budget_id, budget_vo.domain_id)
+        budget_usage_mgr.notify_budget_usage(budget_vo)
+
         return budget_vo
 
     @transaction(append_meta={'authorization.scope': 'PROJECT'})
@@ -175,12 +179,12 @@ def set_notification(self, params):
         domain_id = params['domain_id']
         notifications = params.get('notifications', [])
 
+        budget_vo: Budget = self.budget_mgr.get_budget(budget_id, domain_id)
+
         # Check Notifications
-        self._check_notifications(notifications)
+        self._check_notifications(notifications, budget_vo.project_id, budget_vo.project_group_id)
         params['notifications'] = notifications
 
-        budget_vo: Budget = self.budget_mgr.get_budget(budget_id, domain_id)
-
         return self.budget_mgr.update_budget_by_vo(params, budget_vo)
 
     @transaction(append_meta={'authorization.scope': 'PROJECT'})
@@ -332,7 +336,10 @@ def _convert_planned_limits_data_type(planned_limits):
         return planned_limits_dict
 
     @staticmethod
-    def _check_notifications(notifications):
+    def _check_notifications(notifications, project_id, project_group_id):
+        if project_group_id and project_id is None:
+            raise ERROR_NOTIFICATION_IS_NOT_SUPPORTED_IN_PROJECT_GROUP(target=project_group_id)
+
         for notification in notifications:
             unit = notification.get('unit')
             notification_type = notification.get('notification_type')

src/spaceone/cost_analysis/service/job_service.py

@@ -2,6 +2,7 @@
 import datetime
 import logging
 from datetime import timedelta, datetime
+from dateutil.relativedelta import relativedelta
 
 from spaceone.core.service import *
 from spaceone.core import utils
@@ -440,7 +441,14 @@ def _close_job(self, job_id, domain_id, data_source_id):
                 except Exception as e:
                     _LOGGER.error(f'[_close_job] aggregate cost data error: {e}', exc_info=True)
                     self._rollback_cost_data(job_vo)
-                    self.job_mgr.change_error_status(job_vo, e)
+                    self.job_mgr.change_error_status(job_vo, f'aggregate cost data error: {e}')
+                    raise e
+
+                try:
+                    self._delete_old_cost_data(domain_id, data_source_id)
+                except Exception as e:
+                    _LOGGER.error(f'[_close_job] delete old cost data error: {e}', exc_info=True)
+                    self.job_mgr.change_error_status(job_vo, f'delete old cost data error: {e}')
                     raise e
 
                 try:
@@ -453,9 +461,11 @@ def _close_job(self, job_id, domain_id, data_source_id):
 
                     self._update_last_sync_time(job_vo)
                     self.job_mgr.change_success_status(job_vo)
+
                 except Exception as e:
                     _LOGGER.error(f'[_close_job] cache and budget update error: {e}', exc_info=True)
-                    self.job_mgr.change_error_status(job_vo, e)
+                    self.job_mgr.change_error_status(job_vo, f'cache and budget update error: {e}')
+                    raise e
 
             elif job_vo.status == 'ERROR':
                 self._rollback_cost_data(job_vo)
@@ -488,6 +498,35 @@ def _update_last_sync_time(self, job_vo: Job):
         data_source_vo = self.data_source_mgr.get_data_source(job_vo.data_source_id, job_vo.domain_id)
         self.data_source_mgr.update_data_source_by_vo({'last_synchronized_at': job_vo.created_at}, data_source_vo)
 
+    def _delete_old_cost_data(self, data_source_id, domain_id):
+        now = datetime.utcnow().date()
+        old_billed_month = (now - relativedelta(months=12)).strftime('%Y-%m')
+        old_billed_year = (now - relativedelta(months=36)).strftime('%Y')
+
+        cost_delete_query = {
+            'filter': [
+                {'k': 'billed_month', 'v': old_billed_month, 'o': 'lt'},
+                {'k': 'data_source_id', 'v': data_source_id, 'o': 'eq'},
+                {'k': 'domain_id', 'v': domain_id, 'o': 'eq'}
+            ]
+        }
+
+        cost_vos, total_count = self.cost_mgr.list_costs(cost_delete_query)
+        _LOGGER.debug(f'[_delete_old_cost_data] delete costs (count = {total_count})')
+        cost_vos.delete()
+
+        monthly_cost_delete_query = {
+            'filter': [
+                {'k': 'billed_year', 'v': old_billed_year, 'o': 'lt'},
+                {'k': 'data_source_id', 'v': data_source_id, 'o': 'eq'},
+                {'k': 'domain_id', 'v': domain_id, 'o': 'eq'}
+            ]
+        }
+
+        monthly_cost_vos, total_count = self.cost_mgr.list_monthly_costs(monthly_cost_delete_query)
+        _LOGGER.debug(f'[_delete_old_cost_data] delete monthly costs (count = {total_count})')
+        monthly_cost_vos.delete()
+
     def _delete_changed_cost_data(self, job_vo: Job, start, end, change_filter):
         query = {
             'filter': [