kube-proxy保姆级别源码阅读

// cmd\kube-proxy\proxy.go
func main() {
    // 创建command对象并执行
    command := app.NewProxyCommand()
    if err := command.Execute(); err != nil {
        os.Exit(1)
    }
}


// cmd\kube-proxy\app\server.go
func NewProxyCommand() *cobra.Command {
    // k8s每个组件都有类似的*options对象用来存储用户的配置
    opts := NewOptions()

    cmd := &cobra.Command{
        Run: func(cmd *cobra.Command, args []string) {
            // 如果是windows则配置系统
            if err := initForOS(opts.WindowsService); err != nil {
                klog.Fatalf("failed OS init: %v", err)
            }
            // 填充默认参数
            if err := opts.Complete(); err != nil {
                klog.Fatalf("failed complete: %v", err)
            }
            // 验证参数是否合法
            if err := opts.Validate(); err != nil {
                klog.Fatalf("failed validate: %v", err)
            }
            // 基于所给的参数运行
            if err := opts.Run(); err != nil {
                klog.Exit(err)
            }
        },
    }

    // 应用默认值和添加命令行参数
    var err error
    opts.config, err = opts.ApplyDefaults(opts.config)
    opts.AddFlags(cmd.Flags())
    cmd.MarkFlagFilename("config", "yaml", "yml", "json")
    return cmd
}


func (o *Options) Complete() error {
    // Load the config file here in Complete, so that Validate validates the fully-resolved config.
    if len(o.ConfigFile) > 0 {
        // 读取本地配置文件
        c, err := o.loadConfigFromFile(o.ConfigFile)
        o.config = c

        // 用来监听配置文件是否发生变化, 如果修改，重命名等情况就会触发一个error
        // 会导致kube-proxy退出,因为在pod里面，所以会导致重启
        if err := o.initWatcher(); err != nil {
            return err
        }
    }

    // 
    return utilfeature.DefaultMutableFeatureGate.SetFromMap(o.config.FeatureGates)
}

type Options struct {
    // kube-proxy配置文件位置， 如/var/lib/kube-proxy/config.conf
    ConfigFile string

    // 用来运行kube-proxy所需的配置参数
    config *kubeproxyconfig.KubeProxyConfiguration

    // 一个proxyServer对象 
    proxyServer proxyRun

    // kube-apiserver的地址
    master string
}

// 创建ProxyServer对象并运行循环
func (o *Options) Run() error {
    proxyServer, err := NewProxyServer(o)
    o.proxyServer = proxyServer
    return o.runLoop()
}


// 
func (o *Options) runLoop() error {
    // 启动文件监听器，监听配置文件的是否发生非预期的变化
    if o.watcher != nil {
        o.watcher.Run()
    }

    // proxyServer对象以一个额外的gorouting启动
    go func() {
        err := o.proxyServer.Run()
        o.errCh <- err
    }()

    // 进入死循环，直至发生错误, 才会退出
    for {
        err := <-o.errCh
        if err != nil {
            return err
        }
    }
}

// cmd\kube-proxy\app\server_others.go
func NewProxyServer(o *Options) (*ProxyServer, error) {
    return newProxyServer(o.config, o.CleanupAndExit, o.master)
}


func newProxyServer(
    config *proxyconfigapi.KubeProxyConfiguration,
    cleanupAndExit bool,
    master string) (*ProxyServer, error) {

    // /configz 用来检查运行时的配置
    if c, err := configz.New(proxyconfigapi.GroupName); err == nil {
        c.Set(config)
    }

    // 用来操作iptables和内核的接口
    var iptInterface utiliptables.Interface
    var kernelHandler ipvs.KernelHandler

    // 用来操作iptables的命令行接口
    // 即os/exec的封装，用来执行iptables等命令
    execer := exec.New()
    kernelHandler = ipvs.NewLinuxKernelHandler()
    // 创建可以跟k8s集群交互的client
    client, eventClient, err := createClients(config.ClientConnection, master)
    // 拿到运行节点的ip
    nodeIP := detectNodeIP(client, hostname, config.BindAddress)

    // 用来传播事件的对象, 即kubectl get events
    eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()})
    recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "kube-proxy")
    // 一个用来表示当前节点的引用对象
    nodeRef := &v1.ObjectReference{
        Kind:      "Node",
        Name:      hostname,
        UID:       types.UID(hostname),
        Namespace: "",
    }

    // proxier 代理模式的具体实现
    var proxier proxy.Provider
    var detectLocalMode proxyconfigapi.LocalMode

    // 得到代理模式 
    proxyMode := getProxyMode(string(config.Mode), canUseIPVS, iptables.LinuxKernelCompatTester{})
    // 判断是否检查本地流量，即同一节点的两个pod交互的流量
    detectLocalMode, err = getDetectLocalMode(config)

    // iptables有ipv4和ipv6协议, 默认是只有ipv4
    primaryProtocol := utiliptables.ProtocolIPv4
    iptInterface = utiliptables.New(execer, primaryProtocol)

    var ipt [2]utiliptables.Interface
    // 判断是否同时启用ipv4和ipv6, 默认只有ipv4
    dualStack := utilfeature.DefaultFeatureGate.Enabled(features.IPv6DualStack) && proxyMode != proxyModeUserspace
    if proxyMode == proxyModeIPTables {
        if dualStack {
           // 移除了创建双栈(ipv4+ipv6)的逻辑
        } else { // Create a single-stack proxier.
            // 探测本地流量应该为了TopologyAwareHints特性
            var localDetector proxyutiliptables.LocalTrafficDetector
            localDetector, err = getLocalDetector(detectLocalMode, config, iptInterface, nodeInfo)

            // TODO this has side effects that should only happen when Run() is invoked.
            proxier, err = iptables.NewProxier(
                // 操作iptables命令的接口, 创建/删除/确认等操作。
                iptInterface,
                // 用来操作/proc/sys内核参数，如内存分配策略vm/overcommit_memory
                utilsysctl.New(),
                execer,
                config.IPTables.SyncPeriod.Duration,
                config.IPTables.MinSyncPeriod.Duration,
                config.IPTables.MasqueradeAll,
                int(*config.IPTables.MasqueradeBit),
                localDetector,
                hostname,
                nodeIP,
                recorder,
                healthzServer,
                config.NodePortAddresses,
            )
        }

        proxymetrics.RegisterMetrics()
    }

    // 删除了ipvs相关代码

    useEndpointSlices := true
    return &ProxyServer{
        Client:                 client,
        EventClient:            eventClient,
        IptInterface:           iptInterface,
        IpvsInterface:          ipvsInterface,
        IpsetInterface:         ipsetInterface,
        execer:                 execer,
        Proxier:                proxier,
        Broadcaster:            eventBroadcaster,
        Recorder:               recorder,
        ConntrackConfiguration: config.Conntrack,
        // 用来操作/proc/sys/net/netfilter等参数
        // snat/dnat都需要内核跟踪建立的连接
        Conntracker:            &realConntracker{},
        ProxyMode:              proxyMode,
        NodeRef:                nodeRef,
        MetricsBindAddress:     config.MetricsBindAddress,
        BindAddressHardFail:    config.BindAddressHardFail,
        EnableProfiling:        config.EnableProfiling,
        OOMScoreAdj:            config.OOMScoreAdj,
        ConfigSyncPeriod:       config.ConfigSyncPeriod.Duration,
        HealthzServer:          healthzServer,
        UseEndpointSlices:      useEndpointSlices,
    }, nil
}

func (o *Options) Run() error {
    proxyServer, err := NewProxyServer(o)
    o.proxyServer = proxyServer
    return o.runLoop()
}

func (o *Options) runLoop() error {
    go func() {
        err := o.proxyServer.Run()
        o.errCh <- err
    }()

    for {
        err := <-o.errCh
        if err != nil {
            return err
        }
    }
}

// cmd\kube-proxy\app\server.go
func (s *ProxyServer) Run() error {
    // 用来跳转OOM参数，保证系统在内存紧张的时候不优先kill掉kube-proxy的进程
    var oomAdjuster *oom.OOMAdjuster
    if s.OOMScoreAdj != nil {
        oomAdjuster = oom.NewOOMAdjuster()
        if err := oomAdjuster.ApplyOOMScoreAdj(0, int(*s.OOMScoreAdj)); err != nil {
            klog.V(2).Info(err)
        }
    }

    // 根据kube-proxy命令行参数或配置文件跳转conntrack参数
    // 主要是tcp建立连接的相关参数，比如超时，存活检查(keepalive)等, 以及最大连接数等
    if s.Conntracker != nil {
        max, err := getConntrackMax(s.ConntrackConfiguration)
        if err != nil {
            return err
        }
        if max > 0 {
            err := s.Conntracker.SetMax(max
        }

        // TCP相关参数
        if s.ConntrackConfiguration.TCPEstablishedTimeout != nil && s.ConntrackConfiguration.TCPEstablishedTimeout.Duration > 0 {
            timeout := int(s.ConntrackConfiguration.TCPEstablishedTimeout.Duration / time.Second)
            if err := s.Conntracker.SetTCPEstablishedTimeout(timeout); err != nil {
                return err
            }
        }

        if s.ConntrackConfiguration.TCPCloseWaitTimeout != nil && s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration > 0 {
            timeout := int(s.ConntrackConfiguration.TCPCloseWaitTimeout.Duration / time.Second)
            if err := s.Conntracker.SetTCPCloseWaitTimeout(timeout); err != nil {
                return err
            }
        }
    }

    // 这里创建informer工厂函数，最后创建相应的informer用于监听service,endpointslice等资源

    // 这两个NewRequirement用来过滤掉serviceProxyName和noheadless的endpoint
    noProxyName, err := labels.NewRequirement(apis.LabelServiceProxyName, selection.DoesNotExist, nil)
    if err != nil {
        return err
    }
    noHeadlessEndpoints, err := labels.NewRequirement(v1.IsHeadlessService, selection.DoesNotExist, nil)
    if err != nil {
        return err
    }
    labelSelector := labels.NewSelector()
    labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints)
    informerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.ConfigSyncPeriod,
        informers.WithTweakListOptions(func(options *metav1.ListOptions) {
            options.LabelSelector = labelSelector.String()
        }))

    // 开始创建相应的informer并注册事件函数
    // 依次是service informer, endpointslieces informer
    serviceConfig := config.NewServiceConfig(informerFactory.Core().V1().Services(), s.ConfigSyncPeriod)
    serviceConfig.RegisterEventHandler(s.Proxier)
    go serviceConfig.Run(wait.NeverStop)

    if s.UseEndpointSlices {
        endpointSliceConfig := config.NewEndpointSliceConfig(informerFactory.Discovery().V1().EndpointSlices(), s.ConfigSyncPeriod)
        endpointSliceConfig.RegisterEventHandler(s.Proxier)
        go endpointSliceConfig.Run(wait.NeverStop)
    } else {
        endpointsConfig := config.NewEndpointsConfig(informerFactory.Core().V1().Endpoints(), s.ConfigSyncPeriod)
        endpointsConfig.RegisterEventHandler(s.Proxier)
        go endpointsConfig.Run(wait.NeverStop)
    }

    // 启动所有informer 
    informerFactory.Start(wait.NeverStop)

    // 判断是否启用TopologyAwareHints特性以创建node informer

    // 首次触发一次同步。
    s.birthCry()

    // 最后启动同步规则的循环
    go s.Proxier.SyncLoop()

    // 如果错误出现就退出
    return <-errCh
}

func (s *ProxyServer) birthCry() {
    s.Recorder.Eventf(s.NodeRef, nil, api.EventTypeNormal, "Starting", "StartKubeProxy", "")
}

// pkg\proxy\config\config.go
func NewServiceConfig(serviceInformer coreinformers.ServiceInformer, resyncPeriod time.Duration) *ServiceConfig {
    result := &ServiceConfig{
        listerSynced: serviceInformer.Informer().HasSynced,
    }
    // 创建informer并注册事件函数
    serviceInformer.Informer().AddEventHandlerWithResyncPeriod(
        cache.ResourceEventHandlerFuncs{
            AddFunc:    result.handleAddService,
            UpdateFunc: result.handleUpdateService,
            DeleteFunc: result.handleDeleteService,
        },
        resyncPeriod,
    )

    return result
}

// 将事件回调函数加入eventhandlers中，在每次触发事件的时候调用
func (c *ServiceConfig) RegisterEventHandler(handler ServiceHandler) {
    c.eventHandlers = append(c.eventHandlers, handler)
}

// 等待数据同步完成后，调用OnServiceSynced事件回调函数
func (c *ServiceConfig) Run(stopCh <-chan struct{}) {
    klog.Info("Starting service config controller")

    // 等待数据同步
    if !cache.WaitForNamedCacheSync("service config", stopCh, c.listerSynced) {
        return
    }

    // 调用proxier的OnServiceSynced方法
    for i := range c.eventHandlers {
        klog.V(3).Info("Calling handler.OnServiceSynced()")
        c.eventHandlers[i].OnServiceSynced()
    }
}

// pkg\proxy\iptables\proxier.go
func NewProxier(ipt utiliptables.Interface,
    sysctl utilsysctl.Interface,
    exec utilexec.Interface,
    syncPeriod time.Duration,
    minSyncPeriod time.Duration,
    masqueradeAll bool,
    masqueradeBit int,
    localDetector proxyutiliptables.LocalTrafficDetector,
    hostname string,
    nodeIP net.IP,
    recorder events.EventRecorder,
    healthzServer healthcheck.ProxierHealthUpdater,
    nodePortAddresses []string,
) (*Proxier, error) {
    // 正常情况下，内核不会对地址localnet(127.0.0.1/8)的地址做forwarding, 因为这部分代码被认为是martian.
    // 但是可以通过内核中配置来启用route_localnet
    if err := utilproxy.EnsureSysctl(sysctl, sysctlRouteLocalnet, 1); err != nil {
        return nil, err
    }

    // 确保bridge-nf-call-iptabels=1
    if val, err := sysctl.GetSysctl(sysctlBridgeCallIPTables); err == nil && val != 1 {
        klog.InfoS("Missing br-netfilter module or unset sysctl br-nf-call-iptables; proxy may not work as intended")
    }

    // 对snat数据流做标记
    masqueradeValue := 1 << uint(masqueradeBit)
    masqueradeMark := fmt.Sprintf("%#08x", masqueradeValue)
    klog.V(2).InfoS("Using iptables mark for masquerade", "ipFamily", ipt.Protocol(), "mark", masqueradeMark)

    serviceHealthServer := healthcheck.NewServiceHealthServer(hostname, recorder)

    ipFamily := v1.IPv4Protocol
    if ipt.IsIPv6() {
        ipFamily = v1.IPv6Protocol
    }

    ipFamilyMap := utilproxy.MapCIDRsByIPFamily(nodePortAddresses)
    nodePortAddresses = ipFamilyMap[ipFamily]
    // Log the IPs not matching the ipFamily
    if ips, ok := ipFamilyMap[utilproxy.OtherIPFamily(ipFamily)]; ok && len(ips) > 0 {
        klog.InfoS("Found node IPs of the wrong family", "ipFamily", ipFamily, "ips", strings.Join(ips, ","))
    }

    proxier := &Proxier{
        //各个参数..
    }

    // 瞬时并发数量
    burstSyncs := 2
    klog.V(2).InfoS("Iptables sync params", "ipFamily", ipt.Protocol(), "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs)
    // 创建一个syncRunner 对象，它会保证每个任务之间的时间间隔不大于minSyncPeriod
    // 并且最少maxInterval(这里默认是time.Hour, 一个小时)同步一次
    // 说明kube-proxy至少每个小时会触发一次同步
    // **但是同步不一定代表会刷新规则**
    // syncRunner会控制并发。
    proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, time.Hour, burstSyncs)

    // 通过创建一个KUBE-KUBELET-CANARY的链来检测iptables规则是否被刷掉(iptables flush)
    // 如果这个链不存在了，自然说明规则链被清理掉了。
    go ipt.Monitor(kubeProxyCanaryChain, []utiliptables.Table{utiliptables.TableMangle, utiliptables.TableNAT, utiliptables.TableFilter},
        proxier.syncProxyRules, syncPeriod, wait.NeverStop)

    return proxier, nil
}

func (s *ProxyServer) Run() error {
    // 创建informer等操作....

    s.birthCry()
    // 启动proxier的同步规则的循环
    go s.Proxier.SyncLoop()

    return <-errCh
}

// pkg\proxy\iptables\proxier.go 
func (proxier *Proxier) SyncLoop() {
    // 调用syncRunner
    proxier.syncRunner.Loop(wait.NeverStop)
}

// pkg\util\async\bounded_frequency_runner.go
func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) {
    klog.V(3).Infof("%s Loop running", bfr.name)
    // 重置定时器, 即bfr.timer.C()，下次的启动时间是当前时间加上maxInterval
    bfr.timer.Reset(bfr.maxInterval)
    for {
        select {
        case <-stop:
            bfr.stop()
            klog.V(3).Infof("%s Loop stopping", bfr.name)
            return
        case <-bfr.timer.C():
            bfr.tryRun()
        case <-bfr.run:
            bfr.tryRun()
        case <-bfr.retry:
            bfr.doRetry()
        }
    }
}

func (bfr *BoundedFrequencyRunner) tryRun() {
    bfr.mu.Lock()
    defer bfr.mu.Unlock()

    // 获取令牌控制并发。
    if bfr.limiter.TryAccept() {
        // 这里的fn就是proxier.syncProxyRules
        bfr.fn()
        bfr.lastRun = bfr.timer.Now()
        bfr.timer.Stop()
        bfr.timer.Reset(bfr.maxInterval)
        klog.V(3).Infof("%s: ran, next possible in %v, periodic in %v", bfr.name, bfr.minInterval, bfr.maxInterval)
        return
    }

}

func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
    proxier.OnServiceUpdate(nil, service)
} 

func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
    // proxier.isInitialized在informer首次同步完成之后才会返回true
    if proxier.serviceChanges.Update(oldService, service) && proxier.isInitialized() {
        proxier.Sync()
    }
}

// pkg\proxy\service.go
// 增加对象
// 传递参数 nil, service
// 删除对象
// 传递参数 service, nil
// 更新对象
// 传递参数 oldService, currentService
func (sct *ServiceChangeTracker) Update(previous, current *v1.Service) bool {
    svc := current
    if svc == nil {
        svc = previous
    }
    // 如果previous, current都是nil, 直接返回
    if svc == nil {
        return false
    }
    // 用来定位唯一的service, 在一个集群中namespace+servicename是唯一的
    namespacedName := types.NamespacedName{Namespace: svc.Namespace, Name: svc.Name}

    // 判断是否已经在变更中存在
    change, exists := sct.items[namespacedName]
    // 如果不存在，说明是一个新增操作
    if !exists {
        change = &serviceChange{}
        // 根据service对象创建serviceMap对象
        change.previous = sct.serviceToServiceMap(previous)
        sct.items[namespacedName] = change
    }

    change.current = sct.serviceToServiceMap(current)
    // 判断是否有变化，没变化就没必要加入到变更里面了
    if reflect.DeepEqual(change.previous, change.current) {
        delete(sct.items, namespacedName)
    } else {
        klog.V(2).Infof("Service %s updated: %d ports", namespacedName, len(change.current))
    }
    return len(sct.items) > 0
}

// pkg\proxy\iptables\proxier.go
// 根据changes与当前map来计算最终的代理规则
serviceUpdateResult := proxier.serviceMap.Update(proxier.serviceChanges)
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)


// pkg\proxy\service.go
func (sm ServiceMap) Update(changes *ServiceChangeTracker) (result UpdateServiceMapResult) {
    result.UDPStaleClusterIP = sets.NewString()
    // 应用changes
    sm.apply(changes, result.UDPStaleClusterIP)

    // 用来健康检查的端口
    result.HCServiceNodePorts = make(map[types.NamespacedName]uint16)
    for svcPortName, info := range sm {
        if info.HealthCheckNodePort() != 0 {
            result.HCServiceNodePorts[svcPortName.NamespacedName] = uint16(info.HealthCheckNodePort())
        }
    }
    return result
}

// 将changes里的数据合并到serviceMap里面, 然后将changes置为空
func (sm *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
    for _, change := range changes.items {
        // 合并, 过滤, 删除
        sm.merge(change.current)
        change.previous.filter(change.current)
        sm.unmerge(change.previous, UDPStaleClusterIP)
    }
    // 置为空
    changes.items = make(map[types.NamespacedName]*serviceChange)
    metrics.ServiceChangesPending.Set(0)
}

    // 初始化化空对象
    conntrackCleanupServiceIPs := serviceUpdateResult.UDPStaleClusterIP
    conntrackCleanupServiceNodePorts := sets.NewInt()
    // 基于这些差异结果来插入过时的数据，用于后续清理
    for _, svcPortName := range endpointUpdateResult.StaleServiceNames {
        if svcInfo, ok := proxier.serviceMap[svcPortName]; ok && svcInfo != nil && conntrack.IsClearConntrackNeeded(svcInfo.Protocol()) {
            conntrackCleanupServiceIPs.Insert(svcInfo.ClusterIP().String())
            for _, extIP := range svcInfo.ExternalIPStrings() {
                conntrackCleanupServiceIPs.Insert(extIP)
            }
            for _, lbIP := range svcInfo.LoadBalancerIPStrings() {
                conntrackCleanupServiceIPs.Insert(lbIP)
            }
            nodePort := svcInfo.NodePort()
            if svcInfo.Protocol() == v1.ProtocolUDP && nodePort != 0 {
                klog.V(2).Infof("Stale %s service NodePort %v -> %d", strings.ToLower(string(svcInfo.Protocol())), svcPortName, nodePort)
                conntrackCleanupServiceNodePorts.Insert(nodePort)
            }
        }
    }

// iptablesJumpChains是一个切片，包含各个表的各个链
for _, jump := range iptablesJumpChains {
    if _, err := proxier.iptables.EnsureChain(jump.table, jump.dstChain); err != nil {
        klog.ErrorS(err, "Failed to ensure chain exists", "table", jump.table, "chain", jump.dstChain)
        return
    }
    args := append(jump.extraArgs,
        "-m", "comment", "--comment", jump.comment,
        "-j", string(jump.dstChain),
    )
    if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, jump.table, jump.srcChain, args...); err != nil {
        klog.ErrorS(err, "Failed to ensure chain jumps", "table", jump.table, "srcChain", jump.srcChain, "dstChain", jump.dstChain)
        return
    }
}

// 确保KUBE-MARK-DROP规则链存在
for _, ch := range iptablesEnsureChains {
    if _, err := proxier.iptables.EnsureChain(ch.table, ch.chain); err != nil {
        klog.ErrorS(err, "Failed to ensure chain exists", "table", ch.table, "chain", ch.chain)
        return
    }
}

// 通过iptables-save -t nat/filter命令将相应的表的数据导出
// filter表数据导出
existingFilterChains := make(map[utiliptables.Chain][]byte)
proxier.existingFilterChainsData.Reset()
err := proxier.iptables.SaveInto(utiliptables.TableFilter, proxier.existingFilterChainsData)
if err != nil { // if we failed to get any rules
    klog.ErrorS(err, "Failed to execute iptables-save, syncing all rules")
} else {
    // 将导出数据中的规则链列表导出
    existingFilterChains = utiliptables.GetChainLines(utiliptables.TableFilter, proxier.existingFilterChainsData.Bytes())
}

// nat表数据导出
// 与filter表差不多 
existingNATChains := make(map[utiliptables.Chain][]byte)
proxier.iptablesData.Reset()
err = proxier.iptables.SaveInto(utiliptables.TableNAT, proxier.iptablesData)
if err != nil { // if we failed to get any rules
    klog.ErrorS(err, "Failed to execute iptables-save, syncing all rules")
} else { // otherwise parse the output
    existingNATChains = utiliptables.GetChainLines(utiliptables.TableNAT, proxier.iptablesData.Bytes())
}

{"KUBE-SVC-NPX46M4PTMTKRN6Y": ":KUBE-SVC-NPX46M4PTMTKRN6Y - [0:0]"}

proxier.filterChains.Reset()
proxier.filterRules.Reset()
proxier.natChains.Reset()
proxier.natRules.Reset()

// 写入表头
utilproxy.WriteLine(proxier.filterChains, "*filter")
utilproxy.WriteLine(proxier.natChains, "*nat")


// 在构造的文本中写入规则链和规则
for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeExternalServicesChain, kubeForwardChain, kubeNodePortsChain} {
    if chain, ok := existingFilterChains[chainName]; ok {
        utilproxy.WriteBytesLine(proxier.filterChains, chain)
    } else {
        utilproxy.WriteLine(proxier.filterChains, utiliptables.MakeChainLine(chainName))
    }
}
for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, KubeMarkMasqChain} {
    if chain, ok := existingNATChains[chainName]; ok {
        utilproxy.WriteBytesLine(proxier.natChains, chain)
    } else {
        utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(chainName))
    }
} 

// 插入SNAT规则
utilproxy.WriteLine(proxier.natRules, []string{
    "-A", string(kubePostroutingChain),
    "-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
    "-j", "RETURN",
}...)
// Clear the mark to avoid re-masquerading if the packet re-traverses the network stack.
utilproxy.WriteLine(proxier.natRules, []string{
    "-A", string(kubePostroutingChain),
    // XOR proxier.masqueradeMark to unset it
    "-j", "MARK", "--xor-mark", proxier.masqueradeMark,
}...)
masqRule := []string{
    "-A", string(kubePostroutingChain),
    "-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`,
    "-j", "MASQUERADE",
}
if proxier.iptables.HasRandomFully() {
    masqRule = append(masqRule, "--random-fully")
}
utilproxy.WriteLine(proxier.natRules, masqRule...) 
// 打标记
utilproxy.WriteLine(proxier.natRules, []string{
    "-A", string(KubeMarkMasqChain),
    "-j", "MARK", "--or-mark", proxier.masqueradeMark,
}...)

-A KUBE-POSTROUTING -m mark ! --mark 0x4000/0x4000 -j RETURN
-A KUBE-POSTROUTING -j MARK --set-xmark 0x4000/0x0
-A KUBE-POSTROUTING -m comment --comment "kubernetes service traffic requiring SNAT" -j MASQUER

// 还在使用的nat规则链，用来过滤过时的规则链
activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set
// 必要的数据结构
replacementPortsMap := map[utilnet.LocalPort]utilnet.Closeable{}
readyEndpointChains := make([]utiliptables.Chain, 0)
localEndpointChains := make([]utiliptables.Chain, 0)

// iptables规则参数，比如-m tcp之类，初始化长度为64,一是为了避免内存在分配
// 二是对于大多数情况足够了, 即使超过64也没关系，因为切片可以动态扩容
args := make([]string, 64)

// 计算所有服务的endpoint规则链的总数
proxier.endpointChainsNumber = 0
for svcName := range proxier.serviceMap {
    proxier.endpointChainsNumber += len(proxier.endpointsMap[svcName])
}

// 获取本地地址
localAddrSet := utilproxy.GetLocalAddrSet()
nodeAddresses, err := utilproxy.GetNodeAddresses(proxier.nodePortAddresses, proxier.networkInterfacer)

for svcName, svc := range proxier.serviceMap {
}

// 对象转换
svcInfo, ok := svc.(*serviceInfo)
protocol := strings.ToLower(string(svcInfo.Protocol()))
svcNameString := svcInfo.serviceNameString

// 根据serviceName到endpointsMap找打对应的endpoint
allEndpoints := proxier.endpointsMap[svcName]

// 这里的过滤是为了topology aware endpoint这个特性
allEndpoints = proxy.FilterEndpoints(allEndpoints, svcInfo, proxier.nodeLabels)

// Scan the endpoints list to see what we have. "hasEndpoints" will be true
// if there are any usable endpoints for this service anywhere in the cluster.
var hasEndpoints, hasLocalReadyEndpoints, hasLocalServingTerminatingEndpoints bool
for _, ep := range allEndpoints {
// 判断hasEndpoints, hasLocalReadyEndpoints,hasLocalServingTerminatingEndpoints 
}
useTerminatingEndpoints := !hasLocalReadyEndpoints && hasLocalServingTerminatingEndpoints

// Generate the per-endpoint chains.
readyEndpointChains = readyEndpointChains[:0]
localEndpointChains = localEndpointChains[:0]
for _, ep := range allEndpoints {
    epInfo, ok := ep.(*endpointsInfo)
    if !ok {
        klog.ErrorS(err, "Failed to cast endpointsInfo", "endpointsInfo", ep)
        continue
    }

    endpointChain := epInfo.endpointChain(svcNameString, protocol)
    endpointInUse := false


    // 检查是否已经存在endpoint链，否则就创建
    // endpoint链就是 KUBE-SEP-XXXX
    if chain, ok := existingNATChains[endpointChain]; ok {
        utilproxy.WriteBytesLine(proxier.natChains, chain)
    } else {
        utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(endpointChain))
    }
    // 表示是一个有效的链
    activeNATChains[endpointChain] = true

    args = append(args[:0], "-A", string(endpointChain))
    args = proxier.appendServiceCommentLocked(args, svcNameString)
    // 写入DNAT规则
    utilproxy.WriteLine(proxier.natRules, append(args,
        "-s", utilproxy.ToCIDR(net.ParseIP(epInfo.IP())),
        "-j", string(KubeMarkMasqChain))...)
    // Update client-affinity lists.
    if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
        args = append(args, "-m", "recent", "--name", string(endpointChain), "--set")
    }
    // DNAT to final destination.
    args = append(args, "-m", protocol, "-p", protocol, "-j", "DNAT", "--to-destination", epInfo.Endpoint)
    utilproxy.WriteLine(proxier.natRules, args...)
}

// 确保KUBE-SVC-XXX链存在, 不存在就创建
svcChain := svcInfo.servicePortChainName
if hasEndpoints {
    // Create the per-service chain, retaining counters if possible.
    if chain, ok := existingNATChains[svcChain]; ok {
        utilproxy.WriteBytesLine(proxier.natChains, chain)
    } else {
        utilproxy.WriteLine(proxier.natChains, utiliptables.MakeChainLine(svcChain))
    }
    activeNATChains[svcChain] = true
}



// Capture the clusterIP.
if hasEndpoints {
    args = append(args[:0],
        "-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcNameString),
        "-m", protocol, "-p", protocol,
        "-d", utilproxy.ToCIDR(svcInfo.ClusterIP()),
        "--dport", strconv.Itoa(svcInfo.Port()),
    )
    // 写入KUBE-SVC-XXX链的规则
    utilproxy.WriteRuleLine(proxier.natRules, string(kubeServicesChain), append(args, "-j", string(svcChain))...)
}

-A KUBE-SERVICES -d 10.152.183.1/32 -p tcp -m comment --comment "default/myservice cluster IP" -m tcp --dport 80 -j KUBE-SVC-NPX46M4PTMTKRN6Y
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -m statistic --mode random --probability 0.500000000 -j KUBE-SEP-72LVGSP46NP3XHTG
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -j KUBE-SEP-EDGGJ3GHDFLJOF2D
-A KUBE-SEP-72LVGSP46NP3XHTG -s 10.0.20.4/32 -m comment --comment "default/myservice" -j KUBE-MARK-MASQ
-A KUBE-SEP-72LVGSP46NP3XHTG -p tcp -m comment --comment "default/myservice" -m tcp -j DNAT --to-destination 10.0.20.4:80

for chain := range existingNATChains {
    if !activeNATChains[chain] {
        chainString := string(chain) 
        // 如果不是k8s创建的链就跳
        if !strings.HasPrefix(chainString, "KUBE-SVC-") && !strings.HasPrefix(chainString, "KUBE-SEP-") && !strings.HasPrefix(chainString, "KUBE-FW-") && !strings.HasPrefix(chainString, "KUBE-XLB-") {
            // Ignore chains that aren't ours.
            continue
        }
        // 删除链之前确保链存在
        // -X KUBE-SVC-XXXX, KUBE-SEP-XXXX
        utilproxy.WriteBytesLine(proxier.natChains, existingNATChains[chain])
        utilproxy.WriteLine(proxier.natRules, "-X", chainString)
    }
}

utilproxy.WriteLine(proxier.filterRules, "COMMIT")
utilproxy.WriteLine(proxier.natRules, "COMMIT")

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

err = proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
if err != nil {
    klog.ErrorS(err, "Failed to execute iptables-restore")
    return
}
success = true

iptables-restore --noflush --counters < xxxx

klog.V(4).InfoS("Deleting conntrack stale entries for Services", "ips", conntrackCleanupServiceIPs.UnsortedList())
for _, svcIP := range conntrackCleanupServiceIPs.UnsortedList() {
    if err := conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP); err != nil {
        klog.ErrorS(err, "Failed to delete stale service connections", "ip", svcIP)
    }
}
klog.V(4).InfoS("Deleting conntrack stale entries for Services", "nodeports", conntrackCleanupServiceNodePorts.UnsortedList())
for _, nodePort := range conntrackCleanupServiceNodePorts.UnsortedList() {
    err := conntrack.ClearEntriesForPort(proxier.exec, nodePort, isIPv6, v1.ProtocolUDP)
    if err != nil {
        klog.ErrorS(err, "Failed to clear udp conntrack", "port", nodePort)
    }
}

apiVersion: v1
kind: Service
metadata:
  name: myservice
spec:
  ports:
  - port: 80
    protocol: TCP
    targetPort: 80
  selector:
    app: myservice
  type: ClusterIP

Name:              myservice
Namespace:         default
Labels:            app=myservice
Annotations:       <none>
Selector:          <none>
Type:              ClusterIP
IP Family Policy:  SingleStack
IP Families:       IPv4
IP:                10.152.183.1
IPs:               10.152.183.1
Port:              http  80/TCP
TargetPort:        80/TCP
Endpoints:         10.0.20.4:80, 10.0.22.3:80
Session Affinity:  None
Events:            <none>

# nat表
-A PREROUTING -m comment --comment "kubernetes service portals" -j KUBE-SERVICES
-A KUBE-SERVICES -d 10.152.183.1/32 -p tcp -m comment --comment "default/myservice cluster IP" -m tcp --dport 80 -j KUBE-SVC-NPX46M4PTMTKRN6Y
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -m statistic --mode random --probability 0.500000000 -j KUBE-SEP-72LVGSP46NP3XHTG
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -j KUBE-SEP-EDGGJ3GHDFLJOF2D
# 注意只有源ip是10.0.20.4/32
# 这是为了解决自己访问自己的service, 如果不做特殊处理，那么会发生错误
-A KUBE-SEP-72LVGSP46NP3XHTG -s 10.0.20.4/32 -m comment --comment "default/myservice" -j KUBE-MARK-MASQ
-A KUBE-SEP-72LVGSP46NP3XHTG -p tcp -m comment --comment "default/myservice" -m tcp -j DNAT --to-destination 10.0.20.4:80

# KUBE-MARK-MASQ就是简单的打个标记
-A KUBE-MARK-MASQ -j MARK --set-xmark 0x4000/0x4000

-A POSTROUTING -m comment --comment "kubernetes postrouting rules" -j KUBE-POSTROUTING
-A KUBE-POSTROUTING -m mark ! --mark 0x4000/0x4000 -j RETURN
-A KUBE-POSTROUTING -j MARK --set-xmark 0x4000/0x0
-A KUBE-POSTROUTING -m comment --comment "kubernetes service traffic requiring SNAT" -j MASQUERADE

# nat表
-A PREROUTING -m comment --comment "kubernetes service portals" -j KUBE-SERVICES
-A KUBE-SERVICES -m comment --comment "kubernetes service nodeports; NOTE: this must be the last rule in this chain" -m addrtype --dst
-type LOCAL -j KUBE-NODEPORTS
-A KUBE-NODEPORTS -p tcp -m comment --comment "default/myservice" -m tcp --dport 80 -j KUBE-SVC-NPX46M4PTMTKRN6Y
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -m statistic --mode random --probability 0.500000000 -j KUBE-SEP-72LVGSP46NP3XHTG
-A KUBE-SVC-NPX46M4PTMTKRN6Y -m comment --comment "default/myservice" -j KUBE-SEP-EDGGJ3GHDFLJOF2D
# 注意只有源ip是10.0.20.4/32
-A KUBE-SEP-72LVGSP46NP3XHTG -s 10.0.20.4/32 -m comment --comment "default/myservice" -j KUBE-MARK-MASQ
-A KUBE-SEP-72LVGSP46NP3XHTG -p tcp -m comment --comment "default/myservice" -m tcp -j DNAT --to-destination 10.0.20.4:80