docker/services/mocked_data_service.py

import pandas as pd
from cron_converter import Cron

from commons.constants import ALLOWED_ACTIONS, ACTION_SCALE_DOWN, \
    ACTION_SCALE_UP, ACTION_EMPTY, ACTION_SHUTDOWN, ACTION_SPLIT, \
    ACTION_SCHEDULE
from commons.log_helper import get_logger
from models.base_model import CloudEnum
from services.os_service import OSService
from tests_executor.constants import POINTS_IN_DAY
from tests_executor.utils import generate_timestamp_series, constant_to_series, \
    generate_constant_metric_series, generate_split_series, \
    generate_scheduled_metric_series

_LOG = get_logger('r8s_mocked_data_service')

TAG_TEST_CASE = 'r8s_test_case'
TAG_PERIOD_DAYS = 'r8s_period_days'
TAG_CPU = 'r8s_cpu_load'
TAG_MEMORY = 'r8s_memory_load'
TAG_AVG_DISK_IOPS = 'r8s_avg_disk_iops'
TAG_MAX_DISK_IOPS = 'r8s_max_disk_iops'
TAG_NET_OUTPUT_LOAD = 'r8s_net_output_load'
TAG_STD = 'r8s_std'
TAG_CRON_START = 'r8s_cron_start'
TAG_CRON_STOP = 'r8s_cron_stop'
TAG_PROBABILITY = 'r8s_probability'

ALLOWED_TAGS = [TAG_TEST_CASE, TAG_PERIOD_DAYS, TAG_CPU, TAG_MEMORY,
                TAG_AVG_DISK_IOPS, TAG_MAX_DISK_IOPS, TAG_NET_OUTPUT_LOAD,
                TAG_STD, TAG_CRON_START, TAG_CRON_STOP, TAG_PROBABILITY]
NUMBER_TAGS = [TAG_PERIOD_DAYS, TAG_CPU, TAG_MEMORY, TAG_AVG_DISK_IOPS,
               TAG_MAX_DISK_IOPS, TAG_NET_OUTPUT_LOAD, TAG_STD,
               TAG_PROBABILITY]
DEFAULT_CONFIG = {
    ACTION_SCALE_DOWN: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: 20,
        TAG_MEMORY: 20,
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_STD: 2,
    },
    ACTION_SCALE_UP: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: 80,
        TAG_MEMORY: 80,
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_STD: 2,
    },
    ACTION_EMPTY: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: 45,
        TAG_MEMORY: 55,
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_STD: 2,
    },
    ACTION_SHUTDOWN: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: 5,
        TAG_MEMORY: 3,
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_STD: 1,
    },
    ACTION_SPLIT: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: [25, 80],
        TAG_MEMORY: [25, 80],
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_PROBABILITY: [50, 50],
        TAG_STD: 1,
    },
    ACTION_SCHEDULE: {
        TAG_PERIOD_DAYS: 14,
        TAG_CPU: 50,
        TAG_MEMORY: 50,
        TAG_AVG_DISK_IOPS: -1,
        TAG_MAX_DISK_IOPS: -1,
        TAG_NET_OUTPUT_LOAD: -1,
        TAG_STD: 1,
        TAG_CRON_START: '0 9 * * 0-4',
        TAG_CRON_STOP: '0 18 * * 0-4',
    },
}


class MockedDataService:
    """
    Allows to replace actual instance metrics with mocked data based on
    instance meta tags.Supported tags:
    - r8s_test_case: SCALE_UP/SCALE_DOWN/SHUTDOWN/SCHEDULE/SPLIT -
        contains resulted action
    - r8s_period_days: mocked data period
    - r8s_cpu_load: average cpu load on instance
    - r8s_memory_load: average memory load on instance
    - r8s_std: standard deviation, used to modify load
    - r8s_cron_start (SCHEDULE only): cron that will be used to generate
        instance load
    - r8s_cron_stop (SCHEDULE only):cron that will be used to generate
        instance load
    - r8s_probability (SPLIT only): to split load by speficic percentages
    """

    def __init__(self, os_service: OSService):
        self.os_service = os_service

        self.action_generator_mapping = {
            ACTION_EMPTY: self.generate_constant_load_metrics,
            ACTION_SHUTDOWN: self.generate_constant_load_metrics,
            ACTION_SCALE_UP: self.generate_constant_load_metrics,
            ACTION_SCALE_DOWN: self.generate_constant_load_metrics,
            ACTION_SCHEDULE: self.generate_schedule_load_metrics,
            ACTION_SPLIT: self.generate_split_load_metrics,
        }

    def process(self, instance_meta_mapping, metric_file_paths):
        instance_tags_mapping = self.parse_tags(
            instance_meta_mapping=instance_meta_mapping)

        meta_mapping = {k: v for k, v in instance_tags_mapping.items()
                        if TAG_TEST_CASE in v}
        file_to_meta_mapping = {}

        for file in metric_file_paths:
            instance_id = self.os_service.path_to_instance_id(file_path=file)
            if instance_id in meta_mapping:
                file_to_meta_mapping[file] = meta_mapping[instance_id]

        if not file_to_meta_mapping:
            _LOG.warning(f'No instances with tag \'{TAG_TEST_CASE}\' found.')
            return

        for file_path, instance_meta in file_to_meta_mapping.items():
            _LOG.debug(f'Going to replace metrics by path \'{file_path}\' '
                       f'with mocked metrics by tags: {instance_meta}')
            self.process_instance(instance_meta=instance_meta,
                                  metric_file_path=file_path)

    def process_instance(self, instance_meta, metric_file_path):
        _LOG.debug(f'Filtering instance meta')
        instance_meta = {k: v for k, v in instance_meta.items()
                         if k in ALLOWED_TAGS}
        _LOG.debug(f'Converting tag values to numbers')
        instance_meta = self.values_to_number(instance_meta=instance_meta)

        test_case = instance_meta.get(TAG_TEST_CASE).upper()
        if test_case not in ALLOWED_ACTIONS:
            _LOG.error(f'Invalid test case specified: \'{test_case}\'. '
                       f'Allowed test cases: {ALLOWED_ACTIONS}')
            return

        test_config = DEFAULT_CONFIG.get(test_case).copy()

        for key, value in instance_meta.items():
            if key not in test_config:
                continue
            default_value = test_config.get(key)
            if isinstance(default_value, list) and not isinstance(value, list):
                _LOG.warning(f'Expected \'list\' value for key \'{key}\', '
                             f'skipping')
                continue
            if isinstance(default_value, int) and not isinstance(value,
                                                                 (int, float)):
                _LOG.warning(f'Expected \'int\' value for key \'{key}\', '
                             f'skipping')
                continue
            test_config[key] = value
        generator = self.action_generator_mapping.get(test_case)
        generator(test_config, metric_file_path)

    @staticmethod
    def values_to_number(instance_meta):
        for k, v in instance_meta.items():
            if v and k in NUMBER_TAGS:
                try:
                    if '/' in v:
                        values = v.split('/')
                        values = [int(i) for i in values]
                        instance_meta[k] = values
                    instance_meta[k] = int(v)
                except (TypeError, ValueError):
                    _LOG.error(f'Invalid value specified for \'{k}\' tag, '
                               f'skipping')
        return instance_meta

    def generate_constant_load_metrics(self, config, metric_file_path):
        period_days = config.get(TAG_PERIOD_DAYS)
        length = POINTS_IN_DAY * period_days

        instance_id_series, instance_type_series, timestamp_series, \
        shape_series, shape_size_koef_series = self.generate_common_columns(
            metric_file_path=metric_file_path, length=length
        )

        deviation = config.get(TAG_STD)
        cpu_avg = config.get(TAG_CPU)
        memory_avg = config.get(TAG_MEMORY)
        avg_iops_avg = config.get(TAG_AVG_DISK_IOPS)
        max_iops_avg = config.get(TAG_MAX_DISK_IOPS)
        net_output_avg = config.get(TAG_NET_OUTPUT_LOAD)

        cpu_load_series = generate_constant_metric_series(
            distribution='normal',
            loc=cpu_avg,
            scale=deviation,
            size=length
        )
        memory_load_series = generate_constant_metric_series(
            distribution='normal',
            loc=memory_avg,
            scale=deviation,
            size=length
        )
        net_output_load_series = generate_constant_metric_series(
            distribution='normal',
            loc=net_output_avg,
            scale=deviation,
            size=length
        )
        avg_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=avg_iops_avg,
            scale=deviation,
            size=length
        )
        max_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=max_iops_avg,
            scale=deviation,
            size=length
        )
        df_data = {
            'instance_id': instance_id_series,
            'instance_type': instance_type_series,
            'shape': shape_series,
            'shape_size_koef': shape_size_koef_series,
            'timestamp': timestamp_series,
            'cpu_load': cpu_load_series,
            'memory_load': memory_load_series,
            'net_output_load': net_output_load_series,
            'avg_disk_iops': avg_iops_series,
            'max_disk_iops': max_iops_series,
        }
        df = pd.DataFrame(df_data)
        df.to_csv(metric_file_path, sep=',', index=False)

    def generate_schedule_load_metrics(self, config, metric_file_path):
        period_days = config.get(TAG_PERIOD_DAYS)
        length = POINTS_IN_DAY * period_days

        instance_id_series, instance_type_series, timestamp_series, \
        shape_series, shape_size_koef_series = self.generate_common_columns(
            metric_file_path=metric_file_path, length=length
        )

        deviation = config.get(TAG_STD)
        cpu_avg = config.get(TAG_CPU)
        memory_avg = config.get(TAG_MEMORY)
        avg_iops_avg = config.get(TAG_AVG_DISK_IOPS)
        max_iops_avg = config.get(TAG_MAX_DISK_IOPS)
        net_output_avg = config.get(TAG_NET_OUTPUT_LOAD)
        cron_start = config.get(TAG_CRON_START)
        cron_stop = config.get(TAG_CRON_STOP)

        cron_start_list = self._cron_to_list(cron_start)
        cron_stop_list = self._cron_to_list(cron_stop)

        if not cron_start_list or not cron_stop_list:
            _LOG.error(f'Some of the specified cron strings are not valid')
            return

        work_days, work_hours = self._get_work_days_hours(
            cron_start_list=cron_start_list, cron_stop_list=cron_stop_list)

        cpu_load_series = generate_scheduled_metric_series(
            distribution='normal',
            timestamp_series=timestamp_series,
            work_days=work_days,
            work_hours=work_hours,
            work_kwargs=dict(loc=cpu_avg, scale=deviation),
            idle_kwargs=dict(loc=3, scale=deviation)
        )
        memory_load_series = generate_scheduled_metric_series(
            distribution='normal',
            timestamp_series=timestamp_series,
            work_days=work_days,
            work_hours=work_hours,
            work_kwargs=dict(loc=memory_avg, scale=deviation),
            idle_kwargs=dict(loc=3, scale=deviation)
        )
        net_output_load_series = generate_constant_metric_series(
            distribution='normal',
            loc=net_output_avg,
            scale=deviation,
            size=length
        )
        avg_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=avg_iops_avg,
            scale=deviation,
            size=length
        )
        max_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=max_iops_avg,
            scale=deviation,
            size=length
        )
        df_data = {
            'instance_id': instance_id_series,
            'instance_type': instance_type_series,
            'shape': shape_series,
            'shape_size_koef': shape_size_koef_series,
            'timestamp': timestamp_series,
            'cpu_load': cpu_load_series,
            'memory_load': memory_load_series,
            'net_output_load': net_output_load_series,
            'avg_disk_iops': avg_iops_series,
            'max_disk_iops': max_iops_series,
        }
        df = pd.DataFrame(df_data)
        df.to_csv(metric_file_path, sep=',', index=False)

    def generate_split_load_metrics(self, config, metric_file_path):
        period_days = config.get(TAG_PERIOD_DAYS)
        length = POINTS_IN_DAY * period_days

        instance_id_series, instance_type_series, timestamp_series, \
        shape_series, shape_size_koef_series = self.generate_common_columns(
            metric_file_path=metric_file_path, length=length
        )

        deviation = config.get(TAG_STD)
        cpu_avg = config.get(TAG_CPU)
        memory_avg = config.get(TAG_MEMORY)
        avg_iops_avg = config.get(TAG_AVG_DISK_IOPS)
        max_iops_avg = config.get(TAG_MAX_DISK_IOPS)
        net_output_avg = config.get(TAG_NET_OUTPUT_LOAD)
        probability = config.get(TAG_PROBABILITY)

        cpu_load_series = generate_split_series(
            distribution='normal',
            avg_loads=cpu_avg,
            probabilities=probability,
            scale=deviation,
            size=length
        )
        memory_load_series = generate_split_series(
            distribution='normal',
            avg_loads=memory_avg,
            probabilities=probability,
            scale=deviation,
            size=length
        )
        net_output_load_series = generate_constant_metric_series(
            distribution='normal',
            loc=net_output_avg,
            scale=deviation,
            size=length
        )
        avg_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=avg_iops_avg,
            scale=deviation,
            size=length
        )
        max_iops_series = generate_constant_metric_series(
            distribution='normal',
            loc=max_iops_avg,
            scale=deviation,
            size=length
        )

        df_data = {
            'instance_id': instance_id_series,
            'instance_type': instance_type_series,
            'shape': shape_series,
            'shape_size_koef': shape_size_koef_series,
            'timestamp': timestamp_series,
            'cpu_load': cpu_load_series,
            'memory_load': memory_load_series,
            'net_output_load': net_output_load_series,
            'avg_disk_iops': avg_iops_series,
            'max_disk_iops': max_iops_series,
        }
        df = pd.DataFrame(df_data)
        df.to_csv(metric_file_path, sep=',', index=False)

    def generate_common_columns(self, metric_file_path, length):
        timestamp_series = generate_timestamp_series(
            length=length
        )

        shape_cloud_mapping = {
            CloudEnum.CLOUD_AWS.value: 'c5.4xlarge',
            CloudEnum.CLOUD_AZURE.value: 'Standard_D8_v5',
            CloudEnum.CLOUD_GOOGLE.value: 'n2-standard-8',
        }

        instance_id = self.os_service.path_to_instance_id(
            file_path=metric_file_path)
        cloud = self.os_service.path_to_cloud(file_path=metric_file_path)
        instance_id_series = constant_to_series(
            value=instance_id,
            length=length
        )
        instance_type_series = constant_to_series(
            value=shape_cloud_mapping.get(cloud.upper()),
            length=length
        )
        shape_series = constant_to_series('stub', length)
        shape_size_koef_series = constant_to_series(0.5, length)

        return instance_id_series, instance_type_series, timestamp_series, \
               shape_series, shape_size_koef_series

    @staticmethod
    def _cron_to_list(cron_str):
        try:
            cron = Cron(cron_str)
            return cron.to_list()
        except ValueError:
            _LOG.error(f'Invalid cron string specified: \'{cron_str}\'')

    @staticmethod
    def _get_work_days_hours(cron_start_list, cron_stop_list):
        start_hours = cron_start_list[1]
        stop_hours = cron_stop_list[1]

        start_week_days = cron_start_list[-1]
        stop_week_days = cron_stop_list[-1]

        work_hours = [list(range(start, stop)) for start, stop in
                      zip(start_hours, stop_hours)]
        work_hours = list(set([item for sublist in work_hours
                               for item in sublist]))
        work_days = list(set(start_week_days + stop_week_days))
        return work_days, work_hours

    @staticmethod
    def parse_tags(instance_meta_mapping: dict):
        result = {}

        for instance_id, instance_meta in instance_meta_mapping.items():
            _LOG.debug(f'Parsing instance tags from meta: \'{instance_meta}\'')
            tags_list = instance_meta.get('tags')
            if not tags_list or tags_list and not isinstance(tags_list, list):
                result[instance_id] = {}
                continue
            instance_tags = {}
            for item in tags_list:
                if not isinstance(item, dict):
                    continue
                key = item.get('key')
                value = item.get('value')

                if not isinstance(key, str) or not isinstance(value, str):
                    _LOG.warning(f'Both tag key and value must be strings: '
                                 f'{key}:{value}')
                    continue
                instance_tags[key] = value
            result[instance_id] = instance_tags
        return result