kube-proxy iptables 模式

// pkg/proxy/iptables/proxier.go

// 计算每个 endpoint 的概率
func probability(n int) string {
    if n <= 1 {
        return ""
    }
    // 第 i 个 endpoint 的概率是 1/(n-i+1)
    return fmt.Sprintf("--probability %0.10f", 1.0/float64(n))
}

// pkg/proxy/iptables/proxier.go

type Proxier struct {
    // iptables 接口
    iptables       utiliptables.Interface
    ipset          utilipset.Interface

    // 服务和端点信息
    svcPortMap     proxy.ServicePortMap
    endpointsMap   proxy.EndpointsMap

    // 同步状态
    syncPeriod     time.Duration
    minSyncPeriod  time.Duration
    lastSync       time.Time

    // 节点信息
    nodeIP         net.IP
    hostname       string
    nodePortAddresses []string

    // 规则缓冲
    filterChains  utilproxy.LineBuffer
    filterRules   utilproxy.LineBuffer
    natChains     utilproxy.LineBuffer
    natRules      utilproxy.LineBuffer
}

// syncProxyRules 同步 iptables 规则
func (proxier *Proxier) syncProxyRules() {
    proxier.mu.Lock()
    defer proxier.mu.Unlock()

    // 1. 获取当前需要的服务和端点
    serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
    endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)

    // 2. 初始化规则缓冲
    proxier.filterChains.Reset()
    proxier.filterRules.Reset()
    proxier.natChains.Reset()
    proxier.natRules.Reset()

    // 3. 写入固定链
    proxier.writeIptablesBaseChains()

    // 4. 为每个服务写入规则
    for svcName, svc := range proxier.svcPortMap {
        svcInfo := svc.(*servicePortInfo)

        // 写入 KUBE-SERVICES 规则
        proxier.writeServiceToEndpointRules(svcName, svcInfo)

        // 写入 NodePort 规则
        if svcInfo.NodePort() != 0 {
            proxier.writeNodePortRules(svcName, svcInfo)
        }

        // 写入 LoadBalancer 规则
        for _, ingress := range svcInfo.LoadBalancerIPStrings() {
            proxier.writeLoadBalancerRules(svcName, svcInfo, ingress)
        }
    }

    // 5. 写入 KUBE-MARK-MASQ 和 KUBE-POSTROUTING
    proxier.writeMasqueradeRules()

    // 6. 原子替换规则
    proxier.iptablesData.Reset()
    proxier.writeIptablesData()

    // 使用 iptables-restore 原子更新
    err := proxier.iptables.RestoreAll(
        proxier.iptablesData.Bytes(),
        utiliptables.NoFlushTables,
        utiliptables.RestoreCounters,
    )
    if err != nil {
        klog.ErrorS(err, "Failed to restore iptables rules")
    }

    proxier.lastSync = time.Now()
}

// writeServiceToEndpointRules 写入服务到端点的规则
func (proxier *Proxier) writeServiceToEndpointRules(
    svcName proxy.ServicePortName,
    svcInfo *servicePortInfo) {

    endpoints := proxier.endpointsMap[svcName]
    nEndpoints := len(endpoints)

    if nEndpoints == 0 {
        // 无端点，写入拒绝规则
        proxier.writeServiceNoEndpointRules(svcName, svcInfo)
        return
    }

    // 写入 KUBE-SERVICES 到 KUBE-SVC-XXX 的跳转
    svcChain := svcInfo.ServiceChainName()
    proxier.natRules.Write(
        "-A", string(kubeServicesChain),
        "-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcName.String()),
        "-m", protocol, "-p", protocol,
        "-d", svcInfo.ClusterIP().String(),
        "--dport", strconv.Itoa(svcInfo.Port()),
        "-j", string(svcChain),
    )

    // 写入 KUBE-SVC-XXX 链
    proxier.natChains.Write(utiliptables.MakeChainLine(svcChain))

    // 会话亲和性处理
    if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
        proxier.writeSessionAffinityRules(svcChain, endpoints, svcInfo)
    }

    // 写入负载均衡规则
    for i, ep := range endpoints {
        epInfo := ep.(*endpointInfo)
        sepChain := epInfo.ChainName()

        // 概率计算
        n := nEndpoints - i
        args := []string{
            "-A", string(svcChain),
            "-m", "comment", "--comment", fmt.Sprintf(`"%s -> %s"`, svcName.String(), epInfo.Endpoint),
        }
        if n > 1 {
            args = append(args, "-m", "statistic", "--mode", "random",
                "--probability", probability(n))
        }
        args = append(args, "-j", string(sepChain))
        proxier.natRules.Write(args...)

        // 写入 KUBE-SEP-XXX 链
        proxier.writeEndpointRules(sepChain, epInfo, svcInfo)
    }
}

// writeEndpointRules 写入端点规则
func (proxier *Proxier) writeEndpointRules(
    sepChain utiliptables.Chain,
    epInfo *endpointInfo,
    svcInfo *servicePortInfo) {

    proxier.natChains.Write(utiliptables.MakeChainLine(sepChain))

    // 如果源是本端点，标记 MASQUERADE（处理 hairpin）
    proxier.natRules.Write(
        "-A", string(sepChain),
        "-s", epInfo.IP(),
        "-m", "comment", "--comment", svcInfo.serviceNameString,
        "-j", string(kubeMarkMasqChain),
    )

    // DNAT 规则
    proxier.natRules.Write(
        "-A", string(sepChain),
        "-m", "comment", "--comment", svcInfo.serviceNameString,
        "-m", protocol, "-p", protocol,
        "-j", "DNAT",
        "--to-destination", epInfo.Endpoint,
    )
}

// pkg/proxy/iptables/proxier.go

// OnServiceAdd Service 添加时触发
func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
    proxier.OnServiceUpdate(nil, service)
}

// OnServiceUpdate Service 更新时触发
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
    if proxier.serviceChanges.Update(oldService, service) {
        proxier.Sync()
    }
}

// OnServiceDelete Service 删除时触发
func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
    proxier.OnServiceUpdate(service, nil)
}

// OnEndpointSliceAdd EndpointSlice 添加时触发
func (proxier *Proxier) OnEndpointSliceAdd(endpointSlice *discovery.EndpointSlice) {
    if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) {
        proxier.Sync()
    }
}

// Sync 触发同步
func (proxier *Proxier) Sync() {
    if proxier.isInitialized() {
        proxier.syncRunner.Run()
    }
}

// 使用 BoundedFrequencyRunner 控制同步频率
proxier.syncRunner = async.NewBoundedFrequencyRunner(
    "sync-runner",
    proxier.syncProxyRules,
    minSyncPeriod,  // 最小间隔
    syncPeriod,     // 最大间隔
    burstSyncs,     // 突发次数
)

# kube-proxy 配置
apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: iptables
iptables:
  # 最小同步间隔
  minSyncPeriod: 1s
  # 同步周期
  syncPeriod: 30s
  # 掩码位数（用于 SNAT）
  masqueradeBit: 14
conntrack:
  # 连接追踪最大数
  maxPerCore: 32768
  min: 131072
  # TCP 超时
  tcpEstablishedTimeout: 24h
  tcpCloseWaitTimeout: 1h

# Prometheus 指标
- kubeproxy_sync_proxy_rules_duration_seconds  # 同步耗时
- kubeproxy_sync_proxy_rules_last_timestamp    # 最后同步时间
- kubeproxy_network_programming_duration_seconds # 编程耗时
- kubeproxy_sync_proxy_rules_iptables_total    # iptables 规则数