chaos_framework.py - master - Masimo05/Lab_11

chaos_framework.py
419 строк18 KB
Masimo05
upload files
18 дек 2025, 23:20
18 дек 2025, 23:2079b2819
import time
import random
import subprocess
import psutil
import requests
from datetime import datetime
import pandas as pd
from cloud_client import CloudDataClient

class ChaosFramework:
    def __init__(self, use_localstack=True, client=None):
        # self.client = CloudDataClient(use_localstack)
        if client:
            self.client = client
        else:
            self.client = CloudDataClient(use_localstack)
        self.experiments_log = []
        
    def log_experiment(self, experiment_type, description, success):
        """Логируем результаты эксперимента"""
        experiment = {
            'timestamp': datetime.now().isoformat(),
            'type': experiment_type,
            'description': description,
            'success': success,
            'duration': getattr(self, 'current_duration', 0)
        }
        self.experiments_log.append(experiment)
        print(f"📝 {experiment_type}: {description} - {'✅ УСПЕХ' if success else '❌ ПРОВАЛ'}")
    
    def network_latency(self, duration=30, latency_ms=1000):
        """Добавляем сетевую задержку (эмуляция)"""
        print(f"🌐 Добавляем сетевую задержку {latency_ms}мс на {duration} секунд...")
        
        start_time = time.time()
        self.current_duration = duration
        
        try:
            # Эмулируем задержку путем добавления sleep в критические операции
            original_upload = self.client.upload_csv_to_s3
            
            def delayed_upload(*args, **kwargs):
                time.sleep(latency_ms / 1000)  # Конвертируем мс в секунды
                return original_upload(*args, **kwargs)
            
            # Временно заменяем метод
            self.client.upload_csv_to_s3 = delayed_upload
            
            # Ждем указанное время
            time.sleep(duration)
            
            # Восстанавливаем оригинальный метод
            self.client.upload_csv_to_s3 = original_upload
            
            self.log_experiment(
                "NETWORK_LATENCY",
                f"Задержка {latency_ms}мс в течение {duration}с",
                True
            )
            return True
            
        except Exception as e:
            self.log_experiment(
                "NETWORK_LATENCY",
                f"Задержка {latency_ms}мс в течение {duration}с",
                False
            )
            print(f"❌ Ошибка: {e}")
            return False
    
    def service_failure(self, service_type, failure_duration=20):
        """Эмулируем отказ сервиса"""
        print(f"🔥 Эмулируем отказ {service_type} на {failure_duration} секунд...")
        
        self.current_duration = failure_duration
        
        try:
            if service_type == "S3":
                # Сохраняем оригинальные методы
                original_upload = self.client.upload_csv_to_s3
                original_download = self.client.download_csv_from_s3
                
                # Заменяем на методы, которые всегда падают
                def failing_upload(*args, **kwargs):
                    # Добавляем задержку для эмуляции сетевого timeout
                    time.sleep(1)  # Задержка 1 секунда
                    raise Exception("S3 service unavailable - chaos experiment")
                
                def failing_download(*args, **kwargs):
                    # Добавляем задержку для эмуляции сетевого timeout
                    time.sleep(1)  # Задержка 1 секунда
                    raise Exception("S3 service unavailable - chaos experiment")
                
                self.client.upload_csv_to_s3 = failing_upload
                self.client.download_csv_from_s3 = failing_download
                
                # Ждем указанное время
                time.sleep(failure_duration)
                
                # Восстанавливаем оригинальные методы
                self.client.upload_csv_to_s3 = original_upload
                self.client.download_csv_from_s3 = original_download
            
            elif service_type == "SQS":
                original_send = self.client.send_message
                original_receive = self.client.receive_messages
                
                def failing_send(*args, **kwargs):
                    # Добавляем задержку
                    time.sleep(1)  # Задержка 1 секунда
                    raise Exception("SQS service unavailable - chaos experiment")
                
                def failing_receive(*args, **kwargs):
                    # Добавляем задержку
                    time.sleep(1)  # Задержка 1 секунда
                    raise Exception("SQS service unavailable - chaos experiment")
                
                self.client.send_message = failing_send
                self.client.receive_messages = failing_receive
                
                time.sleep(failure_duration)
                
                self.client.send_message = original_send
                self.client.receive_messages = original_receive
            
            self.log_experiment(
                "SERVICE_FAILURE",
                f"Отказ {service_type} в течение {failure_duration}с",
                True
            )
            return True
            
        except Exception as e:
            self.log_experiment(
                "SERVICE_FAILURE", 
                f"Отказ {service_type} в течение {failure_duration}с",
                False
            )
            print(f"❌ Ошибка: {e}")
            return False
    
    def high_cpu_load(self, duration=30, load_percent=80):
        """Создаем высокую нагрузку на CPU"""
        print(f"🔥 Создаем нагрузку на CPU ({load_percent}%) на {duration} секунд...")
        
        self.current_duration = duration
        start_time = time.time()
        
        try:
            # Функция, которая нагружает CPU
            def cpu_stress():
                while time.time() - start_time < duration:
                    # Вычисления, которые нагружают CPU
                    [x**2 for x in range(10000)]
            
            # Запускаем в отдельном процессе
            import threading
            stress_thread = threading.Thread(target=cpu_stress)
            stress_thread.daemon = True
            stress_thread.start()
            
            # Мониторим нагрузку
            while time.time() - start_time < duration:
                cpu_percent = psutil.cpu_percent(interval=1)
                print(f"⚡ Текущая нагрузка CPU: {cpu_percent}%")
                time.sleep(2)
            
            self.log_experiment(
                "HIGH_CPU_LOAD",
                f"Нагрузка CPU {load_percent}% в течение {duration}с",
                True
            )
            return True
            
        except Exception as e:
            self.log_experiment(
                "HIGH_CPU_LOAD",
                f"Нагрузка CPU {load_percent}% в течение {duration}с", 
                False
            )
            print(f"❌ Ошибка: {e}")
            return False
    
    def memory_pressure(self, duration=30, memory_mb=500):
        """Создаем давление на память"""
        print(f"💾 Создаем давление на память ({memory_mb}MB) на {duration} секунд...")
        
        self.current_duration = duration
        start_time = time.time()
        memory_blocks = []
        
        try:
            # Резервируем память
            block_size = 1024 * 1024  # 1MB
            blocks_needed = memory_mb
            
            for i in range(blocks_needed):
                try:
                    # Создаем блок памяти
                    block = ' ' * block_size
                    memory_blocks.append(block)
                    if i % 50 == 0:  # Каждые 50MB выводим прогресс
                        print(f"📦 Зарезервировано {i}MB из {memory_mb}MB")
                except MemoryError:
                    print("⚠️ Достигнут лимит памяти")
                    break
            
            print(f"✅ Зарезервировано {len(memory_blocks)}MB памяти")
            
            # Ждем указанное время
            time.sleep(duration)
            
            # Освобождаем память
            memory_blocks.clear()
            
            self.log_experiment(
                "MEMORY_PRESSURE", 
                f"Давление на память {memory_mb}MB в течение {duration}с",
                True
            )
            return True
            
        except Exception as e:
            self.log_experiment(
                "MEMORY_PRESSURE",
                f"Давление на память {memory_mb}MB в течение {duration}с",
                False
            )
            print(f"❌ Ошибка: {e}")
            return False
    
    def data_corruption(self, probability=0.3):
        """Эмулируем коррупцию данных"""
        print(f"📉 Эмулируем коррупцию данных с вероятностью {probability*100}%...")
        
        try:
            # Сохраняем оригинальный метод
            original_upload = self.client.upload_csv_to_s3
            
            def corrupting_upload(dataframe, bucket_name, file_key):
                if random.random() < probability:
                    print("💀 Коррумпируем данные...")
                    
                    # Разные типы коррупции данных
                    corruption_type = random.choice([
                        'null_values', 'duplicates', 'wrong_format', 'truncated'
                    ])
                    
                    corrupted_data = dataframe.copy()
                    
                    if corruption_type == 'null_values':
                        # Добавляем пустые значения
                        for col in corrupted_data.columns:
                            if random.random() < 0.2:  # 20% значений в колонке
                                corrupted_data[col] = None
                    
                    elif corruption_type == 'duplicates':
                        # Добавляем дубликаты
                        duplicates = corrupted_data.sample(n=min(5, len(corrupted_data)))
                        corrupted_data = pd.concat([corrupted_data, duplicates])
                    
                    elif corruption_type == 'wrong_format':
                        # Портим формат данных
                        if 'salary' in corrupted_data.columns:
                            corrupted_data['salary'] = 'INVALID_' + corrupted_data['salary'].astype(str)
                    
                    elif corruption_type == 'truncated':
                        # Обрезаем данные
                        corrupted_data = corrupted_data.head(max(1, len(corrupted_data) // 2))
                    
                    print(f"🔧 Применена коррупция: {corruption_type}")
                    return original_upload(corrupted_data, bucket_name, file_key)
                else:
                    return original_upload(dataframe, bucket_name, file_key)
            
            # Временно заменяем метод
            self.client.upload_csv_to_s3 = corrupting_upload
            
            self.log_experiment(
                "DATA_CORRUPTION",
                f"Коррупция данных с вероятностью {probability*100}%",
                True
            )
            return True
            
        except Exception as e:
            self.log_experiment(
                "DATA_CORRUPTION",
                f"Коррупция данных с вероятностью {probability*100}%",
                False
            )
            print(f"❌ Ошибка: {e}")
            return False
    
    def stop_data_corruption(self):
        """Останавливаем коррупцию данных"""
        # Восстанавливаем оригинальный метод
        from cloud_client import CloudDataClient
        original_client = CloudDataClient()
        self.client.upload_csv_to_s3 = original_client.upload_csv_to_s3
        print("✅ Коррупция данных остановлена")
    
    def run_chaos_monkey(self, duration=300, interval=30):
        """Запускаем Chaos Monkey - случайные сбои"""
        print(f"🐵 Запускаем Chaos Monkey на {duration} секунд...")
        print("🎲 Случайные сбои будут происходить каждые 30 секунд")
        
        start_time = time.time()
        experiment_count = 0
        
        while time.time() - start_time < duration:
            # Выбираем случайный эксперимент
            experiments = [
                ('NETWORK_LATENCY', lambda: self.network_latency(20, 500)),
                ('SERVICE_FAILURE_S3', lambda: self.service_failure("S3", 15)),
                ('SERVICE_FAILURE_SQS', lambda: self.service_failure("SQS", 15)),
                ('HIGH_CPU_LOAD', lambda: self.high_cpu_load(20, 70)),
                ('MEMORY_PRESSURE', lambda: self.memory_pressure(20, 300)),
                ('DATA_CORRUPTION', lambda: self.data_corruption(0.4))
            ]
            
            experiment_name, experiment_func = random.choice(experiments)
            
            print(f"\n🎯 Запускаем эксперимент: {experiment_name}")
            success = experiment_func()
            
            if success:
                experiment_count += 1
            
            # Ждем перед следующим экспериментом
            time_left = duration - (time.time() - start_time)
            if time_left > interval:
                print(f"⏰ Следующий эксперимент через {interval} секунд...")
                time.sleep(interval)
            else:
                break
        
        print(f"\n🎉 Chaos Monkey завершил работу")
        print(f"📊 Выполнено экспериментов: {experiment_count}")
        return experiment_count
    
    def generate_report(self):
        """Генерируем отчет по экспериментам"""
        print("\n📊 ГЕНЕРИРУЕМ ОТЧЕТ ПО CHAOS ENGINEERING")
        print("=" * 50)
        
        if not self.experiments_log:
            print("❌ Нет данных для отчета")
            return None
        
        # Создаем DataFrame с результатами
        df = pd.DataFrame(self.experiments_log)
        
        # Статистика по типам экспериментов
        experiment_stats = df.groupby('type').agg({
            'success': ['count', 'sum'],
            'duration': 'mean'
        }).round(2)
        
        experiment_stats['success_rate'] = (experiment_stats[('success', 'sum')] / 
                                          experiment_stats[('success', 'count')] * 100)
        
        print("📈 СТАТИСТИКА ЭКСПЕРИМЕНТОВ:")
        print(experiment_stats)
        
        # Общая статистика
        total_experiments = len(df)
        successful_experiments = df['success'].sum()
        success_rate = (successful_experiments / total_experiments) * 100
        
        print(f"\n🎯 ОБЩАЯ СТАТИСТИКА:")
        print(f"Всего экспериментов: {total_experiments}")
        print(f"Успешных: {successful_experiments}")
        print(f"Успешность: {success_rate:.1f}%")

        stats_converted = {}
        for exp_type in experiment_stats.index:
            row = experiment_stats.loc[exp_type]
            stats_converted[exp_type] = {
                'count': int(row[('success', 'count')]),
                'success_count': int(row[('success', 'sum')]),
                'avg_duration': float(row[('duration', 'mean')]),
                'success_rate': float(row['success_rate'])
            }
        
        # Сохраняем отчет
        report = {
            'timestamp': datetime.now().isoformat(),
            'total_experiments': int(total_experiments),
            'successful_experiments': int(successful_experiments),
            'success_rate': float(success_rate),
            'experiment_details': df.to_dict('records'),
            'statistics': stats_converted
        }
        
        # Сохраняем в файл
        with open('chaos_report.json', 'w') as f:
            import json
            json.dump(report, f, indent=2)
        
        print("✅ Отчет сохранен в chaos_report.json")
        return report

# Пример использования
if __name__ == "__main__":
    chaos = ChaosFramework(use_localstack=True)
    
    print("🎯 ДЕМОНСТРАЦИЯ CHAOS ENGINEERING")
    print("=" * 40)
    
    # Запускаем несколько экспериментов
    chaos.network_latency(15, 800)
    chaos.service_failure("S3", 15)
    chaos.high_cpu_load(20, 60)
    chaos.data_corruption(0.5)
    chaos.stop_data_corruption()
    
    # Генерируем отчет
    chaos.generate_report()