import datetime import glob import json import os from typing import List import numpy as np import pandas import pandas as pd import pytz from pytz import UnknownTimeZoneError from commons import dateparse from commons.constants import JOB_STEP_PROCESS_METRICS, META_FILE_NAME, \ JOB_STEP_INITIALIZE_ALGORITHM, JOB_STEP_VALIDATE_METRICS, COLUMN_CPU_LOAD, \ CSV_EXTENSION from commons.exception import ExecutorException from commons.log_helper import get_logger from commons.profiler import profiler from models.algorithm import Algorithm from models.recommendation_history import RecommendationHistory from services.clustering_service import ClusteringService from services.resize.resize_trend import ResizeTrend _LOG = get_logger('r8s-metrics-service') TIMESTAMP_FREQUENCY = '5Min' DAY_RECORDS = 144 META_KEY_RESOURCE_ID = 'resourceId' META_KEY_CREATE_DATE_TIMESTAMP = 'createDateTimestamp' MINIMUM_DAYS_TO_CUT_INCOMPLETE_EDGE_DAYS = 14 class MetricsService: def __init__(self, clustering_service: ClusteringService): self.clustering_service = clustering_service def calculate_instance_trend(self, df, algorithm: Algorithm) \ -> ResizeTrend: metric_attrs = set(list(algorithm.metric_attributes)) resize_trend = ResizeTrend() for metric in metric_attrs: metric_column = self.get_column(metric_name=metric, df=df) resize_trend.add_metric_trend( metric_name=metric, column=metric_column ) return resize_trend def calculate_instance_trend_multiple( self, algorithm: Algorithm, non_straight_periods, total_length) -> List[ResizeTrend]: result = [] for period_list in non_straight_periods: concat_dt = pd.concat(period_list) period_trend = self.calculate_instance_trend( df=concat_dt, algorithm=algorithm) period_trend.probability = round( len(concat_dt) / total_length, 2) result.append(period_trend) if not result: return result metric_attrs = set(list(algorithm.metric_attributes)) # removes duplicated trends with same metric resize directions result = ResizeTrend.remove_duplicates(trends=result, metric_attrs=metric_attrs) return result @staticmethod def get_threshold_value(column): return column.quantile(.9) @staticmethod def get_column(metric_name, df): try: column = df[metric_name] return column except Exception as e: _LOG.error(f'Invalid metric format. Exception: {e}') raise ExecutorException( step_name=JOB_STEP_PROCESS_METRICS, reason=f'Invalid metric format. Exception: {e}' ) @staticmethod def get_last_period(df, days=7): range_max = df.index.max() range_min = range_max - datetime.timedelta(days=days) # take slice with final N days of data sliced_df = df[(df.index >= range_min) & (df.index <= range_max)] return sliced_df @staticmethod def fill_missing_timestamps(df, diff=TIMESTAMP_FREQUENCY): instance_id = df['instance_id'][0] instance_type = df['instance_type'][0] df = df[~df.index.duplicated(keep='last')] df = df.resample(diff).ffill() df = df.assign(instance_id=instance_id) df = df.assign(instance_type=instance_type) df.fillna(0, inplace=True) return df def validate_metric_file(self, algorithm: Algorithm, metric_file_path): try: df = self.read_metrics(metric_file_path=metric_file_path, algorithm=algorithm, parse_index=False) except Exception as e: _LOG.warning(f'Metric file can not be read: Exception: {e}') raise ExecutorException( step_name=JOB_STEP_VALIDATE_METRICS, reason=f'Metric file can not be read: Exception: {e}' ) column_names = list(df.columns) required_columns_set = set(list(algorithm.required_data_attributes)) file_columns_set = set(column_names) missing_columns = list(required_columns_set - file_columns_set) excess_columns = list(file_columns_set - required_columns_set) if missing_columns or excess_columns: _LOG.error(f'File \'{metric_file_path}\' does not match the ' f'required set of columns') error_message = [] if missing_columns: _LOG.error(f'Missing columns: \'{missing_columns}\'') error_message.append(f'Missing columns: \'{missing_columns}\'') if excess_columns: _LOG.error(f'Excess columns: \'{excess_columns}\'') if error_message: raise ExecutorException( step_name=JOB_STEP_VALIDATE_METRICS, reason=';'.join(error_message) ) absent_metric_attrs = [] metric_attrs = list(algorithm.metric_attributes) for metric_attr in metric_attrs: if (df[metric_attr] == -1).all(): absent_metric_attrs.append(metric_attr) if set(absent_metric_attrs) == set(metric_attrs): _LOG.warning(f'Metric file must contain data for at ' f'least one metric: {", ".join(metric_attrs)}') raise ExecutorException( step_name=JOB_STEP_INITIALIZE_ALGORITHM, reason=f'Metric file must contain data for at ' f'least one metric: {", ".join(metric_attrs)}' ) def load_df(self, path, algorithm: Algorithm, applied_recommendations: List[RecommendationHistory] = None, instance_meta: dict = None): all_attrs = set(list(algorithm.required_data_attributes)) metric_attrs = set(list(algorithm.metric_attributes)) non_metric = all_attrs - metric_attrs non_metric.remove(algorithm.timestamp_attribute) try: df = self.read_metrics(metric_file_path=path, algorithm=algorithm) df = self.trim_from_appliance_date( df=df, applied_recommendations=applied_recommendations) recommendation_settings = algorithm.recommendation_settings timezone_name = recommendation_settings.target_timezone_name if timezone_name: _LOG.debug(f'Converting to timezone \'{timezone_name}\'') try: df = df.tz_convert(timezone_name) except UnknownTimeZoneError: _LOG.error(f'Unknown timezone \'{timezone_name}\'') df = self.fill_missing_timestamps(df=df) df.sort_index(ascending=True, inplace=True) for attr in non_metric: df.drop(attr, inplace=True, axis=1) df = self.discard_start( df=df, algorithm=algorithm, instance_meta=instance_meta ) df_duration_days = (df.index.max() - df.index.min()).days if np.isnan(df_duration_days) or \ df_duration_days < \ algorithm.recommendation_settings.min_allowed_days: _LOG.error( f'Insufficient data. Analysed period must be larger ' f'than a full day with 5-min frequency of records.') raise ExecutorException( step_name=JOB_STEP_INITIALIZE_ALGORITHM, reason=f'Insufficient data. Analysed period must be larger ' f'than a full day with 5-min frequency of records.' ) df = self.get_last_period(df, days=recommendation_settings.max_days) df = self.group_by_time( df=df, step_minutes=recommendation_settings.record_step_minutes) return df except ExecutorException as e: raise e except Exception as e: _LOG.error(f'Error occurred while reading metrics file: {str(e)}') raise ExecutorException( step_name=JOB_STEP_PROCESS_METRICS, reason=f'Unable to read metrics file' ) @staticmethod def discard_start(df: pd.DataFrame, algorithm: Algorithm, instance_meta=None): if instance_meta and 'creationDateTimestamp' in instance_meta: try: creation_date_timestamp = instance_meta[ 'creationDateTimestamp'] _LOG.debug(f'Discarding metrics before timestamp: ' f'{creation_date_timestamp}') creation_dt = datetime.datetime.utcfromtimestamp( creation_date_timestamp // 1000) creation_dt = creation_dt.astimezone( pytz.timezone(df.index.max().tz.zone)) return df[(df.index >= creation_dt)] except Exception as e: _LOG.debug(f'Failed to discard metrics before timestamp. ' f'Exception: {e}') if algorithm.recommendation_settings.discard_initial_zeros: _LOG.debug( f'Going to discard metrics without load from the start.') metric_attrs = algorithm.metric_attributes try: for index, row in df.iterrows(): if not all(row[attr] in (0, -1) for attr in metric_attrs): _LOG.debug(f'Metrics before {index} will be discarded') return df[df.index >= index] except Exception as e: _LOG.debug(f'Failed to discard leading metrics with zeros. ' f'Exception: {e}') return df def merge_metric_files(self, metrics_folder_path, algorithm: Algorithm): metric_files = [y for x in os.walk(metrics_folder_path) for y in glob.glob(os.path.join(x[0], '*.csv'))] instance_id_date_mapping = {} for file in metric_files: path_items = file.split(os.sep)[::-1] instance_id, date = path_items[0:2] instance_id = instance_id.replace(CSV_EXTENSION, '') if instance_id in instance_id_date_mapping: instance_id_date_mapping[instance_id].append(file) else: instance_id_date_mapping[instance_id] = [file] resulted_files = [] for instance_id, files in instance_id_date_mapping.items(): if len(files) == 1: resulted_files.append(files[0]) continue most_recent = max(files) files = sorted(files) csv_to_combine = [self.read_metrics(f, algorithm=algorithm, parse_index=False) for f in files] combined_csv = pd.concat(csv_to_combine) combined_csv.sort_values(algorithm.timestamp_attribute) combined_csv.to_csv(most_recent, index=False) resulted_files.append(most_recent) for file in metric_files: if file not in resulted_files: os.remove(file) return resulted_files @profiler(execution_step=f'instance_clustering') def divide_on_periods(self, df, algorithm: Algorithm): df = self.divide_by_days( df, skip_incomplete_corner_days=True, step_minutes=algorithm.recommendation_settings.record_step_minutes) shutdown_periods = [] low_util_periods = [] good_util_periods = [] over_util_periods = [] centroids = [] for index, df_day in enumerate(df): # print(f'Processing day: {index}/{len(cpu_df)}') shutdown, low, medium, high, day_centroids = self.process_day( df=df_day, algorithm=algorithm) shutdown_periods.extend(shutdown) low_util_periods.extend(low) good_util_periods.extend(medium) over_util_periods.extend(high) centroids.extend(day_centroids) return shutdown_periods, low_util_periods, \ good_util_periods, over_util_periods, centroids @staticmethod def group_by_time(df, step_minutes: int = None, freq='10Min'): if step_minutes: freq = f'{step_minutes}Min' return df.groupby(pd.Grouper(freq=freq)).mean() @staticmethod def divide_by_days(df, skip_incomplete_corner_days: bool, step_minutes: int): df_list = [group[1] for group in df.groupby(df.index.date)] if not df_list: return df_list if len(df_list) > MINIMUM_DAYS_TO_CUT_INCOMPLETE_EDGE_DAYS \ and skip_incomplete_corner_days: last_day_df = df_list[-1] if len(last_day_df) < 24 * 60 // step_minutes: df_list = df_list[:-1] first_day_df = df_list[0] if len(first_day_df) < 24 * 60 // step_minutes: df_list = df_list[1:] return df_list def process_day(self, df: pandas.DataFrame, algorithm: Algorithm): shutdown = [] low_util = [] good_util = [] over_util = [] df_, centroids = self.clustering_service.cluster( df=df, algorithm=algorithm) _LOG.debug(f'Clusters centroids: {centroids}') thresholds = algorithm.recommendation_settings.thresholds for index, centroid in enumerate(centroids): if not centroid: continue cluster: pd.DataFrame = df.loc[df['cluster'] == index] cluster.drop('cluster', axis=1, inplace=True) if centroid[0] < thresholds[0]: shutdown.append(cluster) elif thresholds[0] <= centroid[0] < \ thresholds[1]: low_util.append(cluster) elif thresholds[1] <= centroid[0] < \ thresholds[2]: good_util.append(cluster) else: over_util.append(cluster) shutdown = pd.concat(shutdown) if shutdown else None low_util = pd.concat(low_util) if low_util else None good_util = pd.concat(good_util) if good_util else None over_util = pd.concat(over_util) if over_util else None freq = f'{algorithm.recommendation_settings.record_step_minutes}Min' result = [self.get_time_ranges(cluster, freq=freq) for cluster in (shutdown, low_util, good_util, over_util)] result.append(centroids) return result @staticmethod def get_non_empty_attrs(df: pd.DataFrame, attrs): non_empty = [] for attr in attrs: is_empty = all(value == -1 for value in list(df[attr])) if not is_empty: non_empty.append(attr) return non_empty @staticmethod def filter_by_ranges(row, periods): row_time = row.name.time() for period in periods: if period[0] <= row_time <= period[1]: return row def get_time_ranges(self, df, freq='10Min'): if not isinstance(df, pd.DataFrame) or len(df) == 0: return [] df = df.asfreq(freq) periods = [] period_start = None period_end = None for row in df.itertuples(): cpu_load = getattr(row, COLUMN_CPU_LOAD) if period_start and period_end and np.isnan(cpu_load): periods.append((period_start, period_end)) period_start = None period_end = None elif not period_start and not np.isnan(cpu_load): period_start = row.Index.time() elif period_start and not np.isnan(cpu_load): period_end = row.Index.time() if period_start and period_end: periods.append((period_start, period_end)) return self.build_df_from_periods( df=df, periods=periods ) def build_df_from_periods(self, df, periods: list): dfs = [] for period in periods: df_ = df.copy().apply(self.filter_by_ranges, axis=1, periods=[period], result_type='broadcast') if not isinstance(df_, pd.DataFrame): df_ = df_.to_frame() df_ = df_.dropna(thresh=1) if not df.empty: dfs.append(df_) return dfs @staticmethod def filter_short_periods(periods, min_length_sec=1800): periods = [df for df in periods if (df.index.max() - df.index.min()).total_seconds() >= min_length_sec] periods.sort(key=lambda df: df.index.min()) return periods def get_instance_type(self, metric_file_path, algorithm: Algorithm, instance_type_attr='instance_type'): try: df = self.read_metrics(metric_file_path=metric_file_path, algorithm=algorithm, parse_index=False) return df[instance_type_attr][0] except Exception as e: _LOG.error(f'Failed to extract instance type from metric file. ' f'Error: {e}') raise ExecutorException( step_name=JOB_STEP_PROCESS_METRICS, reason=f'Failed to extract instance type from metric file. ' f'Error: {e}' ) @staticmethod def read_metrics(metric_file_path, algorithm: Algorithm = None, parse_index=True): try: if not parse_index: return pd.read_csv(metric_file_path, **algorithm.read_configuration) return pd.read_csv( metric_file_path, parse_dates=True, date_parser=dateparse, index_col=algorithm.timestamp_attribute, **algorithm.read_configuration) except Exception as e: _LOG.error(f'Error occurred while reading metrics file: {str(e)}') raise ExecutorException( step_name=JOB_STEP_PROCESS_METRICS, reason=f'Unable to read metrics file' ) def read_meta(self, metrics_folder): instance_meta_mapping = {} files = list(glob.iglob(metrics_folder + f'**/**/{META_FILE_NAME}', recursive=True)) _LOG.debug(f'Found \'{len(files)}\' meta files, extracting meta') for file in files: instance_meta_mapping = self._read_meta_file( meta_file_path=file, instance_meta_mapping=instance_meta_mapping) _LOG.debug(f'Loaded meta: {instance_meta_mapping}') return instance_meta_mapping @staticmethod def trim_from_appliance_date(df: pandas.DataFrame, applied_recommendations: List[RecommendationHistory]): if not applied_recommendations: return df last_applied_date = max(item.feedback_dt for item in applied_recommendations) _LOG.debug(f'Trimming metrics from date ' f'\'{last_applied_date.isoformat()}\'') if df.index.tz: last_applied_date = last_applied_date.replace(tzinfo=pytz.UTC) return df[df.index > last_applied_date] @staticmethod def _read_meta_file(meta_file_path: str, instance_meta_mapping: dict): _LOG.debug(f'Processing meta file \'{meta_file_path}\'') try: with open(meta_file_path, 'r') as f: file_meta = json.load(f) except json.decoder.JSONDecodeError: _LOG.error(f'Unable to read meta file \'{meta_file_path}\', ' f'skipping') return instance_meta_mapping if not isinstance(file_meta, list): _LOG.error(f'Invalid meta format: must be a valid list') return instance_meta_mapping for resource_meta in file_meta: resource_id = resource_meta.get(META_KEY_RESOURCE_ID) if not resource_id: continue if resource_id not in instance_meta_mapping: instance_meta_mapping[resource_id] = resource_meta continue timestamp = resource_meta.get(META_KEY_CREATE_DATE_TIMESTAMP) instance_timestamp = instance_meta_mapping[ resource_id].get(META_KEY_CREATE_DATE_TIMESTAMP) # overwrite instance meta with latest meta found if timestamp and instance_timestamp and \ timestamp > instance_timestamp \ or not instance_timestamp: instance_meta_mapping[resource_id].update(resource_meta) return instance_meta_mapping