#!/usr/bin/env python3 import os import time import json import argparse import threading import logging from datetime import datetime logger = logging.getLogger(__name__) class TestRunner(): def __init__(self, path='/mnt/shared', threads=1, duration=10, prepare='true'): self.path = path self.target = os.path.join(path, '.eva_benchmark_file_append.tmp') self.threads = threads or 1 self.duration = duration or 10 self.prepare = prepare self.results = [] self.metrics = [] self.block_size = 1024 # 1kB self.initial_size = 50 * 1024 * 1024 # 50MB self._bytes_written = 0 self._ops = 0 self._lock = threading.Lock() def _create_initial_file(self): alignment = 4096 buf = b"x" * alignment # small buffer for append if os.path.exists(self.target): current_size = os.path.getsize(self.target) logger.debug("Файл существует: %s, %d байт", self.target, current_size) if current_size >= self.initial_size: logger.debug("Размер файла достаточный") # на всякий случай синкаем и идём дальше os.sync() return to_add = self.initial_size - current_size logger.debug("Файл меньше %d, дописываем %d байт", self.initial_size, to_add) with open(self.target, "ab", buffering=0) as f: while to_add > 0: write_size = min(alignment, to_add) f.write(buf[:write_size]) to_add -= write_size f.flush() os.fsync(f.fileno()) logger.debug("Тестовый файл подготовлен") return logger.debug("Создаём тестовый файл на %d байт", self.initial_size) buf = b"x" * 4096 to_add = self.initial_size with open(self.target, "wb") as f: while to_add > 0: write_size = min(len(buf), to_add) f.write(buf[:write_size]) to_add -= write_size f.flush() os.fsync(f.fileno()) logger.debug("Тестовый файл подготовлен") def _generate_log_block(self, thread_id, counter=0): lines = [] base_pattern = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" pattern_len = len(base_pattern) total = 0 while total < self.block_size: ts = datetime.now().isoformat() prefix = f"{ts} [thread-{thread_id}] INFO event_id={counter} " payload_size = 200 payload = "".join( base_pattern[(counter + i) % pattern_len] for i in range(payload_size) ) line = prefix + payload + "\n" encoded = line.encode() lines.append(encoded) total += len(encoded) counter += 1 return counter, b"".join(lines) def _worker(self, stop_event, thread_id): # каунтер для пущей уникальности строк, даже если паттерн повторится counter = 0 while not stop_event.is_set(): counter, data = self._generate_log_block(thread_id, counter) with open(self.target, "ab", buffering=0) as f: f.write(data) with self._lock: self._bytes_written += len(data) self._ops += 1 def run_test(self): if self.prepare != 'skip': self._create_initial_file() if self.prepare == 'only': return stop_event = threading.Event() threads = [] logger.info(f"Тест дозаписи в файл: потоки={self.threads}, длительность={self.duration}s") start = time.time() for i in range(self.threads): t = threading.Thread(target=self._worker,args=(stop_event, i),daemon=True) t.start() threads.append(t) time.sleep(self.duration) stop_event.set() for t in threads: t.join() end = time.time() elapsed = end - start try: os.unlink(self.target) except: pass total_mb = self._bytes_written / (1024 * 1024) mbps = total_mb / elapsed if elapsed else 0 iops = self._ops / elapsed if elapsed else 0 # JSON для benchmark_runner val = round(mbps, 3) ref = 7 fmt = "{x:.2f} MB/s" self.metrics.append({ "title": f"Дозапись файла, операции по 1Кб за 10 секунд, {self.path}, скорость (не менее {fmt.format(x=ref)})", "value": val, "reference": ref, "format": fmt, "condition": "greater" }) val = round(iops, 3) ref = 700 fmt = "{x:.2f} ops/sec" self.metrics.append({ "title": f"Дозапись файла, операции по 1Кб за 10 секунд, {self.path}, операций в секнуду (не менее {fmt.format(x=ref)})", "value": val, "reference": ref, "format": fmt, "condition": "greater" }) # val = round(total_mb, 3) # ref = 70 # fmt = "{x:.2f} MB" # self.metrics.append({ # "title": f"Дозапись файла, операции по 1Кб за 10 секунд, {self.path}, всего записано (не менее {fmt.format(x=ref)})", # "value": val, # "reference": ref, # "format": fmt, # "condition": "greater" # }) # Человеко-читаемый вывод self.results = [ { "human": f"Прошло времени: {elapsed:.2f} sec" }, { "human": f"Записано данных: {total_mb:.2f} MB" }, { "human": f"Скорость записи: {mbps:.2f} MB/s" }, { "human": f"Количество операций за время: {iops:.2f} ops/sec" } ] def configure_logging(debug=False): level = logging.DEBUG if debug else logging.INFO logging.basicConfig( level=level, format="%(asctime)s - [%(levelname)s] - %(module)s.%(funcName)s - %(message)s", ) def parse_args(): parser = argparse.ArgumentParser(description="""Тест дозаписи в файл""") parser.add_argument("--path", type=str, default="/mnt/shared/test.log", help="Путь к тестовому файлу") parser.add_argument("--json", action="store_true", help="Вывести метрики в json") parser.add_argument("--duration", type=int, help="Длительность теста в секундах (погрешность в 1-2 секунды)") parser.add_argument("--threads", type=int, help="Количество потоков, одновременно работающих с файлом") parser.add_argument("--debug", action="store_true", default=False, help='Добавить дебаг (не используется при --json)') parser.add_argument("--prepare", type=str, default="true", choices=['true', 'only', 'skip'], help="Подготовить файлы перед тестами (по-умолчанию - true)") return parser.parse_args() if __name__ == "__main__": args = parse_args() if not args.json: configure_logging(debug=args.debug) runner = TestRunner(path=args.path, prepare=args.prepare, threads=args.threads, duration=args.duration) runner.run_test() if args.json: print(json.dumps(runner.metrics, ensure_ascii=False)) else: if runner.results: logging.info("===== ИТОГ =====") for result in runner.results: logger.info(result['human'])