import json import os from datetime import datetime from typing import Union import numpy as np import pandas as pd from commons.constants import STATUS_ERROR, STATUS_OK, OK_MESSAGE, \ ACTION_SHUTDOWN, ACTION_SCHEDULE, ACTION_SPLIT, ACTION_EMPTY, \ STATUS_POSTPONED from commons.exception import ExecutorException, ProcessingPostponedException from commons.log_helper import get_logger from commons.profiler import profiler from models.algorithm import Algorithm from models.base_model import CloudEnum from models.parent_attributes import ParentMeta from models.shape_price import OSEnum from services.environment_service import EnvironmentService from services.meta_service import MetaService from services.metrics_service import MetricsService from services.recommendation_history_service import \ RecommendationHistoryService from services.resize.resize_service import ResizeService from services.resize.resize_trend import ResizeTrend from services.saving.saving_service import SavingService from services.schedule.schedule_service import ScheduleService _LOG = get_logger('r8s-recommendation-service') RIGHTSIZER_SOURCE = 'RIGHTSIZER' RIGHTSIZER_RESOURCE_TYPE = 'INSTANCE' DEFAULT_SEVERITY = 'MEDIUM' class RecommendationService: def __init__(self, metrics_service: MetricsService, schedule_service: ScheduleService, resize_service: ResizeService, environment_service: EnvironmentService, saving_service: SavingService, meta_service: MetaService, recommendation_history_service: RecommendationHistoryService): self.metrics_service = metrics_service self.schedule_service = schedule_service self.resize_service = resize_service self.environment_service = environment_service self.saving_service = saving_service self.meta_service = meta_service self.recommendation_history_service = recommendation_history_service @profiler(execution_step=f'instance_recommendation_generation') def process_instance(self, metric_file_path, algorithm: Algorithm, reports_dir, instance_meta_mapping=None, parent_meta: Union[None, ParentMeta] = None): _LOG.debug(f'Parsing entity names from metrics file path ' f'\'{metric_file_path}\'') df = None schedule = None recommended_sizes = None resize_action = None savings = None advanced = None customer, cloud, tenant, region, _, instance_id = self._parse_folders( metric_file_path=metric_file_path ) if cloud.upper() != CloudEnum.CLOUD_GOOGLE.value: instance_os = OSEnum.OS_LINUX.value else: # use Linux pricing as a default for all clouds except GCP instance_os = None instance_meta = {} if instance_meta_mapping: instance_meta = instance_meta_mapping.get(instance_id) _LOG.debug(f'Instance meta: {instance_meta}') _LOG.debug(f'Loading past recommendation with feedback') past_recommendations = self.recommendation_history_service. \ get_recommendation_with_feedback(instance_id=instance_id) applied_recommendations = self.recommendation_history_service. \ filter_applied(recommendations=past_recommendations) try: _LOG.debug(f'Loading adjustments from meta') meta_adjustments = self.meta_service.to_adjustments( instance_meta=instance_meta ) applied_recommendations.extend(meta_adjustments) _LOG.debug(f'Loading df') df = self.metrics_service.load_df( path=metric_file_path, algorithm=algorithm, applied_recommendations=applied_recommendations, instance_meta=instance_meta ) _LOG.debug(f'Extracting instance type name') instance_type = self.metrics_service.get_instance_type( metric_file_path=metric_file_path, algorithm=algorithm ) _LOG.debug(f'Dividing into periods with different load') shutdown_periods, low_periods, medium_periods, \ high_periods, centroids = \ self.metrics_service.divide_on_periods( df=df, algorithm=algorithm) _LOG.debug(f'Got {len(high_periods)} high-load, ' f'{len(low_periods)} low-load periods') non_straight_periods, total_length = self.get_non_straight_periods( df=df, grouped_periods=(low_periods, medium_periods, high_periods) ) if non_straight_periods and total_length: _LOG.debug(f'Calculating resize trends for several loads') trends = self.metrics_service. \ calculate_instance_trend_multiple( algorithm=algorithm, non_straight_periods=non_straight_periods, total_length=total_length ) else: _LOG.debug(f'Generating resize trend') if any((low_periods, medium_periods, high_periods)): df_ = pd.concat([*low_periods, *medium_periods, *high_periods]) else: df_ = df trends = self.metrics_service.calculate_instance_trend( df=df_, algorithm=algorithm ) trends = [trends] _LOG.debug(f'Resize trend for instance \'{instance_id}\' has been ' f'calculated. ') _LOG.debug( f'Got {len(shutdown_periods)} shutdown periods to process') _LOG.debug(f'Generating schedule for instance \'{instance_id}\'') if not low_periods and not medium_periods and not high_periods: schedule = [] else: schedule = self.schedule_service.generate_schedule( shutdown_periods=shutdown_periods, recommendation_settings=algorithm.recommendation_settings, instance_id=instance_id, df=df, instance_meta=instance_meta, past_recommendations=applied_recommendations ) _LOG.debug(f'Searching for resize action') resize_action = ResizeTrend.get_resize_action(trends=trends) _LOG.debug(f'Searching for better-fix instance types') compatibility_rule = algorithm.recommendation_settings. \ shape_compatibility_rule past_resize_recommendations = self.recommendation_history_service. \ filter_resize(recommendations=applied_recommendations) if len(trends) == 1: max_recommended_shapes = algorithm.recommendation_settings. \ max_recommended_shapes recommended_sizes = self.resize_service.recommend_size( trend=trends[0], instance_type=instance_type, algorithm=algorithm, cloud=cloud.upper(), instance_meta=instance_meta, resize_action=resize_action, parent_meta=parent_meta, max_results=max_recommended_shapes, shape_compatibility_rule=compatibility_rule, past_resize_recommendations=past_resize_recommendations ) else: recommended_sizes = [] for trend in trends: recommended_size = self.resize_service.recommend_size( trend=trend, instance_type=instance_type, algorithm=algorithm, cloud=cloud.upper(), instance_meta=instance_meta, resize_action=resize_action, parent_meta=parent_meta, max_results=1, shape_compatibility_rule=compatibility_rule, past_resize_recommendations=past_resize_recommendations ) recommended_sizes.extend(recommended_size) recommended_sizes = self.resize_service.add_price( instances=recommended_sizes, customer=customer, region=region, os=instance_os) recommended_sizes = self._cleanup_recommended_shapes( recommended_sizes=recommended_sizes, current_instance_type=instance_type, allow_same_shape=resize_action == ACTION_SPLIT ) recommended_sizes = self.resize_service.sort_shapes( shapes=recommended_sizes, sort_option=algorithm.recommendation_settings.shape_sorting ) _LOG.debug(f'Got {len(recommended_sizes)} ' f'recommended instance types') _LOG.debug(f'Calculate instance stats') stats = self.calculate_instance_stats(df=df) advanced = self.calculate_advanced_stats( df=df, centroids=centroids, algorithm=algorithm ) general_action = self.get_general_action( schedule=schedule, shapes=recommended_sizes, resize_action=resize_action, stats=stats, past_recommendations=past_recommendations ) if not algorithm.recommendation_settings.ignore_savings: _LOG.debug(f'Calculating savings') savings = self.saving_service.calculate_savings( general_actions=general_action, current_shape=instance_type, recommended_shapes=recommended_sizes, schedule=schedule, customer=customer, region=region, os=instance_os ) try: _LOG.debug(f'Saving recommendation to history') history_items = self.recommendation_history_service.create( instance_id=instance_id, job_id=self.environment_service.get_batch_job_id(), customer=customer, tenant=tenant, region=region, schedule=schedule, recommended_shapes=recommended_sizes, current_instance_type=instance_type, savings=savings, actions=general_action, instance_meta=instance_meta ) if history_items: _LOG.debug( f'Saving \'{len(history_items)}\' history items') self.recommendation_history_service.batch_save( recommendations=history_items) except Exception as e: _LOG.error(f'Exception occurred while saving recommendation ' f'to history: {str(e)}') except Exception as e: _LOG.debug(f'Calculate instance stats with exception') stats = self.calculate_instance_stats(df=df, exception=e) general_action = self.get_general_action( schedule=schedule, shapes=recommended_sizes, resize_action=resize_action, stats=stats) _LOG.debug(f'Dumping instance results') item = self.dump_reports( reports_dir=reports_dir, instance_id=instance_id, schedule=schedule, recommended_sizes=recommended_sizes, stats=stats, customer=customer, cloud=cloud, tenant=tenant, region=region, meta=instance_meta, general_action=general_action, savings=savings, advanced=advanced ) return item def dump_error_report(self, reports_dir, metric_file_path, exception): customer, cloud, tenant, region, _, instance_id = self._parse_folders( metric_file_path=metric_file_path ) stats = self.calculate_instance_stats(exception=exception) return self.dump_reports( reports_dir=reports_dir, instance_id=instance_id, schedule=[], recommended_sizes=[], stats=stats, customer=customer, cloud=cloud, tenant=tenant, region=region, meta={}, general_action=STATUS_ERROR ) def dump_reports(self, reports_dir, customer, cloud, tenant, region, stats, instance_id=None, schedule=None, recommended_sizes=None, meta=None, general_action=None, savings=None, advanced=None): if general_action and not isinstance(general_action, list): general_action = [general_action] item = { 'resource_id': instance_id, 'resource_type': RIGHTSIZER_RESOURCE_TYPE, 'source': RIGHTSIZER_SOURCE, 'severity': DEFAULT_SEVERITY, 'recommendation': { 'schedule': schedule, 'recommended_shapes': recommended_sizes, 'savings': savings, 'advanced': advanced, }, 'stats': stats, 'meta': meta, 'general_actions': general_action } self.prettify_recommendation(recommendation_item=item) dir_path = os.path.join(reports_dir, customer, cloud, tenant) os.makedirs(dir_path, exist_ok=True) file_path = os.path.join(dir_path, f'{region}.jsonl') with open(file_path, 'a') as f: f.write(json.dumps(item)) f.write('\n') return item @staticmethod def get_non_straight_periods(df, grouped_periods): total_length = len(df) valid_period_groups = [] total_valid_period_len = 0 for periods in grouped_periods: grouped_len = sum([len(period) for period in periods]) # metrics are in 5min freq, periods are in 10min if grouped_len >= total_length * 0.05: # eq 10%, period metrics are 10min freq, df freq is 5min valid_period_groups.append(periods) total_valid_period_len += grouped_len if not valid_period_groups or len(valid_period_groups) == 1: return None, None return valid_period_groups, total_valid_period_len @staticmethod def calculate_instance_stats(df=None, exception=None): from_date = None to_date = None if df is not None: from_date = min(df.T).to_pydatetime().isoformat() to_date = max(df.T).to_pydatetime().isoformat() if isinstance(exception, ExecutorException): status = STATUS_ERROR message = exception.reason elif isinstance(exception, ProcessingPostponedException): status = STATUS_POSTPONED message = str(exception) elif exception: status = STATUS_ERROR message = f'Unexpected error occurred: {str(exception)}' else: status = STATUS_OK message = OK_MESSAGE return { 'from_date': from_date, 'to_date': to_date, 'status': status, 'message': message } def calculate_advanced_stats(self, df, algorithm, centroids): _LOG.debug(f'Calculating advanced stats') result = {} metric_fields = self._get_metric_fields(df=df, algorithm=algorithm) for metric_field in metric_fields: _LOG.debug(f'Calculating advanced stats for {metric_field}') metric_stats = self._get_metric_advanced_stats( df=df, metric_name=metric_field ) if metric_stats: _LOG.debug(f'{metric_field} advanced stats: {metric_stats}') result[metric_field] = metric_stats _LOG.debug(f'Calculating clustering stats') cluster_stats = self._get_clusters_stats(centroids=centroids) _LOG.debug(f'Clustering stats: {cluster_stats}') result['clusters'] = cluster_stats return result @staticmethod def _parse_folders(metric_file_path): """ Extracts customer, tenant, region, timestamp and instance id from metric file path """ file_name = os.path.basename(metric_file_path) folders = metric_file_path.split(os.sep) instance_id = folders[-1][0:file_name.rindex('.')] timestamp = folders[-2] region = folders[-3] tenant = folders[-4] cloud = folders[-5] customer = folders[-6] return customer, cloud, tenant, region, timestamp, instance_id def get_general_action(self, schedule, shapes, stats, resize_action, past_recommendations: list = None): actions = [] status = stats.get('status', '') if status == STATUS_POSTPONED: return [ACTION_EMPTY] if status != STATUS_OK: return [STATUS_ERROR] shutdown_forbidden = False if past_recommendations: shutdown_forbidden = self.recommendation_history_service. \ is_shutdown_forbidden( recommendations=past_recommendations ) if not schedule and not shutdown_forbidden: return [ACTION_SHUTDOWN] if schedule and not self._is_schedule_always_run(schedule=schedule): actions.append(ACTION_SCHEDULE) if shapes: shape = shapes[0] if 'probability' in shape: actions.append(ACTION_SPLIT) else: actions.append(resize_action) if not actions: return [ACTION_EMPTY] return actions @staticmethod def _is_schedule_always_run(schedule, complete_week=True): if len(schedule) != 1: return False schedule = schedule[0] start = schedule.get('start', '') stop = schedule.get('stop', '') weekdays = schedule.get('weekdays', []) if complete_week: stop_suits = stop.startswith('23:') or stop == '00:00' return start == '00:00' and stop_suits and len(weekdays) == 7 return start == '00:00' and stop.startswith('23:') def prettify_recommendation(self, recommendation_item): recommendation = recommendation_item.get('recommendation') schedule = recommendation.get('schedule') shapes = recommendation.get('recommended_shapes') if shapes is None: recommendation['recommended_shapes'] = [] if schedule is None: recommendation['schedule'] = [] if schedule: schedule.sort(key=lambda k: self._get_schedule_weight(k)) if len(schedule) > 5: schedule = schedule[0:5] recommendation['schedule'] = schedule if not recommendation['advanced']: recommendation.pop('advanced', None) recommendation_item['recommendation'] = recommendation @staticmethod def _get_schedule_weight(schedule_item): start = schedule_item.get('start') stop = schedule_item.get('stop') start_dt = datetime.strptime(start, '%H:%M') stop_dt = datetime.strptime(stop, '%H:%M') delta = (stop_dt - start_dt).total_seconds() return delta * len(schedule_item.get('weekdays')) @staticmethod def _cleanup_recommended_shapes(recommended_sizes, current_instance_type, allow_same_shape=False): result = [] for size in recommended_sizes: instance_type = size.get('name') if size in result: continue if not allow_same_shape and instance_type == current_instance_type: continue result.append(size) return result @staticmethod def _get_metric_fields(df: pd.DataFrame, algorithm: Algorithm): valid_columns = [] for column in list(algorithm.metric_attributes): if any([value not in (0, -1) for value in df['cpu_load']]): valid_columns.append(column) return valid_columns @staticmethod def _get_metric_advanced_stats(df: pd.DataFrame, metric_name): series = df[metric_name] deciles = list(np.quantile(series, np.arange(0.1, 1, 0.1))), deciles = [round(float(decile), 2) for decile in deciles[0]] return { "min": round(float(np.max(series)), 2), "max": round(float(np.max(series)), 2), "mean": round(float(np.mean(series)), 2), "deciles": deciles, "variance": round(float(np.var(series)), 2), "standard_deviation": round(float(np.std(series)), 2) } @staticmethod def _get_clusters_stats(centroids: list): cluster_count_per_day = [len(day_centroids) for day_centroids in centroids] if not centroids: return {} quartiles = list(np.quantile(cluster_count_per_day, np.arange(0.25, 1, 0.25))), quartiles = [round(float(quartile), 2) for quartile in quartiles[0]] return { "avg_per_day": round(float(np.average(cluster_count_per_day)), 2), "max_per_day": round(float(np.max(cluster_count_per_day)), 2), "min_per_day": round(float(np.min(cluster_count_per_day)), 2), "quartiles": quartiles }