kube-proxy nftables 模式

# kube-proxy 配置
apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: nftables
nftables:
  # 同步周期
  syncPeriod: 30s
  # 最小同步间隔
  minSyncPeriod: 1s
  # 伪装位
  masqueradeBit: 14

// pkg/proxy/nftables/proxier.go

type Proxier struct {
    // nftables 接口
    nft knftables.Interface

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

    // 同步控制
    syncRunner   *async.BoundedFrequencyRunner
    syncPeriod   time.Duration

    // 节点信息
    nodeIP   net.IP
    hostname string

    // 规则构建
    staleChains map[string]bool
}

// syncProxyRules 同步 nftables 规则
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. 创建 nftables 事务
    tx := proxier.nft.NewTransaction()

    // 3. 确保基础结构存在
    proxier.ensureBaseRules(tx)

    // 4. 构建服务规则
    activeServices := sets.NewString()
    for svcName, svc := range proxier.svcPortMap {
        svcInfo := svc.(*servicePortInfo)
        proxier.writeServiceRules(tx, svcName, svcInfo, activeServices)
    }

    // 5. 清理过时规则
    proxier.cleanupStaleRules(tx, activeServices)

    // 6. 提交事务
    if err := proxier.nft.Run(context.TODO(), tx); err != nil {
        klog.ErrorS(err, "Failed to sync nftables rules")
    }
}

// writeServiceRules 写入服务规则
func (proxier *Proxier) writeServiceRules(
    tx *knftables.Transaction,
    svcName proxy.ServicePortName,
    svcInfo *servicePortInfo,
    activeServices sets.String) {

    // 获取端点
    endpoints := proxier.endpointsMap[svcName]
    nEndpoints := len(endpoints)

    if nEndpoints == 0 {
        // 无端点，跳过或添加拒绝规则
        return
    }

    // 服务链名称
    svcChain := svcInfo.chainName()
    activeServices.Insert(svcChain)

    // 创建服务链
    tx.Add(&knftables.Chain{
        Name: svcChain,
    })

    // 创建端点链
    epChains := make([]string, nEndpoints)
    for i, ep := range endpoints {
        epInfo := ep.(*endpointInfo)
        epChain := epInfo.chainName()
        epChains[i] = epChain
        activeServices.Insert(epChain)

        // 创建端点链
        tx.Add(&knftables.Chain{
            Name: epChain,
        })

        // 标记 MASQUERADE
        if proxier.needsMasquerade(svcInfo, epInfo) {
            tx.Add(&knftables.Rule{
                Chain: epChain,
                Rule:  "meta mark set mark or 0x4000",
            })
        }

        // DNAT 规则
        tx.Add(&knftables.Rule{
            Chain: epChain,
            Rule: fmt.Sprintf("dnat to %s:%d",
                epInfo.ip, epInfo.port),
        })
    }

    // 负载均衡规则
    if nEndpoints == 1 {
        tx.Add(&knftables.Rule{
            Chain: svcChain,
            Rule:  fmt.Sprintf("jump %s", epChains[0]),
        })
    } else {
        // 使用 numgen 实现随机负载均衡
        vmap := make([]string, nEndpoints)
        for i, chain := range epChains {
            vmap[i] = fmt.Sprintf("%d : goto %s", i, chain)
        }
        tx.Add(&knftables.Rule{
            Chain: svcChain,
            Rule: fmt.Sprintf("numgen random mod %d vmap { %s }",
                nEndpoints, strings.Join(vmap, ", ")),
        })
    }

    // ClusterIP 规则
    tx.Add(&knftables.Element{
        Set: "cluster-ips",
        Key: fmt.Sprintf("%s . %s . %d",
            svcInfo.ClusterIP(), svcInfo.Protocol(), svcInfo.Port()),
    })

    tx.Add(&knftables.Element{
        Map:   "service-endpoints",
        Key:   fmt.Sprintf("%s . %s . %d", svcInfo.ClusterIP(), svcInfo.Protocol(), svcInfo.Port()),
        Value: fmt.Sprintf("jump %s", svcChain),
    })
}

# Prometheus 指标
- kubeproxy_sync_proxy_rules_duration_seconds  # 同步耗时
- kubeproxy_sync_proxy_rules_last_timestamp    # 最后同步时间
- kubeproxy_nftables_sync_total               # nftables 同步次数
- kubeproxy_nftables_sync_errors_total        # 同步错误次数

# 1. 更新 kube-proxy 配置
apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: nftables

# 2. 滚动更新 kube-proxy
kubectl rollout restart ds/kube-proxy -n kube-system

# 3. 验证
kubectl get pods -n kube-system -l k8s-app=kube-proxy -o wide

apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: nftables
nftables:
  syncPeriod: 30s
  minSyncPeriod: 1s
conntrack:
  maxPerCore: 32768
  min: 131072

# 监控 nftables 规则数
- record: kube_proxy:nftables_rules_total
  expr: count(nft_chain_rules)

# 监控同步延迟
- alert: KubeProxyNftablesSyncSlow
  expr: kubeproxy_sync_proxy_rules_duration_seconds > 5
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "kube-proxy nftables sync is slow"