419 lines
14 KiB
Python
419 lines
14 KiB
Python
from fastapi import FastAPI, Header, HTTPException, Request
|
|
from fastapi.responses import StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
import httpx
|
|
from typing import Annotated
|
|
import asyncio
|
|
import json
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
app = FastAPI()
|
|
|
|
DECK_APIS = {
|
|
"prod": "https://deck-api.ngcloud.ru/api/v1",
|
|
"dev": "https://deck-api-dev.ngcloud.ru/api/v1",
|
|
"test": "https://deck-api-test.ngcloud.ru/api/v1",
|
|
}
|
|
DEFAULT_DECK_ENV = "test"
|
|
CACHE_TTL_SECONDS = 180
|
|
GRAPH_CACHE: dict[str, tuple[float, dict]] = {}
|
|
|
|
# Centralized backend policy for table-related derived fields.
|
|
DELETE_POLICY = {
|
|
"status": "suspended",
|
|
"min_days": 14,
|
|
"op_name_patterns": ("suspend", "приост", "замороз"),
|
|
"op_time_fields": ("dtFinish", "dtStart", "dtCreated", "dtState"),
|
|
"fallback_time_fields": ("dtState", "instanceConfigDtUpdated"),
|
|
}
|
|
|
|
|
|
def auth_headers(token: str):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def resolve_deck_api(env_raw: str | None) -> tuple[str, str]:
|
|
env = (env_raw or DEFAULT_DECK_ENV).strip().lower()
|
|
if env not in DECK_APIS:
|
|
env = DEFAULT_DECK_ENV
|
|
return env, DECK_APIS[env]
|
|
|
|
|
|
def _as_text(value) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
if isinstance(value, (int, float, bool)):
|
|
return str(value).strip()
|
|
if isinstance(value, dict):
|
|
for key in ("value", "displayValue", "name", "label", "id", "text"):
|
|
nested = _as_text(value.get(key))
|
|
if nested:
|
|
return nested
|
|
return ""
|
|
|
|
|
|
def extract_platform_name(inst: dict | None) -> str:
|
|
if not isinstance(inst, dict):
|
|
return ""
|
|
|
|
sources = [
|
|
inst.get("parameters"),
|
|
inst.get("params"),
|
|
inst.get("instanceParameters"),
|
|
inst.get("inputParameters"),
|
|
]
|
|
needles = ("platform", "kuber", "k8s", "cluster", "shturval", "штурвал")
|
|
|
|
for src in sources:
|
|
if isinstance(src, dict):
|
|
for key, value in src.items():
|
|
key_s = str(key).strip().lower()
|
|
if any(n in key_s for n in needles):
|
|
txt = _as_text(value)
|
|
if txt:
|
|
return txt
|
|
elif isinstance(src, list):
|
|
for item in src:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key_s = str(item.get("name") or item.get("key") or item.get("code") or "").strip().lower()
|
|
if any(n in key_s for n in needles):
|
|
txt = _as_text(item.get("value") or item.get("displayValue") or item.get("text"))
|
|
if txt:
|
|
return txt
|
|
return ""
|
|
|
|
|
|
def _parse_ts(value) -> float | None:
|
|
if not value:
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
if not isinstance(value, str):
|
|
return None
|
|
raw = value.strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
if raw.endswith("Z"):
|
|
raw = raw[:-1] + "+00:00"
|
|
dt = datetime.fromisoformat(raw)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.timestamp()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _extract_suspend_entered_ts(detail: dict | None) -> float | None:
|
|
if not isinstance(detail, dict):
|
|
return None
|
|
|
|
latest_ts: float | None = None
|
|
operations = detail.get("operations")
|
|
if isinstance(operations, list):
|
|
for op in operations:
|
|
if not isinstance(op, dict):
|
|
continue
|
|
op_name = str(op.get("operation") or op.get("type") or op.get("action") or "").strip().lower()
|
|
if not any(pattern in op_name for pattern in DELETE_POLICY["op_name_patterns"]):
|
|
continue
|
|
for field in DELETE_POLICY["op_time_fields"]:
|
|
ts = _parse_ts(op.get(field))
|
|
if ts is None:
|
|
continue
|
|
if latest_ts is None or ts > latest_ts:
|
|
latest_ts = ts
|
|
break
|
|
|
|
if latest_ts is not None:
|
|
return latest_ts
|
|
|
|
for field in DELETE_POLICY["fallback_time_fields"]:
|
|
ts = _parse_ts(detail.get(field))
|
|
if ts is not None:
|
|
return ts
|
|
return None
|
|
|
|
|
|
def build_stream_row_meta(row: dict, detail: dict | None) -> dict:
|
|
status = str(row.get("explainedStatus") or "").strip().lower()
|
|
meta = {
|
|
"deleteCandidate": {
|
|
"status": status,
|
|
"eligible": False,
|
|
"daysInSuspend": 0,
|
|
"minDays": DELETE_POLICY["min_days"],
|
|
"suspendEnteredAt": "",
|
|
}
|
|
}
|
|
if status != DELETE_POLICY["status"]:
|
|
return meta
|
|
|
|
entered_ts = _extract_suspend_entered_ts(detail)
|
|
if entered_ts is None:
|
|
return meta
|
|
|
|
days = max(0, int((time.time() - entered_ts) // 86400))
|
|
meta["deleteCandidate"]["daysInSuspend"] = days
|
|
meta["deleteCandidate"]["eligible"] = days >= DELETE_POLICY["min_days"]
|
|
meta["deleteCandidate"]["suspendEnteredAt"] = datetime.fromtimestamp(entered_ts, tz=timezone.utc).isoformat().replace("+00:00", "Z")
|
|
return meta
|
|
|
|
|
|
async def fetch_all_instances(token: str, deck_api: str):
|
|
params: dict = {"page": 1, "size": 200}
|
|
all_results = []
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
while True:
|
|
r = await client.get(
|
|
f"{deck_api}/index.cfm/instances",
|
|
headers=auth_headers(token),
|
|
params=params,
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
data = r.json()
|
|
results = data.get("results", [])
|
|
if not isinstance(results, list):
|
|
results = []
|
|
if not results:
|
|
break
|
|
all_results.extend(results)
|
|
params["page"] += 1
|
|
return all_results
|
|
|
|
|
|
@app.get("/api/instances")
|
|
async def list_instances(
|
|
request: Request,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
_, deck_api = resolve_deck_api(x_deck_env)
|
|
# parse status from query string manually (FastAPI list param can fail behind ingress rewrite)
|
|
status_list = request.query_params.getlist("status")
|
|
all_results = await fetch_all_instances(x_deck_token, deck_api)
|
|
if status_list:
|
|
all_results = [i for i in all_results if i.get("explainedStatus") in status_list]
|
|
return {"instances": all_results, "total": len(all_results)}
|
|
|
|
|
|
@app.get("/api/graph")
|
|
async def get_graph(
|
|
request: Request,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
env, deck_api = resolve_deck_api(x_deck_env)
|
|
|
|
status_list = request.query_params.getlist("status")
|
|
root_uid = request.query_params.get("root_uid")
|
|
depth_raw = request.query_params.get("depth", "2")
|
|
try:
|
|
depth = max(1, min(5, int(depth_raw)))
|
|
except ValueError:
|
|
depth = 2
|
|
|
|
status_key = ",".join(sorted(status_list)) if status_list else "__all__"
|
|
cache_key = f"{x_deck_token[:24]}::{env}::{status_key}"
|
|
now = time.time()
|
|
|
|
cached = GRAPH_CACHE.get(cache_key)
|
|
if cached and (now - cached[0] < CACHE_TTL_SECONDS):
|
|
base_graph = cached[1]
|
|
else:
|
|
all_instances = await fetch_all_instances(x_deck_token, deck_api)
|
|
if status_list:
|
|
all_instances = [i for i in all_instances if i.get("explainedStatus") in status_list]
|
|
|
|
uid_to_inst = {
|
|
i.get("instanceUid"): i
|
|
for i in all_instances
|
|
if i.get("instanceUid")
|
|
}
|
|
allowed_uids = set(uid_to_inst.keys())
|
|
|
|
detail_map: dict[str, dict] = {}
|
|
async with httpx.AsyncClient(timeout=25) as shared_client:
|
|
for uid in allowed_uids:
|
|
try:
|
|
r = await shared_client.get(
|
|
f"{deck_api}/index.cfm/instances/{uid}",
|
|
headers=auth_headers(x_deck_token),
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
if r.status_code >= 400:
|
|
continue
|
|
inst = r.json().get("instance")
|
|
if inst:
|
|
detail_map[uid] = inst
|
|
except httpx.TimeoutException:
|
|
continue
|
|
|
|
# Categorise singletons by service name
|
|
_PLATFORM_PATTERNS = ("виртуальный датацентр", "vdc", "virtual data center", "cloud director", "kubernetes", "k8s")
|
|
_SINGLETON_PATTERNS = ("object storage", "s3", "edge", "шлюз периметра", "dns", "container registry", "реестр контейнеров")
|
|
|
|
def _is_platform_svc(svc_name: str) -> bool:
|
|
s = svc_name.strip().lower()
|
|
return any(p in s for p in _PLATFORM_PATTERNS)
|
|
|
|
def _is_singleton_svc(svc_name: str) -> bool:
|
|
s = svc_name.strip().lower()
|
|
return any(p in s for p in _SINGLETON_PATTERNS)
|
|
|
|
# Build realm→platform_uid map: instances whose resourceRealmCnt > 0 ARE platforms
|
|
realm_to_platform: dict[str, str] = {}
|
|
for uid_d, inst_d in detail_map.items():
|
|
cnt = inst_d.get("resourceRealmCnt") or 0
|
|
realm_name = inst_d.get("resourceRealm") or ""
|
|
if cnt and realm_name:
|
|
realm_to_platform[realm_name] = uid_d
|
|
|
|
nodes = []
|
|
for uid_i in allowed_uids:
|
|
i = uid_to_inst.get(uid_i, {})
|
|
detail = detail_map.get(uid_i, {})
|
|
svc = i.get("svc") or ""
|
|
realm_cnt = detail.get("resourceRealmCnt") or 0
|
|
is_aux = bool(detail.get("isAuxiliary"))
|
|
# Determine lane by service name
|
|
if _is_platform_svc(svc):
|
|
lane = "platform"
|
|
elif _is_singleton_svc(svc):
|
|
lane = "singleton"
|
|
else:
|
|
lane = "regular"
|
|
nodes.append(
|
|
{
|
|
"id": uid_i,
|
|
"label": i.get("displayName") or uid_i,
|
|
"status": i.get("explainedStatus") or "unknown",
|
|
"svc": svc,
|
|
"realm": i.get("resourceRealm") or "",
|
|
"lane": lane,
|
|
"isAuxiliary": is_aux,
|
|
"created": i.get("instanceConfigDtCreated") or i.get("dtCreated") or "",
|
|
}
|
|
)
|
|
|
|
# Debug: log lane distribution
|
|
import sys
|
|
_lane_counts = {}
|
|
for _n in nodes:
|
|
_l = _n.get("lane", "unknown")
|
|
_lane_counts[_l] = _lane_counts.get(_l, 0) + 1
|
|
print(f"[GRAPH] Lanes: {_lane_counts}, nodes: {len(nodes)}", flush=True)
|
|
for _n in nodes:
|
|
if _n["lane"] in ("platform", "singleton"):
|
|
print(f" {_n['lane'].upper()}: {_n['label']} | svc={_n['svc']}", flush=True)
|
|
|
|
base_graph = {"nodes": nodes, "edges": []}
|
|
GRAPH_CACHE[cache_key] = (now, base_graph)
|
|
|
|
# Return full graph fast
|
|
if not root_uid:
|
|
return {
|
|
"nodes": base_graph["nodes"],
|
|
"edges": [],
|
|
"total_nodes": len(base_graph["nodes"]),
|
|
"total_edges": 0,
|
|
"cached": bool(cached),
|
|
}
|
|
|
|
node_ids = {n["id"] for n in base_graph["nodes"]}
|
|
if root_uid not in node_ids:
|
|
return {"nodes": [], "edges": [], "total_nodes": 0, "total_edges": 0, "cached": bool(cached)}
|
|
|
|
sub_nodes = [n for n in base_graph["nodes"] if n["id"] == root_uid]
|
|
return {
|
|
"nodes": sub_nodes,
|
|
"edges": [],
|
|
"total_nodes": len(sub_nodes),
|
|
"total_edges": 0,
|
|
"cached": bool(cached),
|
|
}
|
|
|
|
|
|
@app.get("/api/instances/stream")
|
|
async def stream_instances(
|
|
request: Request,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
_, deck_api = resolve_deck_api(x_deck_env)
|
|
|
|
all_instances = await fetch_all_instances(x_deck_token, deck_api)
|
|
uid_to_inst = {
|
|
i.get("instanceUid"): i
|
|
for i in all_instances
|
|
if i.get("instanceUid")
|
|
}
|
|
|
|
async def generate():
|
|
total = len(uid_to_inst)
|
|
loaded = 0
|
|
started_at = time.time()
|
|
|
|
async with httpx.AsyncClient(timeout=25) as shared_client:
|
|
for uid in uid_to_inst:
|
|
detail = None
|
|
try:
|
|
r = await shared_client.get(
|
|
f"{deck_api}/index.cfm/instances/{uid}",
|
|
headers=auth_headers(x_deck_token),
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
if r.status_code < 400:
|
|
detail = r.json().get("instance")
|
|
except Exception:
|
|
detail = None
|
|
|
|
loaded += 1
|
|
row = dict(uid_to_inst.get(uid, {}))
|
|
row["_progress"] = {"loaded": loaded, "total": total}
|
|
row["_meta"] = build_stream_row_meta(row, detail)
|
|
if detail:
|
|
row["_detail"] = detail
|
|
yield json.dumps(row, ensure_ascii=False) + "\n"
|
|
|
|
duration_ms = int((time.time() - started_at) * 1000)
|
|
yield json.dumps({"_done": True, "total": total, "durationMs": duration_ms}, ensure_ascii=False) + "\n"
|
|
|
|
return StreamingResponse(generate(), media_type="application/x-ndjson")
|
|
|
|
|
|
@app.get("/api/instances/{uid}")
|
|
async def get_instance(
|
|
request: Request,
|
|
uid: str,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
_, deck_api = resolve_deck_api(x_deck_env)
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
r = await client.get(
|
|
f"{deck_api}/index.cfm/instances/{uid}",
|
|
headers=auth_headers(x_deck_token),
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
return r.json()
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="/app/static", html=True), name="static")
|