app-autotest/DOCS/sonnet-full-code-review.md

Полный обзор кода app-autotest — запрос к Sonnet

Отправлено: 27.07.2026

---

Контекст

Flask-приложение на Nubes pythonk8s (gunicorn, 1 worker, нет persistent volume). Репозиторий: https://gitea.services.ngcloud.ru/forcloud/app-autotest.git Деплой: git push → managed service редеплоит. Текущая версия: v1.0.48.

ПРОБЛЕМА

CREATE инстанса через UI: инстанс создаётся в Nubes (виден в UI платформы), НО в нашем списке инстансов НЕ появляется. Происходит СТАБИЛЬНО, 3 раза подряд (v1.0.46, v1.0.47, v1.0.48). Пробовали:

• v1.0.46: tracker_add в фоновом daemon-потоке _finish_op
• v1.0.47: tracker_add в _finish_op, но до установки _op_results[OK] (попытка победить гонку)
• v1.0.48: tracker_add синхронно в api_test() до запуска потока, с try/except: pass

НИ ОДИН вариант не сработал. Инстанс в трекере не появляется.

Текущий код на проде показывает 4 инстанса из _INITIAL:

curl-final-test (suspended)
curl-test-144834 (suspended)
autotest-1 (suspended)
autotest-1-first (running)

Новый инстанс (f192b10d-beb0-4a25-b79f-5dbd7de4712e, создан в 09:16) — отсутствует.

---

ВЕСЬ КОД

1. site/app.py — точка входа

import os
from flask import Flask
from routes.main import bp as main_bp
from routes.api import bp as api_bp
from routes.api_test import bp as api_test_bp

VERSION = "1.0.48"

app = Flask(__name__, template_folder="templates", static_folder="static")
app.config["NUBES_API_ENDPOINT"] = os.getenv("NUBES_API_ENDPOINT", "https://lk-api-gateway-dev.ngcloud.ru/api/v1/svc")
app.config["NUBES_API_TOKEN"] = os.getenv("NUBES_API_TOKEN", "")
app.config["VERSION"] = VERSION
app.register_blueprint(main_bp)
app.register_blueprint(api_bp)
app.register_blueprint(api_test_bp)

@app.route("/health")
def health():
    return "OK"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

2. site/api/http_client.py — HTTP клиент

import requests

class HttpClient:
    def __init__(self, endpoint, token):
        self._endpoint = endpoint.rstrip("/")
        self._session = requests.Session()
        self._session.headers.update({
            "Authorization": f"Bearer {token}",
            "User-Agent": "Mozilla/5.0",
        })

    def get(self, path, **kwargs):
        kwargs.setdefault("timeout", 10)
        r = self._session.get(f"{self._endpoint}{path}", **kwargs)
        r.raise_for_status()
        return r.json()

    def post(self, path, data=None, **kwargs):
        kwargs.setdefault("timeout", 30)
        url = f"{self._endpoint}{path}"
        r = self._session.post(url, json=(data if data is not None else {}), **kwargs)
        if not r.ok:
            raise Exception(f"POST {path}: {r.status_code} {r.reason}: {r.text[:200]}")
        result = {}
        try:
            parsed = r.json()
            if isinstance(parsed, dict):
                result = parsed
        except Exception:
            pass
        loc = r.headers.get("Location", "")
        if loc:
            result["_location"] = loc
        return result

3. site/operations/tracker.py — трекер инстансов

"""Трекер созданных инстансов — только те что породило приложение."""
import json
import os
import threading

_LOCK = threading.Lock()
_PATH = "/tmp/instances.json"

_INITIAL = {
    "408b7f96-6ed7-4985-be1b-f5bdcb6ab44d": {"svcId": 1, "displayName": "autotest-1", "instanceUid": "408b7f96-6ed7-4985-be1b-f5bdcb6ab44d"},
    "884356cf-5212-4395-b56b-27c58a5fc1fa": {"svcId": 1, "displayName": "autotest-1-first", "instanceUid": "884356cf-5212-4395-b56b-27c58a5fc1fa"},
    "6528854d-a4ea-428c-9fa4-68e85fa9b3ac": {"svcId": 1, "displayName": "curl-test-144834", "instanceUid": "6528854d-a4ea-428c-9fa4-68e85fa9b3ac"},
    "fdcb5887-39f7-4ef8-a9c0-d55e434a55ff": {"svcId": 1, "displayName": "curl-final-test", "instanceUid": "fdcb5887-39f7-4ef8-a9c0-d55e434a55ff"},
}

def _load():
    try:
        with open(_PATH) as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        data = dict(_INITIAL)
        _save(data)
        return data

def _save(data):
    with open(_PATH, "w") as f:
        json.dump(data, f, indent=2)

def add(instance_uid, svc_id, display_name):
    with _LOCK:
        data = _load()
        data[instance_uid] = {
            "svcId": svc_id,
            "displayName": display_name,
            "instanceUid": instance_uid,
        }
        _save(data)

def remove(instance_uid):
    with _LOCK:
        data = _load()
        data.pop(instance_uid, None)
        _save(data)

def list_all():
    with _LOCK:
        return list(_load().values())

4. site/routes/api_test.py — основной файл с проблемой

from flask import Blueprint, current_app, jsonify, request

from api.http_client import HttpClient
from operations.get_services import get_services, get_service_detail
from operations.get_instances import get_instances
from operations.tracker import add as tracker_add, list_all as tracker_list
from operations.tracker import remove as tracker_remove

bp = Blueprint("api_test", __name__)


def _client():
    token = request.cookies.get("token") or current_app.config["NUBES_API_TOKEN"]
    return HttpClient(current_app.config["NUBES_API_ENDPOINT"], token)


@bp.route("/api/services")
def api_services():
    try:
        raw = get_services(_client())
        svc_list = [{"svcId": s["svcId"], "svc": s["svc"], "svcExtendedName": s.get("svcExtendedName", "")} for s in raw]
        svc_list.sort(key=lambda s: s["svcId"])
        return jsonify(svc_list)
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/instances/list")
def api_instances_list():
    try:
        raw = get_instances(_client())
        inst = [{"instanceUid": i["instanceUid"], "displayName": i["displayName"], "serviceId": i["serviceId"], "svc": i["svc"], "explainedStatus": i.get("explainedStatus", "?")} for i in raw]
        return jsonify(inst)
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/operations/<int:svc_id>")
def api_operations(svc_id):
    try:
        detail = get_service_detail(_client(), svc_id)
        ops = detail.get("operations", [])
        # только отслеживаемые инстансы этого сервиса
        tracked = tracker_list()
        tracked_uids = {t["instanceUid"] for t in tracked if t["svcId"] == svc_id}
        instances = get_instances(_client())
        svc_instances = [i for i in instances
            if i.get("instanceUid") in tracked_uids
            and i.get("explainedStatus") not in ("deleted", "not created")]
        return jsonify({
            "svc": detail.get("svc", ""),
            "operations": [{"svcOperationId": o["svcOperationId"], "operation": o["operation"]} for o in ops],
            "instances": svc_instances,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/params/<int:op_id>")
def api_params(op_id):
    try:
        data = _client().get(f"/instanceOperations/default/{op_id}")
        params = data["svcOperation"]["cfsParams"]
        result = []
        for p in params:
            result.append({
                "svcOperationCfsParamId": p["svcOperationCfsParamId"],
                "name": p.get("svcOperationCfsParam", ""),
                "dataType": p.get("dataType", ""),
                "isRequired": p.get("isRequired", False),
                "defaultValue": p.get("defaultValue"),
                "valueList": p.get("valueList"),
                "dataDescriptor": {k: {"dataType": v.get("dataType",""), "valueList": v.get("valueList",""), "isRequired": v.get("isRequired", False)} for k, v in p.get("dataDescriptor", {}).items()} if p.get("dataDescriptor") else None,
            })
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/test", methods=["POST"])
def api_test():
    """Запустить операцию — возвращает opUid сразу, выполнение в фоне."""
    import threading, time

    data = request.get_json()
    svc_id = data["serviceId"]
    op_name = data["operation"]
    svc_op_id = data["svcOperationId"]
    params = data.get("params", {})
    instance_uid = data.get("instanceUid")
    display_name = data.get("displayName", f"autotest-{svc_id}")

    client = _client()

    try:
        if op_name == "create":
            payload = {"serviceId": svc_id, "displayName": display_name, "descr": ""}
            resp = client.post("/instances", payload)
            instance_uid = resp.get("instanceUid") or _find_uid(resp) or _uid_from_location(resp.get("_location", ""))
            if not instance_uid:
                return jsonify({"status": "FAIL", "error": "Не удалось получить instanceUid"}), 500

            op_payload = {"instanceUid": instance_uid, "operation": "create"}
            op_resp = client.post("/instanceOperations", op_payload)
            op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
            if not op_uid:
                return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500

            for pid, pval in params.items():
                client.post("/instanceOperationCfsParams",
                    {"instanceOperationUid": op_uid, "svcOperationCfsParamId": int(pid), "paramValue": str(pval)})
            client.post(f"/instanceOperations/{op_uid}/run")

            # Записать в трекер СРАЗУ, до фонового потока
            try:
                tracker_add(instance_uid, svc_id, display_name)
            except Exception:
                pass  # ← МОЛЧА ПРОГЛАТЫВАЕТ ОШИБКУ

            # фоном ждать завершения
            threading.Thread(target=_finish_op, args=(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, True), daemon=True).start()
            return jsonify({"status": "RUNNING", "opUid": op_uid, "instanceUid": instance_uid})

        else:
            if not instance_uid:
                return jsonify({"status": "FAIL", "error": "Нет instanceUid"}), 400

            if op_name == "redeploy":
                op_payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id, "operation": op_name}
                op_resp = client.post("/instanceOperations", op_payload)
                op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
                if not op_uid:
                    return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500
                client.post(f"/instanceOperations/{op_uid}/run")
            else:
                op_payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id, "operation": op_name}
                op_resp = client.post("/instanceOperations", op_payload)
                op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
                if not op_uid:
                    return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500
                for pid, pval in params.items():
                    client.post("/instanceOperationCfsParams",
                        {"instanceOperationUid": op_uid, "svcOperationCfsParamId": int(pid), "paramValue": str(pval)})
                client.post(f"/instanceOperations/{op_uid}/run")

            is_delete = (op_name == "delete")
            threading.Thread(target=_finish_op, args=(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, False, is_delete), daemon=True).start()
            return jsonify({"status": "RUNNING", "opUid": op_uid, "instanceUid": instance_uid})

    except Exception as e:
        import traceback
        return jsonify({"status": "FAIL", "error": str(e) + " | " + traceback.format_exc()[-200:]})


# Результаты фоновых операций: opUid → {status, error, stages, duration}
_op_results = {}


def _finish_op(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, is_create, is_delete=False):
    """Фоном ждать dtFinish и сохранить результат."""
    import time
    t0 = time.time()
    deadline = t0 + 300
    while time.time() < deadline:
        try:
            data = client.get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,isInProgress,duration,stages")
        except Exception:
            time.sleep(5)
            continue
        op = data.get("instanceOperation", {})
        dt_finish = op.get("dtFinish")
        _op_results[op_uid] = {
            "status": "RUNNING",
            "stages": op.get("stages", []),
            "duration": round(time.time() - t0, 1),
        }
        if dt_finish and str(dt_finish).strip():
            is_ok = op.get("isSuccessful")
            err = op.get("errorLog") or ""
            print(f"[DEBUG] _finish_op op_uid={op_uid} isSuccessful={is_ok!r}", flush=True)
            if is_ok and is_delete:
                tracker_remove(instance_uid)
            _op_results[op_uid] = {
                "status": "OK" if is_ok else "FAIL",
                "error": str(err) if err else "",
                "stages": op.get("stages", []),
                "duration": round(time.time() - t0, 1),
            }
            return
        time.sleep(5)
    _op_results[op_uid] = {"status": "TIMEOUT", "duration": round(time.time() - t0, 1)}


@bp.route("/api/test/status/<op_uid>")
def api_test_status(op_uid):
    """Получить текущий статус операции (поллинг с UI)."""
    if op_uid in _op_results:
        return jsonify(_op_results[op_uid])
    try:
        data = _client().get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,isInProgress,duration,stages")
        op = data.get("instanceOperation", {})
        dt_finish = op.get("dtFinish")
        done = bool(dt_finish and str(dt_finish).strip())
        return jsonify({
            "status": "OK" if (done and op.get("isSuccessful")) else ("FAIL" if done else "RUNNING"),
            "done": done,
            "isSuccessful": op.get("isSuccessful"),
            "isInProgress": op.get("isInProgress"),
            "duration": op.get("duration"),
            "stages": op.get("stages", []),
            "errorLog": op.get("errorLog"),
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


def _find_uid(d):
    if isinstance(d, dict):
        for k in ("instanceUid", "instanceOperationUid", "uid", "Uid"):
            if k in d:
                return d[k]
    return None


def _uid_from_location(loc):
    parts = loc.rstrip("/").split("/")
    return parts[-1] if parts else None

5. site/routes/main.py — главная страница (фрагмент с /api/operations)

@bp.route("/api/operations/<int:svc_id>")
def api_operations(svc_id):
    try:
        detail = get_service_detail(_client(), svc_id)
        ops = detail.get("operations", [])
        tracked = tracker_list()
        tracked_uids = {t["instanceUid"] for t in tracked if t["svcId"] == svc_id}
        instances = get_instances(_client())
        svc_instances = [i for i in instances
            if i.get("instanceUid") in tracked_uids
            and i.get("explainedStatus") not in ("deleted", "not created")]
        return jsonify({
            "svc": detail.get("svc", ""),
            "operations": [{"svcOperationId": o["svcOperationId"], "operation": o["operation"]} for o in ops],
            "instances": svc_instances,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

6. site/templates/index.html — UI (ключевые функции)

// Инстансы + кнопки операций
async function selectService(svcId){
    selectedInst=null; selectedOp=null; stopPoll();
    document.getElementById('params-card').style.display='none';
    document.getElementById('stages-box').style.display='none';
    const r=await fetch('/api/operations/'+svcId);
    const d=await r.json();
    svcInstances=d.instances||[];
    // ... рендерит список
}

async function executeOp(params){
    stopPoll();
    const displayName=document.getElementById('param-displayname')?.value||'autotest-1';
    // ... очищает форму
    const r=await fetch('/api/test',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({
        serviceId:SVC_ID,
        operation:selectedOp.opName,
        svcOperationId:selectedOp.opId,
        params,
        instanceUid:selectedInst||'',
        displayName
    })});
    const d=await r.json();
    if(d.status==='FAIL'){/* ошибка */ return;}
    // start polling
    const opUid=d.opUid;
    pollTimer=setInterval(async()=>{
        const sr=await fetch('/api/test/status/'+opUid);
        const sd=await sr.json();
        showStages(sd.stages||[]);
        if(sd.status!=='RUNNING'){
            stopPoll();
            // ... показать результат
            if(sd.status==='OK'){
                if(selectedOp.opName==='create'){await selectService(SVC_ID);}  // ← ПЕРЕЗАГРУЖАЕТ ВСЁ
                else{await refreshInstances();}
            }
        }
    },2000);
}

async function refreshInstances(){
    const r=await fetch('/api/operations/'+SVC_ID);
    const d=await r.json();
    svcInstances=d.instances||[];
    // обновить только бейджи статусов
    svcInstances.forEach(i=>{
        const el=document.querySelector(`[data-iuid="${i.instanceUid}"] .badge`);
        if(el){
            el.textContent=i.explainedStatus||'?';
            el.className='badge '+(i.explainedStatus==='running'?'badge-success':'');
        }
    });
}

---

ЧТО ПРОИСХОДИТ ПРИ CREATE (трассировка)

1. UI: executeOp() → POST /api/test {serviceId:1, operation:"create", svcOperationId:18, params:{...}, instanceUid:"", displayName:"autotest-1-lq5x3a"}
2. api_test(): op_name="create", svc_id=1 (int), display_name="autotest-1-lq5x3a" (str)
3. client.post("/instances", ...) → ответ от Nubes → instance_uid = "f192b10d-beb0-4a25-b79f-5dbd7de4712e"
4. client.post("/instanceOperations", ...) → op_uid = "d489348e-5d92-47cd-97f7-25f1c4d65ffc"
5. Параметры → client.post("/instanceOperationCfsParams", ...) для каждого
6. client.post("/instanceOperations/{op_uid}/run") → запуск
7. tracker_add("f192b10d-...", 1, "autotest-1-lq5x3a") ← ЗДЕСЬ ПРОБЛЕМА
8. threading.Thread(target=_finish_op, ...) → фон
9. Возврат {status:"RUNNING", opUid:"d489348e-...", instanceUid:"f192b10d-..."}
10. UI поллит /api/test/status/d489348e-...
11. _finish_op получает isSuccessful=true → _op_results[opUid] = {status:"OK", ...}
12. UI получает OK → вызывает selectService(1) → /api/operations/1 → tracker_list() → 4 инстанса → нового НЕТ

---

ВОПРОСЫ

Критический: почему tracker_add не работает?

1. Может ли json.dump в /tmp/instances.json падать молча на pythonk8s? (диск full, fs ro, quota)
2. Может ли _save записать, но _load прочитать старую версию из-за кеша ФС?
3. Может ли gunicorn preload создавать несколько копий модуля tracker.py с разными _LOCK?
4. Может ли _INITIAL перезаписывать файл при КАЖДОМ _load, если файл повреждён?
5. ГЛАВНЫЙ ВОПРОС: как надёжно сохранять состояние на pythonk8s БЕЗ persistent volume?

Архитектурный

6. Не перейти ли на sqlite3 в /tmp/? Даст ли это атомарность?
7. Не заменить ли /tmp/instances.json на in-memory dict + seed из _INITIAL при старте? (без файла вообще)
8. Как правильно логировать ошибки на pythonk8s чтобы их было видно?

UI

9. После успешного CREATE selectService() скрывает progress card — это бесит. Как обновить только список инстансов не скрывая stages?

---

Что уже пробовали (НЕ ПОМОГЛО)

• v1.0.46: tracker_add в daemon-потоке _finish_op → поток умирает под gunicorn
• v1.0.47: tracker_add в _finish_op, но до _op_results[OK] → не помогло
• v1.0.48: tracker_add синхронно в api_test() ДО потока, try/except: pass → ОШИБКА СКРЫТА