K8S 原始碼探祕之 nginx-ingress 工作原理分析

阿新 • • 發佈：2018-11-29

一、引言

Nginx-ingress 是 Kubernetes 生態中的重要成員，主要負責向外暴露服務，同時提供負載均衡等附加功能；

截至目前，nginx-ingress 已經能夠完成 7/4 層的代理功能（4 層代理基於 ConfigMap，感覺還有改進的空間）；

Nginx 的 7 層反向代理模式，可以簡單用下圖表示：

Nginx 對後端執行的服務（Service1、Service2）提供反向代理，在配置檔案中配置了域名與後端服務 Endpoints 的對應關係。客戶端通過使用 DNS 服務或者直接配置本地的 hosts 檔案，將域名都對映到 Nginx 代理伺服器。當客戶端訪問 service1.com 時，瀏覽器會把包含域名的請求傳送給 nginx 伺服器，nginx 伺服器根據傳來的域名，選擇對應的 Service，這裡就是選擇 Service 1 後端服務，然後根據一定的負載均衡策略，選擇 Service1 中的某個容器接收來自客戶端的請求並作出響應。過程很簡單，nginx 在整個過程中彷彿是一臺根據域名進行請求轉發的“路由器”，這也就是7層代理的整體工作流程了！

對於 Nginx 反向代理做了什麼，我們已經大概瞭解了。在 k8s 系統中，後端服務的變化是十分頻繁的，單純依靠人工來更新nginx 的配置檔案幾乎不可能，nginx-ingress 由此應運而生。Nginx-ingress 通過監視 k8s 的資源狀態變化實現對 nginx 配置檔案的自動更新，下面本文就來分析下其工作原理。

二、nginx-ingress 工作流程分析

首先，上一張整體工作模式架構圖（只關注配置同步更新）

不考慮 nginx 狀態收集等附件功能，nginx-ingress 模組在執行時主要包括三個主體：NginxController、Store、SyncQueue。其中，Store 主要負責從 kubernetes APIServer 收集執行時資訊，感知各類資源（如 ingress、service等）的變化，並及時將更新事件訊息（event）寫入一個環形管道；SyncQueue 協程定期掃描 syncQueue 佇列，發現有任務就執行更新操作，即藉助 Store 完成最新執行資料的拉取，然後根據一定的規則產生新的 nginx 配置，（有些更新必須 reload，就本地寫入新配置，執行 reload），然後執行動態更新操作，即構造 POST 資料，向本地 Nginx Lua 服務模組傳送 post 請求，實現配置更新；NginxController 作為中間的聯絡者，監聽 updateChannel，一旦收到配置更新事件，就向同步佇列 syncQueue 裡寫入一個更新請求。

下邊，我們就結合程式碼來分析一遍以上的流程：

首先，來到程式入口處，cmd/nginx/main.go

func main() {
	...
	ngx := controller.NewNGINXController(conf, mc, fs)
	...
	ngx.Start()
}

為了避免陷入，我們只關注最主要的程式碼。在 main 函式中，程式首先構造了 NginxController，並執行了其 Start 方法，啟動了 Controller 主程式。

我們具體看下，ngx.Start() 到底做了什麼，跟蹤到 internal/ingress/controller/nginx.go#Start()

func (n *NGINXController) Start() {
	...
	n.store.Run(n.stopCh)
	...
	go n.syncQueue.Run(time.Second, n.stopCh)
	...
	for {
		select {
		...
		case event := <-n.updateCh.Out():
			if n.isShuttingDown {
				break
			}
			if evt, ok := event.(store.Event); ok {
				if evt.Type == store.ConfigurationEvent {
					n.syncQueue.EnqueueTask(task.GetDummyObject("configmap-change"))
					continue
				}
				n.syncQueue.EnqueueSkippableTask(evt.Obj)
			} 
		...
	}
}

可以看到，NginxController 首先啟動了 Store 協程，然後啟動了 syncQueue 協程，最後監聽 updateCh，當收到事件後，經過簡單判斷就向 syncQueue 寫入了一個 task。

再來看 Store 協程，跟蹤到 internal/ingress/controller/store/store.go#Run()

func (s k8sStore) Run(stopCh chan struct{}) {
	s.informers.Run(stopCh)
	...
}

可以看到，繼續呼叫了 informer 的 Run 方法，繼續跟蹤，還在這個檔案，移步到 148 行左右

// Run initiates the synchronization of the informers against the API server.
func (i *Informer) Run(stopCh chan struct{}) {
	go i.Endpoint.Run(stopCh)
	go i.Service.Run(stopCh)
	go i.Secret.Run(stopCh)
	go i.ConfigMap.Run(stopCh)
	...
	go i.Ingress.Run(stopCh)
	...
}

我們不難發現，informer 的 Run 方法，會起更多的協程，去監聽不同資源的變化，包括 Endpoint、Service、Secret、ConfigMap、Ingress。我們以 Ingress 為例，跟蹤到其定義處，仍在這個檔案，找到 New() 方法

// New creates a new object store to be used in the ingress controller
func New(... updateCh *channels.RingChannel ...) Storer {
	...
	store.informers.Ingress = infFactory.Extensions().V1beta1().Ingresses().Informer()
	...
	ingEventHandler := 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{}) {
			...
			updateCh.In() <- Event{
				Type: UpdateEvent,
				Obj:  cur,
			}
		},
	}
	...
	store.informers.Ingress.AddEventHandler(ingEventHandler)
	...
}

可以看出，Ingress 協程定義了監聽 ingress 資訊的 informer 物件，並註冊了相關事件的回撥方法，在回撥方法內向之前提到的 updateCh 寫入了事件，進而也就達到了當資源變化時通知 Controller 主程向同步佇列寫入task的目的。

反過頭來，看一下 syncQueue ，首先找到其定義，跟蹤到 internal/ingress/controller/nginx.go#NewNGINXController()

// NewNGINXController creates a new NGINX Ingress controller.
func NewNGINXController(config *Configuration, mc metric.Collector, fs file.Filesystem) *NGINXController {
	...
	n.syncQueue = task.NewTaskQueue(n.syncIngress)
	...
}

不難發現，佇列的建立是通過 task.NewTaskQueue() 完成的，而且傳入了關鍵的處理函式 n.syncIngress

繼續跟蹤到 internal/task/queue.go#NewTaskQueue()

// NewTaskQueue creates a new task queue with the given sync function.
// The sync function is called for every element inserted into the queue.
func NewTaskQueue(syncFn func(interface{}) error) *Queue {
	return NewCustomTaskQueue(syncFn, nil)
}

// NewCustomTaskQueue ...
func NewCustomTaskQueue(syncFn func(interface{}) error, fn func(interface{}) (interface{}, error)) *Queue {
	q := &Queue{
		queue:      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		sync:       syncFn,
		workerDone: make(chan bool),
		fn:         fn,
	}
	...
	return q
}

可以看出，傳入的處理函式 n.syncIngress 被賦值給 Queue 的 sync 屬性了。實際上，syncQueue 的執行就是在反覆執行該方法以消費佇列裡的元素。Queue 的 Run 定義可以在本檔案中找到：

// 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.
func (t *Queue) worker() {
	for {
		key, quit := t.queue.Get()
		...
		if err := t.sync(key); err != nil {
			t.queue.AddRateLimited(Element{
				Key:       item.Key,
				Timestamp: time.Now().UnixNano(),
			})
		} else {
			t.queue.Forget(key)
			t.lastSync = ts
		}

		t.queue.Done(key)
	}
}

同步佇列協程的主要工作就是定期取出佇列裡的元素，並利用傳入的 n.syncIngress （即 t.sync(key)）方法處理佇列裡的元素。

n.syncIngress 方法的定義在 internal/ingress/controller/controller.go#syncIngress()

// 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.
func (n *NGINXController) syncIngress(interface{}) error {

	// 獲取最新配置資訊
	....
	// 構造 nginx 配置
	pcfg := &ingress.Configuration{
		Backends:              upstreams,
		Servers:               servers,
		PassthroughBackends:   passUpstreams,
		BackendConfigChecksum: n.store.GetBackendConfiguration().Checksum,
	}
	...
	// 不能避免 reload，就執行 reload 更新配置
	if !n.IsDynamicConfigurationEnough(pcfg) {
		...
		err := n.OnUpdate(*pcfg)
		...
	}
	...
	// 動態更新配置
	err := wait.ExponentialBackoff(retry, func() (bool, error) {
		err := configureDynamically(pcfg, n.cfg.ListenPorts.Status, n.cfg.DynamicCertificatesEnabled)
		...
	})
	...
}

reload 配置的函式定義位於 internal/ingress/controller/nginx.go#OnUpdate()，具體就不解釋了，讀者可自行查閱。

這裡主要看下動態更新是怎麼完成的，函式定義位於 internal/ingress/controller/nginx.go#configureDynamically()

// configureDynamically encodes new Backends in JSON format and POSTs the
// payload to an internal HTTP endpoint handled by Lua.
func configureDynamically(pcfg *ingress.Configuration, port int, isDynamicCertificatesEnabled bool) error {
	backends := make([]*ingress.Backend, len(pcfg.Backends))

	for i, backend := range pcfg.Backends {
		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
	}

	url := fmt.Sprintf("http://localhost:%d/configuration/backends", port)
	err := post(url, backends)
	if err != nil {
		return err
	}

	if isDynamicCertificatesEnabled {
		err = configureCertificates(pcfg, port)
		if err != nil {
			return err
		}
	}

	return nil
}

可以看出，確實是通過請求 Lua 後端來實現動態配置更新的，使用的是典型的 http post 方法。後續的動態更新動作轉交給 Lua 模組負責。因為 Lua 以模組形式嵌入 Nginx 執行，因此其更新配置也就在一定程度上避免了 reload。

目前來看，具體 Serive Pod 的變化不會導致 reload 的發生，因為服務內容器的負載均衡已經移交給 Lua 處理，而 Service 的增減變化仍通過配置檔案形式更新，會導致 reload 發生。這可以從 nginx 的配置檔案看出：

...
http {
...
   upstream upstream_balancer {
       server 0.0.0.1; # placeholder
       balancer_by_lua_block {
           balancer.balance()
       }
       keepalive 32;
   }

   server {
       server_name wordpress.local ;
       listen 80;
       listen [::]:80;
       set $proxy_upstream_name "-";
       location / {
       ...
           proxy_pass http://upstream_balancer;
           proxy_redirect off;
       }
   }
}

可以看到當添加了 wordpress 服務後，在 nginx 配置中確實添加了一個 server 條目，但該條目中沒有具體指出後端的容器地址，而是指向了一個叫 upstream_balancer 的地址，這個 balancer 其實是由 Lua 動態提供路由的。既然沒有實際的容器後端在配置檔案中進行配置，自然地，服務中容器數量的增減變化也就不必修改 nginx 配置檔案了，這就是免 reload 的關鍵！簡單推測，Lua 模組所做的就是維持一個服務到容器的對映關係，動態地提供負載均衡路由。

官方關於避免 Reload 的介紹：https://kubernetes.github.io/ingress-nginx/how-it-works/#avoiding-reloads

以上就是博主對 Nginx-Ingress 執行機制的簡單理解，歡迎讀者批評指正！

K8S 原始碼探祕之 nginx-ingress 工作原理分析

K8S 原始碼探祕之 nginx-ingress 工作原理分析

K8S 原始碼探祕之 kubeadm join 執行流程分析

K8S 原始碼探祕之 kubeadm init 執行流程分析

K8S 原始碼探祕之 kubelet 建立 Pod 的工作原理

K8S 原始碼探祕之 kubeadm upgrade apply 執行流程分析

K8S 原始碼探祕之 kubelet 同步 Node 狀態（kubelet 心跳機制分析）

spring原始碼分析之@ImportSelector、@Import、ImportResource工作原理分析

shell編程培訓之shell的工作原理

Android 基於Netty的消息推送方案之概念和工作原理(二)

代理服務器之趣談工作原理

SCM之Git的工作原理及常規操作

hadoop之hdfs及其工作原理

Android 基於Netty的訊息推送方案之概念和工作原理(二)

spring原始碼學習之路---IOC實現原理（二）

Linux：驅動之字元裝置工作原理（未完）

uWSGI+django+nginx的工作原理流程

Android的訊息機制之ThreadLocal的工作原理

深入理解Tomcat系列之六 Servlet工作原理

Hadoop 之 MapReduce 的工作原理及其倒排索引的建立

Nginx的工作原理

K8S 原始碼探祕 之 nginx-ingress 工作原理分析

相關推薦

K8S 原始碼探祕之 nginx-ingress 工作原理分析