1. 程式人生 > >K8S 原始碼探祕 之 nginx-ingress 工作原理分析

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

一、引言

       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 執行機制的簡單理解,歡迎讀者批評指正!