3
源码分析 kubernetes kube-proxy 的设计实现
source link: https://xiaorui.cc/archives/7306
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
源码分析 kubernetes kube-proxy 的设计实现
kubernetes kube-proxy 的版本是 v1.27.0
cmd 入口
kube-proxy 启动的过程就不细说了，简单说就是解析传递配置，构建 ProxyServer 服务，最后启动 ProxyServer.
cmd/kube-proxy/proxy.go: main
cmd/kube-proxy/app/server.go: NewProxyCommand
cmd/kube-proxy/app/server.go: ProxyServer.Run()
JavaScript
ProxyServer
在创建 NewProxyServer() 时，根据 proxyMode 实例化不同的 proxier 对象. 在 kube-proxy 里 proxier 有iptables 和 ipvs 两种实现, 以前在 userspace 实现的代理已经在老版本中剔除了.
代码位置: cmd/kube-proxy/app/server.go
// NewProxyServer returns a new ProxyServer.
func NewProxyServer(o *Options) (*ProxyServer, error) {
return newProxyServer(o.config, o.master)
}
func newProxyServer(
config *proxyconfigapi.KubeProxyConfiguration,
master string) (*ProxyServer, error) {
// 实例化 iptables 的 proxy
if proxyMode == proxyconfigapi.ProxyModeIPTables {
if dualStack {
} else {
proxier, err = iptables.NewProxier(
iptInterface,
utilsysctl.New(),
execer,
config.IPTables.SyncPeriod.Duration,
config.IPTables.MinSyncPeriod.Duration,
...
)
}
// 实例化 ipvs 的 proxy
} else if proxyMode == proxyconfigapi.ProxyModeIPVS {
// 实例化 kernel, ipset，ipvs 管理工具
kernelHandler := ipvs.NewLinuxKernelHandler()
ipsetInterface = utilipset.New(execer)
ipvsInterface = utilipvs.New()
if dualStack {
} else {
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)
}
proxier, err = ipvs.NewProxier(
iptInterface,
ipvsInterface,
ipsetInterface,
utilsysctl.New(),
...
)
}
...
}
return &ProxyServer{
Client: client,
EventClient: eventClient,
IptInterface: iptInterface,
IpvsInterface: ipvsInterface,
IpsetInterface: ipsetInterface,
...
}, nil
}
ProxyServer 在调用 Run() 启动时会实例化 informer 监听器，然后一直监听 apiserver 反馈的 service, endpoints, node 变动事件.
func (s *ProxyServer) Run() error {
informerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
informers.WithTweakListOptions(func(options *metav1.ListOptions) {
options.LabelSelector = labelSelector.String()
}))
// 监听 service resource 资源
serviceConfig := config.NewServiceConfig(informerFactory.Core().V1().Services(), s.ConfigSyncPeriod)
serviceConfig.RegisterEventHandler(s.Proxier)
go serviceConfig.Run(wait.NeverStop)
// 监听 endpoints resource 资源
endpointSliceConfig := config.NewEndpointSliceConfig(informerFactory.Discovery().V1().EndpointSlices(), s.ConfigSyncPeriod)
endpointSliceConfig.RegisterEventHandler(s.Proxier)
go endpointSliceConfig.Run(wait.NeverStop)
informerFactory.Start(wait.NeverStop)
// 监听 node 资源更新
currentNodeInformerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
informers.WithTweakListOptions(func(options *metav1.ListOptions) {
options.FieldSelector = fields.OneTermEqualSelector("metadata.name", s.NodeRef.Name).String()
}))
nodeConfig := config.NewNodeConfig(currentNodeInformerFactory.Core().V1().Nodes(), s.ConfigSyncPeriod)
nodeConfig.RegisterEventHandler(s.Proxier)
go nodeConfig.Run(wait.NeverStop)
currentNodeInformerFactory.Start(wait.NeverStop)
s.birthCry()
go s.Proxier.SyncLoop()
return <-errCh
}
如何监听和处理事件 ?
kube-proxy 开启了对 services 和 endpoints 的事件监听，但篇幅原因这里就只聊 Services 的事件监听.
初始化阶段时会实例化 ServiceConfig 对象， 并向 serviceConfig 注册 proxier 对象. ServiceConfig 实现了对 informer 的监听，并向 informer 注册封装的回调接口. 当有 service 的增删改事件时, 调用 proxier 的 OnServiceAdd, OnServiceUpdate, OnServiceDelete 方法.
type ServiceConfig struct {
listerSynced cache.InformerSynced
eventHandlers []ServiceHandler
}
// NewServiceConfig creates a new ServiceConfig.
func NewServiceConfig(serviceInformer coreinformers.ServiceInformer, resyncPeriod time.Duration) *ServiceConfig {
result := &ServiceConfig{
listerSynced: serviceInformer.Informer().HasSynced,
}
serviceInformer.Informer().AddEventHandlerWithResyncPeriod(
cache.ResourceEventHandlerFuncs{
AddFunc: result.handleAddService,
UpdateFunc: result.handleUpdateService,
DeleteFunc: result.handleDeleteService,
},
resyncPeriod,
)
return result
}
// 注册事件处理，其实就是 proxier, 它实现了 ServiceHandler 接口.
func (c *ServiceConfig) RegisterEventHandler(handler ServiceHandler) {
c.eventHandlers = append(c.eventHandlers, handler)
}
func (c *ServiceConfig) Run(stopCh <-chan struct{}) {
// 第一次启动时, 同步下数据到缓存
if !cache.WaitForNamedCacheSync("service config", stopCh, c.listerSynced) {
return
}
// proxier 的 OnServiceSynced 实现是调用 `proxier.syncProxyRules()`
for i := range c.eventHandlers {
c.eventHandlers[i].OnServiceSynced()
}
}
// 处理 service 新增事件
func (c *ServiceConfig) handleAddService(obj interface{}) {
service, ok := obj.(*v1.Service)
for i := range c.eventHandlers {
c.eventHandlers[i].OnServiceAdd(service)
}
}
// 处理 service 更新事件
func (c *ServiceConfig) handleUpdateService(oldObj, newObj interface{}) {
oldService, ok := oldObj.(*v1.Service)
for i := range c.eventHandlers {
c.eventHandlers[i].OnServiceUpdate(oldService, service)
}
}
// 处理 service 删除事件
func (c *ServiceConfig) handleDeleteService(obj interface{}) {
service, ok := obj.(*v1.Service)
for i := range c.eventHandlers {
c.eventHandlers[i].OnServiceDelete(service)
}
}
从 ipvs 的 proxier 里, 分析下 OnServiceAdd, OnServiceUpdate, OnServiceDelete 这几个方法的实现. 其实就是对 proxier 的 serviceChanges 做存储变动对象操作. 需要注意的是，真正处理变动的在 proxier.syncProxyRules() 方法里.
// OnServiceAdd is called whenever creation of new service object is observed.
func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
proxier.OnServiceUpdate(nil, service)
}
// OnServiceUpdate is called whenever modification of an existing service object is observed.
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
if proxier.serviceChanges.Update(oldService, service) && proxier.isInitialized() {
proxier.Sync()
}
}
// OnServiceDelete is called whenever deletion of an existing service object is observed.
func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
proxier.OnServiceUpdate(service, nil)
}
真正干活 Proxier 的实现
在阅读了 kube-proxy 源码后可以发现，Proxier 才是真正干活的类，kube-proxy 里对于 ipvs，iptables, ipset 等工具的操作都在 proxier 里.
在 kube-proxy 中 proxier 有 iptables 和 ipvs 两种实现, 现在已经少有用 iptables 的了，所以就只分析 ipvs proxier 实现.
在实例化 Proxier 对象时会注册一个定时任务，它会周期性调用 syncProxyRules 方法.
func NewProxier(
syncPeriod time.Duration,
minSyncPeriod time.Duration,
masqueradeAll bool,
...
) (*Proxier, error) {
...
proxier := &Proxier{
svcPortMap: make(proxy.ServicePortMap),
endpointsMap: make(proxy.EndpointsMap),
...
}
...
proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
return proxier, nil
}
syncProxyRules 函数是 kube-proxy 实现的核心. 该主要逻辑每次都从头开始遍历 servcie 和 endpoints 所有对象，对比当前配置来进行增删改 ipvs virtualserver 和 realserver, 及绑定解绑 dummy ip地址，还有配置 ipset 和 iptables 的相关策略. 原理大概是这样，下面看细节实现.
代码位置: https://github.com/kubernetes/kubernetes/blob/master/pkg/proxy/ipvs/proxier.go
func (proxier *Proxier) syncProxyRules() {
proxier.mu.Lock()
defer proxier.mu.Unlock()
// 把从 informer 拿到变更的 servcie 结构，更新到 svcPortMap 里, 每次 Update 还会把 changes 清空.
serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
// 把从 informer 拿到变更的 endpoint 结构，更新到 endpointsMap 里, 每次 Update 还会把 changes 清空.
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
// 创建名为 `kube-ipvs0` 的 dummy 类型网络设备
_, err := proxier.netlinkHandle.EnsureDummyDevice(defaultDummyDevice)
if err != nil {
return
}
// 创建 ipset 规则，调用 ipset create 创建 ipset 时，还需指定 size 和 entry 的格式.
for _, set := range proxier.ipsetList {
if err := ensureIPSet(set); err != nil {
return
}
set.resetEntries()
}
// 遍历当前的 services 对象集合
for svcPortName, svcPort := range proxier.svcPortMap {
// 遍历 endpoints 对象
for _, e := range proxier.endpointsMap[svcPortName] {
// 拿到 endpoints 对象
ep, ok := e.(*proxy.BaseEndpointInfo)
epIP := ep.IP()
epPort, err := ep.Port()
// 定义 endpoint 的 ipset entry 结构
entry := &utilipset.Entry{
IP: epIP,
Port: epPort,
Protocol: protocol,
IP2: epIP,
SetType: utilipset.HashIPPortIP,
}
// 在 kubeLoopBackIPSet 配置集合中加入 entry 配置.
proxier.ipsetList[kubeLoopBackIPSet].activeEntries.Insert(entry.String())
}
// 定义 service 的 ipset entry 结构
entry := &utilipset.Entry{
IP: svcInfo.ClusterIP().String(),
Port: svcInfo.Port(),
}
// 把 service ipset entry 加入到 kubeClusterIPSet 配置集合中
proxier.ipsetList[kubeClusterIPSet].activeEntries.Insert(entry.String())
serv := &utilipvs.VirtualServer{
Address: svcInfo.ClusterIP(),
Port: uint16(svcInfo.Port()),
}
// 创建或更新 lvs virtualServer 配置
if err := proxier.syncService(svcPortNameString, serv, true, bindedAddresses); err == nil {
// 创建或更新 lvs realserver 配置
if err := proxier.syncEndpoint(svcPortName, internalNodeLocal, serv); err != nil {}
}
}
// 篇幅不提, 解决 external ip.
for _, externalIP := range svcInfo.ExternalIPStrings() {
...
}
// 篇幅不提，解决 load balancer
for _, ingress := range svcInfo.LoadBalancerIPStrings() {
...
}
// 同步 ipset 配置.
for _, set := range proxier.ipsetList {
set.syncIPSetEntries()
}
// 同步 iptables 的配置.
proxier.writeIptablesRules()
proxier.iptablesData.Reset()
proxier.iptablesData.Write(proxier.natChains.Bytes())
proxier.iptablesData.Write(proxier.natRules.Bytes())
err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
...
// 清理绑定的ip地址
legacyBindAddrs := proxier.getLegacyBindAddr(activeBindAddrs, currentBindAddrs)
// 清理需要删除 service, 逻辑里含有 ipvs vs 和 rs 的清理.
proxier.cleanLegacyService(activeIPVSServices, currentIPVSServices, legacyBindAddrs)
// 遍历不新鲜的 servcies 集合，通过 contrack tool 工具剔除在 contrack 里协议为 UDP 的旧连接.
for _, svcIP := range staleServices.UnsortedList() {
if err := conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP); err != nil {
klog.ErrorS(err, "Failed to delete stale service IP connections", "IP", svcIP)
}
}
}
同步 ipset 配置 ? (syncIPSetEntries)
syncIPSetEntries() 用来同步刷新 ipset 配置，就是把 ip 地址通过 ipset 命令更新到对应的 ipset 集合里.
代码位置: pkg/proxy/ipvs/ipset.go
func (set *IPSet) syncIPSetEntries() {
// 从本机获取 name 相关所有的 ipset entry
appliedEntries, err := set.handle.ListEntries(set.Name)
if err != nil {
return
}
// 放到一个 string set 集合里，为后面做差集做准备
currentIPSetEntries := sets.NewString()
for _, appliedEntry := range appliedEntries {
currentIPSetEntries.Insert(appliedEntry)
}
// 如果不相等的话, 进行同步操作.
if !set.activeEntries.Equal(currentIPSetEntries) {
// 删除遗留的 ipset 元素. 也就是删除本机有，但在激活集合中不存在的 ipset entry.
for _, entry := range currentIPSetEntries.Difference(set.activeEntries).List() {
if err := set.handle.DelEntry(entry, set.Name); err != nil {
...
}
}
// 添加 ipset 元素. 添加本机没有但在激活集合中存在的 ipset entry.
for _, entry := range set.activeEntries.Difference(currentIPSetEntries).List() {
if err := set.handle.AddEntry(entry, &set.IPSet, true); err != nil {
}
}
}
}
ipset 的最终是依赖 exec 去调用 ipset 二进制执行文件实现的. 增删改 ipset 集合时先是拼凑 ipset 命令的参数，然后在传递给 ipset 命令去执行.
代码位置: pkg/util/ipset/ipset.go
const IPSetCmd = "ipset"
type IPSet struct {
Name string
SetType Type
HashSize int
MaxElem int
...
}
func (runner *runner) AddEntry(entry string, set *IPSet, ignoreExistErr bool) error {
args := []string{"add", set.Name, entry}
if ignoreExistErr {
args = append(args, "-exist")
}
if out, err := runner.exec.Command(IPSetCmd, args...).CombinedOutput(); err != nil {
return fmt.Errorf("error adding entry %s, error: %v (%s)", entry, err, out)
}
return nil
}
func (runner *runner) DelEntry(entry string, set string) error {
if out, err := runner.exec.Command(IPSetCmd, "del", set, entry).CombinedOutput(); err != nil {
return fmt.Errorf("error deleting entry %s: from set: %s, error: %v (%s)", entry, set, err, out)
}
return nil
}
同步 ipvs service 配置 ? (syncService)
如果 ipvs vs 在当前节点配置过，且配置没有变动，另外 service ip 已绑定到 dummy 设备上，就一路绿灯无需真实去操作.
如果当前节点不存在该 ipvs VirtualServer, 则需要进行创建，如果已经存在，但配置不同，则进行更新 VirtualServer，另外还需要在本地 kube-ipvs0 设备上绑定 service clusterIP 地址.
k8s 的 ipvs 是通过 moby/ipvs 库实现增删改查操作, 原理是 moby/ipvs 里使用 netlink 实现用户态进程跟内核态进行通信.
func (proxier *Proxier) syncService(svcName string, vs *utilipvs.VirtualServer, bindAddr bool, bindedAddresses sets.String) error {
// 从当前节点里获取该 ipvs vs 配置
appliedVirtualServer, _ := proxier.ipvs.GetVirtualServer(vs)
// 如果当前节点不存在该 ipvs vs 配置，或者 有配置但配置发生了变更.
if appliedVirtualServer == nil || !appliedVirtualServer.Equal(vs) {
// 当前节点不存在该 ipvs vs, 则进行创建
if appliedVirtualServer == nil {
klog.V(3).InfoS("Adding new service", "serviceName", svcName, "virtualServer", vs)
if err := proxier.ipvs.AddVirtualServer(vs); err != nil {
return err
}
} else {
// 如果配置发生变更，则进行 update 更新.
klog.V(3).InfoS("IPVS service was changed", "serviceName", svcName)
if err := proxier.ipvs.UpdateVirtualServer(vs); err != nil {
return err
}
}
}
// 把 service ip 绑定到 dummy 设备上，也就是 kube-ipvs0 设备.
if bindAddr {
// 如果已经绑定了，那就无需再绑定.
if bindedAddresses != nil && bindedAddresses.Has(vs.Address.String()) {
return nil
}
// 绑定 ip 地址, 重复执行也不会返回 error，可理解为幂等操作.
_, err := proxier.netlinkHandle.EnsureAddressBind(vs.Address.String(), defaultDummyDevice)
if err != nil {
return err
}
}
return nil
}
如何删除 ipvs service ? (cleanLegacyService)
对比 active 和主机的 ipvs vs 的配置，如果主机存在某 service 配置，但 active 里没有，则删除并进行解绑 ip.
func (proxier *Proxier) cleanLegacyService(activeServices map[string]bool, currentServices map[string]*utilipvs.VirtualServer, legacyBindAddrs map[string]bool) {
isIPv6 := netutils.IsIPv6(proxier.nodeIP)
// 使用当前的 services 进行遍历
for cs := range currentServices {
svc := currentServices[cs]
...
// 如果激活集合里存在，但当前集合不存在，说明需要删除该 service.
if _, ok := activeServices[cs]; !ok {
klog.V(4).InfoS("Delete service", "virtualServer", svc)
// 调用 ipvs 删除 svc 相关的负载策略
if err := proxier.ipvs.DeleteVirtualServer(svc); err != nil {
klog.ErrorS(err, "Failed to delete service", "virtualServer", svc)
}
addr := svc.Address.String()
if _, ok := legacyBindAddrs[addr]; ok {
klog.V(4).InfoS("Unbinding address", "address", addr)
// 解绑 cluster ip 地址
proxier.netlinkHandle.UnbindAddress(addr, defaultDummyDevice)
delete(legacyBindAddrs, addr)
}
}
}
}
同步 ipvs endpoints 配置 (syncEndpoint)?
对比当前 virtualServer 的 realserver 集合，如果有新增变动，则调用 ipvs.AddRealServer, 有变动则使用 UpdateRealServer 更改.
func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNodeLocalEndpoints bool, vs *utilipvs.VirtualServer) error {
// 获取当前主机的 ipvs vs 配置
appliedVirtualServer, err := proxier.ipvs.GetVirtualServer(vs)
if err != nil {
klog.ErrorS(err, "Failed to get IPVS service")
return err
}
curEndpoints := sets.NewString()
curDests, err := proxier.ipvs.GetRealServers(appliedVirtualServer)
if err != nil {
return err
}
// 把 realserver 放到一个 set string 集合里
for _, des := range curDests {
curEndpoints.Insert(des.String())
}
endpoints := proxier.endpointsMap[svcPortName]
newEndpoints := sets.NewString()
for _, epInfo := range endpoints {
newEndpoints.Insert(epInfo.String())
}
// 添加 realserver
for _, ep := range newEndpoints.List() {
ip, port, err := net.SplitHostPort(ep)
portNum, err := strconv.Atoi(port)
newDest := &utilipvs.RealServer{
Address: netutils.ParseIPSloppy(ip),
Port: uint16(portNum),
Weight: 1,
}
// 更新 realServer
if curEndpoints.Has(ep) {
if proxier.initialSync {
for _, dest := range curDests {
// 如果新旧的权重不同, 则调用 ipvs 接口进行变更
if dest.Weight != newDest.Weight {
err = proxier.ipvs.UpdateRealServer(appliedVirtualServer, newDest)
if err != nil {
continue
}
}
}
}
...
}
// 为 vs 添加新的 realServer
err = proxier.ipvs.AddRealServer(appliedVirtualServer, newDest)
if err != nil {
klog.ErrorS(err, "Failed to add destination", "newDest", newDest)
continue
}
}
...
return nil
}
在 conntrack 里剔除 UDP 旧连接 (ClearEntriesForIP) ?
kube-proxy 在删除 UDP 的 service 时候, 也需要清除这些 ip_conntrack 连接，否则后面的流量还是会被导入的废弃的 pod 上面. 为什么 TCP 不需要在 conntrack 里剔除，而 UDP 需要 ? 这是因为 TCP 本是有状态的连接，当把流量切到其他 pod 时，旧连接会因为对端在关闭时进行挥手 fin，或者不可达时，自然就会开辟新连接，新连接自然使用新的策略. UDP 由于是无状态的连接，所以不受影响, 虽然数据不可达，但照旧能用，所以需要剔除旧的 UDP conntrack 状态.
剔除连接的原理很简单，就是调用 conntrack-tools 来剔除连接，我们也可以使用 conntrack 工具来手动剔除.
比如 conntrack -D -s 114.114.114.114 -p UDP.
func ClearEntriesForIP(execer exec.Interface, ip string, protocol v1.Protocol) error {
parameters := parametersWithFamily(utilnet.IsIPv6String(ip), "-D", "--orig-dst", ip, "-p", protoStr(protocol))
err := Exec(execer, parameters...)
if err != nil && !strings.Contains(err.Error(), NoConnectionToDelete) {
return fmt.Errorf("error deleting connection tracking state for UDP service IP: %s, error: %v", ip, err)
}
return nil
}
func Exec(execer exec.Interface, parameters ...string) error {
conntrackPath, err := execer.LookPath("conntrack")
if err != nil {
return fmt.Errorf("error looking for path of conntrack: %v", err)
}
output, err := execer.Command(conntrackPath, parameters...).CombinedOutput()
return nil
}
ipvs 里 iptables 相关配置
ipvs 里主要的 iptables 的实现都在 proxier.writeIptablesRules() 方法里，代码本不复杂，就是一些 iptables 的规则拼凑，所以不从源码来解析，换个角度直接通过生成的 iptabels 和 chains 流程走向来分析.
另外, kubernetes 在 ipvs 的 README.md 里有详细的 ipvs 和 iptables 的配置, 这里只说 --cluster-cidr=<cidr>, 暂时不深入 --masquerade-all=true 的配置.
代码位置: https://github.com/kubernetes/kubernetes/blob/master/pkg/proxy/ipvs/README.md
iptables chains 流程走向
iptables 配置如下:
# iptables -t nat -nL
Chain PREROUTING (policy ACCEPT)
target prot opt source destination
KUBE-SERVICES all -- 0.0.0.0/0 0.0.0.0/0 /* kubernetes service portals */
Chain OUTPUT (policy ACCEPT)
target prot opt source destination
KUBE-SERVICES all -- 0.0.0.0/0 0.0.0.0/0 /* kubernetes service portals */
Chain POSTROUTING (policy ACCEPT)
target prot opt source destination
KUBE-POSTROUTING all -- 0.0.0.0/0 0.0.0.0/0 /* kubernetes postrouting rules */
Chain KUBE-MARK-MASQ (3 references)
target prot opt source destination
MARK all -- 0.0.0.0/0 0.0.0.0/0 MARK or 0x4000
Chain KUBE-POSTROUTING (1 references)
target prot opt source destination
MASQUERADE all -- 0.0.0.0/0 0.0.0.0/0 /* kubernetes service traffic requiring SNAT */ mark match 0x4000/0x4000
MASQUERADE all -- 0.0.0.0/0 0.0.0.0/0 match-set KUBE-LOOP-BACK dst,dst,src
Chain KUBE-SERVICES (2 references)
target prot opt source destination
KUBE-MARK-MASQ all -- !10.244.16.0/24 0.0.0.0/0 match-set KUBE-CLUSTER-IP dst,dst
ACCEPT all -- 0.0.0.0/0 0.0.0.0/0 match-set KUBE-CLUSTER-IP dst,dst
JavaScript
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK