371 lines
15 KiB
Python
371 lines
15 KiB
Python
#!/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) Крайние условия: сообщения до 256КБ, юникод, пустые тела, пакеты по 10,
|
||
сообщение > лимита (300КБ — ожидается отказ).
|
||
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=250 * 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
|
||
|
||
# сообщение произвольного размера, включая 256КБ и юникод
|
||
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
|
||
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=""))
|
||
# слишком длинное сообщение
|
||
cases.append(lambda: sqs.send_message(QueueUrl=q, MessageBody="x" * (300 * 1024)))
|
||
# 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 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 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
|
||
|
||
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)
|
||
|
||
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())
|