#!/usr/bin/env python3 """ Phantom TLD System Performance Testing Suite Комплексное тестирование производительности и масштабируемости Автор: Phantom Protocol Team 2025 Версия: 1.0.0 """ import asyncio import aiohttp import time import json import random import string import statistics import concurrent.futures import threading from dataclasses import dataclass from typing import List, Dict, Any, Optional import argparse import logging import sys import socket import dns.resolver import dns.query import dns.message import psutil import matplotlib.pyplot as plt import numpy as np # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('tld-performance-test.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) @dataclass class TestConfig: """Конфигурация тестирования""" tld_nodes: List[str] dns_servers: List[str] test_duration: int = 300 # секунд concurrent_users: int = 100 domains_per_user: int = 10 query_rate: int = 1000 # запросов в секунду domain_prefix: str = "test" tld_name: str = "phantom" @dataclass class TestResult: """Результат тестирования""" test_name: str duration: float total_operations: int successful_operations: int failed_operations: int avg_response_time: float min_response_time: float max_response_time: float p95_response_time: float p99_response_time: float throughput: float error_rate: float class PerformanceMonitor: """Мониторинг производительности системы""" def __init__(self): self.metrics = { 'cpu_usage': [], 'memory_usage': [], 'network_io': [], 'disk_io': [], 'timestamps': [] } self.monitoring = False def start_monitoring(self): """Запуск мониторинга""" self.monitoring = True self.monitor_thread = threading.Thread(target=self._monitor_loop) self.monitor_thread.start() def stop_monitoring(self): """Остановка мониторинга""" self.monitoring = False if hasattr(self, 'monitor_thread'): self.monitor_thread.join() def _monitor_loop(self): """Цикл мониторинга""" while self.monitoring: timestamp = time.time() # CPU использование cpu_percent = psutil.cpu_percent(interval=1) # Память memory = psutil.virtual_memory() memory_percent = memory.percent # Сетевой I/O net_io = psutil.net_io_counters() # Дисковый I/O disk_io = psutil.disk_io_counters() self.metrics['timestamps'].append(timestamp) self.metrics['cpu_usage'].append(cpu_percent) self.metrics['memory_usage'].append(memory_percent) self.metrics['network_io'].append({ 'bytes_sent': net_io.bytes_sent, 'bytes_recv': net_io.bytes_recv, 'packets_sent': net_io.packets_sent, 'packets_recv': net_io.packets_recv }) self.metrics['disk_io'].append({ 'read_bytes': disk_io.read_bytes if disk_io else 0, 'write_bytes': disk_io.write_bytes if disk_io else 0 }) time.sleep(5) # Сбор метрик каждые 5 секунд def get_summary(self) -> Dict[str, Any]: """Получение сводки метрик""" if not self.metrics['cpu_usage']: return {} return { 'avg_cpu_usage': statistics.mean(self.metrics['cpu_usage']), 'max_cpu_usage': max(self.metrics['cpu_usage']), 'avg_memory_usage': statistics.mean(self.metrics['memory_usage']), 'max_memory_usage': max(self.metrics['memory_usage']), 'duration': self.metrics['timestamps'][-1] - self.metrics['timestamps'][0] if len(self.metrics['timestamps']) > 1 else 0 } class DomainGenerator: """Генератор доменных имен для тестирования""" @staticmethod def generate_random_domain(prefix: str = "test", tld: str = "phantom") -> str: """Генерация случайного домена""" random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8)) return f"{prefix}-{random_suffix}.{tld}" @staticmethod def generate_domain_batch(count: int, prefix: str = "test", tld: str = "phantom") -> List[str]: """Генерация пакета доменов""" return [DomainGenerator.generate_random_domain(prefix, tld) for _ in range(count)] class TLDSystemTester: """Основной класс для тестирования TLD системы""" def __init__(self, config: TestConfig): self.config = config self.monitor = PerformanceMonitor() self.session = None async def __aenter__(self): """Асинхронный контекст менеджер - вход""" self.session = aiohttp.ClientSession( timeout=aiohttp.ClientTimeout(total=30), connector=aiohttp.TCPConnector(limit=1000) ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Асинхронный контекст менеджер - выход""" if self.session: await self.session.close() async def register_domain(self, domain: str, ipv4: str = "192.168.1.100") -> tuple[bool, float]: """Регистрация домена через API""" start_time = time.time() try: node = random.choice(self.config.tld_nodes) url = f"http://{node}/api/domains" payload = { "domain": domain.split('.')[0], "tld": domain.split('.')[1], "ipv4": ipv4, "ttl": 3600 } async with self.session.post(url, json=payload) as response: success = response.status == 200 or response.status == 201 response_time = time.time() - start_time return success, response_time except Exception as e: logger.error(f"Ошибка регистрации домена {domain}: {e}") response_time = time.time() - start_time return False, response_time async def query_domain(self, domain: str) -> tuple[bool, float]: """DNS запрос домена""" start_time = time.time() try: dns_server = random.choice(self.config.dns_servers) # Создание DNS запроса query = dns.message.make_query(domain, dns.rdatatype.A) # Отправка запроса response = await asyncio.get_event_loop().run_in_executor( None, lambda: dns.query.udp(query, dns_server.split(':')[0], port=int(dns_server.split(':')[1]) if ':' in dns_server else 53, timeout=5) ) success = len(response.answer) > 0 response_time = time.time() - start_time return success, response_time except Exception as e: logger.error(f"Ошибка DNS запроса {domain}: {e}") response_time = time.time() - start_time return False, response_time async def test_domain_registration(self) -> TestResult: """Тест регистрации доменов""" logger.info("Запуск теста регистрации доменов...") self.monitor.start_monitoring() start_time = time.time() # Генерация доменов для тестирования total_domains = self.config.concurrent_users * self.config.domains_per_user domains = DomainGenerator.generate_domain_batch( total_domains, self.config.domain_prefix, self.config.tld_name ) # Асинхронная регистрация доменов tasks = [] for domain in domains: task = self.register_domain(domain) tasks.append(task) # Выполнение с ограничением concurrency semaphore = asyncio.Semaphore(self.config.concurrent_users) async def limited_register(domain): async with semaphore: return await self.register_domain(domain) results = await asyncio.gather(*[limited_register(domain) for domain in domains]) end_time = time.time() self.monitor.stop_monitoring() # Анализ результатов successful = sum(1 for success, _ in results if success) failed = len(results) - successful response_times = [rt for _, rt in results] return TestResult( test_name="Domain Registration", duration=end_time - start_time, total_operations=len(results), successful_operations=successful, failed_operations=failed, avg_response_time=statistics.mean(response_times) if response_times else 0, min_response_time=min(response_times) if response_times else 0, max_response_time=max(response_times) if response_times else 0, p95_response_time=np.percentile(response_times, 95) if response_times else 0, p99_response_time=np.percentile(response_times, 99) if response_times else 0, throughput=successful / (end_time - start_time) if end_time > start_time else 0, error_rate=failed / len(results) if results else 0 ) async def test_dns_queries(self) -> TestResult: """Тест DNS запросов""" logger.info("Запуск теста DNS запросов...") self.monitor.start_monitoring() start_time = time.time() # Предварительная регистрация доменов для тестирования test_domains = DomainGenerator.generate_domain_batch(100, self.config.domain_prefix, self.config.tld_name) logger.info("Регистрация тестовых доменов...") for domain in test_domains: await self.register_domain(domain) # Ожидание распространения записей await asyncio.sleep(10) # Генерация запросов total_queries = self.config.query_rate * (self.config.test_duration // 60) # запросы за минуту async def query_worker(): """Рабочий поток для выполнения запросов""" results = [] queries_per_worker = total_queries // self.config.concurrent_users for _ in range(queries_per_worker): domain = random.choice(test_domains) success, response_time = await self.query_domain(domain) results.append((success, response_time)) # Контроль частоты запросов await asyncio.sleep(60 / self.config.query_rate) return results # Запуск рабочих потоков tasks = [query_worker() for _ in range(self.config.concurrent_users)] worker_results = await asyncio.gather(*tasks) end_time = time.time() self.monitor.stop_monitoring() # Объединение результатов all_results = [] for worker_result in worker_results: all_results.extend(worker_result) # Анализ результатов successful = sum(1 for success, _ in all_results if success) failed = len(all_results) - successful response_times = [rt for _, rt in all_results] return TestResult( test_name="DNS Queries", duration=end_time - start_time, total_operations=len(all_results), successful_operations=successful, failed_operations=failed, avg_response_time=statistics.mean(response_times) if response_times else 0, min_response_time=min(response_times) if response_times else 0, max_response_time=max(response_times) if response_times else 0, p95_response_time=np.percentile(response_times, 95) if response_times else 0, p99_response_time=np.percentile(response_times, 99) if response_times else 0, throughput=successful / (end_time - start_time) if end_time > start_time else 0, error_rate=failed / len(all_results) if all_results else 0 ) async def test_mixed_workload(self) -> TestResult: """Тест смешанной нагрузки (регистрация + запросы)""" logger.info("Запуск теста смешанной нагрузки...") self.monitor.start_monitoring() start_time = time.time() async def mixed_worker(): """Рабочий поток для смешанной нагрузки""" results = [] for _ in range(self.config.domains_per_user): # 70% запросов, 30% регистраций if random.random() < 0.7: # DNS запрос domain = DomainGenerator.generate_random_domain(self.config.domain_prefix, self.config.tld_name) success, response_time = await self.query_domain(domain) else: # Регистрация домена domain = DomainGenerator.generate_random_domain(self.config.domain_prefix, self.config.tld_name) success, response_time = await self.register_domain(domain) results.append((success, response_time)) await asyncio.sleep(0.1) # Небольшая задержка между операциями return results # Запуск рабочих потоков tasks = [mixed_worker() for _ in range(self.config.concurrent_users)] worker_results = await asyncio.gather(*tasks) end_time = time.time() self.monitor.stop_monitoring() # Объединение результатов all_results = [] for worker_result in worker_results: all_results.extend(worker_result) # Анализ результатов successful = sum(1 for success, _ in all_results if success) failed = len(all_results) - successful response_times = [rt for _, rt in all_results] return TestResult( test_name="Mixed Workload", duration=end_time - start_time, total_operations=len(all_results), successful_operations=successful, failed_operations=failed, avg_response_time=statistics.mean(response_times) if response_times else 0, min_response_time=min(response_times) if response_times else 0, max_response_time=max(response_times) if response_times else 0, p95_response_time=np.percentile(response_times, 95) if response_times else 0, p99_response_time=np.percentile(response_times, 99) if response_times else 0, throughput=successful / (end_time - start_time) if end_time > start_time else 0, error_rate=failed / len(all_results) if all_results else 0 ) def generate_report(self, results: List[TestResult], system_metrics: Dict[str, Any]) -> str: """Генерация отчета о тестировании""" report = [] report.append("# Phantom TLD System - Отчет о тестировании производительности") report.append(f"Дата: {time.strftime('%Y-%m-%d %H:%M:%S')}") report.append("") # Конфигурация тестирования report.append("## Конфигурация тестирования") report.append(f"- TLD узлы: {', '.join(self.config.tld_nodes)}") report.append(f"- DNS серверы: {', '.join(self.config.dns_servers)}") report.append(f"- Одновременных пользователей: {self.config.concurrent_users}") report.append(f"- Доменов на пользователя: {self.config.domains_per_user}") report.append(f"- Частота запросов: {self.config.query_rate} запросов/сек") report.append("") # Результаты тестов report.append("## Результаты тестирования") for result in results: report.append(f"### {result.test_name}") report.append(f"- Длительность: {result.duration:.2f} сек") report.append(f"- Всего операций: {result.total_operations}") report.append(f"- Успешных: {result.successful_operations}") report.append(f"- Неудачных: {result.failed_operations}") report.append(f"- Пропускная способность: {result.throughput:.2f} операций/сек") report.append(f"- Частота ошибок: {result.error_rate:.2%}") report.append(f"- Среднее время отклика: {result.avg_response_time*1000:.2f} мс") report.append(f"- P95 время отклика: {result.p95_response_time*1000:.2f} мс") report.append(f"- P99 время отклика: {result.p99_response_time*1000:.2f} мс") report.append("") # Системные метрики if system_metrics: report.append("## Системные метрики") report.append(f"- Среднее использование CPU: {system_metrics.get('avg_cpu_usage', 0):.1f}%") report.append(f"- Максимальное использование CPU: {system_metrics.get('max_cpu_usage', 0):.1f}%") report.append(f"- Среднее использование памяти: {system_metrics.get('avg_memory_usage', 0):.1f}%") report.append(f"- Максимальное использование памяти: {system_metrics.get('max_memory_usage', 0):.1f}%") report.append("") # Рекомендации report.append("## Рекомендации") avg_throughput = statistics.mean([r.throughput for r in results]) avg_error_rate = statistics.mean([r.error_rate for r in results]) if avg_throughput > 1000: report.append("✅ Отличная производительность - система готова к продакшену") elif avg_throughput > 500: report.append("⚠️ Хорошая производительность - рекомендуется оптимизация") else: report.append("❌ Низкая производительность - требуется серьезная оптимизация") if avg_error_rate < 0.01: report.append("✅ Отличная надежность") elif avg_error_rate < 0.05: report.append("⚠️ Приемлемая надежность") else: report.append("❌ Высокая частота ошибок - требуется исследование") return "\n".join(report) async def main(): """Главная функция""" parser = argparse.ArgumentParser(description="Phantom TLD System Performance Testing") parser.add_argument("--tld-nodes", nargs="+", default=["localhost:8053"], help="TLD узлы для тестирования") parser.add_argument("--dns-servers", nargs="+", default=["localhost:53"], help="DNS серверы для тестирования") parser.add_argument("--concurrent-users", type=int, default=100, help="Количество одновременных пользователей") parser.add_argument("--domains-per-user", type=int, default=10, help="Количество доменов на пользователя") parser.add_argument("--query-rate", type=int, default=1000, help="Частота DNS запросов (запросов/сек)") parser.add_argument("--test-duration", type=int, default=300, help="Длительность теста в секундах") parser.add_argument("--output", default="tld-performance-report.md", help="Файл для сохранения отчета") args = parser.parse_args() # Создание конфигурации config = TestConfig( tld_nodes=args.tld_nodes, dns_servers=args.dns_servers, concurrent_users=args.concurrent_users, domains_per_user=args.domains_per_user, query_rate=args.query_rate, test_duration=args.test_duration ) logger.info("Запуск тестирования производительности Phantom TLD System") logger.info(f"Конфигурация: {config}") # Выполнение тестов async with TLDSystemTester(config) as tester: results = [] # Тест регистрации доменов try: result = await tester.test_domain_registration() results.append(result) logger.info(f"Тест регистрации завершен: {result.throughput:.2f} операций/сек") except Exception as e: logger.error(f"Ошибка в тесте регистрации: {e}") # Тест DNS запросов try: result = await tester.test_dns_queries() results.append(result) logger.info(f"Тест DNS запросов завершен: {result.throughput:.2f} операций/сек") except Exception as e: logger.error(f"Ошибка в тесте DNS запросов: {e}") # Тест смешанной нагрузки try: result = await tester.test_mixed_workload() results.append(result) logger.info(f"Тест смешанной нагрузки завершен: {result.throughput:.2f} операций/сек") except Exception as e: logger.error(f"Ошибка в тесте смешанной нагрузки: {e}") # Получение системных метрик system_metrics = tester.monitor.get_summary() # Генерация отчета report = tester.generate_report(results, system_metrics) # Сохранение отчета with open(args.output, 'w', encoding='utf-8') as f: f.write(report) logger.info(f"Отчет сохранен в {args.output}") print(f"\nТестирование завершено. Отчет сохранен в {args.output}") # Вывод краткой сводки if results: avg_throughput = statistics.mean([r.throughput for r in results]) avg_error_rate = statistics.mean([r.error_rate for r in results]) print(f"Средняя пропускная способность: {avg_throughput:.2f} операций/сек") print(f"Средняя частота ошибок: {avg_error_rate:.2%}") if __name__ == "__main__": asyncio.run(main())