endless 如何實現不停機重啟 Go 程序？

2021 年 7 月 21 日
筆記
Go

轉載請聲明出處哦~，本篇文章發佈於luozhiyun的博客：//www.luozhiyun.com/archives/584

前幾篇文章講解了如何實現一個高效的 HTTP 服務，這次我們來看一下如何實現一個永不不停機的 Go 程序。

前提

事情是這樣的，在一天風和日麗的周末，我正在看 TiDB 源碼的時候，有一位胖友找到我說，Go 是不是每次修改都需要重啟才行？由於我才疏學淺不知道有不停機重啟這個東西，所以回答是的。然後他說，那完全沒有 PHP 好用啊，PHP 修改邏輯完之後直接替換一個文件就可以實現發佈，不需要重啟。我當時只能和他說可以多 Pod 部署，金絲雀發佈等等也可以做到整個服務不停機發佈。但是他最後還是帶着得以意笑容離去。

當時看着他離去的身影我就發誓，我要研究一下 Go 語言的不停機重啟，證明不是 Go 不行，而是我不行 [DOGE] [DOGE] [DOGE]，所以就有了這麼一篇文章。

那麼對於一個不停機重啟 Go 程序我們需要解決以下兩個問題：

進程重啟不需要關閉監聽的端口；
既有請求應當完全處理或者超時；

後面我們會看一下 endless 是如何做到這兩點的。

基本概念

下面先簡單介紹一下兩個知識點，以便後面的開展

信號處理

Go 信號通知通過在 Channel 上發送 os.Signal 值來工作。如我們如果使用 Ctrl+C，那麼會觸發 SIGINT 信號，操作系統會中斷該進程的正常流程，並進入相應的信號處理函數執行操作，完成後再回到中斷的地方繼續執行。

func main() {
    sigs := make(chan os.Signal, 1)
    done := make(chan bool, 1)
    // 監聽信號
    signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) 
    go func() {
        // 接收到信號返回
        sig := <-sigs
        fmt.Println()
        fmt.Println(sig)
        done <- true
    }() 
    fmt.Println("awaiting signal")
    // 等待信號的接收
    <-done
    fmt.Println("exiting")
}

通過上述簡單的幾行代碼，我們就可以監聽 SIGINT 和 SIGTERM 信號。當 Go 接收到操作系統發送過來的信號，那麼會將信號值放入到 sigs 管道中進行處理。

Fork 子進程

在Go語言中 exec 包為我們很好的封裝好了 Fork 調用，並且使用它可以使用 ExtraFiles 很好的繼承父進程已打開的文件。

file := netListener.File() // this returns a Dup()
path := "/path/to/executable"
args := []string{
    "-graceful"}
// 產生 Cmd 實例
cmd := exec.Command(path, args...)
// 標準輸出
cmd.Stdout = os.Stdout
// 標準錯誤輸出
cmd.Stderr = os.Stderr
cmd.ExtraFiles = []*os.File{file}
// 啟動命令
err := cmd.Start()
if err != nil {
    log.Fatalf("gracefulRestart: Failed to launch, error: %v", err)
}

通過調用 exec 包的 Command 命令傳入 path（將要執行的命令路徑）、args （命令的參數）即可返回 Cmd 實例，通過 ExtraFiles 字段指定額外被新進程繼承的已打開文件，最後調用 Start 方法創建子進程。

這裡的 netListener.File會通過系統調用 dup 複製一份 file descriptor 文件描述符。

func Dup(oldfd int) (fd int, err error) {
	r0, _, e1 := Syscall(SYS_DUP, uintptr(oldfd), 0, 0)
	fd = int(r0)
	if e1 != 0 {
		err = errnoErr(e1)
	}
	return
}

我們可以看到 dup 的命令介紹：

dup and dup2 create a copy of the file descriptor oldfd.
After successful return of dup or dup2, the old and new descriptors may
be used interchangeably. They share locks, file position pointers and
flags; for example, if the file position is modified by using lseek on
one of the descriptors, the position is also changed for the other.

The two descriptors do not share the close-on-exec flag, however.

通過上面的描述可以知道，返回的新文件描述符和參數 oldfd 指向同一個文件，共享所有的索性、讀寫指針、各項權限或標誌位等。但是不共享關閉標誌位，也就是說 oldfd 已經關閉了，也不影響寫入新的數據到 newfd 中。

上圖顯示了fork一個子進程，子進程複製父進程的文件描述符表。

endless 不停機重啟示例

我這裡稍微寫一下 endless 的使用示例給沒有用過 endless 的同學看看，熟悉 endless 使用的同學可以跳過。

import (
	"log"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/fvbock/endless"
	"github.com/gorilla/mux"
)

func handler(w http.ResponseWriter, r *http.Request) {
	duration, err := time.ParseDuration(r.FormValue("duration"))
	if err != nil {
		http.Error(w, err.Error(), 400)
		return
	}
	time.Sleep(duration)
	w.Write([]byte("Hello World"))
}

func main() {
	mux1 := mux.NewRouter()
	mux1.HandleFunc("/sleep", handler)

	w := sync.WaitGroup{}
	w.Add(1)
	go func() {
		err := endless.ListenAndServe("127.0.0.1:5003", mux1)
		if err != nil {
			log.Println(err)
		}
		log.Println("Server on 5003 stopped")
		w.Done()
	}()
	w.Wait()
	log.Println("All servers stopped. Exiting.")

	os.Exit(0)
}

下面驗證一下 endless 創建的不停機服務：

# 第一次構建項目
go build main.go
# 運行項目，這時就可以做內容修改了
./endless &
# 請求項目，60s後返回
curl "//127.0.0.1:5003/sleep?duration=60s" &
# 再次構建項目，這裡是新內容
go build main.go
# 重啟，17171為pid
kill -1 17171
# 新API請求
curl "//127.0.0.1:5003/sleep?duration=1s"

運行完上面的命令我們可以看到，對於第一個請求返回的是：Hello world，在發送第二個請求之前，我將 handler 裏面的返回值改成了：Hello world2222，然後進行構建重啟。

由於我設置了 60s 才返回第一個請求，第二個請求設置的是 1s 返回，所以這裡會先返回第二個請求的值，然後再返回第一個請求的值。

整個時間線如下所示：

並且在等待第一個請求返回期間，可以看到同時有兩個進程在跑：

$ ps -ef |grep main
root      84636  80539  0 22:25 pts/2    00:00:00 ./main
root      85423  84636  0 22:26 pts/2    00:00:00 ./main

在第一個請求響應之後，我們再看進程可以發現父進程已經關掉了，實現了父子進程無縫切換：

$ ps -ef |grep main
root      85423      1  0 22:26 pts/2    00:00:00 ./main

實現原理

在實現上，我這裡用的是 endless 的實現方案，所以下面原理和代碼都通過它的代碼進行講解。

我們要做的不停機重啟，實現原理如上圖所示：

監聽 SIGHUP 信號；
收到信號時 fork 子進程（使用相同的啟動命令），將服務監聽的 socket 文件描述符傳遞給子進程；
子進程監聽父進程的 socket，這個時候父進程和子進程都可以接收請求；
子進程啟動成功之後發送 SIGTERM 信號給父進程，父進程停止接收新的連接，等待舊連接處理完成（或超時）；
父進程退出，升級完成；

代碼實現

我們從上面的示例可以看出，endless 的入口是 ListenAndServe 函數：

 func ListenAndServe(addr string, handler http.Handler) error {
    // 初始化 server
	server := NewServer(addr, handler)
    // 監聽以及處理請求
	return server.ListenAndServe()
}

這個方法分為兩部分，先是初始化 server，然後再監聽以及處理請求。

初始化 Server

我們首先看一下一個 endless 服務的 Server 結構體是怎樣：

type endlessServer struct {
	// 用於繼承 http.Server 結構
	http.Server
	// 監聽客戶端請求的 Listener
	EndlessListener  net.Listener  
	// 用於記錄還有多少客戶端請求沒有完成
	wg               sync.WaitGroup
	// 用於接收信號的管道
	sigChan          chan os.Signal
	// 用於重啟時標誌本進程是否是為一個新進程
	isChild          bool
	// 當前進程的狀態
	state            uint8 
    ...
}

這個 endlessServer 除了繼承 http.Server 所有字段以外，因為還需要監聽信號以及判斷是不是一個新的進程，所以添加了幾個狀態位的字段：

wg：標記還有多少客戶端請求沒有完成；
sigChan：用於接收信號的管道；
isChild：用於重啟時標誌本進程是否是為一個新進程；
state：當前進程的狀態。

下面我們看看如何初始化 endlessServer ：

func NewServer(addr string, handler http.Handler) (srv *endlessServer) {
	runningServerReg.Lock()
	defer runningServerReg.Unlock()
 	
    socketOrder = os.Getenv("ENDLESS_SOCKET_ORDER")
	// 根據環境變量判斷是不是子進程
	isChild = os.Getenv("ENDLESS_CONTINUE") != "" 
    // 由於支持多 server，所以這裡需要設置一下 server 的順序
    if len(socketOrder) > 0 {
		for i, addr := range strings.Split(socketOrder, ",") {
			socketPtrOffsetMap[addr] = uint(i)
		}
	} else {
		socketPtrOffsetMap[addr] = uint(len(runningServersOrder))
	}

	srv = &endlessServer{
		wg:      sync.WaitGroup{},
		sigChan: make(chan os.Signal),
		isChild: isChild,
		...
		state: STATE_INIT,
		lock:  &sync.RWMutex{},
	}

	srv.Server.Addr = addr
	srv.Server.ReadTimeout = DefaultReadTimeOut
	srv.Server.WriteTimeout = DefaultWriteTimeOut
	srv.Server.MaxHeaderBytes = DefaultMaxHeaderBytes
	srv.Server.Handler = handler
 
	runningServers[addr] = srv
	...
	return
}

這裡初始化都是我們在 net/http 裏面看到的一些常見的參數，包括 ReadTimeout 讀取超時時間、WriteTimeout 寫入超時時間、Handler 請求處理器等，不熟悉的可以看一下這篇：《一文說透 Go 語言 HTTP 標準庫 //www.luozhiyun.com/archives/561 》。

需要注意的是，這裡是通過 ENDLESS_CONTINUE 環境變量來判斷是否是個子進程，這個環境變量會在 fork 子進程的時候寫入。因為 endless 是支持多 server 的，所以需要用 ENDLESS_SOCKET_ORDER變量來判斷一下 server 的順序。

ListenAndServe

func (srv *endlessServer) ListenAndServe() (err error) {
	addr := srv.Addr
	if addr == "" {
		addr = ":http"
	}
	// 異步處理信號量
	go srv.handleSignals()
	// 獲取端口監聽
	l, err := srv.getListener(addr)
	if err != nil {
		log.Println(err)
		return
	}
	// 將監聽轉為 endlessListener
	srv.EndlessListener = newEndlessListener(l, srv)

	// 如果是子進程，那麼發送 SIGTERM 信號給父進程
	if srv.isChild {
		syscall.Kill(syscall.Getppid(), syscall.SIGTERM)
	}

	srv.BeforeBegin(srv.Addr)
	// 響應Listener監聽，執行對應請求邏輯
	return srv.Serve()
}

這個方法其實和 net/http 庫是比較像的，首先獲取端口監聽，然後調用 Serve 處理請求發送過來的數據，大家可以打開文章《一文說透 Go 語言 HTTP 標準庫 //www.luozhiyun.com/archives/561 》對比一下和 endless 的異同。

但是還是有幾點不一樣的，endless 為了做到平滑重啟需要用到信號監聽處理，並且在 getListener 的時候也不一樣，如果是子進程需要繼承到父進程的 listen fd，這樣才能做到不關閉監聽的端口。

handleSignals 信號處理

信號處理主要是信號的一個監聽，然後根據不同的信號循環處理。

func (srv *endlessServer) handleSignals() {
	var sig os.Signal
	// 註冊信號監聽
	signal.Notify(
		srv.sigChan,
		hookableSignals...,
	)
	// 獲取pid
	pid := syscall.Getpid()
	for {
		sig = <-srv.sigChan
		// 在處理信號之前觸發hook
		srv.signalHooks(PRE_SIGNAL, sig)
		switch sig {
		// 接收到平滑重啟信號
		case syscall.SIGHUP:
			log.Println(pid, "Received SIGHUP. forking.")
			err := srv.fork()
			if err != nil {
				log.Println("Fork err:", err)
			} 
		// 停機信號
		case syscall.SIGINT:
			log.Println(pid, "Received SIGINT.")
			srv.shutdown()
		// 停機信號
		case syscall.SIGTERM:
			log.Println(pid, "Received SIGTERM.")
			srv.shutdown()
		...
		// 在處理信號之後觸發hook
		srv.signalHooks(POST_SIGNAL, sig)
	}
}

這一部分的代碼十分簡潔，當我們用kill -1 $pid 的時候這裡 srv.sigChan 就會接收到相應的信號，並進入到 case syscall.SIGHUP 這塊邏輯代碼中。

需要注意的是，在上面的 ListenAndServe 方法中子進程會像父進程發送 syscall.SIGTERM 信號也會在這裡被處理，執行的是 shutdown 停機邏輯。

在進入到 case syscall.SIGHUP 這塊邏輯代碼之後會調用 fork 函數，下面我們再來看看 fork 邏輯：

func (srv *endlessServer) fork() (err error) {
	runningServerReg.Lock()
	defer runningServerReg.Unlock()

	// 校驗是否已經fork過
	if runningServersForked {
		return errors.New("Another process already forked. Ignoring this one.")
	} 
	runningServersForked = true

	var files = make([]*os.File, len(runningServers))
	var orderArgs = make([]string, len(runningServers))
	// 因為有多 server 的情況，所以獲取所有 listen fd
	for _, srvPtr := range runningServers { 
		switch srvPtr.EndlessListener.(type) {
		case *endlessListener: 
			files[socketPtrOffsetMap[srvPtr.Server.Addr]] = srvPtr.EndlessListener.(*endlessListener).File()
		default: 
			files[socketPtrOffsetMap[srvPtr.Server.Addr]] = srvPtr.tlsInnerListener.File()
		}
		orderArgs[socketPtrOffsetMap[srvPtr.Server.Addr]] = srvPtr.Server.Addr
	}
	// 環境變量
	env := append(
		os.Environ(),
    // 啟動endless 的時候，會根據這個參數來判斷是否是子進程
		"ENDLESS_CONTINUE=1",
	)
	if len(runningServers) > 1 {
		env = append(env, fmt.Sprintf(`ENDLESS_SOCKET_ORDER=%s`, strings.Join(orderArgs, ",")))
	}

	// 程序運行路徑
	path := os.Args[0]
	var args []string
	// 參數
	if len(os.Args) > 1 {
		args = os.Args[1:]
	}

	cmd := exec.Command(path, args...)
	// 標準輸出
	cmd.Stdout = os.Stdout
	// 錯誤
	cmd.Stderr = os.Stderr
	cmd.ExtraFiles = files
	cmd.Env = env  
	err = cmd.Start()
	if err != nil {
		log.Fatalf("Restart: Failed to launch, error: %v", err)
	} 
	return
}

fork 這塊代碼首先會根據 server 來獲取不同的 listen fd 然後封裝到 files 列表中，然後在調用 cmd 的時候將文件描述符傳入到 ExtraFiles 參數中，這樣子進程就可以無縫託管到父進程監聽的端口。

需要注意的是，env 參數列表中有一個 ENDLESS_CONTINUE 參數，這個參數會在 endless 啟動的時候做校驗：

func NewServer(addr string, handler http.Handler) (srv *endlessServer) {
	runningServerReg.Lock()
	defer runningServerReg.Unlock()

	socketOrder = os.Getenv("ENDLESS_SOCKET_ORDER")
	isChild = os.Getenv("ENDLESS_CONTINUE") != ""
  ...
}

下面我們再看看接收到 SIGTERM 信號後，shutdown 會怎麼做：

func (srv *endlessServer) shutdown() {
	if srv.getState() != STATE_RUNNING {
		return
	}

	srv.setState(STATE_SHUTTING_DOWN)
    // 默認 DefaultHammerTime 為 60秒
	if DefaultHammerTime >= 0 {
		go srv.hammerTime(DefaultHammerTime)
	}
	// 關閉存活的連接
	srv.SetKeepAlivesEnabled(false)
	err := srv.EndlessListener.Close()
	if err != nil {
		log.Println(syscall.Getpid(), "Listener.Close() error:", err)
	} else {
		log.Println(syscall.Getpid(), srv.EndlessListener.Addr(), "Listener closed.")
	}
}

shutdown 這裡會先將連接關閉，因為這個時候子進程已經啟動了，所以不再處理請求，需要把端口的監聽關了。這裡還會異步調用 srv.hammerTime 方法等待60秒把父進程的請求處理完畢才關閉父進程。

getListener 獲取端口監聽

func (srv *endlessServer) getListener(laddr string) (l net.Listener, err error) {
	// 如果是子進程
	if srv.isChild {
		var ptrOffset uint = 0
		runningServerReg.RLock()
		defer runningServerReg.RUnlock()
		// 這裡還是處理多個 server 的情況
		if len(socketPtrOffsetMap) > 0 {
			// 根據server 的順序來獲取 listen fd 的序號
			ptrOffset = socketPtrOffsetMap[laddr] 
		}
		// fd 0，1，2是預留給 標準輸入、輸出和錯誤的，所以從3開始
		f := os.NewFile(uintptr(3+ptrOffset), "")
		l, err = net.FileListener(f)
		if err != nil {
			err = fmt.Errorf("net.FileListener error: %v", err)
			return
		}
	} else {
		// 父進程 直接返回 listener
		l, err = net.Listen("tcp", laddr)
		if err != nil {
			err = fmt.Errorf("net.Listen error: %v", err)
			return
		}
	}
	return
}

這裡如果是父進程沒什麼好說的，直接創建一個端口監聽並返回就好了。

但是對於子進程來說是有一些繞，首先說一下 os.NewFile 的參數為什麼要從3開始。因為子進程在繼承父進程的 fd 的時候0，1，2是預留給標準輸入、輸出和錯誤的，所以父進程給的第一個fd在子進程里順序排就是從3開始了，又因為 fork 的時候cmd.ExtraFiles 參數傳入的是一個 files，如果有多個 server 那麼會依次從3開始遞增。

如下圖，前三個 fd 是預留給標準輸入、輸出和錯誤的，fd 3 是根據傳入 ExtraFiles 的數組列表依次遞增的。

其實這裡我們也可以用開頭的例子做一下試驗：

# 第一次構建項目
go build main.go
# 運行項目，這時就可以做內容修改了
./endless &
# 這個時候我們看看父進程打開的文件
lsof  -P -p 17116
COMMAND   PID USER   FD      TYPE  DEVICE SIZE/OFF     NODE NAME
...
main    18942 root    0u      CHR   136,2      0t0        5 /dev/pts/2
main    18942 root    1u      CHR   136,2      0t0        5 /dev/pts/2
main    18942 root    2u      CHR   136,2      0t0        5 /dev/pts/2
main    18942 root    3u     IPv4 2223979      0t0      TCP localhost:5003 (LISTEN)
# 請求項目，60s後返回
curl "//127.0.0.1:5003/sleep?duration=60s" & 
# 重啟，17116為父進程pid
kill -1 17116
# 然後我們看一下 main 程序的進程應該有兩個
ps -ef |grep ./main
root      17116  80539  0 04:19 pts/2    00:00:00 ./main
root      18110  17116  0 04:21 pts/2    00:00:00 ./main
# 可以看到子進程pid 為18110,我們看看該進程打開的文件
lsof  -P -p 18110
COMMAND   PID USER   FD      TYPE  DEVICE SIZE/OFF     NODE NAME
...
main    19073 root    0r      CHR     1,3      0t0     1028 /dev/null
main    19073 root    1u      CHR   136,2      0t0        5 /dev/pts/2
main    19073 root    2u      CHR   136,2      0t0        5 /dev/pts/2
main    19073 root    3u     IPv4 2223979      0t0      TCP localhost:5003 (LISTEN)
main    19073 root    4u     IPv4 2223979      0t0      TCP localhost:5003 (LISTEN)
# 新API請求
curl "//127.0.0.1:5003/sleep?duration=1s"

總結

通過上面的介紹，我們通過 endless 學習了在 Go 服務中如何做到不停機也可以重啟服務，相信這個功能在很多場景下都會用到，沒用到的同學也可以嘗試在自己的系統上玩一下。

熱重啟總的來說它允許服務重啟期間，不中斷已經建立的連接，老服務進程不再接受新連接請求，新連接請求將在新服務進程中受理。對於原服務進程中已經建立的連接，也可以將其設為讀關閉，等待平滑處理完連接上的請求及連接空閑後再行退出。

通過這種方式，可以保證已建立的連接不中斷，新的服務進程也可以正常接受連接請求。

Reference

//goteleport.com/blog/golang-ssh-bastion-graceful-restarts/

//grisha.org/blog/2014/06/03/graceful-restart-in-golang/

//stackoverflow.com/questions/28370646/how-do-i-fork-a-go-process/28371586#28371586

//xixiliguo.github.io/post/golang-exec/

//github.com/fvbock/endless

//golang.org/pkg/os/signal/

//stackoverflow.com/questions/11635219/dup2-dup-why-would-i-need-to-duplicate-a-file-descriptor

//www.hitzhangjie.pro/blog/2020-08-28-go程序如何實現熱重啟/

Tags: Go