【kubernetes/k8s源码分析】ingress-nginx 源码解析

比眉伴天荒 2022-05-20 04:34 249阅读 0赞

github地址： https://github.com/kubernetes/ingress-nginx

usr/bin/dumb-init /nginx-ingress-controller --default-backend-service=kube-system/default-http-backend --configmap=kube-system/nginx-configuration --tcp-services-configmap=kube-system/tcp-services --udp-services-configmap=kube-system/udp-services --annotations-prefix=nginx.ingress.kubernetes.io

/nginx-ingress-controller --default-backend-service=kube-system/default-http-backend --configmap=kube-system/nginx-configuration --tcp-services-configmap=kube-system/tcp-services --udp-services-configmap=kube-system/udp-services --annotations-prefix=nginx.ingress.kubernetes.io

# 一 ingress启动流程 #

实例化一个IngressContrller并将其Start

1 main函数

## ##

##  ##

*  创建kubernetes客户端
 *  创建NewNGINXController
 *  启动服务start

func main() {
       kubeClient, err := createApiserverClient(conf.APIServerHost, conf.KubeConfigFile)
       if err != nil {
          handleFatalInitError(err)
       }
    
       ngx := controller.NewNGINXController(conf, fs)
    
       go handleSigterm(ngx, func(code int) {
          os.Exit(code)
       })
    
       mux := http.NewServeMux()
       go registerHandlers(conf.EnableProfiling, conf.ListenPorts.Health, ngx, mux)
    
       err = collector.InitNGINXStatusCollector(conf.Namespace, class.IngressClass, conf.ListenPorts.Status)
    
       if err != nil {
          glog.Fatalf("Error creating metric collector:  %v", err)
       }
    
       err = collector.NewInstance(conf.Namespace, class.IngressClass)
       if err != nil {
          glog.Fatalf("Error creating unix socket server:  %v", err)
       }
    
       ngx.Start()
    }

2 createApiserverClient函数

## ##

# #

*  连接到kubernetes，目的是list-watch

func createApiserverClient(apiserverHost, kubeConfig string) (*kubernetes.Clientset, error) {
       cfg, err := clientcmd.BuildConfigFromFlags(apiserverHost, kubeConfig)
       if err != nil {
          return nil, err
       }
    
       cfg.QPS = defaultQPS
       cfg.Burst = defaultBurst
       cfg.ContentType = "application/vnd.kubernetes.protobuf"
    
       client, err := kubernetes.NewForConfig(cfg)
       if err != nil {
          return nil, err
       }
    
       var v *discovery.Info
    
       // The client may fail to connect to the API server in the first request.
       // https://github.com/kubernetes/ingress-nginx/issues/1968
       defaultRetry := wait.Backoff{
          Steps:    10,
          Duration: 1 * time.Second,
          Factor:   1.5,
          Jitter:   0.1,
       }
    
       var lastErr error
       retries := 0
       glog.V(2).Info("Trying to discover Kubernetes version")
       err = wait.ExponentialBackoff(defaultRetry, func() (bool, error) {
          v, err = client.Discovery().ServerVersion()
    
          if err == nil {
             return true, nil
          }
    
          lastErr = err
          glog.V(2).Infof("Unexpected error discovering Kubernetes version (attempt %v): %v", err, retries)
          retries++
          return false, nil
       })
    
       return client, nil
    }

3 handleSigterm函数

## ##

# #

*  定义信号处理函数

func handleSigterm(ngx *controller.NGINXController, exit exiter) {
       signalChan := make(chan os.Signal, 1)
       signal.Notify(signalChan, syscall.SIGTERM)
       <-signalChan
       glog.Infof("Received SIGTERM, shutting down")
    
       exitCode := 0
       if err := ngx.Stop(); err != nil {
          glog.Infof("Error during shutdown: %v", err)
          exitCode = 1
       }
    
       glog.Infof("Handled quit, awaiting Pod deletion")
       time.Sleep(10 * time.Second)
    
       glog.Infof("Exiting with %v", exitCode)
       exit(exitCode)
    }

4 NewNGINXController函数

## ##

# #

*  做的事情相当多
 *  store.New（4.1讲解）

// NewNGINXController creates a new NGINX Ingress controller.
    func NewNGINXController(config *Configuration, fs file.Filesystem) *NGINXController {

n.store = store.New(
       config.EnableSSLChainCompletion,
       config.Namespace,
       config.ConfigMapName,
       config.TCPConfigMapName,
       config.UDPConfigMapName,
       config.DefaultSSLCertificate,
       config.ResyncPeriod,
       config.Client,
       fs,
       n.updateCh)

4.1 New函数

## ##

# #

*  代码特么多，看主要工作
 *  监控了ingress service endpoint secret configmap资源，资源发生了变化，会立即通知backend，例如nginx等
 *  定义一些资源处理函数

func New(checkOCSP bool,
       namespace, configmap, tcp, udp, defaultSSLCertificate string,
       resyncPeriod time.Duration,
       client clientset.Interface,
       fs file.Filesystem,
       updateCh *channels.RingChannel) Storer {
    
       store.informers.Ingress = infFactory.Extensions().V1beta1().Ingresses().Informer()
       store.listers.Ingress.Store = store.informers.Ingress.GetStore()
    
       store.informers.Endpoint = infFactory.Core().V1().Endpoints().Informer()
       store.listers.Endpoint.Store = store.informers.Endpoint.GetStore()
    
       store.informers.Secret = infFactory.Core().V1().Secrets().Informer()
       store.listers.Secret.Store = store.informers.Secret.GetStore()
    
       store.informers.ConfigMap = infFactory.Core().V1().ConfigMaps().Informer()
       store.listers.ConfigMap.Store = store.informers.ConfigMap.GetStore()
    
       store.informers.Service = infFactory.Core().V1().Services().Informer()
       store.listers.Service.Store = store.informers.Service.GetStore()
    
       
       store.informers.Ingress.AddEventHandler(ingEventHandler)
       store.informers.Endpoint.AddEventHandler(epEventHandler)
       store.informers.Secret.AddEventHandler(secrEventHandler)
       store.informers.ConfigMap.AddEventHandler(cmEventHandler)
       store.informers.Service.AddEventHandler(cache.ResourceEventHandlerFuncs{})
    
       // do not wait for informers to read the configmap configuration
       ns, name, _ := k8s.ParseNameNS(configmap)
       cm, err := client.CoreV1().ConfigMaps(ns).Get(name, metav1.GetOptions{})
       if err != nil {
          glog.Warningf("Unexpected error reading configuration configmap: %v", err)
       }
       store.setConfig(cm)
    
       return store
    }

*  ingEventHandl定义Ingress Event Handler: Add, Delete, Update
 *      ingEventHandler := cache.ResourceEventHandlerFuncs{
           AddFunc: func(obj interface{}) {
              ing := obj.(*extensions.Ingress)
              if !class.IsValid(ing) {
                 a, _ := parser.GetStringAnnotation(class.IngressKey, ing)
                 glog.Infof("ignoring add for ingress %v based on annotation %v with value %v", ing.Name, class.IngressKey, a)
                 return
              }
              recorder.Eventf(ing, corev1.EventTypeNormal, "CREATE", fmt.Sprintf("Ingress %s/%s", ing.Namespace, ing.Name))
        
              store.extractAnnotations(ing)
              store.updateSecretIngressMap(ing)
              store.syncSecrets(ing)
        
              updateCh.In() <- Event{
                 Type: CreateEvent,
                 Obj:  obj,
              }
           },
           DeleteFunc: func(obj interface{}) {
           },
           UpdateFunc: func(old, cur interface{}) {
           },
        }
 *  IngressKey = "kubernetes.io/ingress.class" 只处理Annotation为kubernetes.io/ingress.class满足条件的Ingress，条件满足其中之一：
 *  1. 如果Annotation为空，则要求--ingress-class设置的值为nginx
 *  2. Annotation与--ingress-class设置的值相同
 *  对于Add/Delete/Update都放入RingChannel中的input channel中
 *      func IsValid(ing *extensions.Ingress) bool {
           ingress, ok := ing.GetAnnotations()[IngressKey]
           if !ok {
              glog.V(3).Infof("annotation %v is not present in ingress %v/%v", IngressKey, ing.Namespace, ing.Name)
           }
        
           // we have 2 valid combinations
           // 1 - ingress with default class | blank annotation on ingress
           // 2 - ingress with specific class | same annotation on ingress
           //
           // and 2 invalid combinations
           // 3 - ingress with default class | fixed annotation on ingress
           // 4 - ingress with specific class | different annotation on ingress
           if ingress == "" && IngressClass == DefaultClass {
              return true
           }
        
           return ingress == IngressClass
        }
 *  定义通用event handler，对于Add/Delete/Update都放入RingChannel中的input channel中

epEventHandler := cache.ResourceEventHandlerFuncs{
       AddFunc: func(obj interface{}) {
          updateCh.In() <- Event{
             Type: CreateEvent,
             Obj:  obj,
          }
       },
       DeleteFunc: func(obj interface{}) {
          updateCh.In() <- Event{
             Type: DeleteEvent,
             Obj:  obj,
          }
       },
       UpdateFunc: func(old, cur interface{}) {
          oep := old.(*corev1.Endpoints)
          cep := cur.(*corev1.Endpoints)
          if !reflect.DeepEqual(cep.Subsets, oep.Subsets) {
             updateCh.In() <- Event{
                Type: UpdateEvent,
                Obj:  cur,
             }
          }
       },
    }

4.2

## ##

# #

*  创建Ingress的syncQueue，每往syncQueue插入一个Ingress对象，就会调用syncIngress一次

n.syncQueue = task.NewTaskQueue(n.syncIngress)

4.3

## ##

# #

*  定义nginx.tmpl文件内容发生变化时：重新生成nginx template instance,并赋给NGINXController

onTemplateChange := func() {
          template, err := ngx_template.NewTemplate(tmplPath, fs)
          if err != nil {
             // this error is different from the rest because it must be clear why nginx is not working
             glog.Errorf(`
    -------------------------------------------------------------------------------
    Error loading new template: %v
    -------------------------------------------------------------------------------
    `, err)
             return
          }
    
          n.t = template
          glog.Info("New NGINX configuration template loaded.")
          n.syncQueue.EnqueueTask(task.GetDummyObject("template-change"))
       }

4.3

## ##

# #

*  根据nginx.tmpl生成Template，并将onTemplateChange注册，通过FileWatcher监听nginx.tmpl变化，模板文件位置
    
        tmplPath = "/etc/nginx/template/nginx.tmpl"

ngxTpl, err := ngx_template.NewTemplate(tmplPath, fs)
    if err != nil {
       glog.Fatalf("Invalid NGINX configuration template: %v", err)
    }
    
    n.t = ngxTpl
    
    if _, ok := fs.(filesystem.DefaultFs); !ok {
       // do not setup watchers on tests
       return n
    }
    
    _, err = watch.NewFileWatcher(tmplPath, onTemplateChange)
    if err != nil {
       glog.Fatalf("Error creating file watcher for %v: %v", tmplPath, err)
    }

5 RingChannel结构体

## ##

# #

*  `updateCh` 是一个 RingChannel buffer 长度为 1024，不会阻塞写入，当空间用完时，新的元素将覆盖最早的元素
 *      // RingChannel implements the Channel interface in a way that never blocks the writer.
        // Specifically, if a value is written to a RingChannel when its buffer is full then the oldest
        // value in the buffer is discarded to make room (just like a standard ring-buffer).
        // Note that Go's scheduler can cause discarded values when they could be avoided, simply by scheduling
        // the writer before the reader, so caveat emptor.
        // For the opposite behaviour (discarding the newest element, not the oldest) see OverflowingChannel.
        type RingChannel struct {
           input, output chan interface{}
           length        chan int
           buffer        *queue.Queue
           size          BufferCap
        }

6 syncIngress函数

## ##

# #

*  ListIngress（6.1讲解）
 *  getBackendServers（6.2讲解）

func (n *NGINXController) syncIngress(interface{}) error {
       // sort Ingresses using the ResourceVersion field
       ings := n.store.ListIngresses()
       sort.SliceStable(ings, func(i, j int) bool {
          ir := ings[i].ResourceVersion
          jr := ings[j].ResourceVersion
          return ir < jr
       })
    
       upstreams, servers := n.getBackendServers(ings)
       var passUpstreams []*ingress.SSLPassthroughBackend
    
       for _, server := range servers {
          if !server.SSLPassthrough {
             continue
          }
    
          for _, loc := range server.Locations {
             if loc.Path != rootLocation {
                glog.Warningf("Ignoring SSL Passthrough for location %q in server %q", loc.Path, server.Hostname)
                continue
             }
             passUpstreams = append(passUpstreams, &ingress.SSLPassthroughBackend{
                Backend:  loc.Backend,
                Hostname: server.Hostname,
                Service:  loc.Service,
                Port:     loc.Port,
             })
             break
          }
       }
    
       pcfg := ingress.Configuration{
          Backends:            upstreams,
          Servers:             servers,
          TCPEndpoints:        n.getStreamServices(n.cfg.TCPConfigMapName, apiv1.ProtocolTCP),
          UDPEndpoints:        n.getStreamServices(n.cfg.UDPConfigMapName, apiv1.ProtocolUDP),
          PassthroughBackends: passUpstreams,
    
          ConfigurationChecksum: n.store.GetBackendConfiguration().Checksum,
       }
    
       if n.runningConfig.Equal(&pcfg) {
          glog.V(3).Infof("No configuration change detected, skipping backend reload.")
          return nil
       }
    
       if n.cfg.DynamicConfigurationEnabled && n.IsDynamicConfigurationEnough(&pcfg) {
          glog.Infof("Changes handled by the dynamic configuration, skipping backend reload.")
       } else {
          glog.Infof("Configuration changes detected, backend reload required.")
    
          err := n.OnUpdate(pcfg)
          if err != nil {
             IncReloadErrorCount()
             ConfigSuccess(false)
             glog.Errorf("Unexpected failure reloading the backend:\n%v", err)
             return err
          }
    
          glog.Infof("Backend successfully reloaded.")
          ConfigSuccess(true)
          IncReloadCount()
          setSSLExpireTime(servers)
       }
    
       if n.cfg.DynamicConfigurationEnabled {
          isFirstSync := n.runningConfig.Equal(&ingress.Configuration{})
          go func(isFirstSync bool) {
             if isFirstSync {
                glog.Infof("Initial synchronization of the NGINX configuration.")
    
                // it takes time for NGINX to start listening on the configured ports
                time.Sleep(1 * time.Second)
             }
             err := configureDynamically(&pcfg, n.cfg.ListenPorts.Status)
             if err == nil {
                glog.Infof("Dynamic reconfiguration succeeded.")
             } else {
                glog.Warningf("Dynamic reconfiguration failed: %v", err)
             }
          }(isFirstSync)
       }
    
       n.runningConfig = &pcfg
    
       return nil
    }

6.1 ListIngress函数

## ##

# #

*  主要是从cache中读取，cache实现在vendor/k8s.io/client-go/tools/cache/thread\_safe\_store.go

// ListIngresses returns the list of Ingresses
    func (s k8sStore) ListIngresses() []*extensions.Ingress {
       // filter ingress rules
       var ingresses []*extensions.Ingress
       for _, item := range s.listers.Ingress.List() {
          ing := item.(*extensions.Ingress)
          if !class.IsValid(ing) {
             continue
          }
    
          for ri, rule := range ing.Spec.Rules {
             if rule.HTTP == nil {
                continue
             }
    
             for pi, path := range rule.HTTP.Paths {
                if path.Path == "" {
                   ing.Spec.Rules[ri].HTTP.Paths[pi].Path = "/"
                }
             }
          }
    
          ingresses = append(ingresses, ing)
       }
    
       return ingresses
    }

6.2 getBackendServers函数

## ##

# #

*  这个函数一猜也大概知道结果，就是便利所有ingress，从中提取host，path，以及后端service，端口等

// getBackendServers returns a list of Upstream and Server to be used by the
    // backend.  An upstream can be used in multiple servers if the namespace,
    // service name and port are the same.
    func (n *NGINXController) getBackendServers(ingresses []*extensions.Ingress) ([]*ingress.Backend, []*ingress.Server) {
       du := n.getDefaultUpstream()
       upstreams := n.createUpstreams(ingresses, du)
       servers := n.createServers(ingresses, upstreams, du)
    
       for _, ing := range ingresses {
          
       
       }
    
       aUpstreams := make([]*ingress.Backend, 0, len(upstreams))
    
       for _, upstream := range upstreams {
         
       }
    
       // create the list of upstreams and skip those without Endpoints
       for _, upstream := range upstreams {
          if len(upstream.Endpoints) == 0 {
             continue
          }
          aUpstreams = append(aUpstreams, upstream)
       }
    
       aServers := make([]*ingress.Server, 0, len(servers))
       for _, value := range servers {
          sort.SliceStable(value.Locations, func(i, j int) bool {
             return value.Locations[i].Path > value.Locations[j].Path
          })
          aServers = append(aServers, value)
       }
    
       sort.SliceStable(aUpstreams, func(a, b int) bool {
          return aUpstreams[a].Name < aUpstreams[b].Name
       })
    
       sort.SliceStable(aServers, func(i, j int) bool {
          return aServers[i].Hostname < aServers[j].Hostname
       })
    
       return aUpstreams, aServers
    }

6.2.1 Backend结构体

## ##

# #

*  主要包括kubernets中的service（namespace-name-port）port endpoint信息

/ Backend describes one or more remote server/s (endpoints) associated with a service
    // +k8s:deepcopy-gen=true
    type Backend struct {
       // Name represents an unique apiv1.Service name formatted as <namespace>-<name>-<port>
       Name    string             `json:"name"`
       Service *apiv1.Service     `json:"service,omitempty"`
       Port    intstr.IntOrString `json:"port"`
       // This indicates if the communication protocol between the backend and the endpoint is HTTP or HTTPS
       // Allowing the use of HTTPS
       // The endpoint/s must provide a TLS connection.
       // The certificate used in the endpoint cannot be a self signed certificate
       Secure bool `json:"secure"`
       // SecureCACert has the filename and SHA1 of the certificate authorities used to validate
       // a secured connection to the backend
       SecureCACert resolver.AuthSSLCert `json:"secureCACert"`
       // SSLPassthrough indicates that Ingress controller will delegate TLS termination to the endpoints.
       SSLPassthrough bool `json:"sslPassthrough"`
       // Endpoints contains the list of endpoints currently running
       Endpoints []Endpoint `json:"endpoints,omitempty"`
       // StickySessionAffinitySession contains the StickyConfig object with stickyness configuration
       SessionAffinity SessionAffinityConfig `json:"sessionAffinityConfig"`
       // Consistent hashing by NGINX variable
       UpstreamHashBy string `json:"upstream-hash-by,omitempty"`
       // LB algorithm configuration per ingress
       LoadBalancing string `json:"load-balance,omitempty"`
    }

6.2.2 createUpstreams结构体

## ##

# #

*  会创建ingressBackend，以namespace-service-port 为name

// createUpstreams creates the NGINX upstreams (Endpoints) for each Service
    // referenced in Ingress rules.
    func (n *NGINXController) createUpstreams(data []*extensions.Ingress, du *ingress.Backend) map[string]*ingress.Backend {
       upstreams := make(map[string]*ingress.Backend)
       upstreams[defUpstreamName] = du
    
       for _, ing := range data {
          
          if ing.Spec.Backend != nil {
             defBackend = fmt.Sprintf("%v-%v-%v",
                ing.Namespace,
                ing.Spec.Backend.ServiceName,
                ing.Spec.Backend.ServicePort.String())
    
             glog.V(3).Infof("Creating upstream %q", defBackend)
             upstreams[defBackend] = newUpstream(defBackend)
          }
    
          for _, rule := range ing.Spec.Rules {
             for _, path := range rule.HTTP.Paths {
                name := fmt.Sprintf("%v-%v-%v",
                   ing.Namespace,
                   path.Backend.ServiceName,
                   path.Backend.ServicePort.String())
    
                if _, ok := upstreams[name]; ok {
                   continue
                }
    
                glog.V(3).Infof("Creating upstream %q", name)
                
                upstreams[name].Service = s
             }
          }
       }
    
       return upstreams
    }

6.2.3 createServers结构体

## ##

# #

*  创建一个以host name 到Server的map

// createServers builds a map of host name to Server structs from a map of
    // already computed Upstream structs. Each Server is configured with at least
    // one root location, which uses a default backend if left unspecified.
    func (n *NGINXController) createServers(data []*extensions.Ingress,
       upstreams map[string]*ingress.Backend,
       du *ingress.Backend) map[string]*ingress.Server {
    
       servers := make(map[string]*ingress.Server, len(data))
       aliases := make(map[string]string, len(data))
    
       bdef := n.store.GetDefaultBackend()
      
       // generated on Start() with createDefaultSSLCertificate()
       defaultPemFileName := n.cfg.FakeCertificatePath
       defaultPemSHA := n.cfg.FakeCertificateSHA
    
       // read custom default SSL certificate, fall back to generated default certificate
       defaultCertificate, err := n.store.GetLocalSSLCert(n.cfg.DefaultSSLCertificate)
       if err == nil {
          defaultPemFileName = defaultCertificate.PemFileName
          defaultPemSHA = defaultCertificate.PemSHA
       }
    
       // initialize default server and root location
       servers[defServerName] = &ingress.Server{
          Hostname: defServerName,
          SSLCert: ingress.SSLCert{
             PemFileName: defaultPemFileName,
             PemSHA:      defaultPemSHA,
          },
          Locations: []*ingress.Location{
             {
                Path:         rootLocation,
                IsDefBackend: true,
                Backend:      du.Name,
                Proxy:        ngxProxy,
                Service:      du.Service,
             },
          }}
    
       // initialize all other servers
       for _, ing := range data {
          ingKey := k8s.MetaNamespaceKey(ing)
    
          anns, err := n.store.GetIngressAnnotations(ingKey)
          if err != nil {
             glog.Errorf("Error getting Ingress annotations %q: %v", ingKey, err)
          }
    
          // default upstream name
          un := du.Name
    
          if ing.Spec.Backend != nil {
             defUpstream := fmt.Sprintf("%v-%v-%v", ing.Namespace, ing.Spec.Backend.ServiceName, ing.Spec.Backend.ServicePort.String())
    
             if backendUpstream, ok := upstreams[defUpstream]; ok {
                // use backend specified in Ingress as the default backend for all its rules
                un = backendUpstream.Name
    
                // special "catch all" case, Ingress with a backend but no rule
                defLoc := servers[defServerName].Locations[0]
                if defLoc.IsDefBackend && len(ing.Spec.Rules) == 0 {
                   glog.Infof("Ingress %q defines a backend but no rule. Using it to configure the catch-all server %q",
                      ingKey, defServerName)
    
                   defLoc.IsDefBackend = false
                   defLoc.Backend = backendUpstream.Name
                   defLoc.Service = backendUpstream.Service
                   defLoc.Ingress = ing
    
                   // customize using Ingress annotations
                   defLoc.Logs = anns.Logs
                   defLoc.BasicDigestAuth = anns.BasicDigestAuth
                   defLoc.ClientBodyBufferSize = anns.ClientBodyBufferSize
                   defLoc.ConfigurationSnippet = anns.ConfigurationSnippet
                   defLoc.CorsConfig = anns.CorsConfig
                   defLoc.ExternalAuth = anns.ExternalAuth
                   defLoc.Proxy = anns.Proxy
                   defLoc.RateLimit = anns.RateLimit
                   // TODO: Redirect and rewrite can affect the catch all behavior, skip for now
                   // defLoc.Redirect = anns.Redirect
                   // defLoc.Rewrite = anns.Rewrite
                   defLoc.UpstreamVhost = anns.UpstreamVhost
                   defLoc.Whitelist = anns.Whitelist
                   defLoc.Denied = anns.Denied
                   defLoc.GRPC = anns.GRPC
                   defLoc.LuaRestyWAF = anns.LuaRestyWAF
                   defLoc.InfluxDB = anns.InfluxDB
                } else {
                   glog.V(3).Infof("Ingress %q defines both a backend and rules. Using its backend as default upstream for all its rules.",
                      ingKey)
                }
             }
          }
    
          for _, rule := range ing.Spec.Rules {
             host := rule.Host
             if host == "" {
                host = defServerName
             }
             if _, ok := servers[host]; ok {
                // server already configured
                continue
             }
    
             servers[host] = &ingress.Server{
                Hostname: host,
                Locations: []*ingress.Location{
                   {
                      Path:         rootLocation,
                      IsDefBackend: true,
                      Backend:      un,
                      Proxy:        ngxProxy,
                      Service:      &apiv1.Service{},
                   },
                },
                SSLPassthrough: anns.SSLPassthrough,
                SSLCiphers:     anns.SSLCiphers,
             }
          }
       }
    
       // configure default location, alias, and SSL
       for _, ing := range data {
          ingKey := k8s.MetaNamespaceKey(ing)
    
          anns, err := n.store.GetIngressAnnotations(ingKey)
          if err != nil {
             glog.Errorf("Error getting Ingress annotations %q: %v", ingKey, err)
          }
    
          for _, rule := range ing.Spec.Rules {
             host := rule.Host
             if host == "" {
                host = defServerName
             }
    
             if anns.Alias != "" {
                if servers[host].Alias == "" {
                   servers[host].Alias = anns.Alias
                   if _, ok := aliases["Alias"]; !ok {
                      aliases["Alias"] = host
                   }
                } else {
                   glog.Warningf("Aliases already configured for server %q, skipping (Ingress %q)",
                      host, ingKey)
                }
             }
    
             if anns.ServerSnippet != "" {
                if servers[host].ServerSnippet == "" {
                   servers[host].ServerSnippet = anns.ServerSnippet
                } else {
                   glog.Warningf("Server snippet already configured for server %q, skipping (Ingress %q)",
                      host, ingKey)
                }
             }
    
             // only add SSL ciphers if the server does not have them previously configured
             if servers[host].SSLCiphers == "" && anns.SSLCiphers != "" {
                servers[host].SSLCiphers = anns.SSLCiphers
             }
    
             // only add a certificate if the server does not have one previously configured
             if servers[host].SSLCert.PemFileName != "" {
                continue
             }
    
             if len(ing.Spec.TLS) == 0 {
                glog.V(3).Infof("Ingress %q does not contains a TLS section.", ingKey)
                continue
             }
    
             tlsSecretName := extractTLSSecretName(host, ing, n.store.GetLocalSSLCert)
    
             if tlsSecretName == "" {
                glog.V(3).Infof("Host %q is listed in the TLS section but secretName is empty. Using default certificate.", host)
                servers[host].SSLCert.PemFileName = defaultPemFileName
                servers[host].SSLCert.PemSHA = defaultPemSHA
                continue
             }
    
             secrKey := fmt.Sprintf("%v/%v", ing.Namespace, tlsSecretName)
             cert, err := n.store.GetLocalSSLCert(secrKey)
             if err != nil {
                glog.Warningf("Error getting SSL certificate %q: %v", secrKey, err)
                continue
             }
    
             err = cert.Certificate.VerifyHostname(host)
             if err != nil {
                glog.Warningf("Unexpected error validating SSL certificate %q for server %q: %v", secrKey, host, err)
                glog.Warning("Validating certificate against DNS names. This will be deprecated in a future version.")
                // check the Common Name field
                // https://github.com/golang/go/issues/22922
                err := verifyHostname(host, cert.Certificate)
                if err != nil {
                   glog.Warningf("SSL certificate %q does not contain a Common Name or Subject Alternative Name for server %q: %v",
                      secrKey, host, err)
                   continue
                }
             }
    
             servers[host].SSLCert = *cert
    
             if cert.ExpireTime.Before(time.Now().Add(240 * time.Hour)) {
                glog.Warningf("SSL certificate for server %q is about to expire (%v)", host, cert.ExpireTime)
             }
          }
       }
    
       for alias, host := range aliases {
          if _, ok := servers[alias]; ok {
             glog.Warningf("Conflicting hostname (%v) and alias (%v). Removing alias to avoid conflicts.", host, alias)
             servers[host].Alias = ""
          }
       }
    
       return servers
    }

6.2.4

## ##

# #

*  创建passUpstreams

upstreams, servers := n.getBackendServers(ings)
    var passUpstreams []*ingress.SSLPassthroughBackend
    
    for _, server := range servers {
       if !server.SSLPassthrough {
          continue
       }
    
       for _, loc := range server.Locations {
          if loc.Path != rootLocation {
             glog.Warningf("Ignoring SSL Passthrough for location %q in server %q", loc.Path, server.Hostname)
             continue
          }
          passUpstreams = append(passUpstreams, &ingress.SSLPassthroughBackend{
             Backend:  loc.Backend,
             Hostname: server.Hostname,
             Service:  loc.Service,
             Port:     loc.Port,
          })
          break
       }
    }

6.2.5

## ##

# #

*  如果配置更改调用Onupdate函数(第7章讲解)

if n.cfg.DynamicConfigurationEnabled && n.IsDynamicConfigurationEnough(&pcfg) {
       glog.Infof("Changes handled by the dynamic configuration, skipping backend reload.")
    } else {
       glog.Infof("Configuration changes detected, backend reload required.")
    
       err := n.OnUpdate(pcfg)
       if err != nil {
          IncReloadErrorCount()
          ConfigSuccess(false)
          glog.Errorf("Unexpected failure reloading the backend:\n%v", err)
          return err
       }
    
       glog.Infof("Backend successfully reloaded.")
       ConfigSuccess(true)
       IncReloadCount()
       setSSLExpireTime(servers)
    }

7 OnUpdate函数

## ##

# #

*  填充TemplateConfig结构体
 *  重写写入/etc/nginx/nginx.conf文件
 *  执行nginx -s reload命令

/ OnUpdate is called by the synchronization loop whenever configuration
    // changes were detected. The received backend Configuration is merged with the
    // configuration ConfigMap before generating the final configuration file.
    // Returns nil in case the backend was successfully reloaded.
    func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
       cfg := n.store.GetBackendConfiguration()
       cfg.Resolver = n.resolver
    
       tc := ngx_config.TemplateConfig{
          ProxySetHeaders:             setHeaders,
          AddHeaders:                  addHeaders,
          MaxOpenFiles:                maxOpenFiles,
          BacklogSize:                 sysctlSomaxconn(),
          Backends:                    ingressCfg.Backends,
          PassthroughBackends:         ingressCfg.PassthroughBackends,
          Servers:                     ingressCfg.Servers,
          TCPBackends:                 ingressCfg.TCPEndpoints,
          UDPBackends:                 ingressCfg.UDPEndpoints,
          HealthzURI:                  ngxHealthPath,
          CustomErrors:                len(cfg.CustomHTTPErrors) > 0,
          Cfg:                         cfg,
          IsIPV6Enabled:               n.isIPV6Enabled && !cfg.DisableIpv6,
          NginxStatusIpv4Whitelist:    cfg.NginxStatusIpv4Whitelist,
          NginxStatusIpv6Whitelist:    cfg.NginxStatusIpv6Whitelist,
          RedirectServers:             redirectServers,
          IsSSLPassthroughEnabled:     n.cfg.EnableSSLPassthrough,
          ListenPorts:                 n.cfg.ListenPorts,
          PublishService:              n.GetPublishService(),
          DynamicConfigurationEnabled: n.cfg.DynamicConfigurationEnabled,
          DisableLua:                  n.cfg.DisableLua,
       }
    
       content, err := n.t.Write(tc)
       if err != nil {
          return err
       }
    
       if cfg.EnableOpentracing {
          err := createOpentracingCfg(cfg)
          if err != nil {
             return err
          }
       }
    
       err = n.testTemplate(content)
       if err != nil {
          return err
       }
       err = ioutil.WriteFile(cfgPath, content, file.ReadWriteByUser)
       if err != nil {
          return err
       }
    
       o, err := nginxExecCommand("-s", "reload").CombinedOutput()
       if err != nil {
          return fmt.Errorf("%v\n%v", err, string(o))
       }
    
       return nil
    }