Felix 计算图原理

// 文件: felix/calc/calc_graph.go:114-135

type CalcGraph struct {
    // AllUpdDispatcher 是计算图的输入节点
    AllUpdDispatcher *dispatcher.Dispatcher

    // 指向其他计算图节点的指针
    localEndpointDispatcher *dispatcher.Dispatcher
    activeRulesCalculator   *ActiveRulesCalculator
    ruleScanner             *RuleScanner
    serviceIndex            *serviceindex.ServiceIndex
    ipsetMemberIndex        *labelindex.SelectorAndNamedPortIndex
    hostIPPassthru          *DataplanePassthru
    l3RouteResolver         *L3RouteResolver
    vxlanResolver           *VXLANResolver
    configBatcher           *ConfigBatcher
    profileDecoder          *ProfileDecoder
    encapsulationResolver   *EncapsulationResolver
    policyResolver          *PolicyResolver
}

// 文件: felix/calc/calc_graph.go:543-574

// endpointHostnameFilter 过滤非本机端点
type endpointHostnameFilter struct {
    hostname string
}

func (f *endpointHostnameFilter) OnUpdate(update api.Update) (filterOut bool) {
    switch key := update.Key.(type) {
    case model.WorkloadEndpointKey:
        if key.Hostname != f.hostname {
            filterOut = true  // 过滤掉非本机端点
        }
    case model.HostEndpointKey:
        if key.Hostname != f.hostname {
            filterOut = true
        }
    }
    return
}

// 文件: felix/calc/active_rules_calculator.go:61-97

type ActiveRulesCalculator struct {
    // 所有已知的 Tier/Policy/Profile 缓存
    allTiers        map[string]*model.Tier
    allPolicies     packedmap.Map[model.PolicyKey, *model.Policy]
    allProfileRules packedmap.Deduped[string, *model.ProfileRules]

    // Policy/Profile ID 到匹配端点集合的映射
    policyIDToEndpointKeys  multidict.Multidict[any, any]
    profileIDToEndpointKeys multidict.Multidict[string, any]

    // 标签索引，用于策略选择器与本地端点的匹配
    labelIndex *labelindex.InheritIndex

    // 端点到 Profile ID 的缓存
    endpointKeyToProfileIDs *EndpointKeyToProfileIDMap

    // 回调接口
    RuleScanner          ruleScanner
    PolicyMatchListeners []PolicyMatchListener
}

// 文件: felix/calc/active_rules_calculator.go:352-391

// 当选择器开始匹配端点时调用
func (arc *ActiveRulesCalculator) onMatchStarted(selID, labelId interface{}) {
    polKey := selID.(model.PolicyKey)
    policyWasActive := arc.policyIDToEndpointKeys.ContainsKey(polKey)
    arc.policyIDToEndpointKeys.Put(selID, labelId)

    if !policyWasActive {
        // 策略之前不活跃，现在变为活跃
        log.Debugf("Policy %v now active", polKey)
        policy, known := arc.allPolicies.Get(polKey)
        if !known {
            log.WithField("policy", polKey).Panic("Policy active but missing from allPolicies.")
        }
        // 通知 RuleScanner 策略变为活跃
        arc.sendPolicyUpdate(polKey, policy)
    }

    // 通知 PolicyMatchListeners（包括 PolicyResolver）
    if labelId, ok := labelId.(model.EndpointKey); ok {
        for _, l := range arc.PolicyMatchListeners {
            l.OnPolicyMatch(polKey, labelId)
        }
    }
}

// 当选择器停止匹配端点时调用
func (arc *ActiveRulesCalculator) onMatchStopped(selID, labelId interface{}) {
    polKey := selID.(model.PolicyKey)
    arc.policyIDToEndpointKeys.Discard(selID, labelId)

    if !arc.policyIDToEndpointKeys.ContainsKey(selID) {
        // 策略不再活跃
        log.Debugf("Policy %v no longer active", polKey)
        policy, _ := arc.allPolicies.Get(polKey)
        arc.sendPolicyUpdate(polKey, policy)
    }

    // 通知匹配停止
    if labelId, ok := labelId.(model.EndpointKey); ok {
        for _, l := range arc.PolicyMatchListeners {
            l.OnPolicyMatchStopped(polKey, labelId)
        }
    }
}

// 文件: felix/labelindex/label_inheritance_index.go:96-110

type InheritIndex struct {
    // 端点数据缓存
    itemDataByID         map[interface{}]*itemData
    // 父级（Profile）数据缓存
    parentDataByParentID map[string]*parentData
    // 选择器缓存
    selectorsById        map[interface{}]*selector.Selector

    // 当前匹配关系
    selIdsByLabelId map[interface{}]set.Typed[any]
    labelIdsBySelId map[interface{}]set.Typed[any]

    // 回调函数
    OnMatchStarted MatchCallback
    OnMatchStopped MatchCallback

    // 脏数据集合（需要重新计算）
    dirtyItemIDs set.Typed[any]
}

// 文件: felix/labelindex/label_inheritance_index.go:374-386

func (idx *InheritIndex) updateMatches(
    selId interface{},
    sel *selector.Selector,
    labelId interface{},
    labels parser.Labels,
) {
    // 使用选择器评估标签
    nowMatches := sel.EvaluateLabels(labels)
    if nowMatches {
        idx.storeMatch(selId, labelId)
    } else {
        idx.deleteMatch(selId, labelId)
    }
}

// 文件: felix/calc/rule_scanner.go:191-282

func (rs *RuleScanner) updateRules(
    key interface{},
    inbound, outbound []model.Rule,
    untracked, preDNAT bool,
    origNamespace string,
    origSelector string,
    tier string,
    perfHints []apiv3.PolicyPerformanceHint,
) (parsedRules *ParsedRules) {
    // 提取所有新的选择器和命名端口
    currentUIDToIPSet := make(map[string]*IPSetData)
    parsedInbound := make([]*ParsedRule, len(inbound))

    for ii, rule := range inbound {
        parsed, allIPSets := ruleToParsedRule(&rule)
        parsedInbound[ii] = parsed
        for _, ipSet := range allIPSets {
            currentUIDToIPSet[ipSet.UniqueID()] = ipSet
        }
    }

    // 计算新增和删除的 IP 集合
    addedUids := set.New[string]()
    for uid := range currentUIDToIPSet {
        if !rs.rulesIDToUIDs.Contains(key, uid) {
            addedUids.Add(uid)
        }
    }

    removedUids := set.New[string]()
    rs.rulesIDToUIDs.Iter(key, func(uid string) {
        if _, ok := currentUIDToIPSet[uid]; !ok {
            removedUids.Add(uid)
        }
    })

    // 触发 IP 集合激活/停用事件
    for uid := range addedUids.All() {
        rs.rulesIDToUIDs.Put(key, uid)
        if !rs.uidsToRulesIDs.ContainsKey(uid) {
            ipSet := currentUIDToIPSet[uid]
            rs.ipSetsByUID[uid] = ipSet
            rs.OnIPSetActive(ipSet)  // 通知 IP 集合激活
        }
        rs.uidsToRulesIDs.Put(uid, key)
    }

    for uid := range removedUids.All() {
        rs.rulesIDToUIDs.Discard(key, uid)
        rs.uidsToRulesIDs.Discard(uid, key)
        if !rs.uidsToRulesIDs.ContainsKey(uid) {
            ipSetData := rs.ipSetsByUID[uid]
            delete(rs.ipSetsByUID, uid)
            rs.OnIPSetInactive(ipSetData)  // 通知 IP 集合停用
        }
    }

    return parsedRules
}

// 文件: felix/calc/policy_resolver.go:52-66

type PolicyResolver struct {
    // 策略到端点的映射
    policyIDToEndpointIDs multidict.Multidict[model.PolicyKey, model.EndpointKey]
    // 端点到策略的映射
    endpointIDToPolicyIDs multidict.Multidict[model.EndpointKey, model.PolicyKey]
    // 策略元数据
    allPolicies           map[model.PolicyKey]policyMetadata
    // 排序后的 Tier 数据
    sortedTierData        []*TierInfo
    // 本地端点
    endpoints             map[model.Key]model.Endpoint
    // 需要重新计算的端点
    dirtyEndpoints        set.Set[model.EndpointKey]
    // 策略排序器
    policySorter          *PolicySorter
    // 回调接口
    Callbacks             []PolicyResolverCallbacks
}

// 文件: felix/calc/policy_resolver.go:178-201

func (pr *PolicyResolver) Flush() {
    if !pr.InSync {
        return
    }

    // 处理待处理的策略更新
    pr.pendingPolicyUpdates.Iter(func(polKey model.PolicyKey) error {
        policy, ok := pr.allPolicies[polKey]
        if !ok {
            return nil
        }
        pr.policySorter.UpdatePolicy(polKey, &policy)
        return set.RemoveItem
    })

    // 获取排序后的 Tier 数据
    pr.sortedTierData = pr.policySorter.Sorted()

    // 为每个脏端点发送更新
    for endpointID := range pr.dirtyEndpoints.All() {
        pr.sendEndpointUpdate(endpointID)
    }
    pr.dirtyEndpoints.Clear()
}

// 文件: felix/calc/event_sequencer.go:867-920

func (buf *EventSequencer) Flush() {
    // 1. 首先刷新配置更新（可能触发重启）
    buf.flushReadyFlag()
    buf.flushConfigUpdate()

    // 2. 按依赖顺序刷新添加/更新
    // IP 集合 -> 策略 -> 端点
    buf.flushAddedIPSets()
    buf.flushIPSetDeltas()
    buf.flushPolicyUpdates()
    buf.flushProfileUpdates()
    buf.flushEndpointTierUpdates()

    // 3. 按相反顺序刷新删除
    // 端点 -> 策略 -> IP 集合
    buf.flushEndpointTierDeletes()
    buf.flushProfileDeletes()
    buf.flushPolicyDeletes()
    buf.flushRemovedIPSets()

    // 4. 刷新其他资源
    buf.flushServiceAccounts()
    buf.flushNamespaces()

    // 5. VXLAN 相关（先删后加）
    buf.flushRouteRemoves()
    buf.flushVTEPRemoves()
    buf.flushVTEPAdds()
    buf.flushRouteAdds()

    // 6. 集群范围的更新
    buf.flushHostWireguardDeletes()
    buf.flushHostWireguardUpdates()
    buf.flushHostIPDeletes()
    buf.flushHostIPUpdates()
    buf.flushIPPoolDeletes()
    buf.flushIPPoolUpdates()
    buf.flushEncapUpdate()

    // 7. BGP 和 Service 更新
    if buf.pendingGlobalBGPConfig != nil {
        buf.Callback(buf.pendingGlobalBGPConfig)
        buf.pendingGlobalBGPConfig = nil
    }
    buf.flushServices()
}

// 文件: felix/calc/async_calc_graph.go:72-88

type AsyncCalcGraph struct {
    CalcGraph        *CalcGraph
    inputEvents      chan interface{}
    outputChannels   []chan<- interface{}
    eventSequencer   *EventSequencer
    beenInSync       bool
    needToSendInSync bool
    syncStatusNow    api.SyncStatus
    healthAggregator *health.HealthAggregator

    flushTicks       <-chan time.Time
    healthTicks      <-chan time.Time
    flushLeakyBucket int      // 漏桶限流
    dirty            bool
}

// 文件: felix/calc/async_calc_graph.go:136-194

func (acg *AsyncCalcGraph) loop() {
    log.Info("AsyncCalcGraph running")
    acg.reportHealth()

    for {
        select {
        case update := <-acg.inputEvents:
            switch update := update.(type) {
            case []api.Update:
                // 逐个处理更新，中间报告健康状态
                for i, upd := range update {
                    updStartTime := time.Now()
                    acg.CalcGraph.OnUpdates(update[i : i+1])
                    summaryUpdateTime.Observe(time.Since(updStartTime).Seconds())
                    // 记录处理统计
                    typeName := reflect.TypeOf(upd.Key).Name()
                    countUpdatesProcessed.WithLabelValues(typeName).Inc()
                    acg.reportHealth()
                }
            case api.SyncStatus:
                acg.syncStatusNow = update
                acg.CalcGraph.OnStatusUpdated(update)
                if update == api.InSync && !acg.beenInSync {
                    log.Info("First time we've been in sync")
                    acg.beenInSync = true
                    acg.needToSendInSync = true
                    acg.dirty = true
                    if acg.flushLeakyBucket == 0 {
                        acg.flushLeakyBucket++  // 强制刷新
                    }
                }
            }
            acg.dirty = true

        case <-acg.flushTicks:
            // 定时器：填充漏桶
            if acg.flushLeakyBucket < leakyBucketSize {
                acg.flushLeakyBucket++
            }

        case <-acg.healthTicks:
            acg.reportHealth()
        }
        acg.maybeFlush()
    }
}

// 文件: felix/calc/async_calc_graph.go:206-229

func (acg *AsyncCalcGraph) maybeFlush() {
    if !acg.dirty {
        return
    }
    if acg.flushLeakyBucket > 0 {
        log.Debug("Not throttled: flushing event buffer")
        acg.flushLeakyBucket--

        flushStart := time.Now()
        acg.CalcGraph.Flush()
        acg.eventSequencer.Flush()
        flushDuration := time.Since(flushStart)

        if flushDuration > time.Second {
            log.WithField("time", flushDuration).Info("Flush took over 1s.")
        }

        if acg.needToSendInSync {
            log.Info("First flush after becoming in sync, sending InSync message.")
            acg.onEvent(&proto.InSync{})
            acg.needToSendInSync = false
        }
        acg.dirty = false
    } else {
        log.Debug("Throttled: not flushing event buffer")
    }
}

// 文件: felix/calc/calc_graph.go:149-532

func NewCalculationGraph(
    callbacks PipelineCallbacks,
    cache *LookupsCache,
    conf *config.Config,
    liveCallback func(),
) *CalcGraph {
    hostname := conf.FelixHostname
    cg := &CalcGraph{}

    // 1. 创建全局更新分发器
    allUpdDispatcher := dispatcher.NewDispatcher()
    cg.AllUpdDispatcher = allUpdDispatcher

    // 2. 创建本地端点分发器（带过滤）
    localEndpointDispatcher := dispatcher.NewDispatcher()
    localEndpointFilter := &endpointHostnameFilter{hostname: hostname}
    localEndpointFilter.RegisterWith(localEndpointDispatcher)

    // 3. 创建 ActiveRulesCalculator
    activeRulesCalc := NewActiveRulesCalculator()
    activeRulesCalc.RegisterWith(localEndpointDispatcher, allUpdDispatcher)

    // 4. 创建 RuleScanner
    ruleScanner := NewRuleScanner()
    activeRulesCalc.RuleScanner = ruleScanner
    ruleScanner.RulesUpdateCallbacks = callbacks

    // 5. 创建 IP 集合成员索引
    ipsetMemberIndex := labelindex.NewSelectorAndNamedPortIndex(...)
    ipsetMemberIndex.RegisterWith(allUpdDispatcher)

    // 6. 连接 RuleScanner 和 IPSetMemberIndex
    ruleScanner.OnIPSetActive = func(ipSet *IPSetData) {
        callbacks.OnIPSetAdded(ipSet.UniqueID(), ipSet.DataplaneProtocolType())
        ipsetMemberIndex.UpdateIPSet(ipSet.UniqueID(), ipSet.Selector, ...)
    }
    ruleScanner.OnIPSetInactive = func(ipSet *IPSetData) {
        ipsetMemberIndex.DeleteIPSet(ipSet.UniqueID())
        callbacks.OnIPSetRemoved(ipSet.UniqueID())
    }

    // 7. 创建 PolicyResolver
    polResolver := NewPolicyResolver()
    activeRulesCalc.RegisterPolicyMatchListener(polResolver)
    polResolver.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
    polResolver.RegisterCallback(callbacks)

    // 8. 创建路由解析器（如果需要）
    if conf.BPFEnabled || conf.Encapsulation.VXLANEnabled || ... {
        l3RR := NewL3RouteResolver(hostname, callbacks, ...)
        l3RR.RegisterWith(allUpdDispatcher, localEndpointDispatcher)
    }

    return cg
}

// 文件: felix/daemon/daemon.go（简化）

func (d *Daemon) startCalculationGraph() {
    // 创建输出通道
    outputChannels := []chan<- interface{}{d.dataplaneChannel}

    // 创建异步计算图
    d.asyncCalcGraph = calc.NewAsyncCalcGraph(
        d.config,
        outputChannels,
        d.healthAggregator,
        lookupCache,
    )

    // 启动计算图
    d.asyncCalcGraph.Start()

    // 连接 Syncer
    d.syncer.AddCallbacks(d.asyncCalcGraph)
}

// 文件: felix/calc/active_rules_calculator.go:67-70

// 使用压缩 Map 存储策略（减少内存占用）
allPolicies     packedmap.Map[model.PolicyKey, *model.Policy]
// 使用去重压缩 Map 存储 Profile（相同内容共享存储）
allProfileRules packedmap.Deduped[string, *model.ProfileRules]