Kube-Proxy IPVS 模式源码分析

kube-proxy 整体逻辑结构

这张时序图描述了 kube-proxy 的整体逻辑结构，由于 kub-proxy 组件和其它的 kube-* 组件一样都是使用 pflag 和 cobra 库去构建命令行应用程序。所以先简单介绍下该包的基本使用方式:

 复制代码

funcmain() {
command:= &cobra.Command{
Use:"echo [string to echo]",
Short:"Echo anything to the screen",
Long: `echo is for echoing anything back.Echo works a lot like print, except it has a child command.`,
Args: cobra.MinimumNArgs(1),
Run:func(cmd *cobra.Command, args []string) {
fmt.Println("Print: "+ strings.Join(args," "))
},
}

command.Execute()
}

上面这段代码就是使用 cobra 包的一个最简单的例子，首先初始化 Command 结构，其中该结构中的 Run 就是最终要执行的真正逻辑。当初始化完成 Command 之后，通过 commnad.Execute 去启动应用程序。

现在看上面的图就能比较直观的理解程序的启动机制了，这张图的整体过程就是对 Commnad 结构中的 Run 进行核心逻辑实现。也就是说 kube-proxy 核心逻辑入口就是从这里开始 (Command.Run)。

在 Command.Run 中主要做了如下几件事, 看下面的代码:

 复制代码

// Run runs the specified ProxyServer.
func(o *Options)Run()error{
deferclose(o.errCh)

//....
proxyServer, err := NewProxyServer(o)
iferr !=nil{
returnerr
}

ifo.CleanupAndExit {
returnproxyServer.CleanupAndExit()
}

o.proxyServer = proxyServer
returno.runLoop()
}

1. 对 ProxyServer 实例进行初始化。

2. 如果在启动 kube-proxy 服务时，CleanupAndExit 参数设置为 true, 则会将 userspace, iptables, ipvs 三种模式之前设置的所有规则清除掉，然后直接退出。

3. 如果在启动 kube-proxy 服务时，CleanupAndExit 参数设置为 flase, 则会调用 runLoop 来启动 ProxyServer 服务。

首先先来看看 ProxyServer 的结构定义:

 复制代码

typeProxyServerstruct {
Clientclientset.Interface
EventClientv1core.EventsGetter
IptInterfaceutiliptables.Interface
IpvsInterfaceutilipvs.Interface
IpsetInterfaceutilipset.Interface
execer exec.Interface
Proxierproxy.ProxyProvider
Broadcasterrecord.EventBroadcaster
Recorderrecord.EventRecorder
ConntrackConfigurationkubeproxyconfig.KubeProxyConntrackConfiguration
ConntrackerConntracker// if nil, ignored
ProxyModestring
NodeRef*v1.ObjectReference
CleanupIPVSbool
MetricsBindAddressstring
EnableProfilingbool
OOMScoreAdj*int32
ConfigSyncPeriodtime.Duration
HealthzServer*healthcheck.HealthzServer
}

在 ProxyServer 结构中包含了与 kube-apiserver 通信的 Client、操作 Iptables 的 IptInterface、操作 IPVS 的 IpvsInterface、操作 IpSet 的 IpsetInterface，以及通过 ProxyMode 参数获取基于 userspace, iptables, ipvs 三种方式中的哪种使用的 Proxier。

接下来重点介绍基于 ipvs 模式实现的 Proxier, 在 ipvs 模式下 Proxier 结构的定义:

 复制代码

type Proxier struct {
endpointsChanges *proxy.EndpointChangeTracker
serviceChanges *proxy.ServiceChangeTracker

//...
serviceMap proxy.ServiceMap
endpointsMap proxy.EndpointsMap
portsMap map[utilproxy.LocalPort]utilproxy.Closeable

//...
syncRunner *async.BoundedFrequencyRunner// governs calls to syncProxyRules

//...
iptables utiliptables.Interface
ipvs utilipvs.Interface
ipset utilipset.Interface
exec utilexec.Interface

//...
ipvsScheduler string
}

在 Proxier 结构中，先介绍下 async.BoundedFrequencyRunner, 其它的在介绍 ProxyServer.Run 的时候介绍。

BoundedFrequencyRunner 的定义结构如下:

 复制代码

type BoundedFrequencyRunner struct {
name string// the name of this instance
minIntervaltime.Duration// the min time between runs, modulo bursts
maxIntervaltime.Duration// the max time between runs

run chan struct{}// try an async run

mu sync.Mutex// guards runs of fn and all mutations
fn func()// function to run
lastRuntime.Time// time of last run
timer timer// timer for deferred runs
limiter rateLimiter// rate limiter for on-demand runs
}

BoundedFrequencyRunner 结构中的 run 会异步的去定期的执行任务 fn，比如定期的执行 proxier.syncProxyRules 去创建或者更新 VirtuaServer 和 RealServer 并将 VirtualServer 的 VIP 绑定到 dummy interface(kube-ipvs0)。

下面是在 NewProxier 方法中初始化 BoundedFrequencyRunner 对象的示例:

 复制代码

proxier.syncRunner= async.NewBoundedFrequencyRunner(
"sync-runner", proxier.syncProxyRules,minSyncPeriod,syncPeriod,burstSyncs)

其中:

minSyncPeriod: 规则最小的更新时间

syncPeriod: 规则最大更新时间

proxier.syncProxyRules: 同步规则的实现函数 (也是 kube-proxy 基于 ipvs 同步规则的核心实现)

ProxyServer 启动流程

这部分介绍下 ProxyServer.Run 的逻辑实现，ProxyServer 启动流程如下图所示：

在启动过程中，主要做了下面这几件事情:

1. 启动健康检查服务 HealthzServer.
2. 启动暴露监控指标的 MetricsServer.
3. 如果需要调整系统的 conntrack 相关参数, 则对系统的 conntrack 进行参数调整.
4. 创建一个 informerFactory 实例，后面去通过 informerFactory 获取 kubernetes 的各类资源数据.
5. 创建一个 ServiceConfig 实例，这个实例主要作用是实时的 WATCH kubernetes Service 资源的变化，并加入队列中，用于后续对变化的 Service 进行规则同步。
6. 注册 servier event hander 到 Proxier.
7. 启动 serviceConfig.

接下来详细的介绍下 [4-7] 这几步的流程。

ServiceConfig 的结构定义如下:

 复制代码

typeServiceConfigstruct {
listerSynced cache.InformerSynced
eventHandlers []ServiceHandler
}

ServiceHandler 的结构定义如下:

 复制代码

typeServiceHandler interface {
// OnServiceAddiscalledwhenever creationofnewserviceobject
//isobserved.
OnServiceAdd(service *v1.Service)
// OnServiceUpdateiscalledwhenever modificationofan existing
// serviceobjectisobserved.
OnServiceUpdate(oldService, service *v1.Service)
// OnServiceDeleteiscalledwhenever deletionofan existing service
//objectisobserved.
OnServiceDelete(service *v1.Service)
// OnServiceSyncediscalledonceallthe initial even handlers were
//calledandthe stateisfully propagatedtolocalcache.
OnServiceSynced()
}

创建 ServiceConfig 实例对象的具体实现如下:

 复制代码

funcNewServiceConfig(serviceInformercoreinformers.ServiceInformer,resyncPeriodtime.Duration) *ServiceConfig{
result:= &ServiceConfig{
listerSynced: serviceInformer.Informer().HasSynced,
}

serviceInformer.Informer().AddEventHandlerWithResyncPeriod(
cache.ResourceEventHandlerFuncs{
AddFunc: result.handleAddService,
UpdateFunc: result.handleUpdateService,
DeleteFunc: result.handleDeleteService,
},
resyncPeriod,
)

returnresult
}

• 首先通过执行 serviceInformer.Informer().HasSynced 来将 kubernetes 下的所有 Service 资源同步到缓存 listerSynced 中。
• 其次为 AddEventHandlerWithResyncPeriod 添加针对 Service 对象，添加，更新，删除的事件触发函数。当 Service 有相应的触发动作，就会调用相应的函数：handleAddService、handleUpdateService 和 handleDeleteService。

我们看看 handleAddService 触发函数的实现逻辑，具体代码如下:

 复制代码

func(c *ServiceConfig)handleAddService(objinterface{}){
service, ok := obj.(*v1.Service)
if!ok {
utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
return
}
fori :=rangec.eventHandlers {
klog.V(4).Info("Calling handler.OnServiceAdd")
c.eventHandlers[i].OnServiceAdd(service)
}
}

当 watch 到 kubernetes 集群中有新的 Service 被创建之后，会触发 handleAddService 函数，并在该函数中遍历 eventHandlers 分别去调用 OnServiceAdd 来对 proxier 结构中的 serviceChanages 进行更新并去同步相应的规则。

OnServiceAdd 的具体实现逻辑如下:

 复制代码

// 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){
ifproxier.serviceChanges.Update(oldService,service)&&proxier.isInitialized(){
proxier.syncRunner.Run()
}
}

ServiceChangeTracker 的结构定义如下:

 复制代码

// ServiceChangeTracker carries state aboutuncommittedchangestoan arbitrary numberof
// Services, keyedbytheir namespaceandname.
typeServiceChangeTracker struct {
//lockprotects items.
locksync.Mutex
// items maps a servicetoits serviceChange.
items map[types.NamespacedName]*serviceChange
// makeServiceInfo allows proxiertoinject customized informationwhenprocessing service.
makeServiceInfo makeServicePortFunc
// isIPv6Mode indicatesifchange trackerisunder IPv6/IPv4 mode. Nil meansnotapplicable.
isIPv6Mode *bool
recorderrecord.EventRecorder
}

serviceChanage 的结构定义如下:

 复制代码

// serviceChange containsallchangestoservices that happened since proxy rules were synced. For a single object,
// changes are accumulated, i.e. previous isstatefrombefore applying the changes,
// current isstateafter applyingallof the changes.
type serviceChange struct {
previous ServiceMap
current ServiceMap
}

到这里在回过头来看上面的基于 IPVS 实现的 Proxier 的整体流程就完全通了，ProxyServer.Run 函数在启动时，通过 kubernetes LIST/WATCH 机制去实时的感知 kubernetes 集群 Service 资源的变化，然后不断的在更新 Proxier 结构中的 ServiceChanges，然后将变化的 Service 保存在 ServiceChanges 结构中的 ServiceMap 中，给后续的 async.BoundedFrequencyRunner 去执行同步规则函数 syncProxyRules 来使用。

8. endpointConfig 的实现机制和 serviceConfig 的机制完全一样，这里就不在详细的介绍了。

9. 上面做的所有预处理工作，会在 informerFactory.Start 这步生效。

10. birthCry 的作用就是通过 event 的方式通知 kubernetes, kube-proxy 这边的所有准备工作都处理好了，我要启动了。

 复制代码

s.Recorder.Eventf(s.NodeRef, api.EventTypeNormal, "Starting", "Starting kube-proxy.")
}

11. 最终通过 SyncLoop 启动 kube-proxy 服务，并立刻执行 syncProxyRules 先来一遍同步再说. 之后便会通过异步的方式定期的去同步 IPVS, Iptables, Ipset 的规则。

而 syncProxyRules 函数是 kube-proxy 实现的核心。主体逻辑是遍历 ServiceMap 并遍历 ServiceMap 下的 endpointsMap 及创建的 Service 类型 (如: CLusterIP, Loadbalancer, NodePort) 去分别创建相应的 IPVS 规则。

syncProxyRules 的函数实现定义如下:

 复制代码

func(proxier *Proxier)syncProxyRules(){
//.....

// Build IPVS rules for each service.
forsvcName, svc :=rangeproxier.serviceMap {
//......

// Handle traffic that loops back to the originator with SNAT.
for_, e :=rangeproxier.endpointsMap[svcName] {
//....
}

// Capture the clusterIP.
// ipset call
entry := &utilipset.Entry{
IP: svcInfo.ClusterIP().String(),
Port: svcInfo.Port(),
Protocol: protocol,
SetType: utilipset.HashIPPort,
}
// add service Cluster IP:Port to kubeServiceAccess ip set for the purpose of solving hairpin.
// proxier.kubeServiceAccessSet.activeEntries.Insert(entry.String())
ifvalid := proxier.ipsetList[kubeClusterIPSet].validateEntry(entry); !valid {
klog.Errorf("%s", fmt.Sprintf(EntryInvalidErr, entry, proxier.ipsetList[kubeClusterIPSet].Name))
continue
}
proxier.ipsetList[kubeClusterIPSet].activeEntries.Insert(entry.String())
// ipvs call
serv := &utilipvs.VirtualServer{
Address: svcInfo.ClusterIP(),
Port:uint16(svcInfo.Port()),
Protocol:string(svcInfo.Protocol()),
Scheduler: proxier.ipvsScheduler,
}
// Set session affinity flag and timeout for IPVS service
ifsvcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
serv.Flags |= utilipvs.FlagPersistent
serv.Timeout =uint32(svcInfo.StickyMaxAgeSeconds())
}
// We need to bind ClusterIP to dummy interface, so set `bindAddr` parameter to `true` in syncService()
iferr := proxier.syncService(svcNameString, serv,true); err ==nil{
activeIPVSServices[serv.String()] =true
activeBindAddrs[serv.Address.String()] =true
// ExternalTrafficPolicy only works for NodePort and external LB traffic, does not affect ClusterIP
// So we still need clusterIP rules in onlyNodeLocalEndpoints mode.
iferr := proxier.syncEndpoint(svcName,false, serv); err !=nil{
klog.Errorf("Failed to sync endpoint for service: %v, err: %v", serv, err)
}
}else{
klog.Errorf("Failed to sync service: %v, err: %v", serv, err)
}

// Capture externalIPs.
for_, externalIP :=rangesvcInfo.ExternalIPStrings() {
//....
}

// Capture load-balancer ingress.
for_, ingress :=rangesvcInfo.LoadBalancerIPStrings() {
//.....
}

ifsvcInfo.NodePort() !=0{
//....
}
}

// sync ipset entries
for_, set :=rangeproxier.ipsetList {
set.syncIPSetEntries()
}

// Tail call iptables rules for ipset, make sure only call iptables once
// in a single loop per ip set.
proxier.writeIptablesRules()

// Sync iptables rules.
//NOTE:NoFlushTables is used so we don't flush non-kubernetes chains in the table.
proxier.iptablesData.Reset()
proxier.iptablesData.Write(proxier.natChains.Bytes())
proxier.iptablesData.Write(proxier.natRules.Bytes())
proxier.iptablesData.Write(proxier.filterChains.Bytes())
proxier.iptablesData.Write(proxier.filterRules.Bytes())

}

总结

kube-proxy 的代码逻辑还是比较简洁的，整体的思想就是 kube-proxy 服务去 watch kubernetes 集群的 Service 和 Endpoint 对象，当这两个资源对象有状态变化时，会把它们保存在 ServiceMap 和 EndPonintMap 中，然后会通过 async.BoundedFrequencyRunner 去异步的执行 syncProxyRules 去下发规则。

本文转载自公众号 360 云计算（ID：hulktalk）。

原文链接：

https://mp.weixin.qq.com/s?__biz=MzU4ODgyMDI0Mg==&mid=2247486894&idx=1&sn=c39bafbcc79e6ea0a25fcb077a0b1128&chksm=fdd7b7d3caa03ec520bb4ef2ec98c498a1646e38f66d684b1124fe1aa2eb841bf4f2f080dec9&scene=27#wechat_redirect