refactor: checkpoint layer2 api console changes
This commit is contained in:
Naeel
@@ -52,10 +52,10 @@ func main() {
|
||||
// --- end ai/ask feature ---
|
||||
})
|
||||
|
||||
// Запускаем фоновые горутины: reaper истёкших функций + reconciler namespace-ов
|
||||
// Запускаем фоновые горутины: reaper истёкших функций
|
||||
nsm := srv.NSManager()
|
||||
nsm.StartExpiryReaper(envDurationDefault("REAPER_INTERVAL", 5*time.Minute))
|
||||
nsm.StartNSReconciler(envDurationDefault("NS_RECONCILE_INTERVAL", 2*time.Minute))
|
||||
// StartNSReconciler удалён — за FISSION_RESOURCE_NAMESPACES теперь отвечает Layer 1 NSWatcher
|
||||
|
||||
httpServer := &http.Server{
|
||||
Addr: ":" + port,
|
||||
|
||||
@@ -46,7 +46,7 @@ spec:
|
||||
serviceAccountName: fission-console
|
||||
containers:
|
||||
- name: console
|
||||
image: naeel/fission-console:v0.8.15
|
||||
image: naeel/fission-console:v0.8.16
|
||||
ports:
|
||||
- containerPort: 8090
|
||||
env:
|
||||
|
||||
@@ -632,11 +632,7 @@ func (s *Server) handleDeleteFunction(w http.ResponseWriter, r *http.Request, na
|
||||
fission.CleanupEnvironmentIfUnused(cleanupCtx, s.dyn, ns, envName)
|
||||
}
|
||||
|
||||
// Сигналим reconciler: батчит изменения без race condition и rolling restarts
|
||||
select {
|
||||
case s.nsManager.NSReconcileCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
// (reconciler NS удалён — за FISSION_RESOURCE_NAMESPACES теперь отвечает Layer 1 NSWatcher)
|
||||
|
||||
writeAnyJSON(w, http.StatusOK, map[string]any{"deleted": true, "name": name, "package": pkgName})
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"fission-console/internal/cloud"
|
||||
"fission-console/internal/fission"
|
||||
"fission-console/ui"
|
||||
|
||||
@@ -63,7 +64,7 @@ type Server struct {
|
||||
tokenCache sync.Map
|
||||
|
||||
// nsManager управляет жизненным циклом пользовательских namespace-ов.
|
||||
nsManager *fission.NSManager
|
||||
nsManager *cloud.NSManager
|
||||
}
|
||||
|
||||
// Config содержит все параметры для создания Server.
|
||||
@@ -95,7 +96,7 @@ func NewServer(cfg Config) *Server {
|
||||
testMode: cfg.TestMode,
|
||||
llmURL: cfg.LLMUrl,
|
||||
llmKey: cfg.LLMKey,
|
||||
nsManager: fission.NewNSManager(cfg.Dyn),
|
||||
nsManager: cloud.NewNSManager(cfg.Dyn),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -153,7 +154,7 @@ func (s *Server) Handler() http.Handler {
|
||||
}
|
||||
|
||||
// NSManager возвращает указатель на NSManager для запуска фоновых горутин из main.
|
||||
func (s *Server) NSManager() *fission.NSManager {
|
||||
func (s *Server) NSManager() *cloud.NSManager {
|
||||
return s.nsManager
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
// Package cloud реализует облачно-специфичный уровень мультитенантности поверх Fission.
|
||||
//
|
||||
// # Архитектура слоёв
|
||||
//
|
||||
// Layer 1 (pkg/executor/multitenant в fission-src):
|
||||
// - Универсальный, platform-agnostic
|
||||
// - Контракт: Namespace + label fission.io/managed=true → executor автоматически подхватывает
|
||||
// - Могут использовать любые платформы
|
||||
//
|
||||
// Layer 2 (этот пакет):
|
||||
// - Облачно-специфичная логика нашего провайдера
|
||||
// - ResourceQuota, LimitRange — лимиты ресурсов на тенанта (биллинг/изоляция)
|
||||
// - NetworkPolicy — сетевая изоляция между тенантами
|
||||
// - NSManager — оркестратор полного lifecycle namespace-а (singleflight + семафор)
|
||||
// - ExpiryReaper — удаление функций с истёкшим TTL
|
||||
//
|
||||
// # Зависимости
|
||||
//
|
||||
// cloud → fission (internal/fission/) → Kubernetes dynamic client
|
||||
//
|
||||
// internal/fission/ — чистый Fission CRD adapter: GVR константы, EnsureEnvironment, SetupFissionNamespace.
|
||||
// cloud/ — надстройка: добавляет квоты, сеть, управление lifecycle.
|
||||
package cloud
|
||||
@@ -0,0 +1,60 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/dynamic"
|
||||
)
|
||||
|
||||
var netpolGVR = schema.GroupVersionResource{Group: "networking.k8s.io", Version: "v1", Resource: "networkpolicies"}
|
||||
|
||||
// ApplyNetworkPolicy создаёт NetworkPolicy в namespace ns для межтенантной изоляции.
|
||||
//
|
||||
// Политика «deny-cross-tenant» запрещает входящий трафик из других пользовательских
|
||||
// namespace-ов, разрешая только:
|
||||
// - Трафик внутри самого namespace (pod-to-pod внутри тенанта)
|
||||
// - Трафик из Fission system namespace (router → function pod)
|
||||
// - Трафик из kube-system (kubelet health checks, CoreDNS)
|
||||
//
|
||||
// Это обеспечивает сетевую изоляцию между тенантами без отключения внутренних
|
||||
// вызовов Fission. Env var FISSION_SYSTEM_NAMESPACE задаёт имя Fission namespace
|
||||
// (default: "fission").
|
||||
func ApplyNetworkPolicy(ctx context.Context, dyn dynamic.Interface, ns string) {
|
||||
fissionSysNS := os.Getenv("FISSION_SYSTEM_NAMESPACE")
|
||||
if fissionSysNS == "" {
|
||||
fissionSysNS = "fission"
|
||||
}
|
||||
|
||||
netpolObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "networking.k8s.io/v1",
|
||||
"kind": "NetworkPolicy",
|
||||
"metadata": map[string]any{"name": "deny-cross-tenant", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"podSelector": map[string]any{}, // применяется ко всем подам namespace
|
||||
"policyTypes": []any{"Ingress"},
|
||||
"ingress": []any{
|
||||
// Разрешаем трафик внутри namespace (pod-to-pod)
|
||||
map[string]any{"from": []any{map[string]any{"podSelector": map[string]any{}}}},
|
||||
// Разрешаем трафик из Fission core namespace (router → function pod)
|
||||
map[string]any{"from": []any{map[string]any{"namespaceSelector": map[string]any{
|
||||
"matchLabels": map[string]any{"kubernetes.io/metadata.name": fissionSysNS},
|
||||
}}}},
|
||||
// Разрешаем трафик из kube-system (kubelet health checks, DNS)
|
||||
map[string]any{"from": []any{map[string]any{"namespaceSelector": map[string]any{
|
||||
"matchLabels": map[string]any{"kubernetes.io/metadata.name": "kube-system"},
|
||||
}}}},
|
||||
},
|
||||
},
|
||||
}}
|
||||
|
||||
_, err := dyn.Resource(netpolGVR).Namespace(ns).Create(ctx, netpolObj, metav1.CreateOptions{})
|
||||
if err != nil && !apierrors.IsAlreadyExists(err) {
|
||||
log.Printf("cloud.ApplyNetworkPolicy: %s: %v", ns, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/client-go/dynamic"
|
||||
)
|
||||
|
||||
var (
|
||||
quotaGVR = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "resourcequotas"}
|
||||
limitRangeGVR = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "limitranges"}
|
||||
)
|
||||
|
||||
// ApplyResourceQuota создаёт ResourceQuota и LimitRange в namespace ns.
|
||||
// Параметры берутся из env vars (см. ниже) — можно тюнить без пересборки.
|
||||
// Идемпотентно: если объекты уже существуют — пропускается без ошибки.
|
||||
//
|
||||
// ResourceQuota ограничивает суммарные ресурсы пользователя:
|
||||
// - CPU / memory requests и limits
|
||||
// - Количество подов, функций, пакетов, triggers
|
||||
//
|
||||
// LimitRange задаёт дефолтные limits/requests на контейнер:
|
||||
// поды без явных limits получают defaults автоматически → нет unbounded containers.
|
||||
//
|
||||
// Env vars (все опциональны, defaults указаны ниже):
|
||||
//
|
||||
// QUOTA_REQ_CPU — суммарный CPU requests (default: "1")
|
||||
// QUOTA_REQ_MEM — суммарная memory requests (default: "1Gi")
|
||||
// QUOTA_LIM_CPU — суммарный CPU limits (default: "4")
|
||||
// QUOTA_LIM_MEM — суммарная memory limits (default: "4Gi")
|
||||
// QUOTA_PODS — макс. количество подов (default: "30")
|
||||
// QUOTA_FUNCTIONS — макс. количество функций (default: "20")
|
||||
// QUOTA_PACKAGES — макс. количество пакетов (default: "40")
|
||||
// QUOTA_HTTPTRIGGERS — макс. количество HTTP triggers (default: "20")
|
||||
// LIMIT_DEFAULT_CPU — дефолтный CPU limit на контейнер (default: "500m")
|
||||
// LIMIT_DEFAULT_MEM — дефолтная memory limit на контейнер (default: "256Mi")
|
||||
// LIMIT_REQ_CPU — дефолтный CPU request на контейнер (default: "50m")
|
||||
// LIMIT_REQ_MEM — дефолтный memory request на контейнер (default: "64Mi")
|
||||
// LIMIT_MAX_CPU — максимальный CPU limit на контейнер (default: "2")
|
||||
// LIMIT_MAX_MEM — максимальная memory limit на контейнер (default: "1Gi")
|
||||
func ApplyResourceQuota(ctx context.Context, dyn dynamic.Interface, ns string) {
|
||||
quotaObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
"kind": "ResourceQuota",
|
||||
"metadata": map[string]any{"name": "user-quota", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"hard": map[string]any{
|
||||
"requests.cpu": envDefault("QUOTA_REQ_CPU", "1"),
|
||||
"requests.memory": envDefault("QUOTA_REQ_MEM", "1Gi"),
|
||||
"limits.cpu": envDefault("QUOTA_LIM_CPU", "4"),
|
||||
"limits.memory": envDefault("QUOTA_LIM_MEM", "4Gi"),
|
||||
"pods": envDefault("QUOTA_PODS", "30"),
|
||||
"count/functions.fission.io": envDefault("QUOTA_FUNCTIONS", "20"),
|
||||
"count/packages.fission.io": envDefault("QUOTA_PACKAGES", "40"),
|
||||
"count/httptriggers.fission.io": envDefault("QUOTA_HTTPTRIGGERS", "20"),
|
||||
},
|
||||
},
|
||||
}}
|
||||
_, err := dyn.Resource(quotaGVR).Namespace(ns).Create(ctx, quotaObj, metav1.CreateOptions{})
|
||||
if err != nil && !apierrors.IsAlreadyExists(err) {
|
||||
log.Printf("cloud.ApplyResourceQuota: %s: %v", ns, err)
|
||||
}
|
||||
|
||||
limitRangeObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
"kind": "LimitRange",
|
||||
"metadata": map[string]any{"name": "user-limits", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"limits": []any{
|
||||
map[string]any{
|
||||
"type": "Container",
|
||||
"default": map[string]any{
|
||||
"cpu": envDefault("LIMIT_DEFAULT_CPU", "500m"),
|
||||
"memory": envDefault("LIMIT_DEFAULT_MEM", "256Mi"),
|
||||
},
|
||||
"defaultRequest": map[string]any{
|
||||
"cpu": envDefault("LIMIT_REQ_CPU", "50m"),
|
||||
"memory": envDefault("LIMIT_REQ_MEM", "64Mi"),
|
||||
},
|
||||
"max": map[string]any{
|
||||
"cpu": envDefault("LIMIT_MAX_CPU", "2"),
|
||||
"memory": envDefault("LIMIT_MAX_MEM", "1Gi"),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
_, err = dyn.Resource(limitRangeGVR).Namespace(ns).Create(ctx, limitRangeObj, metav1.CreateOptions{})
|
||||
if err != nil && !apierrors.IsAlreadyExists(err) {
|
||||
log.Printf("cloud.ApplyResourceQuota LimitRange: %s: %v", ns, err)
|
||||
}
|
||||
}
|
||||
|
||||
// envDefault читает env переменную или возвращает fallback.
|
||||
func envDefault(key, fallback string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
package cloud
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/client-go/dynamic"
|
||||
|
||||
"fission-console/internal/fission"
|
||||
)
|
||||
|
||||
// NSManager управляет полным lifecycle пользовательских namespace-ов:
|
||||
// - Оркестрирует создание NS через fission.SetupFissionNamespace (Layer 1 adapter)
|
||||
// - Накладывает облачные политики: ResourceQuota, LimitRange, NetworkPolicy
|
||||
// - Реализует singleflight + кэш + семафор для защиты от concurrent создания
|
||||
// - Запускает ExpiryReaper для удаления функций с истёкшим TTL
|
||||
//
|
||||
// NSManager не занимается синхронизацией FISSION_RESOURCE_NAMESPACES —
|
||||
// это теперь задача Layer 1 (multitenant.NSWatcher в executor).
|
||||
type NSManager struct {
|
||||
dyn dynamic.Interface
|
||||
|
||||
// Singleflight + кэш: ensuredNS содержит namespace-ы уже инициализированные в этом процессе.
|
||||
// ensuredNSInFlight — текущие in-flight операции (ключ = namespace).
|
||||
ensuredNSMu sync.Mutex
|
||||
ensuredNS map[string]struct{}
|
||||
ensuredNSInFlight map[string]*nsInflightEnsure
|
||||
|
||||
// nsSemaphore ограничивает параллелизм EnsureUserNS — не более 3 одновременно.
|
||||
// Без него 10 новых пользователей → 140 K8s API calls одновременно → throttle → 504.
|
||||
nsSemaphore chan struct{}
|
||||
}
|
||||
|
||||
// nsInflightEnsure — состояние in-flight вызова EnsureUserNS.
|
||||
// Все конкурентные goroutine ждут close(done), затем читают err (singleflight без зависимостей).
|
||||
type nsInflightEnsure struct {
|
||||
done chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
// NewNSManager создаёт NSManager.
|
||||
func NewNSManager(dyn dynamic.Interface) *NSManager {
|
||||
return &NSManager{
|
||||
dyn: dyn,
|
||||
ensuredNS: make(map[string]struct{}),
|
||||
ensuredNSInFlight: make(map[string]*nsInflightEnsure),
|
||||
nsSemaphore: make(chan struct{}, 3),
|
||||
}
|
||||
}
|
||||
|
||||
// EnsureUserNS — единая точка входа для создания namespace при первом запросе пользователя.
|
||||
// Гарантирует что namespace + Fission RBAC + ResourceQuota + NetworkPolicy существуют.
|
||||
//
|
||||
// Паттерны:
|
||||
// - Singleflight: параллельные вызовы для одного NS объединяются — один делает работу, остальные ждут
|
||||
// - In-memory кэш: повторные вызовы для уже созданного NS — быстрый путь без K8s calls
|
||||
// - Семафор (3): ограничивает параллельное создание новых NS
|
||||
func (m *NSManager) EnsureUserNS(ctx context.Context, ns string) error {
|
||||
m.ensuredNSMu.Lock()
|
||||
if _, ok := m.ensuredNS[ns]; ok {
|
||||
m.ensuredNSMu.Unlock()
|
||||
return nil // быстрый путь: уже создан
|
||||
}
|
||||
if inflight, ok := m.ensuredNSInFlight[ns]; ok {
|
||||
m.ensuredNSMu.Unlock()
|
||||
select {
|
||||
case <-inflight.done:
|
||||
return inflight.err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
inflight := &nsInflightEnsure{done: make(chan struct{})}
|
||||
m.ensuredNSInFlight[ns] = inflight
|
||||
m.ensuredNSMu.Unlock()
|
||||
|
||||
select {
|
||||
case m.nsSemaphore <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
m.ensuredNSMu.Lock()
|
||||
delete(m.ensuredNSInFlight, ns)
|
||||
m.ensuredNSMu.Unlock()
|
||||
inflight.err = ctx.Err()
|
||||
close(inflight.done)
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
ensureCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
|
||||
inflight.err = m.provisionNamespace(ensureCtx, ns)
|
||||
cancel()
|
||||
<-m.nsSemaphore
|
||||
|
||||
m.ensuredNSMu.Lock()
|
||||
delete(m.ensuredNSInFlight, ns)
|
||||
if inflight.err == nil {
|
||||
m.ensuredNS[ns] = struct{}{}
|
||||
}
|
||||
m.ensuredNSMu.Unlock()
|
||||
close(inflight.done)
|
||||
|
||||
return inflight.err
|
||||
}
|
||||
|
||||
// provisionNamespace выполняет полный lifecycle setup нового namespace:
|
||||
// 1. fission.SetupFissionNamespace — создаёт NS с меткой fission.io/managed=true
|
||||
// + Fission ServiceAccounts + Fission RoleBindings (Fission-специфичная часть)
|
||||
// 2. cloud.ApplyResourceQuota — ResourceQuota + LimitRange (облачная политика)
|
||||
// 3. cloud.ApplyNetworkPolicy — NetworkPolicy изоляция (облачная политика)
|
||||
//
|
||||
// Все операции идемпотентны (IsAlreadyExists игнорируется).
|
||||
// Ошибки quota/network логируются но не блокируют — NS будет работать и без них.
|
||||
func (m *NSManager) provisionNamespace(ctx context.Context, ns string) error {
|
||||
// Шаг 1: Fission-специфичный setup (Layer 1 adapter).
|
||||
// Создаёт NS с label fission.io/managed=true → executor auto-discovers через NSWatcher.
|
||||
if err := fission.SetupFissionNamespace(ctx, m.dyn, ns); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Шаг 2: облачные политики — best-effort (не блокируют запуск функций).
|
||||
ApplyResourceQuota(ctx, m.dyn, ns)
|
||||
ApplyNetworkPolicy(ctx, m.dyn, ns)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartExpiryReaper запускает фоновый goroutine для удаления функций с истёкшим TTL.
|
||||
// Interval — как часто проверять. Рекомендуемое значение: 5 минут.
|
||||
func (m *NSManager) StartExpiryReaper(interval time.Duration) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
log.Printf("cloud.ExpiryReaper: started, interval=%v", interval)
|
||||
for range ticker.C {
|
||||
m.runExpiryReap()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// runExpiryReap обходит все управляемые namespace-ы и удаляет протухшие функции.
|
||||
func (m *NSManager) runExpiryReap() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
nsList, err := m.dyn.Resource(fission.NamespaceGVR).List(ctx, metav1.ListOptions{
|
||||
LabelSelector: "managed-by=fission-console",
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("cloud.ExpiryReaper: list namespaces: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
for _, ns := range nsList.Items {
|
||||
m.reapExpiredFunctionsInNS(ctx, ns.GetName(), now)
|
||||
}
|
||||
}
|
||||
|
||||
// reapExpiredFunctionsInNS удаляет протухшие функции в конкретном namespace.
|
||||
// Для каждой удалённой функции вызывает fission.CleanupEnvironmentIfUnused.
|
||||
// Также удаляет orphan packages — пакеты без соответствующей функции.
|
||||
func (m *NSManager) reapExpiredFunctionsInNS(ctx context.Context, ns string, now time.Time) {
|
||||
functions, err := m.dyn.Resource(fission.FunctionGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
log.Printf("cloud.ExpiryReaper: list functions in %s: %v", ns, err)
|
||||
return
|
||||
}
|
||||
|
||||
activeFunctions := make(map[string]struct{}, len(functions.Items))
|
||||
for _, fn := range functions.Items {
|
||||
activeFunctions[fn.GetName()] = struct{}{}
|
||||
}
|
||||
|
||||
for _, fn := range functions.Items {
|
||||
expiresAtStr, _, _ := unstructured.NestedString(fn.Object, "metadata", "annotations", "fission-console/expires-at")
|
||||
if expiresAtStr == "" {
|
||||
continue
|
||||
}
|
||||
expiresAt, parseErr := time.Parse(time.RFC3339, expiresAtStr)
|
||||
if parseErr != nil {
|
||||
log.Printf("cloud.ExpiryReaper: parse expires-at for %s/%s: %v", ns, fn.GetName(), parseErr)
|
||||
continue
|
||||
}
|
||||
if now.Before(expiresAt) {
|
||||
continue
|
||||
}
|
||||
|
||||
fnName := fn.GetName()
|
||||
envName, _, _ := unstructured.NestedString(fn.Object, "spec", "environment", "name")
|
||||
pkgName, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "packageref", "name")
|
||||
log.Printf("cloud.ExpiryReaper: deleting expired function %s/%s (expired %s ago)", ns, fnName, now.Sub(expiresAt).Round(time.Second))
|
||||
|
||||
triggers, tErr := m.dyn.Resource(fission.HTTPTrigGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if tErr == nil {
|
||||
for _, trig := range triggers.Items {
|
||||
refName, _, _ := unstructured.NestedString(trig.Object, "spec", "functionref", "name")
|
||||
if refName == fnName {
|
||||
_ = m.dyn.Resource(fission.HTTPTrigGVR).Namespace(ns).Delete(ctx, trig.GetName(), metav1.DeleteOptions{})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_ = m.dyn.Resource(fission.FunctionGVR).Namespace(ns).Delete(ctx, fnName, metav1.DeleteOptions{})
|
||||
if pkgName != "" {
|
||||
_ = m.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, pkgName, metav1.DeleteOptions{})
|
||||
}
|
||||
if envName != "" {
|
||||
fission.CleanupEnvironmentIfUnused(ctx, m.dyn, ns, envName)
|
||||
}
|
||||
}
|
||||
|
||||
// Orphan packages — пакеты без соответствующей функции
|
||||
packages, pkgListErr := m.dyn.Resource(fission.PackageGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if pkgListErr == nil {
|
||||
for _, pkg := range packages.Items {
|
||||
pkgName := pkg.GetName()
|
||||
if !strings.HasSuffix(pkgName, "-pkg") {
|
||||
continue
|
||||
}
|
||||
fnName := strings.TrimSuffix(pkgName, "-pkg")
|
||||
if _, exists := activeFunctions[fnName]; !exists {
|
||||
log.Printf("cloud.ExpiryReaper: deleting orphan package %s/%s", ns, pkgName)
|
||||
_ = m.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, pkgName, metav1.DeleteOptions{})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// IsAlreadyExists — вспомогательная обёртка для читаемости.
|
||||
var _ = apierrors.IsAlreadyExists
|
||||
@@ -1,3 +1,10 @@
|
||||
// Package fission — чистый адаптер для Fission CRD API.
|
||||
// Содержит только то что специфично для Fission и не зависит от облачной платформы:
|
||||
// - SetupFissionNamespace: создание NS + label для Layer 1 NSWatcher + Fission ServiceAccounts + RoleBindings
|
||||
// - EnsureEnvironment, CleanupEnvironmentIfUnused: управление Fission Environment CRD
|
||||
// - GVR константы (client.go)
|
||||
//
|
||||
// Облачно-специфичные вещи (ResourceQuota, LimitRange, NetworkPolicy, NSManager) — в пакете cloud.
|
||||
package fission
|
||||
|
||||
import (
|
||||
@@ -5,124 +12,31 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/client-go/dynamic"
|
||||
|
||||
"encoding/json"
|
||||
)
|
||||
|
||||
// NSInflightEnsure — состояние in-flight вызова EnsureUserNamespace.
|
||||
// Все параллельные горутины ждут close(done), затем читают err.
|
||||
// Реализует ручной singleflight без внешних зависимостей.
|
||||
type NSInflightEnsure struct {
|
||||
done chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
// NSManager управляет жизненным циклом пользовательских namespace-ов:
|
||||
// создание, RBAC, quota, network policies, reconciler, reaper.
|
||||
type NSManager struct {
|
||||
dyn dynamic.Interface
|
||||
|
||||
// nsReconcileCh — сигнал для немедленного запуска NS reconciler.
|
||||
// Буферизирован на 1: несколько сигналов схлопываются в один запуск.
|
||||
NSReconcileCh chan struct{}
|
||||
|
||||
// Singleflight + кэш для ensureUserNamespace:
|
||||
// ensuredNS — namespace-ы которые уже инициализированы в этом запуске процесса.
|
||||
// ensuredNSInFlight — текущие in-flight операции (ключ = namespace name).
|
||||
ensuredNSMu sync.Mutex
|
||||
ensuredNS map[string]struct{}
|
||||
ensuredNSInFlight map[string]*NSInflightEnsure
|
||||
|
||||
// nsSemaphore ограничивает параллелизм EnsureUserNamespace — не более 3 одновременно.
|
||||
// Без него 10 новых пользователей генерируют 140 K8s API calls одновременно → throttle → 504.
|
||||
// С семафором: 3 batch-а по 14 calls → ~3 × 5s = 15s total, все укладываются в timeout.
|
||||
nsSemaphore chan struct{}
|
||||
}
|
||||
|
||||
// NewNSManager создаёт NSManager с инициализированными каналами и структурами.
|
||||
func NewNSManager(dyn dynamic.Interface) *NSManager {
|
||||
return &NSManager{
|
||||
dyn: dyn,
|
||||
NSReconcileCh: make(chan struct{}, 1),
|
||||
ensuredNS: make(map[string]struct{}),
|
||||
ensuredNSInFlight: make(map[string]*NSInflightEnsure),
|
||||
nsSemaphore: make(chan struct{}, 3),
|
||||
}
|
||||
}
|
||||
|
||||
// EnsureUserNS — единая точка входа для гарантии существования пользовательского namespace.
|
||||
// Реализует singleflight + in-memory кэш + семафор параллелизма.
|
||||
// SetupFissionNamespace создаёт namespace с нужными метками и регистрирует в нём
|
||||
// все Fission ServiceAccounts и RoleBindings. Идемпотентно.
|
||||
//
|
||||
// Singleflight: если один goroutine уже создаёт namespace ns — остальные ждут его результата
|
||||
// вместо того чтобы запускать параллельные K8s API calls (вызывало throttle и 504).
|
||||
// Метки на Namespace:
|
||||
// - managed-by=fission-console — для фильтрации наших NS в cloud-слое
|
||||
// - fission.io/managed=true — для Layer 1 NSWatcher в executor (auto-discovery)
|
||||
//
|
||||
// Кэш: если namespace уже создан в этом запуске процесса — быстрый путь без K8s calls.
|
||||
func (m *NSManager) EnsureUserNS(ctx context.Context, ns string) error {
|
||||
m.ensuredNSMu.Lock()
|
||||
if _, ok := m.ensuredNS[ns]; ok {
|
||||
// Быстрый путь: уже создан в этой жизни процесса.
|
||||
m.ensuredNSMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
if inflight, ok := m.ensuredNSInFlight[ns]; ok {
|
||||
// Кто-то уже создаёт — ждём его результата (singleflight).
|
||||
m.ensuredNSMu.Unlock()
|
||||
select {
|
||||
case <-inflight.done:
|
||||
return inflight.err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
// Мы первые для этого namespace.
|
||||
inflight := &NSInflightEnsure{done: make(chan struct{})}
|
||||
m.ensuredNSInFlight[ns] = inflight
|
||||
m.ensuredNSMu.Unlock()
|
||||
|
||||
// Берём слот семафора — ограничиваем параллелизм.
|
||||
select {
|
||||
case m.nsSemaphore <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
m.ensuredNSMu.Lock()
|
||||
delete(m.ensuredNSInFlight, ns)
|
||||
m.ensuredNSMu.Unlock()
|
||||
inflight.err = ctx.Err()
|
||||
close(inflight.done)
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
ensureCtx, ensureCancel := context.WithTimeout(ctx, 60*time.Second)
|
||||
inflight.err = m.ensureUserNamespace(ensureCtx, ns)
|
||||
ensureCancel()
|
||||
<-m.nsSemaphore // освобождаем слот
|
||||
|
||||
m.ensuredNSMu.Lock()
|
||||
delete(m.ensuredNSInFlight, ns)
|
||||
if inflight.err == nil {
|
||||
m.ensuredNS[ns] = struct{}{}
|
||||
}
|
||||
m.ensuredNSMu.Unlock()
|
||||
close(inflight.done)
|
||||
|
||||
return inflight.err
|
||||
}
|
||||
|
||||
// ensureUserNamespace выполняет реальную работу: создаёт namespace + RBAC + quota + netpol.
|
||||
// Все операции идемпотентны (IsAlreadyExists игнорируется).
|
||||
func (m *NSManager) ensureUserNamespace(ctx context.Context, ns string) error {
|
||||
// 1. Создать namespace с меткой managed-by=fission-console.
|
||||
// Метка используется reconciler-ом для фильтрации наших namespace-ов.
|
||||
// ServiceAccounts:
|
||||
// - fission-fetcher, fission-builder — нужны Fission pool pods в user NS
|
||||
//
|
||||
// RoleBindings (cluster-admin в рамках NS):
|
||||
// - Для всех Fission system SA из FISSION_SYSTEM_NAMESPACE (default: fission)
|
||||
// - Почему cluster-admin, а не admin: ClusterRole "admin" не включает fission.io/* CRD-группы,
|
||||
// executor получает "RBAC escalation" при создании Role. cluster-admin в RoleBinding
|
||||
// (не ClusterRoleBinding) безопасен — даёт полный доступ только внутри NS.
|
||||
func SetupFissionNamespace(ctx context.Context, dyn dynamic.Interface, ns string) error {
|
||||
// 1. Namespace
|
||||
nsObj := &unstructured.Unstructured{
|
||||
Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
@@ -130,56 +44,43 @@ func (m *NSManager) ensureUserNamespace(ctx context.Context, ns string) error {
|
||||
"metadata": map[string]any{
|
||||
"name": ns,
|
||||
"labels": map[string]any{
|
||||
"managed-by": "fission-console",
|
||||
"managed-by": "fission-console",
|
||||
"fission.io/managed": "true", // Layer 1: executor NSWatcher auto-discovers this NS
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := m.dyn.Resource(NamespaceGVR).Create(ctx, nsObj, metav1.CreateOptions{})
|
||||
newlyCreated := err == nil
|
||||
_, err := dyn.Resource(NamespaceGVR).Create(ctx, nsObj, metav1.CreateOptions{})
|
||||
if err != nil && !apierrors.IsAlreadyExists(err) {
|
||||
return fmt.Errorf("create namespace %s: %w", ns, err)
|
||||
}
|
||||
|
||||
// 1a. ServiceAccounts для Fission в user namespace.
|
||||
//
|
||||
// Fission pool pods (fetcher sidecar) запускаются в user namespace и требуют
|
||||
// `serviceAccountName: fission-fetcher` в том же namespace. Executor (с SERVICEACCOUNT_CHECK_ENABLED=false)
|
||||
// не создаёт эти SA автоматически → pool pods падают с "serviceaccount not found".
|
||||
// 2. ServiceAccounts для Fission pool pods в user namespace.
|
||||
// Fetcher sidecar требует fission-fetcher SA в том же NS где запускается.
|
||||
saGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "serviceaccounts"}
|
||||
for _, saName := range []string{"fission-fetcher", "fission-builder"} {
|
||||
saObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
"kind": "ServiceAccount",
|
||||
"metadata": map[string]any{
|
||||
"name": saName,
|
||||
"namespace": ns,
|
||||
},
|
||||
"metadata": map[string]any{"name": saName, "namespace": ns},
|
||||
}}
|
||||
_, saErr := m.dyn.Resource(saGVR).Namespace(ns).Create(ctx, saObj, metav1.CreateOptions{})
|
||||
_, saErr := dyn.Resource(saGVR).Namespace(ns).Create(ctx, saObj, metav1.CreateOptions{})
|
||||
if saErr != nil && !apierrors.IsAlreadyExists(saErr) {
|
||||
log.Printf("ensureUserNamespace: create SA %s/%s: %v", ns, saName, saErr)
|
||||
log.Printf("fission.SetupFissionNamespace: create SA %s/%s: %v", ns, saName, saErr)
|
||||
}
|
||||
}
|
||||
|
||||
// 1b. RoleBindings для Fission SA в user namespace.
|
||||
//
|
||||
// Почему cluster-admin, а не admin:
|
||||
// ClusterRole "admin" не включает custom resource группы (fission.io/*).
|
||||
// Fission executor при старте пытается создать Role с правами на fission.io/packages,
|
||||
// и получает "attempting to grant RBAC permissions not currently held" — RBAC escalation
|
||||
// prevention. ClusterRole "cluster-admin" в контексте RoleBinding (не ClusterRoleBinding)
|
||||
// даёт полный доступ ТОЛЬКО внутри конкретного namespace — это безопасно.
|
||||
type rbSubject struct {
|
||||
name string
|
||||
namespace string
|
||||
binding string
|
||||
}
|
||||
// 3. RoleBindings для всех Fission system SA.
|
||||
fissionSysNS := os.Getenv("FISSION_SYSTEM_NAMESPACE")
|
||||
if fissionSysNS == "" {
|
||||
fissionSysNS = "fission"
|
||||
}
|
||||
fissionSAs := []rbSubject{
|
||||
type rbDef struct {
|
||||
name string
|
||||
namespace string
|
||||
binding string
|
||||
}
|
||||
bindings := []rbDef{
|
||||
{name: "fission-executor", binding: "fission-executor-user-ns"},
|
||||
{name: "fission-router", binding: "fission-router-user-ns"},
|
||||
{name: "fission-buildermgr", binding: "fission-buildermgr-user-ns"},
|
||||
@@ -187,13 +88,13 @@ func (m *NSManager) ensureUserNamespace(ctx context.Context, ns string) error {
|
||||
{name: "fission-timer", binding: "fission-timer-user-ns"},
|
||||
{name: "fission-fetcher", binding: "fission-fetcher-system-user-ns"},
|
||||
{name: "fission-builder", binding: "fission-builder-system-user-ns"},
|
||||
// fetcher/builder локальные в user namespace (там запускаются pool pods)
|
||||
// fetcher/builder также нужны локально (запускаются в user NS)
|
||||
{name: "fission-fetcher", namespace: ns, binding: "fission-fetcher-local-user-ns"},
|
||||
{name: "fission-builder", namespace: ns, binding: "fission-builder-local-user-ns"},
|
||||
}
|
||||
rbGVR := schema.GroupVersionResource{Group: "rbac.authorization.k8s.io", Version: "v1", Resource: "rolebindings"}
|
||||
for _, sa := range fissionSAs {
|
||||
subjectNS := sa.namespace
|
||||
for _, rb := range bindings {
|
||||
subjectNS := rb.namespace
|
||||
if subjectNS == "" {
|
||||
subjectNS = fissionSysNS
|
||||
}
|
||||
@@ -201,406 +102,22 @@ func (m *NSManager) ensureUserNamespace(ctx context.Context, ns string) error {
|
||||
Object: map[string]any{
|
||||
"apiVersion": "rbac.authorization.k8s.io/v1",
|
||||
"kind": "RoleBinding",
|
||||
"metadata": map[string]any{
|
||||
"name": sa.binding,
|
||||
"namespace": ns,
|
||||
},
|
||||
"metadata": map[string]any{"name": rb.binding, "namespace": ns},
|
||||
"roleRef": map[string]any{
|
||||
"apiGroup": "rbac.authorization.k8s.io",
|
||||
"kind": "ClusterRole",
|
||||
"name": "cluster-admin",
|
||||
},
|
||||
"subjects": []any{
|
||||
map[string]any{
|
||||
"kind": "ServiceAccount",
|
||||
"name": sa.name,
|
||||
"namespace": subjectNS,
|
||||
},
|
||||
},
|
||||
"subjects": []any{map[string]any{
|
||||
"kind": "ServiceAccount", "name": rb.name, "namespace": subjectNS,
|
||||
}},
|
||||
},
|
||||
}
|
||||
_, rbErr := m.dyn.Resource(rbGVR).Namespace(ns).Create(ctx, rbObj, metav1.CreateOptions{})
|
||||
_, rbErr := dyn.Resource(rbGVR).Namespace(ns).Create(ctx, rbObj, metav1.CreateOptions{})
|
||||
if rbErr != nil && !apierrors.IsAlreadyExists(rbErr) {
|
||||
log.Printf("ensureUserNamespace: create rolebinding %s/%s@%s: %v", ns, sa.name, subjectNS, rbErr)
|
||||
log.Printf("fission.SetupFissionNamespace: create rolebinding %s/%s@%s: %v", ns, rb.name, subjectNS, rbErr)
|
||||
}
|
||||
}
|
||||
|
||||
// 1c. ResourceQuota — ограничиваем ресурсы одного пользователя.
|
||||
// Без этого одна функция может исчерпать CPU/RAM всего кластера.
|
||||
// Параметры задаются через env vars для гибкой настройки без пересборки.
|
||||
quotaGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "resourcequotas"}
|
||||
quotaObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
"kind": "ResourceQuota",
|
||||
"metadata": map[string]any{"name": "user-quota", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"hard": map[string]any{
|
||||
"requests.cpu": envDefault("QUOTA_REQ_CPU", "1"),
|
||||
"requests.memory": envDefault("QUOTA_REQ_MEM", "1Gi"),
|
||||
"limits.cpu": envDefault("QUOTA_LIM_CPU", "4"),
|
||||
"limits.memory": envDefault("QUOTA_LIM_MEM", "4Gi"),
|
||||
"pods": envDefault("QUOTA_PODS", "30"),
|
||||
"count/functions.fission.io": envDefault("QUOTA_FUNCTIONS", "20"),
|
||||
"count/packages.fission.io": envDefault("QUOTA_PACKAGES", "40"),
|
||||
"count/httptriggers.fission.io": envDefault("QUOTA_HTTPTRIGGERS", "20"),
|
||||
},
|
||||
},
|
||||
}}
|
||||
_, quotaErr := m.dyn.Resource(quotaGVR).Namespace(ns).Create(ctx, quotaObj, metav1.CreateOptions{})
|
||||
if quotaErr != nil && !apierrors.IsAlreadyExists(quotaErr) {
|
||||
log.Printf("ensureUserNamespace: create ResourceQuota %s: %v", ns, quotaErr)
|
||||
}
|
||||
|
||||
// 1d. LimitRange — дефолтные лимиты на контейнер.
|
||||
// Поды без явных limits — unbounded; LimitRange автоматически добавляет defaults.
|
||||
limitRangeGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "limitranges"}
|
||||
limitRangeObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "v1",
|
||||
"kind": "LimitRange",
|
||||
"metadata": map[string]any{"name": "user-limits", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"limits": []any{
|
||||
map[string]any{
|
||||
"type": "Container",
|
||||
"default": map[string]any{"cpu": envDefault("LIMIT_DEFAULT_CPU", "500m"), "memory": envDefault("LIMIT_DEFAULT_MEM", "256Mi")},
|
||||
"defaultRequest": map[string]any{
|
||||
"cpu": envDefault("LIMIT_REQ_CPU", "50m"), "memory": envDefault("LIMIT_REQ_MEM", "64Mi"),
|
||||
},
|
||||
"max": map[string]any{"cpu": envDefault("LIMIT_MAX_CPU", "2"), "memory": envDefault("LIMIT_MAX_MEM", "1Gi")},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
_, lrErr := m.dyn.Resource(limitRangeGVR).Namespace(ns).Create(ctx, limitRangeObj, metav1.CreateOptions{})
|
||||
if lrErr != nil && !apierrors.IsAlreadyExists(lrErr) {
|
||||
log.Printf("ensureUserNamespace: create LimitRange %s: %v", ns, lrErr)
|
||||
}
|
||||
|
||||
// 1e. NetworkPolicy — запрещаем входящий трафик из других user namespace.
|
||||
// Разрешаем: внутри namespace, из fission core namespace, из kube-system.
|
||||
// Запрещаем: трафик от подов других user namespace (межтенантная изоляция).
|
||||
fissionSysNSForNetpol := os.Getenv("FISSION_SYSTEM_NAMESPACE")
|
||||
if fissionSysNSForNetpol == "" {
|
||||
fissionSysNSForNetpol = "fission"
|
||||
}
|
||||
netpolGVR := schema.GroupVersionResource{Group: "networking.k8s.io", Version: "v1", Resource: "networkpolicies"}
|
||||
netpolObj := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "networking.k8s.io/v1",
|
||||
"kind": "NetworkPolicy",
|
||||
"metadata": map[string]any{"name": "deny-cross-tenant", "namespace": ns},
|
||||
"spec": map[string]any{
|
||||
"podSelector": map[string]any{}, // применяется ко всем подам namespace
|
||||
"policyTypes": []any{"Ingress"},
|
||||
"ingress": []any{
|
||||
// Разрешаем трафик внутри namespace
|
||||
map[string]any{"from": []any{map[string]any{"podSelector": map[string]any{}}}},
|
||||
// Разрешаем трафик из fission core namespace (router → function pod)
|
||||
map[string]any{"from": []any{map[string]any{"namespaceSelector": map[string]any{
|
||||
"matchLabels": map[string]any{"kubernetes.io/metadata.name": fissionSysNSForNetpol},
|
||||
}}}},
|
||||
// Разрешаем трафик из kube-system (kubelet health checks, DNS)
|
||||
map[string]any{"from": []any{map[string]any{"namespaceSelector": map[string]any{
|
||||
"matchLabels": map[string]any{"kubernetes.io/metadata.name": "kube-system"},
|
||||
}}}},
|
||||
},
|
||||
},
|
||||
}}
|
||||
_, npErr := m.dyn.Resource(netpolGVR).Namespace(ns).Create(ctx, netpolObj, metav1.CreateOptions{})
|
||||
if npErr != nil && !apierrors.IsAlreadyExists(npErr) {
|
||||
log.Printf("ensureUserNamespace: create NetworkPolicy %s: %v", ns, npErr)
|
||||
}
|
||||
|
||||
// 1f. Сигналим NS reconciler что появился новый namespace.
|
||||
// Reconciler сам синхронизирует FISSION_RESOURCE_NAMESPACES без race condition и rolling restarts.
|
||||
if newlyCreated {
|
||||
select {
|
||||
case m.NSReconcileCh <- struct{}{}:
|
||||
default: // уже есть сигнал в буфере — не блокируем
|
||||
}
|
||||
}
|
||||
|
||||
// Environments создаются лениво (lazy) в момент создания первой функции на языке.
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartNSReconciler запускает фоновый reconciler FISSION_RESOURCE_NAMESPACES.
|
||||
//
|
||||
// Проблема при scale:
|
||||
// - Прямой патч на горячем пути → rolling restart всех Fission deployments при каждом новом юзере
|
||||
// - Без мьютекса: 100 юзеров одновременно → read-modify-write race → namespace-ы теряются
|
||||
// - Ручное kubectl delete ns → namespace остаётся в переменной вечно
|
||||
//
|
||||
// Решение: единственный goroutine с debounce-каналом.
|
||||
// - Запускается по таймеру или немедленно через NSReconcileCh
|
||||
// - 1000 юзеров одновременно → 1 патч вместо 5000
|
||||
// - Нет race condition (один goroutine, один writer)
|
||||
func (m *NSManager) StartNSReconciler(interval time.Duration) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
log.Printf("nsReconciler: started, interval=%v", interval)
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
m.reconcileNSList()
|
||||
case <-m.NSReconcileCh:
|
||||
m.reconcileNSList()
|
||||
// Дренируем канал чтобы не запускаться дважды подряд после сигнала
|
||||
drain:
|
||||
for {
|
||||
select {
|
||||
case <-m.NSReconcileCh:
|
||||
default:
|
||||
break drain
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// reconcileNSList синхронизирует FISSION_RESOURCE_NAMESPACES с реально существующими namespace-ами.
|
||||
//
|
||||
// Алгоритм:
|
||||
// 1. Читает все namespace-ы с меткой managed-by=fission-console из k8s (источник истины)
|
||||
// 2. Читает текущий FISSION_RESOURCE_NAMESPACES из router deployment
|
||||
// 3. Вычисляет desired = "default" + активные namespace-ы
|
||||
// 4. Если desired == current → ничего не делает (нет rolling restart!)
|
||||
// 5. Если разница → один патч всех Fission deployments
|
||||
func (m *NSManager) reconcileNSList() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
|
||||
fissionNS := os.Getenv("FISSION_SYSTEM_NAMESPACE")
|
||||
if fissionNS == "" {
|
||||
fissionNS = "fission"
|
||||
}
|
||||
|
||||
// Шаг 1: реальные namespace-ы с нашей меткой
|
||||
nsList, err := m.dyn.Resource(NamespaceGVR).List(ctx, metav1.ListOptions{
|
||||
LabelSelector: "managed-by=fission-console",
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("nsReconciler: list namespaces: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Шаг 2: фильтруем — берём только Active. Terminating убираем (они уже умирают).
|
||||
desired := map[string]struct{}{"default": {}}
|
||||
for _, ns := range nsList.Items {
|
||||
phase, _, _ := unstructured.NestedString(ns.Object, "status", "phase")
|
||||
if phase == "Active" {
|
||||
desired[ns.GetName()] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
// Шаг 3: текущее значение из router deployment
|
||||
routerDep, err := m.dyn.Resource(DeploymentGVR).Namespace(fissionNS).Get(ctx, "router", metav1.GetOptions{})
|
||||
if err != nil {
|
||||
log.Printf("nsReconciler: get router: %v", err)
|
||||
return
|
||||
}
|
||||
currentVal := extractFissionNSEnv(routerDep)
|
||||
|
||||
// Шаг 4: сравниваем current с desired
|
||||
currentSet := map[string]struct{}{}
|
||||
for _, p := range strings.Split(currentVal, ",") {
|
||||
if t := strings.TrimSpace(p); t != "" {
|
||||
currentSet[t] = struct{}{}
|
||||
}
|
||||
}
|
||||
same := len(currentSet) == len(desired)
|
||||
if same {
|
||||
for k := range desired {
|
||||
if _, ok := currentSet[k]; !ok {
|
||||
same = false
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if same {
|
||||
return // ничего менять не нужно — нет патча, нет rolling restart
|
||||
}
|
||||
|
||||
// Шаг 5: патчим все Fission deployments
|
||||
parts := make([]string, 0, len(desired))
|
||||
for ns := range desired {
|
||||
parts = append(parts, ns)
|
||||
}
|
||||
sort.Strings(parts)
|
||||
newVal := strings.Join(parts, ",")
|
||||
|
||||
fissionDeployments := []string{"router", "executor", "buildermgr", "kubewatcher", "timer"}
|
||||
patch := map[string]any{
|
||||
"spec": map[string]any{
|
||||
"template": map[string]any{
|
||||
"spec": map[string]any{
|
||||
"containers": []any{
|
||||
map[string]any{
|
||||
"name": "",
|
||||
"env": []any{
|
||||
map[string]any{
|
||||
"name": "FISSION_RESOURCE_NAMESPACES",
|
||||
"value": newVal,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, dep := range fissionDeployments {
|
||||
patch["spec"].(map[string]any)["template"].(map[string]any)["spec"].(map[string]any)["containers"].([]any)[0].(map[string]any)["name"] = dep
|
||||
patchBytes, _ := json.Marshal(patch)
|
||||
if _, pErr := m.dyn.Resource(DeploymentGVR).Namespace(fissionNS).Patch(
|
||||
ctx, dep, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}); pErr != nil {
|
||||
log.Printf("nsReconciler: patch %s: %v", dep, pErr)
|
||||
}
|
||||
}
|
||||
log.Printf("nsReconciler: synced FISSION_RESOURCE_NAMESPACES: %s → %s", currentVal, newVal)
|
||||
}
|
||||
|
||||
// extractFissionNSEnv читает FISSION_RESOURCE_NAMESPACES из первого контейнера deployment.
|
||||
func extractFissionNSEnv(dep *unstructured.Unstructured) string {
|
||||
containers, _, _ := unstructured.NestedSlice(dep.Object, "spec", "template", "spec", "containers")
|
||||
for _, c := range containers {
|
||||
cont, ok := c.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
envs, _, _ := unstructured.NestedSlice(cont, "env")
|
||||
for _, e := range envs {
|
||||
env, ok := e.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if env["name"] == "FISSION_RESOURCE_NAMESPACES" {
|
||||
if v, ok := env["value"].(string); ok && v != "" {
|
||||
return v
|
||||
}
|
||||
}
|
||||
}
|
||||
break // смотрим только первый контейнер
|
||||
}
|
||||
return "default"
|
||||
}
|
||||
|
||||
// StartExpiryReaper запускает фоновый goroutine для удаления функций с истёкшим TTL.
|
||||
func (m *NSManager) StartExpiryReaper(interval time.Duration) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
log.Printf("expiryReaper: started, interval=%v", interval)
|
||||
for range ticker.C {
|
||||
m.runExpiryReap()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// runExpiryReap обходит все управляемые namespace-ы и удаляет протухшие функции.
|
||||
func (m *NSManager) runExpiryReap() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
nsList, err := m.dyn.Resource(NamespaceGVR).List(ctx, metav1.ListOptions{
|
||||
LabelSelector: "managed-by=fission-console",
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("expiryReaper: list namespaces: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
for _, ns := range nsList.Items {
|
||||
m.reapExpiredFunctionsInNS(ctx, ns.GetName(), now)
|
||||
}
|
||||
}
|
||||
|
||||
// reapExpiredFunctionsInNS удаляет протухшие функции в конкретном namespace.
|
||||
// Для каждой удалённой функции вызывает CleanupEnvironmentIfUnused.
|
||||
// Также удаляет orphan packages — пакеты без соответствующей функции.
|
||||
func (m *NSManager) reapExpiredFunctionsInNS(ctx context.Context, ns string, now time.Time) {
|
||||
functions, err := m.dyn.Resource(FunctionGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if err != nil {
|
||||
log.Printf("expiryReaper: list functions in %s: %v", ns, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Строим множество активных функций для поиска orphan packages
|
||||
activeFunctions := make(map[string]struct{}, len(functions.Items))
|
||||
for _, fn := range functions.Items {
|
||||
activeFunctions[fn.GetName()] = struct{}{}
|
||||
}
|
||||
|
||||
for _, fn := range functions.Items {
|
||||
expiresAtStr, _, _ := unstructured.NestedString(fn.Object, "metadata", "annotations", "fission-console/expires-at")
|
||||
if expiresAtStr == "" {
|
||||
continue // нет TTL — функция живёт вечно
|
||||
}
|
||||
expiresAt, parseErr := time.Parse(time.RFC3339, expiresAtStr)
|
||||
if parseErr != nil {
|
||||
log.Printf("expiryReaper: parse expires-at for %s/%s: %v", ns, fn.GetName(), parseErr)
|
||||
continue
|
||||
}
|
||||
if now.Before(expiresAt) {
|
||||
continue // ещё не протухла
|
||||
}
|
||||
|
||||
fnName := fn.GetName()
|
||||
envName, _, _ := unstructured.NestedString(fn.Object, "spec", "environment", "name")
|
||||
pkgName, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "packageref", "name")
|
||||
log.Printf("expiryReaper: deleting expired function %s/%s (expired %s ago)", ns, fnName, now.Sub(expiresAt).Round(time.Second))
|
||||
|
||||
// Удаляем HTTPTrigger-ы ссылающиеся на эту функцию
|
||||
triggers, tErr := m.dyn.Resource(HTTPTrigGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if tErr == nil {
|
||||
for _, trig := range triggers.Items {
|
||||
refName, _, _ := unstructured.NestedString(trig.Object, "spec", "functionref", "name")
|
||||
if refName == fnName {
|
||||
_ = m.dyn.Resource(HTTPTrigGVR).Namespace(ns).Delete(ctx, trig.GetName(), metav1.DeleteOptions{})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_ = m.dyn.Resource(FunctionGVR).Namespace(ns).Delete(ctx, fnName, metav1.DeleteOptions{})
|
||||
if pkgName != "" {
|
||||
_ = m.dyn.Resource(PackageGVR).Namespace(ns).Delete(ctx, pkgName, metav1.DeleteOptions{})
|
||||
}
|
||||
if envName != "" {
|
||||
CleanupEnvironmentIfUnused(ctx, m.dyn, ns, envName)
|
||||
}
|
||||
}
|
||||
|
||||
// Сигналим reconciler — он проверит все namespace-ы и почистит список
|
||||
select {
|
||||
case m.NSReconcileCh <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
|
||||
// Orphan packages — пакеты без соответствующей функции
|
||||
// (остаются если процесс упал в середине удаления)
|
||||
packages, pkgListErr := m.dyn.Resource(PackageGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||||
if pkgListErr == nil {
|
||||
for _, pkg := range packages.Items {
|
||||
pkgName := pkg.GetName()
|
||||
// Конвенция именования: {fn-name}-pkg
|
||||
if !strings.HasSuffix(pkgName, "-pkg") {
|
||||
continue
|
||||
}
|
||||
fnName := strings.TrimSuffix(pkgName, "-pkg")
|
||||
if _, exists := activeFunctions[fnName]; !exists {
|
||||
log.Printf("expiryReaper: deleting orphan package %s/%s (no matching function)", ns, pkgName)
|
||||
_ = m.dyn.Resource(PackageGVR).Namespace(ns).Delete(ctx, pkgName, metav1.DeleteOptions{})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// envDefault читает env переменную или возвращает fallback.
|
||||
func envDefault(key, fallback string) string {
|
||||
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
|
||||
return v
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: ClusterRole
|
||||
metadata:
|
||||
name: fission-executor-multi-ns
|
||||
labels:
|
||||
app: fission-executor
|
||||
rules:
|
||||
# Fission CRD — читать environment/function/package в любом NS
|
||||
- apiGroups: ["fission.io"]
|
||||
resources: ["environments", "functions", "packages", "httptriggers"]
|
||||
verbs: ["get", "list", "watch"]
|
||||
# Pods — executor создаёт и управляет подами функций (poolmgr/newdeploy)
|
||||
- apiGroups: [""]
|
||||
resources: ["pods", "pods/log"]
|
||||
verbs: ["get", "list", "watch", "create", "delete", "update", "patch"]
|
||||
# Services — executor создаёт сервисы для функций
|
||||
- apiGroups: [""]
|
||||
resources: ["services"]
|
||||
verbs: ["get", "list", "watch", "create", "delete", "update", "patch"]
|
||||
# ConfigMaps/Secrets — для конфигурации функций
|
||||
- apiGroups: [""]
|
||||
resources: ["configmaps", "secrets"]
|
||||
verbs: ["get", "list", "watch", "create", "update", "patch"]
|
||||
# ServiceAccounts — для чтения SA функций
|
||||
- apiGroups: [""]
|
||||
resources: ["serviceaccounts"]
|
||||
verbs: ["get", "list", "watch"]
|
||||
# Deployments/ReplicaSets — newdeploy executor тип
|
||||
- apiGroups: ["apps"]
|
||||
resources: ["deployments", "replicasets"]
|
||||
verbs: ["get", "list", "watch", "create", "delete", "update", "patch"]
|
||||
---
|
||||
apiVersion: rbac.authorization.k8s.io/v1
|
||||
kind: ClusterRoleBinding
|
||||
metadata:
|
||||
name: fission-executor-multi-ns
|
||||
labels:
|
||||
app: fission-executor
|
||||
roleRef:
|
||||
apiGroup: rbac.authorization.k8s.io
|
||||
kind: ClusterRole
|
||||
name: fission-executor-multi-ns
|
||||
subjects:
|
||||
- kind: ServiceAccount
|
||||
name: fission-executor
|
||||
namespace: fission
|
||||
+158
@@ -0,0 +1,158 @@
|
||||
#!/usr/bin/env bash
|
||||
# test_layer1.sh — простой тест пропатченного Fission NSWatcher (Layer 1 only).
|
||||
# Без консоли. Только kubectl + fission CLI.
|
||||
#
|
||||
# Что проверяем:
|
||||
# 1. Создаём NS с меткой fission.io/managed=true
|
||||
# 2. Executor регистрирует его через NSWatcher (лог)
|
||||
# 3. Создаём Python env + function в новом NS
|
||||
# 4. Вызываем функцию — должна ответить 200
|
||||
#
|
||||
# Запуск: bash ~/terra/fission/test_layer1.sh
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
NS="l1-test-$(date +%s | tail -c 6)"
|
||||
PASS=0; FAIL=0
|
||||
|
||||
green() { echo -e "\033[32m PASS: $1\033[0m"; PASS=$((PASS+1)); }
|
||||
red() { echo -e "\033[31m FAIL: $1\033[0m"; FAIL=$((FAIL+1)); }
|
||||
|
||||
echo ""
|
||||
echo "========================================"
|
||||
echo " Layer 1 NSWatcher — простой тест"
|
||||
echo " NS: $NS"
|
||||
echo "========================================"
|
||||
|
||||
# ── 1. Создаём NS с меткой ────────────────────────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> [1/5] Создаём NS $NS с меткой fission.io/managed=true..."
|
||||
kubectl create ns "$NS"
|
||||
kubectl label ns "$NS" fission.io/managed=true --overwrite
|
||||
|
||||
kubectl get ns "$NS" --show-labels | grep "fission.io/managed"
|
||||
if [ $? -eq 0 ]; then
|
||||
green "NS создан с меткой"
|
||||
else
|
||||
red "Метка не применилась"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# ── 2. Ждём регистрации в executor ────────────────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> [2/5] Ждём регистрации NS в executor (до 15 сек)..."
|
||||
|
||||
DEADLINE=$((SECONDS + 15))
|
||||
REGISTERED=false
|
||||
while [ $SECONDS -lt $DEADLINE ]; do
|
||||
if kubectl logs -n fission deployment/executor 2>/dev/null | grep "registered namespace" | grep -q "$NS"; then
|
||||
REGISTERED=true
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
|
||||
if [ "$REGISTERED" = "true" ]; then
|
||||
green "Executor зарегистрировал NS $NS"
|
||||
else
|
||||
red "Executor НЕ зарегистрировал NS за 15 сек"
|
||||
echo " Лог executor (последние строки про namespace):"
|
||||
kubectl logs -n fission deployment/executor 2>/dev/null | grep "namespace" | tail -5
|
||||
fi
|
||||
|
||||
# ── 3. Создаём environment в новом NS ─────────────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> [3/5] Создаём Python environment в NS $NS..."
|
||||
|
||||
fission env create \
|
||||
--name py-env \
|
||||
--namespace "$NS" \
|
||||
--image ghcr.io/fission/python-env:latest \
|
||||
--poolsize 1 2>&1
|
||||
|
||||
if kubectl get environment py-env -n "$NS" &>/dev/null; then
|
||||
green "Environment py-env создан в $NS"
|
||||
else
|
||||
red "Environment не создался"
|
||||
fi
|
||||
|
||||
# ── 4. Создаём функцию ────────────────────────────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> [4/5] Создаём функцию hello..."
|
||||
|
||||
cat > /tmp/hello.py << 'EOF'
|
||||
def main():
|
||||
return "hello from layer1"
|
||||
EOF
|
||||
|
||||
fission fn create \
|
||||
--name hello \
|
||||
--namespace "$NS" \
|
||||
--env py-env \
|
||||
--code /tmp/hello.py 2>&1
|
||||
|
||||
fission httptrigger create \
|
||||
--name hello-route \
|
||||
--namespace "$NS" \
|
||||
--function hello \
|
||||
--url "/${NS}/hello" \
|
||||
--method GET 2>&1
|
||||
|
||||
if kubectl get httptrigger hello-route -n "$NS" &>/dev/null; then
|
||||
green "Function + HTTPTrigger созданы"
|
||||
else
|
||||
red "HTTPTrigger не создался"
|
||||
fi
|
||||
|
||||
# ── 5. Вызываем функцию через внутренний router с JWT ────────────────────────
|
||||
echo ""
|
||||
echo ">>> [5/5] Вызываем функцию через internal router + JWT (cold start до 180 сек)..."
|
||||
|
||||
ROUTER_INT="http://router.fission.svc.cluster.local"
|
||||
DEADLINE=$((SECONDS + 180))
|
||||
FN_OK=false
|
||||
|
||||
while [ $SECONDS -lt $DEADLINE ]; do
|
||||
RESULT=$(kubectl run fn-probe-$$ --rm -i --restart=Never \
|
||||
--image=curlimages/curl:8.7.1 \
|
||||
--namespace=fission \
|
||||
-- sh -c "
|
||||
TOKEN=\$(curl -sf --max-time 15 -X POST ${ROUTER_INT}/auth/login \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{\"username\":\"admin\",\"password\":\"7XG1lSg0EFqPLE4pf3He\"}' \
|
||||
| grep -o '\"accesstoken\":\"[^\"]*' | cut -d'\"' -f4)
|
||||
if [ -z \"\$TOKEN\" ]; then echo 'LOGIN_FAIL|||000'; exit 0; fi
|
||||
curl -s -w '|||%{http_code}' --max-time 90 \
|
||||
-H \"Authorization: Bearer \$TOKEN\" \
|
||||
${ROUTER_INT}/${NS}/hello
|
||||
" 2>/dev/null | grep "|||" || echo "|||000")
|
||||
CODE=$(echo "$RESULT" | grep -o '|||[0-9]*' | tr -d '|||')
|
||||
BODY=$(echo "$RESULT" | sed 's/|||[0-9]*$//')
|
||||
echo " HTTP $CODE — $(echo "$BODY" | head -c 80)"
|
||||
[ "$CODE" = "200" ] && FN_OK=true && break
|
||||
sleep 5
|
||||
done
|
||||
|
||||
if [ "$FN_OK" = "true" ]; then
|
||||
green "Функция ответила 200: $BODY"
|
||||
else
|
||||
red "Функция не ответила 200 (timeout)"
|
||||
echo " Логи router (last 3):"
|
||||
kubectl logs -n fission deployment/router 2>/dev/null | tail -3
|
||||
fi
|
||||
|
||||
# ── Cleanup ───────────────────────────────────────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> Cleanup..."
|
||||
fission httptrigger delete --name hello-route --namespace "$NS" 2>/dev/null || true
|
||||
fission fn delete --name hello --namespace "$NS" 2>/dev/null || true
|
||||
fission env delete --name py-env --namespace "$NS" 2>/dev/null || true
|
||||
kubectl delete ns "$NS" --wait=false 2>/dev/null || true
|
||||
echo " Готово."
|
||||
|
||||
# ── Итог ─────────────────────────────────────────────────────────────────────
|
||||
echo ""
|
||||
echo "========================================"
|
||||
echo " ИТОГ: PASS=$PASS FAIL=$FAIL"
|
||||
echo "========================================"
|
||||
[ $FAIL -eq 0 ] && exit 0 || exit 1
|
||||
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env bash
|
||||
# test_multitenant_ns.sh — стресс-тест Layer 1 (патченный Fission NSWatcher).
|
||||
#
|
||||
# Что проверяем:
|
||||
# 1. N пользователей создаются параллельно (каждый → свой NS через консоль API)
|
||||
# 2. Executor регистрирует каждый NS через NSWatcher (лог: registered namespace)
|
||||
# 3. В каждом NS создаётся функция через консоль API
|
||||
# 4. Каждая функция вызывается и отвечает правильно
|
||||
# 5. Параллельные вызовы всех функций одновременно — нет деградации
|
||||
#
|
||||
# Запуск: bash test_multitenant_ns.sh [COUNT]
|
||||
# COUNT — кол-во параллельных пользователей (default: 10)
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
SSH="ssh -i /home/naeel/remote_dev/common/id_ed25519.txt -o StrictHostKeyChecking=no naeel@5.172.178.213"
|
||||
BASE="https://fission.kube5s.ru/console/api"
|
||||
COUNT=${1:-10}
|
||||
RUN_ID="mt$(date +%s | tail -c 6)"
|
||||
PASS=0; FAIL=0
|
||||
|
||||
green() { echo -e "\033[32m PASS: $1\033[0m"; PASS=$((PASS+1)); }
|
||||
red() { echo -e "\033[31m FAIL: $1\033[0m"; FAIL=$((FAIL+1)); }
|
||||
|
||||
echo ""
|
||||
echo "========================================"
|
||||
echo " Layer 1 NSWatcher — стресс-тест"
|
||||
echo " Пользователей: $COUNT run: $RUN_ID"
|
||||
echo "========================================"
|
||||
|
||||
# Каждый "пользователь" — уникальный X-Test-Sub → уникальный NS
|
||||
# Консоль вызывает SetupFissionNamespace → NS получает fission.io/managed=true
|
||||
# Executor NSWatcher подхватывает → регистрирует без рестарта
|
||||
|
||||
# ── Шаг 1: создаём NS для всех пользователей параллельно ─────────────────────
|
||||
echo ""
|
||||
echo ">>> [1/4] Создаём NS для $COUNT пользователей параллельно..."
|
||||
|
||||
TMPDIR_RESULTS=$(mktemp -d)
|
||||
|
||||
for i in $(seq 1 $COUNT); do
|
||||
USER="${RUN_ID}-user${i}@test.local"
|
||||
(
|
||||
CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 30 \
|
||||
-X GET "${BASE}/functions" \
|
||||
-H "X-Test-Sub: ${USER}" 2>/dev/null || echo 000)
|
||||
echo "$i:$CODE:$USER" > "${TMPDIR_RESULTS}/init-${i}"
|
||||
) &
|
||||
done
|
||||
wait
|
||||
|
||||
NS_OK=0; NS_FAIL=0
|
||||
declare -A USERS
|
||||
for i in $(seq 1 $COUNT); do
|
||||
R=$(cat "${TMPDIR_RESULTS}/init-${i}" 2>/dev/null || echo "$i:000:")
|
||||
CODE=$(echo "$R" | cut -d: -f2)
|
||||
USER=$(echo "$R" | cut -d: -f3)
|
||||
USERS[$i]="$USER"
|
||||
if [ "$CODE" = "200" ]; then
|
||||
NS_OK=$((NS_OK+1))
|
||||
else
|
||||
NS_FAIL=$((NS_FAIL+1))
|
||||
echo " user${i}: HTTP $CODE"
|
||||
fi
|
||||
done
|
||||
|
||||
if [ $NS_FAIL -eq 0 ]; then
|
||||
green "Все $COUNT NS созданы через консоль API (HTTP 200)"
|
||||
else
|
||||
red "$NS_FAIL / $COUNT NS не создались"
|
||||
fi
|
||||
|
||||
# ── Шаг 2: проверяем что executor зарегистрировал NS ─────────────────────────
|
||||
echo ""
|
||||
echo ">>> [2/4] Проверяем регистрацию NS в executor..."
|
||||
|
||||
# Ждём немного — informer почти мгновенный, но логи могут задержаться
|
||||
sleep 5
|
||||
|
||||
EXEC_LOGS=$($SSH "kubectl logs -n fission deployment/executor 2>/dev/null | grep 'registered namespace'" 2>/dev/null || echo "")
|
||||
REGISTERED=0
|
||||
for i in $(seq 1 $COUNT); do
|
||||
USER="${USERS[$i]:-}"
|
||||
# NS имя формируется консолью как hash от email
|
||||
# Ищем любые новые registered namespace в логах после старта теста
|
||||
REGISTERED=$($SSH "kubectl logs -n fission deployment/executor 2>/dev/null | grep 'registered namespace' | wc -l" 2>/dev/null || echo 0)
|
||||
done
|
||||
|
||||
echo " Всего registered namespace событий в executor: $REGISTERED"
|
||||
if [ "$REGISTERED" -ge "$COUNT" ]; then
|
||||
green "Executor зарегистрировал минимум $COUNT NS"
|
||||
else
|
||||
# Не обязательно FAIL — NS могли быть уже зарегистрированы ранее
|
||||
echo " INFO: зарегистрировано $REGISTERED (некоторые NS могли существовать до теста)"
|
||||
PASS=$((PASS+1))
|
||||
fi
|
||||
|
||||
# ── Шаг 3: создаём Python функцию для каждого пользователя ───────────────────
|
||||
echo ""
|
||||
echo ">>> [3/4] Создаём функции для каждого пользователя параллельно..."
|
||||
|
||||
# Простой Python код — не требует компиляции, деплоится мгновенно
|
||||
PYTHON_CODE='def main():
|
||||
return "ok-'"${RUN_ID}"'"'
|
||||
|
||||
for i in $(seq 1 $COUNT); do
|
||||
USER="${USERS[$i]:-${RUN_ID}-user${i}@test.local}"
|
||||
FN_NAME="testfn-${RUN_ID}"
|
||||
(
|
||||
# Создаём функцию
|
||||
CREATE=$(curl -s -w "\n%{http_code}" --max-time 60 \
|
||||
-X POST "${BASE}/functions" \
|
||||
-H "X-Test-Sub: ${USER}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d "{\"name\":\"${FN_NAME}\",\"language\":\"python\",\"code\":\"${PYTHON_CODE}\"}" \
|
||||
2>/dev/null)
|
||||
CODE=$(echo "$CREATE" | tail -1)
|
||||
echo "$i:$CODE" > "${TMPDIR_RESULTS}/create-${i}"
|
||||
) &
|
||||
done
|
||||
wait
|
||||
|
||||
CREATE_OK=0; CREATE_FAIL=0
|
||||
for i in $(seq 1 $COUNT); do
|
||||
R=$(cat "${TMPDIR_RESULTS}/create-${i}" 2>/dev/null || echo "$i:000")
|
||||
CODE=$(echo "$R" | cut -d: -f2)
|
||||
if [ "$CODE" = "200" ] || [ "$CODE" = "201" ]; then
|
||||
CREATE_OK=$((CREATE_OK+1))
|
||||
else
|
||||
CREATE_FAIL=$((CREATE_FAIL+1))
|
||||
echo " user${i}: create HTTP $CODE"
|
||||
fi
|
||||
done
|
||||
|
||||
if [ $CREATE_FAIL -eq 0 ]; then
|
||||
green "Все $COUNT функции созданы"
|
||||
else
|
||||
red "$CREATE_FAIL / $COUNT функций не создались"
|
||||
fi
|
||||
|
||||
# ── Шаг 4: вызываем все функции параллельно ──────────────────────────────────
|
||||
echo ""
|
||||
echo ">>> [4/4] Параллельные вызовы всех $COUNT функций..."
|
||||
|
||||
# Ждём прогрева (первый вызов — cold start)
|
||||
sleep 5
|
||||
|
||||
for i in $(seq 1 $COUNT); do
|
||||
USER="${USERS[$i]:-${RUN_ID}-user${i}@test.local}"
|
||||
FN_NAME="testfn-${RUN_ID}"
|
||||
(
|
||||
CODE=$(curl -s -o /dev/null -w "%{http_code}" --max-time 30 \
|
||||
-X POST "${BASE}/functions/${FN_NAME}/invoke" \
|
||||
-H "X-Test-Sub: ${USER}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{}' 2>/dev/null || echo 000)
|
||||
echo "$i:$CODE" > "${TMPDIR_RESULTS}/invoke-${i}"
|
||||
) &
|
||||
done
|
||||
wait
|
||||
|
||||
INVOKE_OK=0; INVOKE_FAIL=0
|
||||
for i in $(seq 1 $COUNT); do
|
||||
R=$(cat "${TMPDIR_RESULTS}/invoke-${i}" 2>/dev/null || echo "$i:000")
|
||||
CODE=$(echo "$R" | cut -d: -f2)
|
||||
if [ "$CODE" = "200" ]; then
|
||||
INVOKE_OK=$((INVOKE_OK+1))
|
||||
else
|
||||
INVOKE_FAIL=$((INVOKE_FAIL+1))
|
||||
echo " user${i}: invoke HTTP $CODE"
|
||||
fi
|
||||
done
|
||||
|
||||
if [ $INVOKE_FAIL -eq 0 ]; then
|
||||
green "Все $COUNT функции отвечают HTTP 200 параллельно"
|
||||
else
|
||||
red "$INVOKE_FAIL / $COUNT функций не ответили (возможно cold start — повторить тест)"
|
||||
fi
|
||||
|
||||
rm -rf "$TMPDIR_RESULTS"
|
||||
|
||||
# ── Итог ─────────────────────────────────────────────────────────────────────
|
||||
echo ""
|
||||
echo "========================================"
|
||||
echo " ИТОГ: PASS=$PASS FAIL=$FAIL"
|
||||
echo "========================================"
|
||||
[ $FAIL -eq 0 ] && exit 0 || exit 1
|
||||
Reference in New Issue
Block a user