diff --git a/HISTORY/2026-08-14-session-log.md b/HISTORY/2026-08-14-session-log.md index dc4e88b..144745a 100644 --- a/HISTORY/2026-08-14-session-log.md +++ b/HISTORY/2026-08-14-session-log.md @@ -932,3 +932,28 @@ P3.8 — PASS; P3.7 (RSS 2ч) — мониторинг в процессе. 3. Гипотеза выброса: платформенные «паузы» сети к поду Nubes (ранее фиксировали зависания ~51с на больших POST из-за MTU/MSS 1448) — но при 512 байтах причина требует отдельного расследования (tcpdump/тайминги платформы). + +## 2026-08-15 — ДЛИННОЕ сравнение с ЛОКАЛИ (30 мин), shared-SQS vs YMQ + +**Скрипт**: `tests/long_compare_local.py` (ping-pong: пул K=10, раунды +попеременно shared→ymq→shared..., receive→delete→send; retries=0, +read_timeout=30с; лог `tests/long_compare.log`). Локаль (WSL), 12:27–12:57. + +**Итог (317 раундов):** +- shared-SQS: send n=302 p50=90мс p95=269мс max=2785мс; receive n=293 p50=89мс + p95=279мс max=2577мс; delete n=293 p50=90мс p95=103мс max=2462мс; + **errors=49 ReadTimeoutError (30с)**. +- YMQ: send n=327 p50=138мс p95=157мс max=355мс; receive n=316 p50=150мс + p95=421мс max=2473мс; delete n=316 p50=138мс p95=152мс max=2105мс; + errors=1 ProxyConnectionError. + +**Выводы (факты):** +1. По p50 наш быстрее: 89–90мс против 138–150мс (~1.5–1.7 раза). +2. НО у нас **49 запросов из 888 (5.5%) вообще не получили ответ за 30с** + (каждый таймаут — 30с простоя); у Яндекса 1 ошибка из 959. +3. Полезной работы за 30 мин: наш 888 операций, Яндекс 959 — таймауты съели + преимущество по скорости. +4. Тренд: таймауты равномерны весь тест (~1 в 37с) — системная проблема + сетевого пути локаль→шлюз Nubes, не код сервиса (с ВМ их нет). +5. Выбросы max (2.5–2.8с) есть у ОБОИХ: наш send max=2785мс, YMQ receive + max=2473мс — сетевые, не сервисные. diff --git a/tests/long_compare_local.py b/tests/long_compare_local.py new file mode 100644 index 0000000..5df9f69 --- /dev/null +++ b/tests/long_compare_local.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +"""ДЛИННОЕ сравнение shared-SQS vs Yandex YMQ — с ЛОКАЛИ, по таймеру. + +Схема (честная — обе очереди в одних сетевых условиях, раунды попеременно): + 1. Старт: seed K сообщений в каждую очередь. + 2. Раунд для каждой очереди: receive(1) -> delete -> send(1) (восполнение пула). + 3. Раунды чередуются (наш, яндекс, наш, яндекс...) до истечения DURATION. + +Метрики: p50/p95/max для send/receive/delete, ошибки по типам (с таймаутами), +op/s. Прогресс каждые 60с пишется в LOG и stdout. + +Env: SHARED_AK, SHARED_SK, YMQ_AK, YMQ_SK, YMQ_QUEUE_URL, DURATION (сек, default 1800) +""" +import os +import sys +import time +import uuid + +import boto3 +from botocore.config import Config + +DURATION = int(os.environ.get("DURATION", "1800")) +K = 10 +BODY = "x" * 512 +LOG = "tests/long_compare.log" + +SHARED_EP = "https://sqs.containerk8s.dev.nubes.ru" +YMQ_EP = "https://message-queue.api.cloud.yandex.net" + +CFG = Config(connect_timeout=15, read_timeout=30, retries={"max_attempts": 0}) + + +def mk(endpoint, region, ak, sk): + return boto3.client("sqs", endpoint_url=endpoint, region_name=region, + aws_access_key_id=ak, aws_secret_access_key=sk, config=CFG) + + +class Stat: + def __init__(self): + self.vals = {"send": [], "receive": [], "delete": []} + self.errs = {} # type -> count + self.ops = {"send": 0, "receive": 0, "delete": 0} + + def op(self, name, fn): + self.ops[name] += 1 + t0 = time.time() + try: + fn() + self.vals[name].append(time.time() - t0) + except Exception as e: + k = type(e).__name__ + self.errs[k] = self.errs.get(k, 0) + 1 + + +def stat_line(st, label): + out = [label] + for name in ("send", "receive", "delete"): + v = sorted(st.vals[name]) + n = len(v) + if n: + out.append("%s: n=%d p50=%.0fms p95=%.0fms max=%.0fms" % + (name, n, v[n // 2] * 1000, v[int(n * .95)] * 1000, v[-1] * 1000)) + else: + out.append("%s: n=0" % name) + out.append("errors=%s" % (st.errs if st.errs else "0")) + return " | ".join(out) + + +def logp(msg): + line = "%s %s" % (time.strftime("%H:%M:%S"), msg) + print(line, flush=True) + try: + with open(LOG, "a") as f: + f.write(line + "\n") + except Exception: + pass + + +def main(): + shared = mk(SHARED_EP, "us-east-1", os.environ["SHARED_AK"], os.environ["SHARED_SK"]) + ymq = mk(YMQ_EP, "ru-central1", os.environ["YMQ_AK"], os.environ["YMQ_SK"]) + ymq_url = os.environ["YMQ_QUEUE_URL"] + + # временная очередь shared + qname = "longcmp-%s" % uuid.uuid4().hex[:8] + shared_url = shared.create_queue(QueueName=qname)["QueueUrl"] + + def drain(client, url): + while True: + try: + r = client.receive_message(QueueUrl=url, MaxNumberOfMessages=10) + msgs = r.get("Messages", []) + except Exception: + break + if not msgs: + break + for m in msgs: + try: + client.delete_message(QueueUrl=url, ReceiptHandle=m["ReceiptHandle"]) + except Exception: + pass + + drain(shared, shared_url) + drain(ymq, ymq_url) + + ss = Stat() + ys = Stat() + sts = {"shared": (shared, shared_url, ss), "ymq": (ymq, ymq_url, ys)} + + # seed K в обе очереди + for key in ("shared", "ymq"): + c, u, st = sts[key] + for _ in range(K): + st.op("send", lambda c=c, u=u: c.send_message(QueueUrl=u, MessageBody=BODY)) + + logp("LONG COMPARE START: duration=%ds K=%d shared=%s" % (DURATION, K, shared_url)) + logp("ymq=%s" % ymq_url) + + deadline = time.time() + DURATION + last_prog = time.time() + rounds = 0 + + def round_for(key): + nonlocal rounds + c, u, st = sts[key] + # receive (замер) + получить сообщения для delete + got = [] + t0 = time.time() + st.ops["receive"] += 1 + try: + r = c.receive_message(QueueUrl=u, MaxNumberOfMessages=1, VisibilityTimeout=30) + st.vals["receive"].append(time.time() - t0) + got = r.get("Messages", []) + except Exception as e: + st.errs[type(e).__name__] = st.errs.get(type(e).__name__, 0) + 1 + for m in got: + st.op("delete", lambda c=c, u=u, m=m: c.delete_message(QueueUrl=u, ReceiptHandle=m["ReceiptHandle"])) + st.op("send", lambda c=c, u=u: c.send_message(QueueUrl=u, MessageBody=BODY)) + + while time.time() < deadline: + round_for("shared") + round_for("ymq") + rounds += 1 + if time.time() - last_prog >= 60: + last_prog = time.time() + logp("progress %d/%ds" % (int(time.time() - (deadline - DURATION)), DURATION)) + logp(" " + stat_line(ss, "SHARED")) + logp(" " + stat_line(ys, "YMQ")) + + logp("LONG COMPARE DONE: rounds=%d" % rounds) + logp(stat_line(ss, "FINAL SHARED")) + logp(stat_line(ys, "FINAL YMQ")) + + # cleanup + drain(shared, shared_url) + drain(ymq, ymq_url) + try: + shared.delete_queue(QueueUrl=shared_url) + logp("cleanup: shared temp queue deleted, both drained") + except Exception as e: + logp("cleanup err: %s" % str(e)[:100]) + + +if __name__ == "__main__": + main()