130 lines
5.3 KiB
Python
130 lines
5.3 KiB
Python
# 2026-03-18 (обновлено: фильтрация SLESS_EXCLUDE, читаемый вывод через "#"-ключ)
|
|
# funcs_list.py — HTTP-функция: список всех пользовательских функций с их статусами.
|
|
# Вызывает внутренний REST API оператора (ClusterIP, без TLS).
|
|
# Объединяет данные функций и триггеров в один ответ; скрывает служебные функции.
|
|
#
|
|
# 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 имена функций, которые не надо показывать
|
|
# Пример: "funcs,event-writer,event-monitor,event-cleaner"
|
|
#
|
|
# Формат вывода: JSON-объект, где каждая функция содержит поле "#" — краткий комментарий.
|
|
# При pretty-print (python3 -m json.tool) выглядит как читаемый список с аннотациями.
|
|
|
|
import os
|
|
import requests
|
|
|
|
|
|
def _short_comment(fn, http_triggers, cron_triggers):
|
|
"""Генерирует однострочный комментарий-описание функции по её метаданным."""
|
|
phase = fn.get("phase", "")
|
|
runtime = fn.get("runtime", "")
|
|
|
|
if http_triggers:
|
|
active_str = "активна" if http_triggers[0].get("active") else "неактивна"
|
|
return f"HTTP endpoint ({runtime}) — {phase}, {active_str}"
|
|
elif cron_triggers:
|
|
schedule = cron_triggers[0].get("schedule", "?")
|
|
active_str = "активна" if cron_triggers[0].get("active") else "неактивна"
|
|
return f"Cron '{schedule}' ({runtime}) — {phase}, {active_str}"
|
|
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("/")
|
|
|
|
# Имена функций, которые не должны присутствовать в выводе.
|
|
# Включает саму себя ("funcs") и служебные функции других примеров.
|
|
exclude = {
|
|
n.strip()
|
|
for n in os.environ.get("SLESS_EXCLUDE", "").split(",")
|
|
if n.strip()
|
|
}
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
fns_resp = requests.get(
|
|
f"{api_url}/v1/namespaces/{namespace}/functions",
|
|
headers=headers,
|
|
timeout=10,
|
|
)
|
|
fns_resp.raise_for_status()
|
|
|
|
trs_resp = requests.get(
|
|
f"{api_url}/v1/namespaces/{namespace}/triggers",
|
|
headers=headers,
|
|
timeout=10,
|
|
)
|
|
trs_resp.raise_for_status()
|
|
|
|
# Индекс триггеров по имени функции
|
|
triggers_by_fn = {}
|
|
for tr in trs_resp.json():
|
|
fn_name = tr.get("function") or tr.get("functionRef")
|
|
if fn_name:
|
|
triggers_by_fn.setdefault(fn_name, []).append(tr)
|
|
|
|
result = []
|
|
for fn in fns_resp.json():
|
|
name = fn["name"]
|
|
if name in exclude:
|
|
continue
|
|
|
|
http_triggers = [
|
|
t for t in triggers_by_fn.get(name, []) if t.get("type") == "http"
|
|
]
|
|
cron_triggers = [
|
|
t for t in triggers_by_fn.get(name, []) if t.get("type") == "cron"
|
|
]
|
|
is_active = any(
|
|
t.get("enabled", True) and t.get("active", False)
|
|
for t in triggers_by_fn.get(name, [])
|
|
)
|
|
|
|
entry = {
|
|
# "#" — первый ключ: служит визуальным комментарием при pretty-print
|
|
"#": _short_comment(fn, http_triggers, cron_triggers),
|
|
"name": name,
|
|
"runtime": fn.get("runtime"),
|
|
"phase": fn.get("phase"),
|
|
"active": is_active,
|
|
}
|
|
|
|
# URL вычисляем из SLESS_EXTERNAL_URL если задан — state может хранить старый домен
|
|
if http_triggers:
|
|
if ext_url:
|
|
entry["url"] = f"{ext_url}/fn/{namespace}/{name}"
|
|
else:
|
|
entry["url"] = http_triggers[0].get("url", "")
|
|
|
|
if cron_triggers:
|
|
entry["cron"] = cron_triggers[0].get("schedule", "")
|
|
|
|
if fn.get("message"):
|
|
entry["message"] = fn["message"]
|
|
|
|
# created_at и last_built_at — доступны после обновления оператора до v0.1.32+
|
|
if fn.get("created_at"):
|
|
entry["created_at"] = fn["created_at"]
|
|
if fn.get("last_built_at"):
|
|
entry["last_built_at"] = fn["last_built_at"]
|
|
|
|
result.append(entry)
|
|
|
|
# Сортировка: активные вверх, затем по имени
|
|
result.sort(key=lambda f: (not f["active"], f["name"]))
|
|
|
|
return {
|
|
"namespace": namespace,
|
|
"count": len(result),
|
|
"functions": result,
|
|
}
|
|
|