Kube-Proxy IPVS模式源码分析

func main() {
  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()
}

// Run runs the specified ProxyServer.
func (o *Options) Run() error {
  defer close(o.errCh)
  
  //....
  proxyServer, err := NewProxyServer(o)
  if err != nil {
    return err
  }

  if o.CleanupAndExit {
    return proxyServer.CleanupAndExit()
  }

  o.proxyServer = proxyServer
  return o.runLoop()
}

type ProxyServer struct {
  Client                 clientset.Interface 
  EventClient            v1core.EventsGetter
  IptInterface           utiliptables.Interface
  IpvsInterface          utilipvs.Interface
  IpsetInterface         utilipset.Interface
  execer                 exec.Interface
  Proxier                proxy.ProxyProvider
  Broadcaster            record.EventBroadcaster
  Recorder               record.EventRecorder
  ConntrackConfiguration kubeproxyconfig.KubeProxyConntrackConfiguration
  Conntracker            Conntracker // if nil, ignored
  ProxyMode              string
  NodeRef                *v1.ObjectReference
  CleanupIPVS            bool
  MetricsBindAddress     string
  EnableProfiling        bool
  OOMScoreAdj            *int32
  ConfigSyncPeriod       time.Duration
  HealthzServer          *healthcheck.HealthzServer
}

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
}

type BoundedFrequencyRunner struct {
  name        string        // the name of this instance
  minInterval time.Duration // the min time between runs, modulo bursts
  maxInterval time.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
  lastRun time.Time   // time of last run
  timer   timer       // timer for deferred runs
  limiter rateLimiter // rate limiter for on-demand runs
}

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

type ServiceConfig struct {
  listerSynced  cache.InformerSynced
  eventHandlers []ServiceHandler
}

type ServiceHandler interface {
  // OnServiceAdd is called whenever creation of new service object
  // is observed.
  OnServiceAdd(service *v1.Service)
  // OnServiceUpdate is called whenever modification of an existing
  // service object is observed.
  OnServiceUpdate(oldService, service *v1.Service)
  // OnServiceDelete is called whenever deletion of an existing service
  // object is observed.
  OnServiceDelete(service *v1.Service)
  // OnServiceSynced is called once all the initial even handlers were
  // called and the state is fully propagated to local cache.
  OnServiceSynced()
}

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
}

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

// 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.syncRunner.Run()
  }
}

// ServiceChangeTracker carries state about uncommitted changes to an arbitrary number of
// Services, keyed by their namespace and name.
type ServiceChangeTracker struct {
  // lock protects items.
  lock sync.Mutex
  // items maps a service to its serviceChange.
  items map[types.NamespacedName]*serviceChange
  // makeServiceInfo allows proxier to inject customized information when processing service.
  makeServiceInfo makeServicePortFunc
  // isIPv6Mode indicates if change tracker is under IPv6/IPv4 mode. Nil means not applicable.
  isIPv6Mode *bool
  recorder   record.EventRecorder
}

// serviceChange contains all changes to services that happened since proxy rules were synced.  For a single object,
// changes are accumulated, i.e. previous is state from before applying the changes,
// current is state after applying all of the changes.
type serviceChange struct {
  previous ServiceMap
  current  ServiceMap
}

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

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

  // Build IPVS rules for each service.
  for svcName, svc := range proxier.serviceMap {
    //......

    // Handle traffic that loops back to the originator with SNAT.
    for _, e := range proxier.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())
    if valid := 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
    if svcInfo.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()
    if err := 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.
      if err := 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 := range svcInfo.ExternalIPStrings() {
      //....
    }

    // Capture load-balancer ingress.
    for _, ingress := range svcInfo.LoadBalancerIPStrings() {
      //.....
    }

    if svcInfo.NodePort() != 0 {
      //....
    }
  }

  // sync ipset entries
  for _, set := range proxier.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())

}