loadtest/site/app.py

"""
LoadTest - сервис проверки загрузки файлов на платформу Nubes (Flask).

ЦЕЛЬ: проверить, доходят ли файлы (в т.ч. большие) до бэкенда или нет.
Файл принимается в ПАМЯТЬ (без записи на диск) и возвращается МАКСИМУМ статистики.

Обязательная структура для платформы Nubes:
  repo/
  ├── requirements.txt
  └── site/
      ├── app.py           <- этот файл (точка входа: python site/app.py)
      ├── static/style.css
      └── templates/index.html
"""
import hashlib
import time

from flask import Flask, jsonify, render_template, request

# Версия сборки. ВЫВОДИТСЯ на страницу и в /health - чтобы сразу видеть,
# какая версия реально задеплоена на платформу.
VERSION = "2.0.0"

# Явно указываем папки шаблонов и статики -
# платформа запускает `python site/app.py` из корня репо, иначе Flask не найдёт шаблоны.
app = Flask(__name__, template_folder="templates", static_folder="static")

# ВАЖНО: НЕ ставим MAX_CONTENT_LENGTH (по умолчанию лимита нет).
# Цель - проверить РЕАЛЬНЫЕ лимиты платформы/прокси, а не собственный лимит Flask.


@app.route("/")
def index():
    """Главная страница - форма выбора файла и загрузки."""
    return render_template("index.html", version=VERSION)


@app.route("/health")
def health():
    """Проверка живости для платформы + версия сборки."""
    return "ok v" + VERSION


@app.route("/upload", methods=["POST"])
def upload():
    """Приём файла в память + сбор максимальной статистики."""
    t0 = time.time()  # момент начала обработки запроса на сервере

    # Метаданные запроса (заголовки, адрес клиента) - для диагностики.
    meta = {
        "server_start": time.strftime("%Y-%m-%d %H:%M:%S"),
        "remote_addr": request.remote_addr or "",
        "content_length_header": request.headers.get("Content-Length", ""),
        "content_type_header": request.headers.get("Content-Type", ""),
        "user_agent": request.headers.get("User-Agent", ""),
    }

    # Валидация: должен прийти multipart-файл в поле "file".
    if "file" not in request.files:
        return jsonify({"ok": False, "error": "Нет поля 'file' в запросе", "meta": meta}), 400

    f = request.files["file"]
    if not f or f.filename == "":
        return jsonify({"ok": False, "error": "Пустое имя файла", "meta": meta}), 400

    # Читаем ВЕСЬ файл в память и замеряем время чтения.
    t_read = time.time()
    data = f.read()
    read_s = time.time() - t_read

    size = len(data)            # сколько байт РЕАЛЬНО дошло до сервера
    total_s = time.time() - t0  # полное время обработки запроса на сервере

    result = {
        "ok": True,
        "file": {
            "name": f.filename,
            "mime": f.mimetype or "",
            "size_bytes": size,
            "size_mb": round(size / (1024 * 1024), 3),
        },
        "server": {
            "read_ms": round(read_s * 1000, 2),                     # время чтения в память
            "total_ms": round(total_s * 1000, 2),                   # полное время обработки
            "speed_mb_s": round(size / (1024 * 1024) / read_s, 3) if read_s > 0 else 0,
        },
        "md5": hashlib.md5(data).hexdigest(),  # контрольная сумма принятых байт
        "meta": meta,
    }
    result["version"] = VERSION  # версия сборки прямо в ответе
    return jsonify(result)


@app.route("/messages", methods=["POST"])
def messages():
    """Приём СООБЩЕНИЯ в память (для проверки очередей/ИоТ).

    От клиента приходит сырое тело запроса (raw body) нужного размера.
    Задача - проверить, ДОХОДИТ ли сообщение нужного размера до бэкенда
    (в отличие от /upload здесь нет multipart - только тело, как в очередях).
    """
    t0 = time.time()

    meta = {
        "server_start": time.strftime("%Y-%m-%d %H:%M:%S"),
        "remote_addr": request.remote_addr or "",
        # Заголовки, которые обычно шлёт очередь/ИоТ-клиент.
        "content_length_header": request.headers.get("Content-Length", ""),
        "content_type_header": request.headers.get("Content-Type", ""),
        "user_agent": request.headers.get("User-Agent", ""),
        "x_amz_sns_message_type": request.headers.get("X-Amz-Sns-Message-Type", ""),
        "ce_specversion": request.headers.get("Ce-Specversion", ""),
    }

    # Читаем ВСЁ сырое тело сообщения в память и замеряем время.
    t_read = time.time()
    data = request.get_data()   # сырое тело без разбора (как в SQS/IoT)
    read_s = time.time() - t_read

    size = len(data)            # сколько байт РЕАЛЬНО дошло до бэкенда
    total_s = time.time() - t0

    result = {
        "ok": True,
        "message": {
            "size_bytes": size,
            "size_kb": round(size / 1024, 3),
            "size_mb": round(size / (1024 * 1024), 3),
            "prefix": data[:64].decode("utf-8", "replace"),  # первые 64 байта для наглядности
        },
        "server": {
            "read_ms": round(read_s * 1000, 2),
            "total_ms": round(total_s * 1000, 2),
            "speed_mb_s": round(size / (1024 * 1024) / read_s, 3) if read_s > 0 else 0,
        },
        "md5": hashlib.md5(data).hexdigest(),  # контрольная сумма принятых байт
        "meta": meta,
    }
    result["version"] = VERSION  # версия сборки прямо в ответе
    return jsonify(result)


@app.route("/fetch")
def fetch_url():
    """ВРЕМЕННЫЙ: исходящий GET по URL (проверка egress из кластера).

    Нужен для проверки схемы «Flask сам тянет большой файл с ВМ».
    Делает GET по переданному url и возвращает HTTP-статус + размер ответа.
    """
    import requests

    url = request.args.get("url", "")
    if not url:
        return jsonify({"ok": False, "error": "нет параметра url"}), 400

    t0 = time.time()
    try:
        r = requests.get(url, timeout=120)
        size = len(r.content)
        return jsonify({
            "ok": True,
            "url": url,
            "status": r.status_code,
            "size_bytes": size,
            "size_mb": round(size / (1024 * 1024), 3),
            "total_ms": round((time.time() - t0) * 1000, 2),
        })
    except Exception as e:
        return jsonify({"ok": False, "error": type(e).__name__ + ": " + str(e), "url": url}), 502


if __name__ == "__main__":
    # ОБЯЗАТЕЛЬНО для платформы: `python site/app.py` должен стартовать сервер.
    # debug=False - прод-режим, без reloader (иначе поднимается второй процесс).
    app.run(host="0.0.0.0", port=5000, debug=False)