kube-proxy IPVS 模式

# kube-proxy 配置
apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: ipvs
ipvs:
  # 调度算法
  scheduler: rr
  # 同步周期
  syncPeriod: 30s
  # 最小同步间隔
  minSyncPeriod: 2s
  # 会话亲和超时（秒）
  tcpTimeout: 0s
  tcpFinTimeout: 0s
  udpTimeout: 0s

// pkg/proxy/ipvs/proxier.go

// syncService 同步 Service 到 IPVS
func (proxier *Proxier) syncService(
    svcName string,
    vs *utilipvs.VirtualServer,
    bindAddr bool,
    bindedAddresses sets.String) error {

    // 检查虚拟服务器是否存在
    appliedVirtualServer, err := proxier.ipvs.GetVirtualServer(vs)
    if err != nil {
        return err
    }

    if appliedVirtualServer == nil {
        // 创建新的虚拟服务器
        klog.V(3).InfoS("Adding new service", "service", svcName,
            "address", vs.Address, "protocol", vs.Protocol, "port", vs.Port)

        if err := proxier.ipvs.AddVirtualServer(vs); err != nil {
            return err
        }
    } else if !appliedVirtualServer.Equal(vs) {
        // 更新现有虚拟服务器
        klog.V(3).InfoS("Updating service", "service", svcName,
            "address", vs.Address, "protocol", vs.Protocol, "port", vs.Port)

        if err := proxier.ipvs.UpdateVirtualServer(vs); err != nil {
            return err
        }
    }

    // 绑定 IP 到 kube-ipvs0 接口
    if bindAddr {
        if !bindedAddresses.Has(vs.Address.String()) {
            if err := proxier.netlinkHandle.EnsureAddressBind(vs.Address.String(), DefaultDummyDevice); err != nil {
                return err
            }
        }
    }

    return nil
}

// pkg/util/ipvs/ipvs.go

// VirtualServer IPVS 虚拟服务器
type VirtualServer struct {
    Address   net.IP
    Protocol  string
    Port      uint16
    Scheduler string
    Flags     ServiceFlags
    Timeout   uint32
}

// RealServer IPVS 真实服务器
type RealServer struct {
    Address      net.IP
    Port         uint16
    Weight       int
    ActiveConn   int
    InactiveConn int
}

// pkg/proxy/ipvs/proxier.go

// createKubeIPVS0 创建 kube-ipvs0 接口
func createKubeIPVS0() error {
    // 使用 dummy 类型接口
    kl := &netlink.Dummy{
        LinkAttrs: netlink.LinkAttrs{
            Name: DefaultDummyDevice, // kube-ipvs0
        },
    }

    if err := netlink.LinkAdd(kl); err != nil {
        return err
    }

    return netlink.LinkSetUp(kl)
}

// bindServiceAddress 绑定 Service IP 到接口
func (proxier *Proxier) bindServiceAddress(ip net.IP) error {
    addr := &netlink.Addr{
        IPNet: &net.IPNet{
            IP:   ip,
            Mask: net.CIDRMask(32, 32), // /32 地址
        },
    }

    link, err := netlink.LinkByName(DefaultDummyDevice)
    if err != nil {
        return err
    }

    return netlink.AddrAdd(link, addr)
}

// pkg/proxy/ipvs/proxier.go

// createIPSet 创建 IPSet 集合
func (proxier *Proxier) createIPSet() {
    // ClusterIP 集合
    proxier.ipsetList[kubeClusterIPSet] = NewIPSet(proxier.ipset,
        kubeClusterIPSet, utilipset.HashIPPort, false, utilipset.ProtocolFamilyIPV4)

    // Loopback 集合
    proxier.ipsetList[kubeLoopBackIPSet] = NewIPSet(proxier.ipset,
        kubeLoopBackIPSet, utilipset.HashIPPortIP, false, utilipset.ProtocolFamilyIPV4)

    // NodePort 集合
    proxier.ipsetList[kubeNodePortSetTCP] = NewIPSet(proxier.ipset,
        kubeNodePortSetTCP, utilipset.BitmapPort, false, utilipset.ProtocolFamilyIPV4)
}

// syncIPSetEntries 同步 IPSet 条目
func (proxier *Proxier) syncIPSetEntries() {
    // 添加 ClusterIP
    for svcName, svc := range proxier.svcPortMap {
        entry := &utilipset.Entry{
            IP:       svc.ClusterIP().String(),
            Port:     svc.Port(),
            Protocol: string(svc.Protocol()),
            SetType:  utilipset.HashIPPort,
        }
        proxier.ipsetList[kubeClusterIPSet].activeEntries.Insert(entry.String())
    }

    // 同步到内核
    for _, set := range proxier.ipsetList {
        set.syncIPSetEntries()
    }
}

// pkg/proxy/ipvs/proxier.go

// syncEndpoint 同步端点到 IPVS RealServer
func (proxier *Proxier) syncEndpoint(
    svcName string,
    onlyNodeLocalEndpoints bool,
    vs *utilipvs.VirtualServer) error {

    // 获取当前 RealServer 列表
    curRealServers, err := proxier.ipvs.GetRealServers(vs)
    if err != nil {
        return err
    }
    curRSMap := make(map[string]*utilipvs.RealServer)
    for _, rs := range curRealServers {
        curRSMap[rs.String()] = rs
    }

    // 获取期望的端点列表
    endpoints := proxier.endpointsMap[svcName]
    newRSMap := make(map[string]*utilipvs.RealServer)

    for _, ep := range endpoints {
        epInfo := ep.(*endpointInfo)

        // 检查是否只使用本地端点
        if onlyNodeLocalEndpoints && !epInfo.IsLocal() {
            continue
        }

        rs := &utilipvs.RealServer{
            Address: net.ParseIP(epInfo.ip),
            Port:    uint16(epInfo.port),
            Weight:  1,
        }
        newRSMap[rs.String()] = rs
    }

    // 添加新的 RealServer
    for key, rs := range newRSMap {
        if _, exists := curRSMap[key]; !exists {
            klog.V(3).InfoS("Adding new real server", "server", rs.String())
            if err := proxier.ipvs.AddRealServer(vs, rs); err != nil {
                return err
            }
        }
    }

    // 删除不需要的 RealServer
    for key, rs := range curRSMap {
        if _, exists := newRSMap[key]; !exists {
            klog.V(3).InfoS("Deleting real server", "server", rs.String())
            if err := proxier.ipvs.DeleteRealServer(vs, rs); err != nil {
                return err
            }
        }
    }

    return nil
}

// 会话亲和性使用 IPVS 持久连接实现
// 配置示例: ipvsadm -A -t 10.96.0.100:80 -s rr -p 10800

func (proxier *Proxier) createVirtualServer(svc *servicePortInfo) *utilipvs.VirtualServer {
    vs := &utilipvs.VirtualServer{
        Address:   svc.ClusterIP(),
        Port:      uint16(svc.Port()),
        Protocol:  string(svc.Protocol()),
        Scheduler: proxier.ipvsScheduler,
    }

    // 会话亲和性
    if svc.SessionAffinityType() == v1.ServiceAffinityClientIP {
        vs.Flags |= utilipvs.FlagPersistent
        vs.Timeout = uint32(svc.StickyMaxAgeSeconds())
    }

    return vs
}

// IPVS 连接状态由内核管理
// kube-proxy 只负责配置虚拟服务器和真实服务器

// 连接超时配置
type IPVSTimeout struct {
    TCPTimeout     time.Duration
    TCPFinTimeout  time.Duration
    UDPTimeout     time.Duration
}

func (proxier *Proxier) syncIPVSTimeout() error {
    timeout := &utilipvs.IPVSTimeout{
        TCPTimeout:     proxier.ipvsConfig.TCPTimeout,
        TCPFinTimeout:  proxier.ipvsConfig.TCPFinTimeout,
        UDPTimeout:     proxier.ipvsConfig.UDPTimeout,
    }
    return proxier.ipvs.SetTimeout(timeout)
}

# kube-proxy IPVS 指标
- kubeproxy_sync_proxy_rules_ipvs_services_total  # IPVS 服务数
- kubeproxy_sync_proxy_rules_ipvs_endpoints_total # IPVS 端点数
- kubeproxy_ipvs_conn_total                       # IPVS 连接数

# 如果 IPVS 有问题，可以回退
apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: iptables  # 改回 iptables

apiVersion: kubeproxy.config.k8s.io/v1alpha1
kind: KubeProxyConfiguration
mode: ipvs
ipvs:
  # 大多数场景使用轮询
  scheduler: rr
  # 或使用最少连接
  # scheduler: lc
  syncPeriod: 30s
  minSyncPeriod: 2s
  # 会话亲和超时
  tcpTimeout: 0s
  tcpFinTimeout: 0s
  udpTimeout: 0s
  # 排除特定接口
  excludeCIDRs:
    - 10.0.0.0/8
conntrack:
  maxPerCore: 32768
  min: 131072