#!/usr/bin/env python3 """tests/load_test.py — длительная нагрузочная проверка устойчивости shared-sqs. Режимы работы в каждом цикле: 1) Корректные операции: create/get-url/attributes/send/batch/receive(long poll)/ change-visibility/delete/batch-delete/tags/purge/delete-queue + FIFO. 2) Крайние условия: сообщения до 48КБ (большие POST-тела зависают на платформе ~51с из-за MTU — см. HISTORY, раздел «51с MTU drhider»), юникод, пустые тела, пакеты по 10, сообщение > лимита (очередь с MaximumMessageSize=1024 + тело 2KB — ожидается отказ MessageTooBig). 3) Некорректные условия (ожидаемые отказы с известными кодами): несуществующая очередь, битый ReceiptHandle, некорректные атрибуты, повторный purge, >10 записей в batch, дубли Id в batch, пустой batch, FIFO без MessageGroupId, некорректные имена очередей. Контроль: отдельный поток каждые 10с проверяет GET /health. Каждые 30с — строка прогресса. В конце — сводка по кодам ошибок и латентности. Переменные окружения: ENDPOINT_URL (по умолчанию https://sqs.containerk8s.dev.nubes.ru) REGION (us-east-1) DURATION_SECONDS (1800 = 30 минут) WORKERS (4) AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY — обязательны """ import json import os import random import string import sys import threading import time import urllib.request import boto3 from botocore.config import Config from botocore.exceptions import ClientError ENDPOINT = os.environ.get("ENDPOINT_URL", "https://sqs.containerk8s.dev.nubes.ru") REGION = os.environ.get("REGION", "us-east-1") DURATION = int(os.environ.get("DURATION_SECONDS", "1800")) WORKERS = int(os.environ.get("WORKERS", "4")) HEALTH_EVERY = 10 # Коды ошибок, которые ожидаемы для некорректных условий. EXPECTED_CODES = { "NonExistentQueue", "QueueDeletedRecently", "PurgeQueueInProgress", "ReceiptHandleIsInvalid", "InvalidParameterValue", "InvalidAttributeValue", "InvalidAttributeName", "MissingParameter", "ReadCountOutOfRange", "TooManyEntriesInBatchRequest", "BatchEntryIdsNotDistinct", "EmptyBatchRequest", "InvalidBatchEntryId", "MessageTooLong", "InvalidMessageContents", "AWS.SimpleQueueService.NonExistentQueue", "AWS.SimpleQueueService.InvalidParameterValue", "AWS.SimpleQueueService.InvalidAttributeName", "AWS.SimpleQueueService.InvalidAttributeValue", "AWS.SimpleQueueService.PurgeQueueInProgress", "AWS.SimpleQueueService.TooManyEntriesInBatchRequest", "AWS.SimpleQueueService.BatchEntryIdsNotDistinct", "AWS.SimpleQueueService.EmptyBatchRequest", "AWS.SimpleQueueService.InvalidBatchEntryId", } stats = { "valid_ok": 0, "expected_err": 0, "unexpected_fail": 0, "integrity_fail": 0, "soft_warn": 0, # некорректный запрос неожиданно прошёл "health_fail": 0, "ops_total": 0, "err_by_code": {}, } stats_lock = threading.Lock() start_ts = time.time() stop = threading.Event() LATENCY = [] LATENCY_LOCK = threading.Lock() def add_latency(sec): with LATENCY_LOCK: LATENCY.append(sec) if len(LATENCY) > 100000: del LATENCY[:50000] def bump(key, n=1, code=None): with stats_lock: stats[key] += n stats["ops_total"] += n if code: stats["err_by_code"][code] = stats["err_by_code"].get(code, 0) + 1 def record_expected(e: ClientError): code = e.response.get("Error", {}).get("Code", "Unknown") bump("expected_err", code=code) def record_unexpected(e): code = getattr(e, "response", None) and e.response.get("Error", {}).get("Code", "Unknown") or type(e).__name__ bump("unexpected_fail", code=code) def rand_body(max_size=48 * 1024): size = random.randint(0, max_size) return "".join(random.choices(string.ascii_letters + "абвгд\x00\xff😀", k=size)) def make_client(): cfg = Config(connect_timeout=10, read_timeout=25, retries={"max_attempts": 2}) return boto3.client("sqs", endpoint_url=ENDPOINT, region_name=REGION, config=cfg) def health_monitor(): while not stop.is_set(): try: with urllib.request.urlopen( urllib.request.Request(ENDPOINT + "/health", headers={"User-Agent": "load-test"}), timeout=10, ) as r: body = r.read().decode("utf-8", "replace") if r.status != 200 or '"status":"ok"' not in body: bump("health_fail") except Exception: bump("health_fail") stop.wait(HEALTH_EVERY) def valid_cycle(sqs, wid, it): q = f"load-{wid}-{it}-{random.randint(0, 10**9)}" url = sqs.create_queue(QueueName=q)["QueueUrl"] try: assert sqs.get_queue_url(QueueName=q)["QueueUrl"] == url # сообщение произвольного размера до 48КБ, включая юникод body = rand_body() sqs.send_message(QueueUrl=url, MessageBody=body) bump("valid_ok") # batch до 10 entries = [{"Id": str(i), "MessageBody": f"b{i}-{random.randint(0, 10**6)}"} for i in range(random.randint(1, 10))] r = sqs.send_message_batch(QueueUrl=url, Entries=entries) assert len(r["Successful"]) == len(entries) # receive с long poll, сверка целостности msgs = None for _ in range(6): msgs = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=10, WaitTimeSeconds=random.choice([1, 5, 20])).get("Messages", []) if msgs: break time.sleep(1) if not msgs: raise AssertionError("сообщения не получены") bodies = {m["Body"] for m in msgs} if body not in bodies: # тело могло уйти в другой receive предыдущего потока — считаем предупреждением bump("soft_warn") # change visibility + delete sqs.change_message_visibility(QueueUrl=url, ReceiptHandle=msgs[0]["ReceiptHandle"], VisibilityTimeout=1) sqs.delete_message(QueueUrl=url, ReceiptHandle=msgs[0]["ReceiptHandle"]) handles = [{"Id": str(i), "ReceiptHandle": m["ReceiptHandle"]} for i, m in enumerate(msgs[1:])] if handles: sqs.delete_message_batch(QueueUrl=url, Entries=handles) # теги sqs.tag_queue(QueueUrl=url, Tags={"load": "1"}) sqs.list_queue_tags(QueueUrl=url) sqs.untag_queue(QueueUrl=url, TagKeys=["load"]) # атрибуты attrs = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["All"])["Attributes"] assert "VisibilityTimeout" in attrs sqs.set_queue_attributes(QueueUrl=url, Attributes={"VisibilityTimeout": str(random.randint(0, 43200))}) finally: try: sqs.delete_queue(QueueUrl=url) except ClientError: pass def fifo_cycle(sqs, wid, it): q = f"load-{wid}-{it}.fifo" url = sqs.create_queue(QueueName=q, Attributes={"FifoQueue": "true"})["QueueUrl"] try: sqs.send_message(QueueUrl=url, MessageBody="fifo", MessageGroupId="g", MessageDeduplicationId=str(it)) # дубль с тем же DedupId — сервис должен не создать дубль (допустим любой из ответов) sqs.send_message(QueueUrl=url, MessageBody="fifo-dup", MessageGroupId="g", MessageDeduplicationId=str(it)) msgs = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=10).get("Messages", []) if msgs: sqs.delete_message(QueueUrl=url, ReceiptHandle=msgs[0]["ReceiptHandle"]) bump("valid_ok") finally: try: sqs.delete_queue(QueueUrl=url) except ClientError: pass def invalid_cases(sqs): cases = [] # несуществующая очередь cases.append(lambda: sqs.send_message(QueueUrl=ENDPOINT + "/no-such-queue", MessageBody="x")) cases.append(lambda: sqs.receive_message(QueueUrl=ENDPOINT + "/no-such-queue")) cases.append(lambda: sqs.delete_queue(QueueUrl=ENDPOINT + "/no-such-queue")) cases.append(lambda: sqs.get_queue_url(QueueName="no-such-queue-load-test")) # битый ReceiptHandle cases.append(lambda: sqs.delete_message(QueueUrl=ENDPOINT + "/no-such-queue", ReceiptHandle="broken")) # ФИКС: раньше delete_queue стоял в finally и выполнялся ДО прогона сценариев — # сценарии атрибутов/batch/purge/300KB/FIFO тестировали УДАЛЁННУЮ очередь # (NonExistentQueue) вместо заявленных нарушений. Теперь очереди создаются # до прогона и удаляются ПОСЛЕ него. q = None fq = None qlow = None try: # некорректные атрибуты q = sqs.create_queue(QueueName=f"invalid-{random.randint(0, 10**9)}")["QueueUrl"] cases.append(lambda: sqs.set_queue_attributes(QueueUrl=q, Attributes={"VisibilityTimeout": "43201"})) cases.append(lambda: sqs.set_queue_attributes(QueueUrl=q, Attributes={"VisibilityTimeout": "not-a-number"})) cases.append(lambda: sqs.set_queue_attributes(QueueUrl=q, Attributes={"BadAttribute": "1"})) # повторный purge def purge_twice(): sqs.purge_queue(QueueUrl=q) sqs.purge_queue(QueueUrl=q) cases.append(purge_twice) # batch-нарушения cases.append(lambda: sqs.send_message_batch(QueueUrl=q, Entries=[{"Id": str(i), "MessageBody": "x"} for i in range(11)])) cases.append(lambda: sqs.send_message_batch(QueueUrl=q, Entries=[{"Id": "dup", "MessageBody": "a"}, {"Id": "dup", "MessageBody": "b"}])) cases.append(lambda: sqs.send_message_batch(QueueUrl=q, Entries=[])) # пустое тело cases.append(lambda: sqs.send_message(QueueUrl=q, MessageBody="")) # превышение лимита размера: очередь с MaximumMessageSize=1024 + # тело 2KB. Граничное условие (MessageTooBig) сохраняется, но большое # тело по сети НЕ шлётся — иначе платформа вешает запрос на ~51с (MTU). qlow = sqs.create_queue(QueueName=f"invalid-{random.randint(0, 10**9)}", Attributes={"MaximumMessageSize": "1024"})["QueueUrl"] cases.append(lambda: sqs.send_message(QueueUrl=qlow, MessageBody="x" * 2048)) # FIFO без MessageGroupId fq = sqs.create_queue(QueueName=f"invalid-{random.randint(0, 10**9)}.fifo", Attributes={"FifoQueue": "true"})["QueueUrl"] cases.append(lambda: sqs.send_message(QueueUrl=fq, MessageBody="x", MessageDeduplicationId="d")) for c in cases: t0 = time.time() try: c() bump("soft_warn") except ClientError as e: add_latency(time.time() - t0) code = e.response.get("Error", {}).get("Code", "Unknown") if code in EXPECTED_CODES: record_expected(e) else: record_unexpected(e) except Exception as e: record_unexpected(e) finally: # Очереди удаляются ПОСЛЕ прогона сценариев. if qlow is not None: try: sqs.delete_queue(QueueUrl=qlow) except ClientError: pass if fq is not None: try: sqs.delete_queue(QueueUrl=fq) except ClientError: pass if q is not None: try: sqs.delete_queue(QueueUrl=q) except ClientError: pass def worker(wid): sqs = make_client() it = 0 while not stop.is_set(): it += 1 t0 = time.time() try: valid_cycle(sqs, wid, it) add_latency(time.time() - t0) except ClientError as e: code = e.response.get("Error", {}).get("Code", "Unknown") if code in EXPECTED_CODES: record_expected(e) else: record_unexpected(e) except Exception as e: record_unexpected(e) t0 = time.time() try: fifo_cycle(sqs, wid, it) add_latency(time.time() - t0) except ClientError as e: code = e.response.get("Error", {}).get("Code", "Unknown") if code in EXPECTED_CODES: record_expected(e) else: record_unexpected(e) except Exception as e: record_unexpected(e) try: invalid_cases(sqs) except Exception as e: record_unexpected(e) time.sleep(random.uniform(0.01, 0.05)) def cleanup_all_queues(sqs): """Удаляет ВСЕ очереди тенанта. Вызывается ПЕРЕД прогоном (стартовый мусор прошлых прогонов) и ПОСЛЕ (остатки от циклов, упавших на сбое — без этого копятся очереди и тест упирается в лимит MaxQueues тенанта).""" try: urls = sqs.list_queues().get("QueueUrls", []) except Exception: urls = [] deleted = 0 for u in urls: try: sqs.delete_queue(QueueUrl=u) deleted += 1 except Exception: pass if deleted: print(f"cleanup: удалено очередей {deleted}", flush=True) def progress_printer(): while not stop.is_set(): stop.wait(30) with stats_lock: s = dict(stats) el = int(time.time() - start_ts) print(f"[{el:>5}s] ops={s['ops_total']} ok={s['valid_ok']} " f"expected_err={s['expected_err']} unexpected={s['unexpected_fail']} " f"soft_warn={s['soft_warn']} health_fail={s['health_fail']}", flush=True) def main(): print(f"Нагрузочный тест: {ENDPOINT}, длительность {DURATION}s, воркеров {WORKERS}", flush=True) if not os.environ.get("AWS_ACCESS_KEY_ID") or not os.environ.get("AWS_SECRET_ACCESS_KEY"): print("Нужны AWS_ACCESS_KEY_ID и AWS_SECRET_ACCESS_KEY") return 1 # Чистим стартовый мусор прошлых прогонов. cleanup_all_queues(make_client()) threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(WORKERS)] hthread = threading.Thread(target=health_monitor, daemon=True) pthread = threading.Thread(target=progress_printer, daemon=True) for t in threads: t.start() hthread.start() pthread.start() stop.wait(DURATION) stop.set() for t in threads: t.join(timeout=30) # Чистим остатки: очереди от циклов, упавших на сбоях. cleanup_all_queues(make_client()) el = int(time.time() - start_ts) with stats_lock: s = dict(stats) print(f"\n=== Завершено через {el}s ===", flush=True) print(f"Операций всего: {s['ops_total']}") print(f"Корректных (ok): {s['valid_ok']}") print(f"Ожидаемых отказов: {s['expected_err']}") print(f"НЕОЖИДАННЫХ отказов: {s['unexpected_fail']}") print(f"Некорректное прошло: {s['soft_warn']}") print(f"Сбоев /health: {s['health_fail']}") if s["err_by_code"]: print("Коды ошибок:") for code, n in sorted(s["err_by_code"].items(), key=lambda x: -x[1]): print(f" {code}: {n}") with LATENCY_LOCK: if LATENCY: print(f"Латентность операций: min={min(LATENCY):.3f}s avg={sum(LATENCY)/len(LATENCY):.3f}s max={max(LATENCY):.3f}s") verdict = "УСТОЙЧИВО" if s["unexpected_fail"] == 0 and s["health_fail"] == 0 else "ЕСТЬ ПРОБЛЕМЫ" print(f"Вердикт: {verdict}") return 0 if verdict == "УСТОЙЧИВО" else 1 if __name__ == "__main__": sys.exit(main())