主要逻辑

nginx controller 入口函数

// file:k8s.io/ingress-nginx/nginx/main.go
func main() {
    // step1: 初始化日志组件
    klog.InitFlags(nil)

      ......

    // step2:创建必要的目录
    err = file.CreateRequiredDirectories()
      ......

    // step 3 :初始化ApiserverClient
    kubeClient, err := createApiserverClient(conf.APIServerHost, conf.RootCAFile, conf.KubeConfigFile)
      ......

    // step4: 检查service配置
    if len(conf.DefaultService) > 0 {
        err := checkService(conf.DefaultService, kubeClient)
          ......

        klog.Infof("Validated %v as the default backend.", conf.DefaultService)
    }

    if len(conf.PublishService) > 0 {
        err := checkService(conf.PublishService, kubeClient)
          ......
    }

    // step5:获取namespace
    if conf.Namespace != "" {
        _, err = kubeClient.CoreV1().Namespaces().Get(context.TODO(), conf.Namespace, metav1.GetOptions{})
        if err != nil {
            klog.Fatalf("No namespace with name %v found: %v", conf.Namespace, err)
        }
    }

    // step6: 创建默认证书
    conf.FakeCertificate = ssl.GetFakeSSLCert()
    klog.Infof("SSL fake certificate created %v", conf.FakeCertificate.PemFileName)

    // step7: 检查是否支持v1beta API 、k8s 版本是否高于1.18.0
    k8s.IsNetworkingIngressAvailable, k8s.IsIngressV1Ready = k8s.NetworkingIngressAvailable(kubeClient)
    if !k8s.IsNetworkingIngressAvailable {
        klog.Warningf("Using deprecated \"k8s.io/api/extensions/v1beta1\" package because Kubernetes version is < v1.14.0")
    }

    if k8s.IsIngressV1Ready {
          ......
    }

    conf.Client = kubeClient

    // step8: 注册prometheus
    reg := prometheus.NewRegistry()

    reg.MustRegister(prometheus.NewGoCollector())
    reg.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{
        PidFn:        func() (int, error) { return os.Getpid(), nil },
        ReportErrors: true,
    }))
    ......

    // step9:启动profile
    if conf.EnableProfiling {
        go registerProfiler()
    }

    // step10: 实例化nginxcontroller （*）
    ngx := controller.NewNGINXController(conf, mc)

    // step11: 启动健康探测和metrics API
    mux := http.NewServeMux()
    registerHealthz(nginx.HealthPath, ngx, mux)
    registerMetrics(reg, mux)

    go startHTTPServer(conf.ListenPorts.Health, mux)

    // step12: 启动nginx master进程
    go ngx.Start()
  ......
}

nginx controller 初始化

// NewNGINXController creates a new NGINX Ingress controller.
func NewNGINXController(config *Configuration, mc metric.Collector) *NGINXController {
    // 初始化 event broadcaster
    eventBroadcaster := record.NewBroadcaster()
    eventBroadcaster.StartLogging(klog.Infof)
    eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
        Interface: config.Client.CoreV1().Events(config.Namespace),
    })

    // 获取/etc/resolv.conf 中的nameserver 列表
    h, err := dns.GetSystemNameServers()
    if err != nil {
        klog.Warningf("Error reading system nameservers: %v", err)
    }

    // 实例化NGINXController
    n := &NGINXController{
        isIPV6Enabled: ing_net.IsIPv6Enabled(),

        resolver:        h,
        cfg:             config,
        syncRateLimiter: flowcontrol.NewTokenBucketRateLimiter(config.SyncRateLimit, 1),

        recorder: eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{
            Component: "nginx-ingress-controller",
        }),

        stopCh:   make(chan struct{}),
        updateCh: channels.NewRingChannel(1024),

        ngxErrCh: make(chan error),

        stopLock: &sync.Mutex{},

        runningConfig: new(ingress.Configuration),

        Proxy: &TCPProxy{},

        metricCollector: mc,

        command: NewNginxCommand(),
    }

    // 启动webhook 服务
    if n.cfg.ValidationWebhook != "" {
        n.validationWebhookServer = &http.Server{
            Addr:      config.ValidationWebhook,
            Handler:   adm_controller.NewAdmissionControllerServer(&adm_controller.IngressAdmission{Checker: n}),
            TLSConfig: ssl.NewTLSListener(n.cfg.ValidationWebhookCertPath, n.cfg.ValidationWebhookKeyPath).TLSConfig(),
        }
    }

    // 获取pod runtime信息
    pod, err := k8s.GetPodDetails(config.Client)
    if err != nil {
        klog.Fatalf("unexpected error obtaining pod information: %v", err)
    }
    n.podInfo = pod

    // 实例化store(本地缓存)
    n.store = store.New(
        config.Namespace,
        config.ConfigMapName,
        config.TCPConfigMapName,
        config.UDPConfigMapName,
        config.DefaultSSLCertificate,
        config.ResyncPeriod,
        config.Client,
        n.updateCh,
        pod,
        config.DisableCatchAll)

    // 创建同步队列
    n.syncQueue = task.NewTaskQueue(n.syncIngress)

    ... ...

    // 格式化template配置模板
    onTemplateChange := func() {
        template, err := ngx_template.NewTemplate(nginx.TemplatePath)
        if err != nil {
            // this error is different from the rest because it must be clear why nginx is not working
            klog.Errorf(`
-------------------------------------------------------------------------------
Error loading new template: %v
-------------------------------------------------------------------------------
`, err)
            return
        }

        // 若模板格式化正确，则更新到nginxcontroller 对象中，并往同步队列发送一个template-change事件
        n.t = template
        klog.Info("New NGINX configuration template loaded.")
        n.syncQueue.EnqueueTask(task.GetDummyObject("template-change"))
    }

    // 首次启动加载配置模板文件
    ngxTpl, err := ngx_template.NewTemplate(nginx.TemplatePath)
    ......

    n.t = ngxTpl

    // 监听模板文件变化
    // 监听 /etc/nginx/template/nginx.tmpl 模板文件是否有变化，有变化则调用onTemplateChange
    _, err = watch.NewFileWatcher(nginx.TemplatePath, onTemplateChange)
 ... ...

    // 监听/etc/nginx/geoip/ 目录下配置文件变化
    filesToWatch := []string{}
    err = filepath.Walk("/etc/nginx/geoip/", func(path string, info os.FileInfo, err error) error {
        ......
        filesToWatch = append(filesToWatch, path)
        ......
    })

    ......

    for _, f := range filesToWatch {
        _, err = watch.NewFileWatcher(f, func() {
            klog.Infof("File %v changed. Reloading NGINX", f)
            // 配置文件有变化则往同步队列发送一个file-change 事件
            n.syncQueue.EnqueueTask(task.GetDummyObject("file-change"))
        })
        ......
    }

    return n
}

ingress controller 结构体

type NGINXController struct {
    // pod runtime 信息
    podInfo *k8s.PodInfo
    // 配置信息
    cfg *Configuration
    // 事件通知器
    recorder record.EventRecorder
    // 同步队列
    syncQueue *task.Queue
    // 同步状态
    syncStatus status.Syncer
    // 同步限流器
    syncRateLimiter flowcontrol.RateLimiter

    stopLock *sync.Mutex

    stopCh   chan struct{}
    // 更新环状channel
    updateCh *channels.RingChannel

    // 接受nginx 错误信息channel
    ngxErrCh chan error

    // 当前配置文件
    runningConfig *ingress.Configuration
    // nginx 配置模板文件
    t ngx_template.TemplateWriter
    // nameserver 列表
    resolver []net.IP
    // 是否启用ipv6
    isIPV6Enabled bool
    // 是否关闭
    isShuttingDown bool
    // TCP代理
    Proxy *TCPProxy
  // 本地缓存
    store store.Storer
  // metrics 收集器
    metricCollector metric.Collector
  // webhook
    validationWebhookServer *http.Server
    // nginx 二进制命令
    command NginxExecTester
}

ngx.Start()

ngx.Start() 主要做3个事情
启动store 协程
启动syncQueue协程
监听updateCh

当从updateCh 见到变化事件时，向syncQueue 发送一个task

// file：internal/ingress/controller/nginx.go
// Start starts a new NGINX master process running in the foreground.
func (n *NGINXController) Start() {
    klog.Info("Starting NGINX Ingress controller")
    // 初始化同步informers 及secret
    n.store.Run(n.stopCh)

    // we need to use the defined ingress class to allow multiple leaders
    // in order to update information about ingress status
    // 定义节点选举ID （ingress class 用于区分不同集群）
    // 使用定义的ingress class 来允许多个leader节点更新ingress状态
    electionID := fmt.Sprintf("%v-%v", n.cfg.ElectionID, class.DefaultClass)
    if class.IngressClass != "" {
        electionID = fmt.Sprintf("%v-%v", n.cfg.ElectionID, class.IngressClass)
    }

 // leader节点选举
    setupLeaderElection(&leaderElectionConfig{
        ......
    })

    cmd := n.command.ExecCommand()

    ......

    if n.cfg.EnableSSLPassthrough {
        n.setupSSLProxy()
    }

    // 启动nginx
    klog.Info("Starting NGINX process")
    n.start(cmd)

    // 启动同步队列
    go n.syncQueue.Run(time.Second, n.stopCh)

    // force initial sync
    // 发送initial-sync 事件
    n.syncQueue.EnqueueTask(task.GetDummyObject("initial-sync"))

    // In case of error the temporal configuration file will
    // be available up to five minutes after the error
    // 每隔5分钟删除临时配置文件
    go func() {
        for {
            time.Sleep(5 * time.Minute)
            err := cleanTempNginxCfg()
            ......
        }
    }()

    ......

    for {
        select {
        case err := <-n.ngxErrCh:
            if n.isShuttingDown {
                return
            }

            // if the nginx master process dies, the workers continue to process requests
            // until the failure of the configured livenessProbe and restart of the pod.
            // master 进程挂掉时，workerInc进程将继续处理请求，直到配置的liveness探针探测失败
            if process.IsRespawnIfRequired(err) {
                return
            }

        // 循环从updateCh里面获取事件
        case event := <-n.updateCh.Out():
            if n.isShuttingDown {
                break
            }

            if evt, ok := event.(store.Event); ok {
                klog.V(3).Infof("Event %v received - object %v", evt.Type, evt.Obj)
                if evt.Type == store.ConfigurationEvent {
                    // TODO: is this necessary? Consider removing this special case
                    n.syncQueue.EnqueueTask(task.GetDummyObject("configmap-change"))
                    continue
                }
                // 放入可忽略的同步队列
                n.syncQueue.EnqueueSkippableTask(evt.Obj)
            } else {
                klog.Warningf("Unexpected event type received %T", event)
            }
        case <-n.stopCh:
            return
        }
    }
}

事件类型

const (
    // CreateEvent event associated with new objects in an informer
    CreateEvent EventType = "CREATE"
    // UpdateEvent event associated with an object update in an informer
    UpdateEvent EventType = "UPDATE"
    // DeleteEvent event associated when an object is removed from an informer
    DeleteEvent EventType = "DELETE"
    // ConfigurationEvent event associated when a controller configuration object is created or updated
    ConfigurationEvent EventType = "CONFIGURATION"
)

同步队列
结构体

// Queue manages a time work queue through an independent worker that invokes the
// given sync function for every work item inserted.
// The queue uses an internal timestamp that allows the removal of certain elements
// which timestamp is older than the last successful get operation.
type Queue struct {
    // queue is the work queue the worker polls
    queue workqueue.RateLimitingInterface
    // sync is called for each item in the queue
    sync func(interface{}) error
    // workerDone is closed when the worker exits
    workerDone chan bool
    // fn makes a key for an API object
    fn func(obj interface{}) (interface{}, error)
    // lastSync is the Unix epoch time of the last execution of 'sync'
    lastSync int64
}

队列类型
(1) 可忽略队列 EnqueueSkippableTask
(2) 不可忽略队列

// EnqueueTask enqueues ns/name of the given api object in the task queue.
func (t *Queue) EnqueueTask(obj interface{}) {
    t.enqueue(obj, false)
}

// EnqueueSkippableTask enqueues ns/name of the given api object in
// the task queue that can be skipped
func (t *Queue) EnqueueSkippableTask(obj interface{}) {
    t.enqueue(obj, true)
}

// 入队列
// enqueue enqueues ns/name of the given api object in the task queue.
func (t *Queue) enqueue(obj interface{}, skippable bool) {
    if t.IsShuttingDown() {
        klog.Errorf("queue has been shutdown, failed to enqueue: %v", obj)
        return
    }

    ts := time.Now().UnixNano()
    if !skippable {
        // make sure the timestamp is bigger than lastSync
        ts = time.Now().Add(24 * time.Hour).UnixNano()
    }
    klog.V(3).Infof("queuing item %v", obj)
    key, err := t.fn(obj)
    if err != nil {
        klog.Errorf("%v", err)
        return
    }
    t.queue.Add(Element{
        Key:       key,
        Timestamp: ts,
    })
}

store 协程

// file : k8s.io/ingress-nginx/internal/controller/store/store.go
// Run initiates the synchronization of the informers and the initial
// synchronization of the secrets.
func (s *k8sStore) Run(stopCh chan struct{}) {
    // start informers
    s.informers.Run(stopCh)
}

调用了informers.Run()方法
起多个协程去监听ingress、secret、endpoint、service、configmap、pod 的变化

// Run initiates the synchronization of the informers against the API server.
func (i *Informer) Run(stopCh chan struct{}) {
    // 启动secret、endpoint、service、configmap、pod 的informer
    go i.Secret.Run(stopCh)
    go i.Endpoint.Run(stopCh)
    go i.Service.Run(stopCh)
    go i.ConfigMap.Run(stopCh)
    go i.Pod.Run(stopCh)

        ......

    time.Sleep(1 * time.Second)

    go i.Ingress.Run(stopCh)
    ......
}

这里以监听 ingress 变化为例，接着分析具体实现

// New creates a new object store to be used in the ingress controller
func New(
    namespace, configmap, tcp, udp, defaultSSLCertificate string,
    resyncPeriod time.Duration,
    client clientset.Interface,
    updateCh *channels.RingChannel,
    pod *k8s.PodInfo,
    disableCatchAll bool) Storer {

    store := &k8sStore{
        informers:             &Informer{},
        listers:               &Lister{},
        sslStore:              NewSSLCertTracker(),
        updateCh:              updateCh,
        backendConfig:         ngx_config.NewDefault(),
        syncSecretMu:          &sync.Mutex{},
        backendConfigMu:       &sync.RWMutex{},
        secretIngressMap:      NewObjectRefMap(),
        defaultSSLCertificate: defaultSSLCertificate,
        pod:                   pod,
    }

    ......

    // k8sStore fulfills resolver.Resolver interface
    // 格式化annotation
    store.annotations = annotations.NewAnnotationExtractor(store)

    store.listers.IngressWithAnnotation.Store = cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc)

    ......

    // create informers factory, enable and assign required informers
    // informer 工厂函数
    infFactory := informers.NewSharedInformerFactoryWithOptions(client, resyncPeriod,
        informers.WithNamespace(namespace),
        informers.WithTweakListOptions(tweakListOptionsFunc))

    if k8s.IsNetworkingIngressAvailable {
        store.informers.Ingress = infFactory.Networking().V1beta1().Ingresses().Informer()
    } else {
        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()

    labelSelector := labels.SelectorFromSet(store.pod.Labels)

    // list and watch 机制
    store.informers.Pod = cache.NewSharedIndexInformer(
        &cache.ListWatch{
            ListFunc: func(options metav1.ListOptions) (k8sruntime.Object, error) {
                options.LabelSelector = labelSelector.String()
                return client.CoreV1().Pods(store.pod.Namespace).List(context.TODO(), options)
            },
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                options.LabelSelector = labelSelector.String()
                return client.CoreV1().Pods(store.pod.Namespace).Watch(context.TODO(), options)
            },
        },
        &corev1.Pod{},
        resyncPeriod,
        cache.Indexers{},
    )
    store.listers.Pod.Store = store.informers.Pod.GetStore()

    ingDeleteHandler := func(obj interface{}) {
        ing, ok := toIngress(obj)
        if !ok {
            // If we reached here it means the ingress was deleted but its final state is unrecorded.
            tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
            if !ok {
                klog.Errorf("couldn't get object from tombstone %#v", obj)
                return
            }
            ing, ok = tombstone.Obj.(*networkingv1beta1.Ingress)
            if !ok {
                klog.Errorf("Tombstone contained object that is not an Ingress: %#v", obj)
                return
            }
        }

        if !class.IsValid(ing) {
            klog.Infof("ignoring delete for ingress %v based on annotation %v", ing.Name, class.IngressKey)
            return
        }
        if isCatchAllIngress(ing.Spec) && disableCatchAll {
            klog.Infof("ignoring delete for catch-all ingress %v/%v because of --disable-catch-all", ing.Namespace, ing.Name)
            return
        }
        recorder.Eventf(ing, corev1.EventTypeNormal, "DELETE", fmt.Sprintf("Ingress %s/%s", ing.Namespace, ing.Name))

        store.listers.IngressWithAnnotation.Delete(ing)

        key := k8s.MetaNamespaceKey(ing)
        store.secretIngressMap.Delete(key)

        updateCh.In() <- Event{
            Type: DeleteEvent,
            Obj:  obj,
        }
    }

    ingEventHandler := cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            ing, _ := toIngress(obj)
            if !class.IsValid(ing) {
                a, _ := parser.GetStringAnnotation(class.IngressKey, ing)
                klog.Infof("ignoring add for ingress %v based on annotation %v with value %v", ing.Name, class.IngressKey, a)
                return
            }
            if isCatchAllIngress(ing.Spec) && disableCatchAll {
                klog.Infof("ignoring add for catch-all ingress %v/%v because of --disable-catch-all", ing.Namespace, ing.Name)
                return
            }
            recorder.Eventf(ing, corev1.EventTypeNormal, "CREATE", fmt.Sprintf("Ingress %s/%s", ing.Namespace, ing.Name))

            store.syncIngress(ing)
            store.updateSecretIngressMap(ing)
            store.syncSecrets(ing)

            updateCh.In() <- Event{
                Type: CreateEvent,
                Obj:  obj,
            }
        },
        DeleteFunc: ingDeleteHandler,
        UpdateFunc: func(old, cur interface{}) {
            oldIng, _ := toIngress(old)
            curIng, _ := toIngress(cur)

            validOld := class.IsValid(oldIng)
            validCur := class.IsValid(curIng)
            if !validOld && validCur {
                if isCatchAllIngress(curIng.Spec) && disableCatchAll {
                    klog.Infof("ignoring update for catch-all ingress %v/%v because of --disable-catch-all", curIng.Namespace, curIng.Name)
                    return
                }

                klog.Infof("creating ingress %v based on annotation %v", curIng.Name, class.IngressKey)
                recorder.Eventf(curIng, corev1.EventTypeNormal, "CREATE", fmt.Sprintf("Ingress %s/%s", curIng.Namespace, curIng.Name))
            } else if validOld && !validCur {
                klog.Infof("removing ingress %v based on annotation %v", curIng.Name, class.IngressKey)
                ingDeleteHandler(old)
                return
            } else if validCur && !reflect.DeepEqual(old, cur) {
                if isCatchAllIngress(curIng.Spec) && disableCatchAll {
                    klog.Infof("ignoring update for catch-all ingress %v/%v and delete old one because of --disable-catch-all", curIng.Namespace, curIng.Name)
                    ingDeleteHandler(old)
                    return
                }

                recorder.Eventf(curIng, corev1.EventTypeNormal, "UPDATE", fmt.Sprintf("Ingress %s/%s", curIng.Namespace, curIng.Name))
            } else {
                klog.V(3).Infof("No changes on ingress %v/%v. Skipping update", curIng.Namespace, curIng.Name)
                return
            }

            store.syncIngress(curIng)
            store.updateSecretIngressMap(curIng)
            store.syncSecrets(curIng)

            updateCh.In() <- Event{
                Type: UpdateEvent,
                Obj:  cur,
            }
        },
    }

    secrEventHandler := cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {...},
        UpdateFunc: func(old, cur interface{}) {...},
        DeleteFunc: func(obj interface{}) {...},
    }

    epEventHandler := cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {...},
        DeleteFunc: func(obj interface{}) {...},
        UpdateFunc: func(old, cur interface{}) {...},
    }

    ......

    cmEventHandler := cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {...},
        UpdateFunc: func(old, cur interface{}) {...},
    }

    podEventHandler := cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {...},
        UpdateFunc: func(old, cur interface{}) {...},
        DeleteFunc: func(obj interface{}) {...},
    }

    serviceHandler := cache.ResourceEventHandlerFuncs{
        UpdateFunc: func(old, cur interface{}) {...},
    }

 // 注册各种类型的eventHandler
    store.informers.Ingress.AddEventHandler(ingEventHandler)
    store.informers.Endpoint.AddEventHandler(epEventHandler)
    store.informers.Secret.AddEventHandler(secrEventHandler)
    store.informers.ConfigMap.AddEventHandler(cmEventHandler)
    store.informers.Service.AddEventHandler(serviceHandler)
    store.informers.Pod.AddEventHandler(podEventHandler)

    // do not wait for informers to read the configmap configuration
    ns, name, _ := k8s.ParseNameNS(configmap)
    cm, err := client.CoreV1().ConfigMaps(ns).Get(context.TODO(), name, metav1.GetOptions{})
    if err != nil {
        klog.Warningf("Unexpected error reading configuration configmap: %v", err)
    }

    store.setConfig(cm)
    return store
}

可以看到，每种类型的informer 基本都有相关的回调方法，包括:
AddFunc: func(obj interface{}) {...},
UpdateFunc: func(old, cur interface{}) {...},
DeleteFunc: func(obj interface{}) {...},

每个方法里面都会往updateCh 写入不同类型的事件（CreateEvent、DeleteEvent、UpdateEvent）
这一步跟store 协程协同工作，informer 通过list&watch 方法监听资源变化，一旦资源有变化则向updateCh 里面写入事件，store 协程循环监听updateCh变化，一旦收到事件则往syncQueue 写入一个task

队列消费

// file : k8s.io/ingress-controller/internal/ingress/controller/nginx.go
// 初始化Queue
n.syncQueue = task.NewTaskQueue(n.syncIngress)

// NewTaskQueue creates a new task queue with the given sync function.
// The sync function is called for every element inserted into the queue.
// 对于每个插入进来的项目都会调用sync function
func NewTaskQueue(syncFn func(interface{}) error) *Queue {
    return NewCustomTaskQueue(syncFn, nil)
}

// NewCustomTaskQueue  
func NewCustomTaskQueue(syncFn func(interface{}) error, fn func(interface{}) (interface{}, error)) *Queue {
    // syncFn(也就是syncIngress)被赋值到Queue.sync 
    q := &Queue{
        queue:      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
        sync:       syncFn,
        workerDone: make(chan bool),
        fn:         fn,
    }

    if fn == nil {
        q.fn = q.defaultKeyFunc
    }

    return q
}

消费Queue队列
核心方法：
t.queue.Get() -> t.sync()

// file: k8s.io/ingress-nginx/internal/ingress/controller/nginx.go
func (n *NGINXController) Start() {
    ......
    go n.syncQueue.Run(time.Second, n.stopCh)
    ......
}

// file: k8s.io/ingress-nginx/internal/task/queue.go
// Run starts processing elements in the queue
func (t *Queue) Run(period time.Duration, stopCh <-chan struct{}) {
   wait.Until(t.worker, period, stopCh)
}

// worker processes work in the queue through sync.
// 消费Queue队列
func (t *Queue) worker() {
   for {
      key, quit := t.queue.Get()
      ......
      ts := time.Now().UnixNano()

      item := key.(Element)
      // 比对最后一次同步的时间戳与Queue中取出item里面带的时间戳，如果小于最后一次同步时间戳则忽略改变更
      if t.lastSync > item.Timestamp {
         klog.V(3).Infof("skipping %v sync (%v > %v)", item.Key, t.lastSync, item.Timestamp)
         t.queue.Forget(key)
         t.queue.Done(key)
         continue
      }

      klog.V(3).Infof("syncing %v", item.Key)
      // 调用syncIngress
      if err := t.sync(key); err != nil {
         klog.Warningf("requeuing %v, err %v", item.Key, err)
         t.queue.AddRateLimited(Element{
            Key:       item.Key,
            Timestamp: time.Now().UnixNano(),
         })
      } else {
         t.queue.Forget(key)
         t.lastSync = ts
      }

      t.queue.Done(key)
   }
}

syncIngress 工作原理

比对线上在跑的配置跟新生成的配置是否相同，并判断是否能够动态重载配置（仅更新endpoint），减少nginx频繁reload带来性能损耗.
pcfg ：当前格式化出来的配置
n.runningConfig ： 当前线上环境运行的配置

比对pcfg 和 n.runningConfig 配置，判断是否可以动态更新配置（仅endpoint列表变化）
(1)支持动态更新配置：调用n.configureDynamically(pcfg)
将backend 列表以json格式post 到/configuration/backends 这个LUA Handler，动态更新endpoint 列表

(2)不支持动态更新配置，调用 n.OnUpdate(*pcfg)
生成临时配置文件
检测临时配置文件语法
diff 临时配置文件与当前线上配置文件
删除临时配置文件
将新生成的配置写入线上配置文件
执行nginx -s reload 重载配置

// file: k8s.io/ingress-nginx/internal/ingress/controller/controller.go
// syncIngress collects all the pieces required to assemble the NGINX
// configuration file and passes the resulting data structures to the backend
// (OnUpdate) when a reload is deemed necessary.
// 组装nginx 配置文件
// 需要reload 时，调用OnUpdate
func (n *NGINXController) syncIngress(interface{}) error {
    ......
   ings := n.store.ListIngresses(nil)
   // 格式化新配置
   hosts, servers, pcfg := n.getConfiguration(ings)
   ......

   // 判断配置是否有变化
   if n.runningConfig.Equal(pcfg) {
      klog.V(3).Infof("No configuration change detected, skipping backend reload.")
      return nil
   }

   ......

   // 配置有变化，则判断是否需要reload nginx
   if !n.IsDynamicConfigurationEnough(pcfg) {
      klog.Infof("Configuration changes detected, backend reload required.")

      // 生成checksum hash值
      hash, _ := hashstructure.Hash(pcfg, &hashstructure.HashOptions{
         TagName: "json",
      })

      pcfg.ConfigurationChecksum = fmt.Sprintf("%v", hash)

      //调用onUpdate 方法
      err := n.OnUpdate(*pcfg)
      ......

      klog.Infof("Backend successfully reloaded.")
      ......
   }

   // 是否首次同步（ingress.Configuration 结构体是否为空）
   isFirstSync := n.runningConfig.Equal(&ingress.Configuration{})
   if isFirstSync {
      // For the initial sync it always takes some time for NGINX to start listening
      // For large configurations it might take a while so we loop and back off
      // 首次初始化需要耗费一定的时间，睡眠1秒
      klog.Info("Initial sync, sleeping for 1 second.")
      time.Sleep(1 * time.Second)
   }

   // 重试机制
   retry := wait.Backoff{
      Steps:    15,
      Duration: 1 * time.Second,
      Factor:   0.8,
      Jitter:   0.1,
   }

   err := wait.ExponentialBackoff(retry, func() (bool, error) {
      // 动态更新nginx 配置
      err := n.configureDynamically(pcfg)
      if err == nil {
         klog.V(2).Infof("Dynamic reconfiguration succeeded.")
         return true, nil
      }

      klog.Warningf("Dynamic reconfiguration failed: %v", err)
      return false, err
   })
   ......

   n.runningConfig = pcfg

   return nil
}

判断是否可以动态更新配置
不需要reload的场景

1. endpoint 变化

需要reload的场景

1. 新增ingress
2. 新增证书配置
3. ingress 增加/删除 PATH
4. 删除ingress、service、secret
5. Secret 更新
6. 部分annotation变更，造成上述状态更新

// file: k8s.io/ingress-contoller/internal/ingress/controller/nginx.go
// IsDynamicConfigurationEnough returns whether a Configuration can be
// dynamically applied, without reloading the backend.
// 判断是否nginx 可以动态重载，不需要执行reload
func (n *NGINXController) IsDynamicConfigurationEnough(pcfg *ingress.Configuration) bool {
   copyOfRunningConfig := *n.runningConfig
   copyOfPcfg := *pcfg

   copyOfRunningConfig.Backends = []*ingress.Backend{}
   copyOfPcfg.Backends = []*ingress.Backend{}

   clearL4serviceEndpoints(©OfRunningConfig)
   clearL4serviceEndpoints(©OfPcfg)

   copyOfRunningConfig.ControllerPodsCount = 0
   copyOfPcfg.ControllerPodsCount = 0

   clearCertificates(©OfRunningConfig)
   clearCertificates(©OfPcfg)

   return copyOfRunningConfig.Equal(©OfPcfg)
}

不能动态更新，调用nginx 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.
// 当监听到配置发生变化，同步循环将调用OnUdate
// 接收到的backend 配置会跟当前配置的configmap 进行合并
func (n *NGINXController) OnUpdate(ingressCfg ingress.Configuration) error {
   cfg := n.store.GetBackendConfiguration()
   cfg.Resolver = n.resolver

   // 生成临时配置
   content, err := n.generateTemplate(cfg, ingressCfg)
   ......

   // 检查配置是否正确
   err = n.testTemplate(content)
   ......

   if klog.V(2) {
      src, _ := ioutil.ReadFile(cfgPath)
      if !bytes.Equal(src, content) {
         tmpfile, err := ioutil.TempFile("", "new-nginx-cfg")
         if err != nil {
            return err
         }
         defer tmpfile.Close()
         // 创建临时配置文件
         err = ioutil.WriteFile(tmpfile.Name(), content, file.ReadWriteByUser)
         ......
         // diff 比对生成的临时配置跟当前生效配置
         diffOutput, err := exec.Command("diff", "-I", "'# Configuration.*'", "-u", cfgPath, tmpfile.Name()).CombinedOutput()
         ......

         klog.Infof("NGINX configuration diff:\n%v", string(diffOutput))

         // 删除临时配置文件
         os.Remove(tmpfile.Name())
      }
   }
   // 将新配置写入cfgPath
   err = ioutil.WriteFile(cfgPath, content, file.ReadWriteByUser)
   ......

   // reload nginx
   o, err := n.command.ExecCommand("-s", "reload").CombinedOutput()
   ......

   return nil
}

动态更新

// file: k8s.io/ingress-contoller/internal/ingress/controller/nginx.go
// configureDynamically encodes new Backends in JSON format and POSTs the
// payload to an internal HTTP endpoint handled by Lua.
// 以json 的格式封装backend 列表并post 到lua API
func (n *NGINXController) configureDynamically(pcfg *ingress.Configuration) error {
   backendsChanged := !reflect.DeepEqual(n.runningConfig.Backends, pcfg.Backends)
   if backendsChanged {
       // 更新endpoint 列表
      err := configureBackends(pcfg.Backends)
      ......
   }

   // 比对TCP/UDP endpoint 列表
   streamConfigurationChanged := !reflect.DeepEqual(n.runningConfig.TCPEndpoints, pcfg.TCPEndpoints) || !reflect.DeepEqual(n.runningConfig.UDPEndpoints, pcfg.UDPEndpoints)
   if streamConfigurationChanged {
      err := updateStreamConfiguration(pcfg.TCPEndpoints, pcfg.UDPEndpoints)
      ......
   }

   if n.runningConfig.ControllerPodsCount != pcfg.ControllerPodsCount {
      // post pod 数目
      statusCode, _, err := nginx.NewPostStatusRequest("/configuration/general", "application/json", ingress.GeneralConfig{
         ControllerPodsCount: pcfg.ControllerPodsCount,
      })
      ......
   }
   // 比对servers 变化
   serversChanged := !reflect.DeepEqual(n.runningConfig.Servers, pcfg.Servers)
   if serversChanged {
      err := configureCertificates(pcfg.Servers)
      ......
   }

   return nil
}

以JSON 格式 POST 调用LUA Handler /configuration/backends

// file: k8s.io/ingress-nginx/internal/controller/nginx.go
func configureBackends(rawBackends []*ingress.Backend) error {
   backends := make([]*ingress.Backend, len(rawBackends))

   for i, backend := range rawBackends {
      var service *apiv1.Service
      if backend.Service != nil {
         service = &apiv1.Service{Spec: backend.Service.Spec}
      }
      luaBackend := &ingress.Backend{
         Name:                 backend.Name,
         Port:                 backend.Port,
         SSLPassthrough:       backend.SSLPassthrough,
         SessionAffinity:      backend.SessionAffinity,
         UpstreamHashBy:       backend.UpstreamHashBy,
         LoadBalancing:        backend.LoadBalancing,
         Service:              service,
         NoServer:             backend.NoServer,
         TrafficShapingPolicy: backend.TrafficShapingPolicy,
         AlternativeBackends:  backend.AlternativeBackends,
      }

      var endpoints []ingress.Endpoint
      for _, endpoint := range backend.Endpoints {
         endpoints = append(endpoints, ingress.Endpoint{
            Address: endpoint.Address,
            Port:    endpoint.Port,
         })
      }

      luaBackend.Endpoints = endpoints
      backends[i] = luaBackend
   }

    // 更新endpoint 列表
   statusCode, _, err := nginx.NewPostStatusRequest("/configuration/backends", "application/json", backends)
   if err != nil {
      return err
   }

   if statusCode != http.StatusCreated {
      return fmt.Errorf("unexpected error code: %d", statusCode)
   }

   return nil
}