95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
# 2026-03-18 (обновлено: plain text вывод; фильтрация SLESS_EXCLUDE)
|
|
# funcs_list.py — HTTP-функция: список пользовательских функций, человекочитаемый plain text.
|
|
# Вызывает внутренний REST API оператора (ClusterIP, без TLS).
|
|
# Возвращает str → python runtime отдаёт text/plain напрямую без json.dumps.
|
|
#
|
|
# Env vars:
|
|
# SLESS_API_URL — URL оператора (http://sless-operator.sless.svc.cluster.local:9090)
|
|
# SLESS_NAMESPACE — namespace пользователя (sless-{hex16})
|
|
# SLESS_TOKEN — JWT токен для /v1/ API
|
|
# SLESS_EXTERNAL_URL — публичный базовый URL (https://sless.kube5s.ru)
|
|
# SLESS_EXCLUDE — comma-separated имена функций, которые не показывать
|
|
|
|
import os
|
|
import requests
|
|
|
|
SEP = "─" * 52
|
|
|
|
|
|
def _comment(fn, http_trigs, cron_trigs):
|
|
phase = fn.get("phase", "?")
|
|
runtime = fn.get("runtime", "?")
|
|
if http_trigs:
|
|
active = "активна" if http_trigs[0].get("active") else "неактивна"
|
|
return f"HTTP endpoint ({runtime}) — {phase}, {active}"
|
|
elif cron_trigs:
|
|
schedule = cron_trigs[0].get("schedule", "?")
|
|
active = "активна" if cron_trigs[0].get("active") else "неактивна"
|
|
return f"Cron '{schedule}' ({runtime}) — {phase}, {active}"
|
|
else:
|
|
return f"Job/runner без триггера ({runtime}) — {phase}"
|
|
|
|
|
|
def list_all(event):
|
|
api_url = os.environ["SLESS_API_URL"].rstrip("/")
|
|
namespace = os.environ["SLESS_NAMESPACE"]
|
|
token = os.environ["SLESS_TOKEN"]
|
|
ext_url = os.environ.get("SLESS_EXTERNAL_URL", "").rstrip("/")
|
|
exclude = {n.strip() for n in os.environ.get("SLESS_EXCLUDE", "").split(",") if n.strip()}
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
fns = requests.get(f"{api_url}/v1/namespaces/{namespace}/functions", headers=headers, timeout=10)
|
|
trs = requests.get(f"{api_url}/v1/namespaces/{namespace}/triggers", headers=headers, timeout=10)
|
|
fns.raise_for_status()
|
|
trs.raise_for_status()
|
|
|
|
trig_idx = {}
|
|
for tr in trs.json():
|
|
fn_name = tr.get("function") or tr.get("functionRef")
|
|
if fn_name:
|
|
trig_idx.setdefault(fn_name, []).append(tr)
|
|
|
|
items = []
|
|
for fn in fns.json():
|
|
name = fn["name"]
|
|
if name in exclude:
|
|
continue
|
|
http_t = [t for t in trig_idx.get(name, []) if t.get("type") == "http"]
|
|
cron_t = [t for t in trig_idx.get(name, []) if t.get("type") == "cron"]
|
|
is_active = any(t.get("enabled", True) and t.get("active", False) for t in trig_idx.get(name, []))
|
|
items.append((fn, http_t, cron_t, is_active))
|
|
|
|
# Сортировка: активные вверх, затем по имени
|
|
items.sort(key=lambda x: (not x[3], x[0]["name"]))
|
|
|
|
lines = []
|
|
for fn, http_t, cron_t, is_active in items:
|
|
name = fn["name"]
|
|
lines.append(SEP)
|
|
lines.append(f" {_comment(fn, http_t, cron_t)}")
|
|
lines.append(f" name: {name}")
|
|
lines.append(f" runtime: {fn.get('runtime', '?')}")
|
|
lines.append(f" phase: {fn.get('phase', '?')}")
|
|
lines.append(f" active: {'да' if is_active else 'нет'}")
|
|
|
|
if http_t:
|
|
url = f"{ext_url}/fn/{namespace}/{name}" if ext_url else http_t[0].get("url", "")
|
|
lines.append(f" url: {url}")
|
|
if cron_t:
|
|
lines.append(f" cron: {cron_t[0].get('schedule', '?')}")
|
|
if fn.get("created_at"):
|
|
lines.append(f" created: {fn['created_at']}")
|
|
if fn.get("last_built_at"):
|
|
lines.append(f" built: {fn['last_built_at']}")
|
|
if fn.get("message"):
|
|
lines.append(f" message: {fn['message']}")
|
|
|
|
lines.append(SEP)
|
|
lines.append(f" namespace: {namespace} | total: {len(items)}")
|
|
lines.append(SEP)
|
|
|
|
# Возвращаем str — python runtime отдаст text/plain напрямую
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|