02. krt：Istio 自己造了一个控制器运行时 原创

上一篇把 istiod 装配起来了。装配阶段往 Environment 里塞了一大堆控制器，其中最新的一批不是用 client-go 写的，而是挂在 pkg/kube/krt 上——一个 Istio 自己造的控制器运行时。

这一篇回答一个问题：一个 virtualservice 变了，最深处那个派生对象是怎么被重算的，以及为什么只重算了它一个。

读完你能做到两件事：看到任何 krt.NewCollection(...) 能立刻说出它的输入是谁、什么时候重算、重算会不会往外发事件；用一条 curl 把整张依赖图 dump 出来，对着它查"我这个配置到底卡在哪一级"。

源码基线 istio/istio 1.31.4，定位用文件路径 + 函数名，不用行号。

阅读提示：这一篇是全课程里唯一一篇框架课。第 05~09、22、26 篇的控制器代码都长在它上面，所以宁可慢一点。

一、先看一段真实代码

不看框架，先看一段你在 pilot/pkg/config/kube/gateway/controller.go 里能直接找到的代码。Gateway API 的 Gateway 最终要变成 Istio 自己的 config.Config：

GatewayConfigs := krt.NewCollection(Gateways, func(ctx krt.HandlerContext, i Gateway) *config.Config {
	if i.Valid {
		return i.Config
	}
	return nil
}, opts.WithName("GatewayConfigs")...)

三行，干完了一整件事：Gateways 这个集合里任何一个元素变了，就用它重新跑一遍这个函数，把结果放进 GatewayConfigs 集合。

没有 AddEventHandler，没有 workqueue，没有 Reconcile(key)，没有 Get(key) 回读——这个函数拿到的就是完整对象本身。这就是 krt 想解决的问题：把"事件 → 入队 → 取数 → 计算 → 写回"这条每次都手写的流水线，压缩成一个纯函数。

对比一下 krt 自己仓库里那个基准测试的写法（pkg/kube/krt/bench_test.go），同一件事的两种实现：

// krt 版：声明式，函数只关心自己的输入
Workloads := krt.NewCollection(Pods, func(ctx krt.HandlerContext, p *v1.Pod) *Workload {
	if p.Status.PodIP == "" {
		return nil
	}
	services := ServicesByNamespace.Fetch(ctx, p.Namespace, krt.FilterSelectsNonEmpty(p.GetLabels()))
	return &Workload{Named: krt.NewNamed(p), IP: p.Status.PodIP, ServiceNames: /* ... */}
})

// legacy 版：命令式，每个细节都要自己接
func (l *legacy) Reconcile(key types.NamespacedName) error {
	pod := l.pods.Get(key.Name, key.Namespace)
	// ... 自己判断 nil、自己算、自己 diff、自己发事件
}

c.pods.AddEventHandler(controllers.ObjectHandler(c.queue.AddObject))
c.services.AddEventHandler(controllers.FromEventHandler(func(e controllers.Event) {
	// Service 变更时，还要自己找出哪些 Pod 受影响，挨个塞进队列
	o := e.Latest().(*v1.Service)
	for _, pod := range c.pods.List(o.GetNamespace(), klabels.SelectorFromValidatedSet(o.Spec.Selector)) {
		c.queue.AddObject(pod)
	}
}))
go c.queue.Run(stop)

legacy 版最后那几行，就是 krt 想替你干掉的东西。"一个 Service 变了，我要把哪些 Pod 重算一遍"——这件事在 client-go 里是每个控制器作者各自回答一次的送命题，答漏就是脏数据，答多就是白干活。

二、五个原语，一张图

krt 的 API 面很窄。pkg/kube/krt 里真正要记的东西：

原语	签名形状	干什么
Collection[T]	GetKey / List / Register / RegisterBatch / Metadata	一切的基础：一堆有 key 的对象 + 一个事件流
NewInformer[I]	从 kube.Client 拉	把 K8s 的 informer 包成 Collection（外部世界的入口）
NewSingleton[O]	func(ctx) *O	零对一：全局单值，比如 mesh 配置
NewCollection[I,O]	func(ctx, I) *O	一对零/一：最常见的那一档
NewManyCollection[I,O]	func(ctx, I) []O	一对多：一个 Service 出 N 个 Endpoint
Fetch[T](ctx, c, filters...)	任意变换函数里调用	声明二级依赖：我除了主输入，还依赖这个集合

Collection 的接口只有这几行（core.go）：

type Collection[T any] interface {
	GetKey(k string) *T
	List() []T
	EventStream[T]
	Metadata() Metadata
}

EventStream[T] 再补上 Register / RegisterBatch / HasSynced / WaitUntilSynced。就这些。 Singleton[T] 不过是个"永远只有一个元素"的 Collection。

flowchart LR
  INF["NewInformer<br/>K8s 对象"] --> C1["Collection I<br/>主输入"]
  ST["NewStatic<br/>静态/文件/触发器"] --> C1
  C1 --> NC["NewCollection<br/>一对零或一"]
  C1 --> NM["NewManyCollection<br/>一对多"]
  C1 --> NS["NewSingleton<br/>零对一"]
  C2["Collection J<br/>二级来源"] -.->|"Fetch 声明依赖"| NC
  NC --> OUT["Collection O<br/>派生结果"]
  classDef hot fill:#fff3bf,stroke:#f08c00,color:#663c00
  class NC hot

图里那条虚线是整篇的题眼。实线是"主依赖"，它决定基数（一对一、一对多、零对一），也就是"这个变换函数为谁跑"。虚线是"二级依赖"，由 Fetch 在函数体里当场声明——它不决定基数，只决定"什么时候也要重算一遍"。

只有 Fetch 一条路能往外看。 这条限制是硬性的，README 里单独列了一节 Transformation constraints：

• Any querying of other Collections must be done through krt.Fetch.
• Querying other data stores that may change is not permitted.
• Transformations may be called at any time, including many times for the same inputs.

翻译过来：变换函数必须是纯的、无状态的、可以随时被重跑的。 违反不会 panic，会得到"陈旧数据"——最难受的那类故障。

三、核心：一次变更，只重算该重算的那几个

这一节是整篇的重点。挑一个最小但真实的场景：1000 个 Pod、50 个 Service，改一个 Service 的 selector。

flowchart TD
  P["informer/Pods<br/>1000 个"] --> W["Workloads<br/>Pod + 它属于哪些 Service"]
  S["informer/Services<br/>50 个"] -.->|"Fetch + FilterSelectsNonEmpty"| W
  W --> H["下游 handler"]
  classDef hot fill:#fff3bf,stroke:#f08c00,color:#663c00
  class S hot

朴素做法是"Service 变了，把 1000 个 Pod 全部重算"。krt 没这么干。它走的是 collection.go 里的三段：记录 → 反查 → 只跑命中的。

3.1 第一段：每次变换跑完，把依赖记账

Fetch() 不是简简单单 return c.List()。它做的第一件事是报到（fetch.go）：

func fetch[T any](ctx HandlerContext, cc Collection[T], allowMissingContext bool, opts ...FetchOption) []T {
	c := cc.(internalCollection[T])
	d := &dependency{
		id:             c.uid(),
		collectionName: c.name(),
		filter:         &filter{},
	}
	for _, o := range opts {
		o(d)   // 把 FilterKey / FilterIndex / FilterLabel ... 记进 d.filter
	}
	if ctx != nil {
		h := ctx.(registerDependency)
		// Important: register before we List(), so we cannot miss any events
		h.registerDependency(d, c, func(f erasedEventHandler) HandlerRegistration {
			ff := func(o []Event[T]) { f(slices.Map(o, castEvent[T, any])) }
			return c.RegisterBatch(ff, false)
		})
	}
	// ... 然后才是真正的 List / GetKey / 索引查找 + 过滤
}

两个细节值得停下来说：

第一，registerDependency 在 List() 之前。 注释写得很直白：register before we List(), so we cannot miss any events。顺序反了，就存在一个"我已经读了旧列表、但新事件还没订阅上"的窗口，那条事件永久丢失，派生对象永久陈旧——而且没有任何报错。

第二，registerDependency 里的 ctx 是每次变换调用新建的。 看一下 handleChangedPrimaryInputEvents 里的写法（collection.go）：

for idx, a := range items {
	if a.Event == controllers.EventDelete {
		continue  // 删除不用跑变换，直接清
	}
	i := a.Latest()
	iKey := getTypedKey(i)

	ctx := &collectionDependencyTracker[I, O]{manyCollection: h, key: iKey}
	results := slices.GroupUnique(h.transformation(ctx, i), getTypedKey[O])
	recomputedResults[idx] = results
	pendingDepStateUpdates[iKey] = ctx   // 这次变换攒下来的依赖，稍后整体写回
}

每个输入对象 I，一个 collectionDependencyTracker。 变换函数里每次 Fetch 都往这个 tracker 的 d []*dependency 里追加一条。函数跑完，这份清单就是"I 依赖谁"的完整答案。

这里有个很容易忽略的设计：依赖是每次重算都重新生成的，不是一次性注册的。因为一个 Pod 的 labels 变了，它匹配的 Service 集合也会变——依赖本身是数据驱动的。

3.2 第二段：给依赖建反查索引

dependencyState.update() 把 I → []dependency 存下来，同时顺便建反查索引：

func (i dependencyState[I]) update(key Key[I], deps []*dependency) {
	i.delete(key)              // 先撤掉这个 I 的旧反查条目
	i.objectDependencies[key] = deps
	for _, d := range deps {
		if depKeys, typ, extractor, filterID, ok := d.filter.reverseIndexKey(); ok {
			for _, depKey := range depKeys {
				k := indexedDependency{id: d.id, key: depKey, typ: typ}
				sets.InsertOrNew(i.indexedDependencies, k, key)   // 反查表：依赖项 → 哪些 I
			}
			// ...
		}
	}
}

reverseIndexKey() 的返回值决定这条依赖值不值得建索引（filter.go）：

func (f *filter) reverseIndexKey() ([]string, indexedDependencyType, objectKeyExtractor, collectionUID, bool) {
	if f.keys.Len() > 0 {                    // FilterKey / FilterKeys / FilterObjectName
		return f.keys.List(), getKeyType, getKeyExtractor, 0, true
	}
	if f.index != nil {                      // FilterIndex
		return []string{f.index.key}, indexType, f.index.extractKeys, f.index.filterUID, true
	}
	return nil, unknownIndexType, nil, 0, false   // 其余（label / selects / generic）不建
}

这张表就是答案：

你写的 Fetch 过滤条件	能不能精确定位受影响的 I	代价
FilterKey / FilterObjectName / FilterIndex	能，反查表直接给出候选	建索引，O(1) 反查
FilterLabel / FilterSelects / FilterSelectsNonEmpty / FilterGeneric	不能，退回全量扫描	每次都要遍历该集合的全部 I 重算一遍过滤

这就是 krt 性能的全部秘密，也是你能主动优化的那个旋钮。 同一次变更，用 FilterKey 和用 FilterLabel，工作量差三个数量级。

3.3 第三段：变更来了，先问"关我什么事"

主输入变了走 onPrimaryInputEvent，二级依赖变了走 onSecondaryDependencyEvent，后者第一步就是找受影响的主输入（collection.go）：

func (i dependencyState[I]) changedInputKeys(sourceCollection collectionUID, events []Event[any]) sets.Set[Key[I]] {
	changedInputKeys := sets.Set[Key[I]]{}

	// 先看这个 sourceCollection 有没有反查索引
	extractorKeys := []extractorKey{}
	for k := range i.indexedDependenciesExtractor {
		if k.typ == doNotIndex && k.uid == sourceCollection {
			extractorKeys = nil   // 有"无法索引"的依赖混在里面 → 直接放弃索引
			break
		}
		if k.typ != unknownIndexType && k.uid == sourceCollection {
			extractorKeys = append(extractorKeys, k)
		}
	}
	for _, ev := range events {
		foundAny := false
		for _, ekey := range extractorKeys {
			if extractor, f := i.indexedDependenciesExtractor[ekey]; f {
				foundAny = true
				for _, item := range ev.Items() {          // 用提取器算出候选 key
					for _, key := range extractor(item) {
						for iKey := range i.indexedDependencies[{id: sourceCollection, key: key, typ: ekey.typ}] {
							if changed := objectChanged(i.objectDependencies[iKey], sourceCollection, ev, true); changed {
								changedInputKeys.Insert(iKey)
							}
						}
					}
				}
			}
		}
		if !foundAny {
			for iKey, dependencies := range i.objectDependencies {   // 兜底：全扫
				if changed := objectChanged(dependencies, sourceCollection, ev, false); changed {
					changedInputKeys.Insert(iKey)
				}
			}
		}
	}
	return changedInputKeys
}

注意那个 doNotIndex 的短路：只要你对同一个集合既用了可索引的过滤器、又用了不可索引的过滤器，索引就整个作废。 一个 FilterGeneric 就能毁掉一整个集合的反查优化。

再看 objectChanged() 里三个连续判断，顺序本身就是设计：

for _, dep := range dependencies {
	if dep.id != sourceCollection { continue }        // 不是这条依赖的来源，跳过
	if dep.filter.SuppressChange(ev) { continue }     // 显式声明"这块变了也不算变"，跳过
	for _, item := range ev.Items() {
		if dep.filter.Matches(item, preFiltered) { return true }   // 命中就走
	}
}

ev.Items() 同时返回旧值和新值——原来匹配、现在不匹配（或者反过来）都算变化。少算这一半，就会出现"label 改了但副产物没更新"的经典 bug。

sequenceDiagram
  participant SRC as Services
  participant COL as Workloads
  participant IDX as 反查索引
  SRC->>COL: Service 变更
  Note over COL: 走 onSecondaryDependencyEvent
  COL->>IDX: changedInputKeys
  IDX-->>COL: 命中的 Pod key（可能只有 2 个）
  COL->>COL: 对这 2 个 Pod 重跑变换
  COL->>COL: Equal 比对，相同就不发事件
  COL->>COL: 分发 Workload 事件

3.4 第四段：Equals 决定"要不要往外喊"

重算完，结果不一定变。handleChangedPrimaryInputEvents 里逐 key 比：

newRes, newExists := results[key]
oldRes, oldExists := h.collectionState.outputs[key]
if newExists && oldExists {
	if Equal(newRes, oldRes) {
		continue      // NOP change, skip
	}
	e.Event = controllers.EventUpdate
}

Equal 的判定顺序在 internal.go：

func Equal[O any](a, b O) bool {
	if ak, ok := any(a).(Equaler[O]); ok { return ak.Equals(b) }     // ① 自定义 Equaler（最优先）
	if ak, ok := any(a).(Equaler[*O]); ok { return ak.Equals(&b) }
	if pk, ok := any(&a).(Equaler[O]); ok { return pk.Equals(b) }
	if pk, ok := any(&a).(Equaler[*O]); ok { return pk.Equals(&b) }

	ap, ok := any(a).(proto.Message)
	if ok {
		if reflect.TypeOf(ap.ProtoReflect().Interface()) == reflect.TypeOf(ap) {
			return proto.Equal(ap, any(b).(proto.Message))            // ② protobuf
		}
		panic(fmt.Sprintf("unable to compare object %T; perhaps it is embedding a protobuf? ...", a))
	}
	return reflect.DeepEqual(a, b)                                    // ③ 兜底
}

四级优先级，落地成三条你必须知道的规则：

规则一：Equals 是链式传播的总闸。 一个派生对象只要 Equals 返回 true，它下游的整条链都不会动。所以判断"什么算变化"的权力，完全握在类型作者手里。

实际用法就是这样，Gateway API 那一层的类型（gateway_collection.go）：

func (g Gateway) Equals(other Gateway) bool {
	return g.Config.Equals(other.Config) &&
		g.Valid == other.Valid // TODO: ok to ignore parent/parentInfo?
}

Parent 和 ParentInfo 被故意排除在比较之外。 这两个字段变了不影响输出，排除掉就能少一次下游重算。这个 TODO 说明他们自己也在权衡——Equals 写松了会丢更新，写严了会白重算，没有中间地带。

规则二：proto 嵌在 struct 里会 panic，不是慢慢变脏。 Equal 里那个 panic("unable to compare object %T; perhaps it is embedding a protobuf?") 是故意的。reflect.DeepEqual 比较 proto 是不可靠的（内部有 XXX_ 缓存字段、unknown fields 顺序），所以宁可炸也不给你错答案。遇到这个 panic，出路是自己实现 Equals。

规则三：默认的 reflect.DeepEqual 很贵。 1000 个对象、每个对象几十个字段，全量深比较会出现在 profile 顶部。所以 krt 的类型通常做成"只装用得到的字段"的小 struct——Workload 里没有整个 *v1.Pod，只有 Named、IP、ServiceNames。

四、拿 krt 当 client-go 用，踩的四个坑

上面是"它怎么工作"，这一节是"你会怎么把它写坏"。四个坑都在真实代码里有正反对照。

4.1 坑一：把 Singleton 当万能，一个变了全量重算

README 里有一节专门讲这件事，叫 Picking a collection type。它先给了一个能用但错的写法：

// ❌ 看起来最省事：一次 Fetch 全量，自己循环
SimplePods := krt.NewSingleton[SimplePod](func(ctx krt.HandlerContext) *[]SimplePod {
	res := []SimplePod{}
	for _, pod := range krt.Fetch(ctx, Pod) {
		res = append(res, SimplePod{Name: pod.Name})
	}
	return &res
})   // 结果是 Collection[[]SimplePod]

问题不在能不能跑，在于：任何一个 Pod 变一下，1000 个 SimplePod 全部重算；而且下游拿到的是一整个 []SimplePod，想按 label 筛都筛不了——只能自己循环。

正确的写法是把 Pod 提升成主依赖：

// ✅ 一对零/一：一个 Pod 变了只重算它自己
SimplePods := krt.NewCollection(Pods, func(ctx krt.HandlerContext, pod *v1.Pod) *SimplePod {
	return &SimplePod{Name: pod.Name}
})

README 给的经验法则一句话：

As a rule of thumb, if your Collection type is a list, you most likely should be using a different type to flatten the list.

Collection[[]X] 基本就是设计错了的信号。 例外是"这个列表本身是一个不可分割的原子集合，永远不会被单独查询"——这时候不要用切片，用 type PodContainers struct { Name string; Containers []string } 把它包成一个对象。

4.2 坑二：返回值是 *O 但内部还在变

NewCollection 的签名是 func(ctx, I) *O，返回 nil 就表示"这个输入不产出对象"（用于过滤）。这条语义很容易用错成"暂时没有，先返回 nil 等会儿再说"——不会有"等会儿"。返回 nil 意味着这条路彻底断开，旧输出会被当成 delete 清掉。

反过来，如果返回的对象里塞了指针，Equals 会去深比较它指的东西，而且下一次重算你会修改同一份内存，Old 和 New 就变成同一个对象，Equal 恒为 true，更新永远发不出去。规则很简单：变换函数返回的必须是新构造的、不含外泄指针的值。

4.3 坑三：在变换函数里读了 krt.Fetch 之外的东西

这条 README 用了最强硬的措辞，但它在代码里完全不报错。典型翻车场景：

// ❌ mesh 配置改了，这个集合不会重算
c := krt.NewCollection(Pods, func(ctx krt.HandlerContext, p *v1.Pod) *Workload {
	if meshConfig.EnableSomething {   // 直接读全局对象，没走 Fetch
		// ...
	}
	return &Workload{...}
})

从框架视角，这次变换的依赖清单里只有 Pods——Collection 根本不知道你偷看了 mesh 配置。所以 mesh 改了，没有人通知它重算。表现是"改完 mesh 配置，一部分 Pod 生效了一部分没生效，重启 istiod 全好"。

正确做法是把 mesh 配置也变成一个 Collection。Istio 自己就是这么干的（pilot/pkg/bootstrap/mesh.go）：

col := s.getMeshConfiguration(args, fileWatcher)
col.AsCollection().WaitUntilSynced(s.internalStop)   // 同步等它好
s.environment.Watcher = meshwatcher.ConfigAdapter(col)

然后在需要它的变换里 krt.FetchOne(ctx, meshConfig)。

4.4 坑四：Fetch 不带过滤器

// ❌ 每次重算都拉全量，然后在函数体里自己 if
all := krt.Fetch(ctx, AllServices)
for _, s := range all { if s.Namespace == p.Namespace { ... } }

能跑，但把两件事都搞砸了：一是没有反查索引（FilterGeneric/无过滤 → unknownIndexType → 全扫），任何 Service 变更都会触发全部 Pod 重算；二是 List() 出来的对象还要再复制一遍。

// ✅ 用 namespace 索引，走 FilterIndex
services := ServicesByNamespace.Fetch(ctx, p.Namespace, krt.FilterSelectsNonEmpty(p.GetLabels()))

krt.NewNamespaceIndex(Services) 建的是按 namespace 的索引，FilterIndex 走的是 indexType 分支——反查表精确到 namespace，Service 变更只会命中同 namespace 的 Pod。

顺带一个坑中坑：FilterLabel 和 FilterSelects 看着像，语义相反。

过滤器	判断式（Matches 里）	空参数时
FilterLabel(labels)	labels ⊆ object.Labels	匹配一切
FilterSelects(labels)	object.Spec.Selector ⊆ labels（对象的选择器被目标标签满足）	匹配一切
FilterSelectsNonEmpty(labels)	object.Spec.Selector.Match(labels)	一个都不匹配

前两个是"子集包含"，第三个是"真正的 selector 匹配"——FilterSelectsNonEmpty 才是"这个 Service 会选中这组 Pod"的准确表达。

workloads 那种场景必须用 FilterSelectsNonEmpty——一个 Pod 的 namespace 里如果有个 selector: {} 的 Service（选中一切），用 FilterSelects 会把全 namespace 的 Pod 都算成它的端点。

五、除了三个构造函数，还有五个常用的

主线上还没出现的几个类型，用到的频率不比 NewCollection 低。

5.1 NewStatic / NewStaticCollection：外部世界怎么进来

K8s 对象有 informer，但 mesh 配置、网络配置、磁盘上的文件没有。krt 的统一入口是静态集合（static.go）：

func NewStatic[T any](initial *T, startSynced bool, opts ...CollectionOption) StaticSingleton[T] {
	val := new(atomic.Pointer[T])
	val.Store(initial)
	x := &static[T]{val: val, synced: &atomic.Bool{}, id: nextUID(), eventHandlers: &handlers[T]{}}
	x.synced.Store(startSynced)
	// ...
}

写进去就触发事件，Set() 会自己 diff 出 add/update/delete：

func (d *static[T]) Set(now *T) {
	old := d.val.Swap(now)
	if old == now {
		return
	}
	events := toEvents(old, now)      // nil→值 = Add，值→nil = Delete，key 变了 = Delete+Add
	for _, h := range d.eventHandlers.Get() {
		h(events)
	}
}

toEvents() 里那个"key 变了就拆成 delete + add"的分支容易被忽略，但没有它，下游会看到一个 key 从 old 跳到 new 的 update 事件，而它持有的映射表里根本没有 new 这个 key——manyCollection 里那句 deletion of non-existent object 的错误日志就是为这类情况准备的。

Istio 里跟文件打交道的用法长这样（pkg/kube/krt/files/files.go），用一个 RecomputeTrigger 把"文件变了"翻译成"重算"：

cur := atomic.NewPointer(&cfg)
trigger := krt.NewRecomputeTrigger(true, opts...)
sc := krt.NewSingleton[T](func(ctx krt.HandlerContext) *T {
	trigger.MarkDependant(ctx)       // 声明：我依赖这个触发器
	return cur.Load()
}, opts...)
sc.AsCollection().WaitUntilSynced(stop)
watchFile(fileWatcher, filename, stop, func() {
	cfg, err := readFile(filename)
	if err != nil { log.Warnf("failed to update: %v", err); return }
	cur.Store(&cfg)
	trigger.TriggerRecomputation()
})

RecomputeTrigger 的实现很朴素，但有一个绕坑的细节（recomputetrigger.go）：

func (r *RecomputeTrigger) TriggerRecomputation() {
	v := r.i.Inc()
	r.inner.Set(ptr.Of(v))     // 每次写一个不同的数，保证事件不会被 Equal 吞掉
}

注释说得很清楚：krt will suppress events for unchanged resources. To workaround this, we constantly change and int each time。如果这里写的是 Set(ptr.Of(int32(1)))，第二次触发就没事件了——因为 Equal(1, 1) == true。这就是前面说的"Equals 是总闸"的一个活体样本。

RecomputeTrigger 在 README 里被明确定性为逃生舱：

RecomputeTrigger works around this by allowing an explicit call to recompute a collection; the caller must be sure to call Trigger() any time the state changes.

翻译：用它就等于自己给自己签了张欠条。忘调一次，就是永久脏数据。

5.2 JoinCollection / JoinWithMergeCollection：多来源合并

Istio 里同一类对象常有多来源：集群 1 的 Service + 集群 2 的 Service；K8s CRD 的 VirtualService + ServiceEntry 里内嵌的。

JoinCollection 的语义是后者让位（join.go 注释）：

// JoinCollection combines multiple Collection[T] into a single
// Collection[T]. Key conflicts are resolved by picking the item
// produced by the first collections in the list of input collections.

冲突处理全靠 refreshEvents()，这里有个我认为最值得抄的设计——不存对象，只存优先级：

// refreshEvents refreshes events by checking the current state of all collections
// to determine which collection is authoritative for each key. This implements
// conflict resolution without storing objects.

具体做法是收事件时先反查一遍"比我优先级高的集合里有没有这个 key"（join.go）：

if ev.Event == controllers.EventDelete {
	if hasHigherPriority {
		continue      // 这个 key 本来就不归你管，你删了也不影响输出
	}
	// 我是权威源，但底层集合还有没有备份？有就把 DELETE 降级成 UPDATE
	if fallbackObj != nil {
		result = append(result, Event[T]{Event: controllers.EventUpdate, Old: ev.Old, New: fallbackObj})
	}
}

把 DELETE 降级成 UPDATE（回落到低优先级源的版本）这一步，是最容易漏的。 漏了它，集群 1 的同名 Service 一删，集群 2 的那份就永久消失，直到有人重启 istiod。

JoinWithMergeCollection 是变体：不丢弃冲突项，而是把同 key 的所有对象收齐交给 merge func([]T) *T 自己决定（mergejoin.go）。Gateway API 合并 HTTPRoute 用的就是它。

5.3 NewIndex：读路径的加速器

前面讲的 indexedDependencies 是给写路径（变化来了找谁重算）用的；krt.NewIndex 是给读路径用的——在 Fetch 时把可能的候选集先缩小。

httpRoutesByInferencePool := krt.NewIndex(inputs.HTTPRoutes, "inferencepool-route", indexHTTPRouteByInferencePool)

一个索引对象不能被直接 List，但可以：

• idx.Lookup(k) —— 同步查
• idx.Fetch(ctx, k, filters...) —— 带依赖注册的查（推荐）
• idx.AsCollection(opts...) —— 变成一个 Collection[IndexObject[K, O]]，用于继续派生

NewNamespaceIndex() 是最常用的特化。注意 FilterIndex 要求索引和过滤条件匹配——如果你在 Fetch 里用了索引，反查走 indexType；如果只用了 FilterGeneric，那前面说的 doNotIndex 会把整个集合的索引优化干掉。

5.4 NewStatusCollection：派生物要反写状态

Gateway API 和 Istio 的很多 CRD 都要求控制器把状态写回 status 字段。这类"从输入派生出资源 + 同时产生一条状态"的模式，krt 有专门的构造函数（status.go）：

func NewStatusManyCollection[I controllers.Object, IStatus, O any](
	c Collection[I],
	hf TransformationMultiStatus[I, IStatus, O],
	opts ...CollectionOption,
) (Collection[ObjectWithStatus[I, IStatus]], Collection[O])

分成两个集合返回：一个装 (原对象, 状态)，一个装真正的产物。关键在于输入被删除时那个补偿逻辑：

cleanupOnRemoval := func(i []Event[I]) {
	for _, e := range i {
		if e.Event == controllers.EventDelete {
			status.DeleteObject(GetKey(e.Latest()))     // 输入没了，状态也得清
		}
	}
}

注释解释了为什么必须单独写：When input is deleted, the transformation function wouldn't run.——变换函数只对"存在的输入"跑，删除是一条不经过函数体的路径。所有"派生对象需要清理"的逻辑都必须挂在删除路径上，写在变换函数里永远不会执行。

六、Sync：krt 里最容易写错的一环

上一篇讲过 istiod 启动时卡在 Waiting for caches to be synced 的症状。krt 的 sync 语义比 informer 复杂，因为一个集合的"就绪"要等它的输入就绪，而输入可能又是一层派生。

manyCollection 的同步逻辑全在 runQueue() 里：

func (h *manyCollection[I, O]) runQueue() {
	c := h.parent
	// Wait for primary dependency to be ready
	if !c.WaitUntilSynced(h.stop) {
		return
	}
	// Now register to our primary collection. On any event, we will enqueue the update.
	parentReg := c.RegisterBatch(func(o []Event[I]) { /* ... */ }, true)
	// Wait for everything initial state to be enqueued
	if !parentReg.WaitUntilSynced(h.stop) {
		parentReg.UnregisterHandler()
		return
	}
	// ...
	h.queue.Run(h.stop)
}

三步，每一步都不能省：

1. 等主输入 synced（c.WaitUntilSynced）
2. 注册并等初始状态全部入队（parentReg.WaitUntilSynced）
3. 跑队列，队列处理完初始状态时回调 close(h.synced)

第 3 步那个"处理完才 close"是靠 queue.NewWithSync 实现的：

h.queue = queue.NewWithSync(func() {
	close(h.synced)
	h.log.Infof("%v synced (uid %v)", h.name(), h.uid())
}, h.collectionName)

于是**"链"这个概念在 krt 里是显式的**：A synced → B 才能开始处理初始状态 → B 处理完才 synced → C 才能开始。链条上任一级卡住，下游全部 WaitUntilSynced 返回 false（或者等到 stop 关闭）。

这也是为什么 NewCollection 的构造是异步启动的：newManyCollection 最后一行就是 go h.runQueue()。构造函数立刻返回，同步在后台慢慢爬。

flowchart LR
  I["informer/Pods<br/>synced"] --> C["Workloads<br/>runQueue"]
  C -->|"① 等 parent synced"| C2["② 注册 + 等初始入队"]
  C2 --> C3["③ queue.Run + 处理完 close(synced)"]
  C3 --> D["下游 Collection<br/>这时才开始"]

反例：直接 List() 一个还没 synced 的集合。List() 永远返回"当前状态"，不阻塞、不报错——刚启动时就是空。

// ❌ 拿到的是空集合，而且没有任何提示
items := someCollection.List()

// ✅ 想确定拿到完整数据，必须显式等
someCollection.WaitUntilSynced(stop)

Istio 自己在启动阶段就是这么小心处理的（pilot/pkg/bootstrap/mesh.go）：

col := s.getMeshConfiguration(args, fileWatcher)
col.AsCollection().WaitUntilSynced(s.internalStop)   // 注意是 internalStop，不是 stop
s.environment.Watcher = meshwatcher.ConfigAdapter(col)

注释解释了为什么用 internalStop：We need to get mesh configuration up-front, before we start anything, so we use internalStop rather than scheduling a task to run later.——装配阶段要等，就只能用那个还没被 Start() 接管的 stop。

还有一个 RegisterBatch(f, runExistingState bool) 的语义坑：runExistingState=true 时，集合会先把当前状态作为一批 Add 事件推给 handler，然后再解锁新事件，中间不会混入重复的 Add。实现是"持读锁发初始状态"（collection.go）：

h.mu.RLock()
defer h.mu.RUnlock()
events := make([]Event[O], 0, len(h.collectionState.outputs))
for _, o := range h.collectionState.outputs {
	events = append(events, Event[O]{New: &o, Event: controllers.EventAdd})
}
return h.eventHandlers.Insert(f, h, events, h.stop)

但 informer 类型不遵守这个参数（informer.go 里写得很直接）：

// Note: runExistingState is NOT respected here.
// Informer doesn't expose a way to do that.

所以从 informer 来的集合，注册 handler 时拿到的初始状态是异步的、逐条的，不存在"完整一批"。写测试时这一点尤其要注意。

七、动手验证：把依赖图 dump 出来

这一节不需要改代码，一个 demo profile 的 istiod 就够。

7.1 一条命令看全图

/debug/krtz 是 istiod 内置的 krt 状态 dump（pilot/pkg/xds/debug.go 注册的 /debug/krtz，krtz() 直接把 s.krtDebugger 序列化）：

kubectl -n istio-system port-forward deploy/istiod 8080:8080 >/dev/null 2>&1 &
sleep 1

# 有多少个集合，各叫什么
curl -s localhost:8080/debug/krtz | jq -r '.[].name' | sort | head -40

# 只看 Gateway API 那条链
curl -s localhost:8080/debug/krtz | jq -r '.[] | select(.name|startswith("gateway/")) | "\(.uid)\t\(.name)\tsynced=\(.state.synced)\toutputs=\(.state.outputs|length)"'

返回结构就是 debug.go 里的 DebugCollection.MarshalJSON：

{ "uid": 132, "name": "gateway/GatewayConfigs", "state": { "outputs": {...}, "inputs": {...}, "synced": true } }

state 是 CollectionDump，三个字段正好对应前面讲的三件事：

字段	含义	拿它查什么
outputs	输出 key → 输出对象	所以"本来该有却没有"就是这一级没产出
inputs	输入 key → {outputs, dependencies}	dependencies 就是这次变换 Fetch 过的集合名
synced	是否就绪	卡启动时先看这个

flowchart TD
  A["/debug/krtz"] --> B["按 name 找到集合"]
  B --> C{"state.synced?"}
  C -->|false| D["上游没就绪：往上找 parent 集合"]
  C -->|true| E{"outputs 里有目标 key?"}
  E -->|没有| F["看 inputs[key].dependencies<br/>缺哪个集合的 Fetch"]
  E -->|有但内容旧| G["查 Equals：Old 与 New 被判相等"]

这张决策树是这一篇最实用的产出。"配置没生效"的排障路径从'猜哪个 controller'变成'按集合逐级往下走'。

7.2 看一个具体的依赖

挑一条你最关心的输入，把它的依赖清单打出来：

curl -s localhost:8080/debug/krtz \
  | jq -r '.[] | select(.name=="gateway/GatewayConfigs") | .state.inputs | to_entries[0].value.dependencies[]'

dependencies 来自 dump() 里的这段（collection.go）：

for k, deps := range h.dependencyState.objectDependencies {
	depss := make([]string, 0, len(deps))
	for _, dep := range deps {
		depss = append(depss, dep.collectionName)
	}
	inputs[string(k)] = InputDump{Outputs: output, Dependencies: depss}
}

如果你在 dependencies 里没看到某个集合，那说明变换函数没用 Fetch 去读它——这就是第四节坑三的现场取证。反之，如果 dependencies 里出现了你完全没预料到的集合，去看看代码里是不是混进了一个宽过滤的 Fetch。

7.3 打开 krt 自己的日志

krt 注册了独立的作用域，可以单独开：

kubectl -n istio-system set env deploy/istiod KRT_LOG_LEVEL=debug
kubectl -n istio-system rollout status deploy/istiod

# 看同步链
kubectl -n istio-system logs deploy/istiod | grep "synced (uid" | head -30

# 看二级依赖命中情况：event size 多少、影响了几个对象
kubectl -n istio-system logs deploy/istiod | grep "impacts" | head -20

event size N, impacts M objects 这行 debug 日志（collection.go 的 onSecondaryDependencyEvent，h.log.Debugf）是性能问题的第一现场：

event size 1, impacts 2 objects      ← 索引生效，只重算 2 个
event size 1, impacts 1000 objects   ← 索引没生效，全量重算

看到 impacts 数量等于集合大小，就去检查 Fetch 是不是用了 FilterLabel / FilterGeneric 这类不可索引的过滤器。

7.4 自己跑基准

krt 自带和"手写最优控制器"的对照基准：

git clone --depth 1 -b 1.31.4 https://github.com/istio/istio.git && cd istio
go test ./pkg/kube/krt -run XXX -bench BenchmarkControllers -benchmem

两个子用例 Controllers/krt 和 Controllers/legacy，场景是 1000 Pod + 50 Service，然后连续更新 1000 个 Pod。README 里记录的结论是开销约 10%：

Controllers/krt-8     13.4ms     Controllers/legacy-8  11.4ms
Controllers/krt-8     15.2MB     Controllers/legacy-8  12.9MB

这个数字最有意思的地方不是"只慢 10%"，而是**"legacy"那一边是理想化的手写实现**——真的手写控制器，很少能做到它那么干净。README 自己的比喻是汇编和高级语言：汇编总能更快，但编译器在一个地方做的优化，每个手写实现都要各做一遍。

7.5 用 conformance 测试给自定义集合兜底

如果你打算在自己的项目里复刻这套东西，conformance_test.go 是现成的测试套。它定义了一个 Rig[T] 接口，任何新集合类型只要实现它，就能跑全套：

type Rig[T any] interface {
	krt.Collection[T]
	CreateObject(key string)
	// ReplaceKey removes oldKey and adds newKey.
	ReplaceKey(oldKey, newKey string)
}

// 关键断言：List() 从 [a,b,c] 变成 [a,b,d] 时，
// 必须看到 c 的 delete 和 d 的 add，不能只有一次 update
func runReplaceConformance[T any](t *testing.T, collection Rig[T]) {
	collection.ReplaceKey("a/b", "a/c")
	handler.WaitUnordered("delete/a/b", "add/a/c")
}

现在它覆盖了 informer、static list、static singleton、join、manyCollection、files 六种实现。这个测试套透露了 krt 最被低估的一点：它把"事件流的正确性"变成了一份可枚举、可测试的契约——换一种集合实现，只要过了这套 conformance，下游代码就不用改。手写控制器里，这份契约通常是口口相传的。

八、为什么 Istio 不用 client-go

到这里可以正面回答开篇那个问题了。pkg/kube/krt/README.md 里给了两条官方理由，但只说对了一半。

官方理由一：client-go 的 informer 只能处理 K8s 类型。 mesh 配置、网络配置、磁盘上的文件，客户端驱动的 CRD，全都不是 K8s informer 能装的东西。krt 把这些统一成 Collection 之后，一条 Fetch 就能跨源依赖——meshwatcher.NewCollection() 把一个配置文件变成一个集合，下游完全不知道它的数据是来自 API Server 还是来自 /etc/istio/config/mesh。

官方理由二：抽象层次。 控制器作者只写 Input -> Output，状态管理交给框架。这一条听起来像客套，但落到代码上就是：NewCollection(Gateways, func(ctx, i Gateway) *config.Config) 这一行，替代了 client-go 版的 handler 注册 + workqueue + Get 回读 + diff + 事件分发，大约 40 行。

然后是我认为更重要的两条，README 里没写：

理由三：Fetch 把"反向索引"变成了默认行为。 client-go 版的"Service 变了要重算哪些 Pod"，每个控制器作者自己实现一遍，实现方式通常是 pods.List(ns, selector) 全扫——因为写反查表麻烦。krt 把它变成了 Fetch(ctx, c, FilterIndex(...)) 的副作用：用对的过滤器，索引自动就有；用错的过滤器，全扫。这不是魔法，是把"优化"变成了 API 设计的一部分。

理由四：派生链可以被 dump 出来。/debug/krtz 能给出整张图，是因为 krt 真的知道这张图。client-go 写完的控制器，依赖关系只存在于作者脑子里——"改了这个配置为什么那个对象没更新"只能靠读代码。

至于代价，README 说得很实：

This library is currently "experimental" and is not used in Istio production yet. The intent is this will be slowly rolled out to controllers that will benefit from it and are lower risk; likely the ambient controller will be the first target.

这段是写作时 README 的原文，但 1.31.4 的代码已经走到另一条路上了：pilot/pkg/config/kube/gateway/controller.go 全线 krt，gateway_collection.go 里 krt.NewStatusManyCollection、krt.Fetch、krt.NewCollection 铺满，mesh 配置本身也是 krt 集合。Ambient 的 WDS 那条链（第 26 篇）是 krt 落地最深的地方——README 这段文字只是没跟着代码更新。

README 还留了一段"未来的想法"值得一提，因为它承认了这套设计的漏点：

Today, the library stores a mapping of Input -> Dependencies (map[Key[I]][]dependency). Often times, there are common dependencies amongst keys. For example, a namespace filter probably has many less unique values than unique input objects.

翻译：map[Key[I]][]dependency 这个结构在 1000 个 Pod 就是 1000 份重复的依赖清单。用 FilterIndex(nsIndex, "ns-1") 的 1000 个 Pod，其实只依赖 2 个 namespace，但存了 1000 遍 []dependency{...}。这是用内存换精确性的交易——精确到"只重算这 2 个"，代价是每个输入都记一份账。

九、收口

这篇只回答了一个问题：一个变更是怎么被精确地传播到该传播的地方的。 答案是一条四段流水线：

1. 记账：每次变换跑完，把 Fetch 攒下来的依赖写回 I → []dependency；
2. 反查：变更来了，先查反查索引，把"可能受影响的 I"缩到几个；
3. 重算：只对这几个 I 重跑变换函数；
4. 过滤：用 Equal 比对结果，相同就不往外发事件——Equals 是整条链的总闸。

带走四件事：

• Collection 是唯一的抽象，Fetch 是唯一的依赖通道。 想读别的集合，只能 Fetch；偷看全局变量不会报错，只会得到陈旧数据。
• FilterKey / FilterIndex 建反查索引，FilterLabel / FilterGeneric 不建。 这一个选择决定一次变更重算 2 个对象还是 1000 个。日志里的 impacts N objects 就是判据。
• Equals 写宽松了丢更新，写严了白重算。 自己实现 Equals 时只比"会影响输出"的字段；嵌了 proto 又没实现 Equaler 会直接 panic，这是故意的。
• "就绪"在 krt 里是一条链：输入 synced → 注册 → 初始状态入队 → 处理完才 synced。直接 List() 一个没 synced 的集合，拿到的是空且不报错。

下一篇回到配置这一侧：装配阶段那些控制器最终都要产出一个东西——config.Config。 一个 virtualservice.yaml 从 YAML 变成 config.Config 要走几道门：kind 注册表、pkg/config 的 schema、kube/crd 客户端、informer，最后一个 PushRequest.ConfigsUpdated。顺带厘清那个最常见的误解——MeshConfig 存在 ConfigMap 里，但它走的是独立的 mesh watcher，和 VirtualService 完全是两条路。