layer1: add namespace snapshot api step 1
This commit is contained in:
@@ -0,0 +1,44 @@
|
|||||||
|
# 2026-04-26 — NamespaceManager rewrite, step 1
|
||||||
|
|
||||||
|
## Цель шага
|
||||||
|
|
||||||
|
Начать bounded rewrite Layer 1 без большого взрыва по коду.
|
||||||
|
Первый шаг deliberately узкий:
|
||||||
|
|
||||||
|
- не менять lifecycle namespace onboarding;
|
||||||
|
- не трогать watcher-ы executor/router/buildermgr;
|
||||||
|
- не менять контракты `AddNamespace`;
|
||||||
|
- убрать первые прямые проходы по общей mutable map `FissionResourceNS`.
|
||||||
|
|
||||||
|
## Почему именно так
|
||||||
|
|
||||||
|
Сейчас multi-tenant логика уже динамическая, но многие старые code path все еще читают
|
||||||
|
`DefaultNSResolver().FissionResourceNS` напрямую. Это опасно по двум причинам:
|
||||||
|
|
||||||
|
1. map общая и mutable, а dynamic onboarding меняет ее во время работы процесса;
|
||||||
|
2. часть helper-ов и startup path продолжают жить как будто список namespace-ов immutable.
|
||||||
|
|
||||||
|
Полный rewrite в один шаг дал бы слишком большой blast radius. Поэтому сначала вводится
|
||||||
|
thread-safe snapshot API в namespace layer, а затем существующие потребители переводятся
|
||||||
|
на него по одному.
|
||||||
|
|
||||||
|
## План шага 1
|
||||||
|
|
||||||
|
1. Добавить в `pkg/utils/namespace.go` методы snapshot для plain namespaces и namespaces with options.
|
||||||
|
2. Перевести `pkg/utils/informer.go` на snapshot API.
|
||||||
|
3. Перевести startup factory path в `pkg/executor/executor.go` на snapshot API.
|
||||||
|
4. Добавить unit tests для snapshot behavior.
|
||||||
|
5. Прогнать `go test ./pkg/utils/... ./pkg/executor/...`.
|
||||||
|
|
||||||
|
## Ожидаемый эффект
|
||||||
|
|
||||||
|
- меньше прямых чтений общей map;
|
||||||
|
- появление базового API, через который дальше можно выносить единый NamespaceManager;
|
||||||
|
- нулевое изменение внешнего поведения на этом шаге.
|
||||||
|
|
||||||
|
## Что НЕ делаем на этом шаге
|
||||||
|
|
||||||
|
- не исправляем watcher lifecycle;
|
||||||
|
- не добавляем remove/delete semantics;
|
||||||
|
- не трогаем router race и buildermgr dedup bug;
|
||||||
|
- не меняем RBAC.
|
||||||
@@ -294,7 +294,7 @@ func StartExecutor(ctx context.Context, clientGen crd.ClientGeneratorInterface,
|
|||||||
logger.Info("Starting executor", zap.String("instanceID", executorInstanceID))
|
logger.Info("Starting executor", zap.String("instanceID", executorInstanceID))
|
||||||
|
|
||||||
finformerFactory := make(map[string]genInformer.SharedInformerFactory, 0)
|
finformerFactory := make(map[string]genInformer.SharedInformerFactory, 0)
|
||||||
for _, ns := range utils.DefaultNSResolver().FissionResourceNS {
|
for _, ns := range utils.DefaultNSResolver().Snapshot() {
|
||||||
finformerFactory[ns] = genInformer.NewFilteredSharedInformerFactory(fissionClient, time.Minute*30, ns, nil)
|
finformerFactory[ns] = genInformer.NewFilteredSharedInformerFactory(fissionClient, time.Minute*30, ns, nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -23,7 +23,7 @@ import (
|
|||||||
|
|
||||||
func GetInformersForNamespaces(client versioned.Interface, defaultSync time.Duration, kind string) map[string]cache.SharedIndexInformer {
|
func GetInformersForNamespaces(client versioned.Interface, defaultSync time.Duration, kind string) map[string]cache.SharedIndexInformer {
|
||||||
informers := make(map[string]cache.SharedIndexInformer)
|
informers := make(map[string]cache.SharedIndexInformer)
|
||||||
for _, ns := range DefaultNSResolver().FissionResourceNS {
|
for _, ns := range DefaultNSResolver().Snapshot() {
|
||||||
factory := genInformer.NewFilteredSharedInformerFactory(client, defaultSync, ns, nil).Core().V1()
|
factory := genInformer.NewFilteredSharedInformerFactory(client, defaultSync, ns, nil).Core().V1()
|
||||||
switch kind {
|
switch kind {
|
||||||
case fv1.CanaryConfigResource:
|
case fv1.CanaryConfigResource:
|
||||||
@@ -52,7 +52,7 @@ func GetInformersForNamespaces(client versioned.Interface, defaultSync time.Dura
|
|||||||
func GetK8sInformersForNamespaces(client kubernetes.Interface, defaultSync time.Duration, kind string) map[string]cache.SharedIndexInformer {
|
func GetK8sInformersForNamespaces(client kubernetes.Interface, defaultSync time.Duration, kind string) map[string]cache.SharedIndexInformer {
|
||||||
informers := make(map[string]cache.SharedIndexInformer)
|
informers := make(map[string]cache.SharedIndexInformer)
|
||||||
namespaces := DefaultNSResolver()
|
namespaces := DefaultNSResolver()
|
||||||
for _, ns := range namespaces.FissionNSWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
for _, ns := range namespaces.SnapshotWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
||||||
factory := k8sInformers.NewSharedInformerFactoryWithOptions(client, defaultSync, k8sInformers.WithNamespace(ns))
|
factory := k8sInformers.NewSharedInformerFactoryWithOptions(client, defaultSync, k8sInformers.WithNamespace(ns))
|
||||||
switch kind {
|
switch kind {
|
||||||
case fv1.Deployments:
|
case fv1.Deployments:
|
||||||
@@ -77,7 +77,7 @@ func GetK8sInformersForNamespaces(client kubernetes.Interface, defaultSync time.
|
|||||||
func GetInformerEventChecker(ctx context.Context, client kubernetes.Interface, reason string) map[string]cache.SharedInformer {
|
func GetInformerEventChecker(ctx context.Context, client kubernetes.Interface, reason string) map[string]cache.SharedInformer {
|
||||||
informers := make(map[string]cache.SharedInformer)
|
informers := make(map[string]cache.SharedInformer)
|
||||||
namespaces := DefaultNSResolver()
|
namespaces := DefaultNSResolver()
|
||||||
for _, ns := range namespaces.FissionNSWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
for _, ns := range namespaces.SnapshotWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
||||||
informers[ns] = cache.NewSharedInformer(
|
informers[ns] = cache.NewSharedInformer(
|
||||||
&cache.ListWatch{
|
&cache.ListWatch{
|
||||||
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
||||||
@@ -100,7 +100,7 @@ func GetInformerFactoryByExecutor(client kubernetes.Interface, labels labels.Sel
|
|||||||
informerFactory := make(map[string]k8sInformers.SharedInformerFactory)
|
informerFactory := make(map[string]k8sInformers.SharedInformerFactory)
|
||||||
|
|
||||||
namespaces := DefaultNSResolver()
|
namespaces := DefaultNSResolver()
|
||||||
for _, ns := range namespaces.FissionNSWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
for _, ns := range namespaces.SnapshotWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) {
|
||||||
factory := k8sInformers.NewSharedInformerFactoryWithOptions(client, defaultResync,
|
factory := k8sInformers.NewSharedInformerFactoryWithOptions(client, defaultResync,
|
||||||
k8sInformers.WithTweakListOptions(func(options *metav1.ListOptions) {
|
k8sInformers.WithTweakListOptions(func(options *metav1.ListOptions) {
|
||||||
options.LabelSelector = labels.String()
|
options.LabelSelector = labels.String()
|
||||||
|
|||||||
+26
-2
@@ -2,6 +2,7 @@ package utils
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"os"
|
"os"
|
||||||
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
@@ -96,10 +97,23 @@ func (nsr *NamespaceResolver) AddNamespace(ns string) bool {
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
nsr.FissionResourceNS[ns] = ns
|
nsr.FissionResourceNS[ns] = ns
|
||||||
nsr.Logger.Info("dynamically added namespace to resolver", zap.String("namespace", ns))
|
if nsr.Logger != nil {
|
||||||
|
nsr.Logger.Info("dynamically added namespace to resolver", zap.String("namespace", ns))
|
||||||
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Snapshot returns a stable copy of the currently registered resource namespaces.
|
||||||
|
// The returned slice is detached from the internal mutable map and safe to iterate.
|
||||||
|
func (nsr *NamespaceResolver) Snapshot() []string {
|
||||||
|
nsr.mu.RLock()
|
||||||
|
defer nsr.mu.RUnlock()
|
||||||
|
|
||||||
|
namespaces := listNamespaces(nsr.FissionResourceNS)
|
||||||
|
sort.Strings(namespaces)
|
||||||
|
return namespaces
|
||||||
|
}
|
||||||
|
|
||||||
func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]string {
|
func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]string {
|
||||||
var options options
|
var options options
|
||||||
for _, opt := range option {
|
for _, opt := range option {
|
||||||
@@ -122,10 +136,20 @@ func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]
|
|||||||
if options.defaultNs && nsr.DefaultNamespace != "" {
|
if options.defaultNs && nsr.DefaultNamespace != "" {
|
||||||
fissionResourceNS[nsr.DefaultNamespace] = nsr.DefaultNamespace
|
fissionResourceNS[nsr.DefaultNamespace] = nsr.DefaultNamespace
|
||||||
}
|
}
|
||||||
nsr.Logger.Debug("fission resource namespaces", zap.Any("namespaces", listNamespaces(fissionResourceNS)))
|
if nsr.Logger != nil {
|
||||||
|
nsr.Logger.Debug("fission resource namespaces", zap.Any("namespaces", listNamespaces(fissionResourceNS)))
|
||||||
|
}
|
||||||
return fissionResourceNS
|
return fissionResourceNS
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SnapshotWithOptions returns a stable slice copy of Fission namespaces after applying
|
||||||
|
// optional builder/function/default namespace expansion.
|
||||||
|
func (nsr *NamespaceResolver) SnapshotWithOptions(option ...option) []string {
|
||||||
|
namespaces := listNamespaces(nsr.FissionNSWithOptions(option...))
|
||||||
|
sort.Strings(namespaces)
|
||||||
|
return namespaces
|
||||||
|
}
|
||||||
|
|
||||||
func GetNamespaces() map[string]string {
|
func GetNamespaces() map[string]string {
|
||||||
namespaces := make(map[string]string)
|
namespaces := make(map[string]string)
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package utils
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"os"
|
"os"
|
||||||
|
"reflect"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -177,6 +178,44 @@ func TestNamespaceResolver(t *testing.T) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
t.Run("Snapshot returns stable copy", func(t *testing.T) {
|
||||||
|
nsr := &NamespaceResolver{
|
||||||
|
FissionResourceNS: map[string]string{
|
||||||
|
"ns-b": "ns-b",
|
||||||
|
"ns-a": "ns-a",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshot := nsr.Snapshot()
|
||||||
|
expected := []string{"ns-a", "ns-b"}
|
||||||
|
if !reflect.DeepEqual(expected, snapshot) {
|
||||||
|
t.Fatalf("expected snapshot %v, got %v", expected, snapshot)
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshot[0] = "mutated"
|
||||||
|
if nsr.FissionResourceNS["ns-a"] != "ns-a" {
|
||||||
|
t.Fatalf("snapshot mutated internal namespace map")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("SnapshotWithOptions includes expanded namespaces once", func(t *testing.T) {
|
||||||
|
nsr := &NamespaceResolver{
|
||||||
|
FunctionNamespace: "fn-ns",
|
||||||
|
BuilderNamespace: "builder-ns",
|
||||||
|
DefaultNamespace: "default",
|
||||||
|
FissionResourceNS: map[string]string{
|
||||||
|
"default": "default",
|
||||||
|
"user-ns": "user-ns",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshot := nsr.SnapshotWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs())
|
||||||
|
expected := []string{"builder-ns", "default", "fn-ns", "user-ns"}
|
||||||
|
if !reflect.DeepEqual(expected, snapshot) {
|
||||||
|
t.Fatalf("expected snapshot with options %v, got %v", expected, snapshot)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
t.Run("getNamespace", func(t *testing.T) {
|
t.Run("getNamespace", func(t *testing.T) {
|
||||||
for _, test := range []struct {
|
for _, test := range []struct {
|
||||||
name string
|
name string
|
||||||
|
|||||||
Reference in New Issue
Block a user