怎么使用Go重構(gòu)流式日志網(wǎng)關(guān)呢？

項(xiàng)目背景

分享之前，先來簡(jiǎn)單介紹下該項(xiàng)目在流式日志處理鏈路中所處的位置。



流式日志網(wǎng)關(guān)的主要功能是提供 HTTP 接口，接收 CDN 邊緣節(jié)點(diǎn)上報(bào)的各類日志（訪問日志/報(bào)錯(cuò)日志/計(jì)費(fèi)日志等），將日志作預(yù)處理并分流到多個(gè)的 Kafka 集群和 Topic 中。

越來越多的客戶要求提供實(shí)時(shí)日志支持，業(yè)務(wù)量的增加讓機(jī)器資源的消耗也與日俱增，最先暴露出了流式日志處理鏈路的一大瓶頸——帶寬資源。 可以通過給集群擴(kuò)充更多的機(jī)器來提升集群總傳輸帶寬，但基于成本考量，重中之重是先優(yōu)化網(wǎng)關(guān)程序。

舊版網(wǎng)關(guān)項(xiàng)目

項(xiàng)目代號(hào) Chopper ，其基于另一個(gè)內(nèi)部 OpenResty 項(xiàng)目框架來開發(fā)的。其亮點(diǎn)功能有：支持從 Consul 、Redis 等其他外部系統(tǒng)熱加載配置及動(dòng)態(tài)生效；能夠加載 Lua 腳本實(shí)現(xiàn)靈活的日志預(yù)處理能力。 其 Kafka 生產(chǎn)者客戶端基于 doujiang24/lua-resty-kafka 實(shí)現(xiàn)。經(jīng)過實(shí)踐考驗(yàn)，Chopper 的吞吐量是滿足現(xiàn)階段需求的。

存在的問題

1. 關(guān)鍵依賴庫的社區(qū)活躍度低 lua-resty-kafka 的社區(qū)活躍度較低，至今仍然處在實(shí)驗(yàn)階段；而且它用作 Kafka 生產(chǎn)者客戶端目前沒有支持消息壓縮功能，而這在其他語言實(shí)現(xiàn)的 Kafka 客戶端中都是標(biāo)準(zhǔn)的選項(xiàng)。

2. 內(nèi)存使用不節(jié)制 單實(shí)例部署配置 4 核 8 G，僅少量請(qǐng)求訪問后，內(nèi)存占用就穩(wěn)定在 2G 而沒有釋放。

3. 配置文件可維護(hù)性差 實(shí)際線上用到 Consul 作為配置中心，采用篇幅很長(zhǎng)的 JSON 格式配置文件，不利于運(yùn)維。另外在 Consul 修改配置沒有回退功能，是一個(gè)高風(fēng)險(xiǎn)操作。 好在目前日志網(wǎng)關(guān)的功能并不復(fù)雜，所以我們決定重構(gòu)它。

新項(xiàng)目啟動(dòng)

眾所周知， Go 語言擁有獨(dú)特的高并發(fā)模型、較低的上手難度和豐富的第三方生態(tài)。而且我們小組成員都有 Go 項(xiàng)目的開發(fā)經(jīng)驗(yàn)，所以我們選擇使用基于 Go 語言的技術(shù)棧來重新構(gòu)建 Chopper 項(xiàng)目，所以新項(xiàng)目命名為 chopper-go 。

需求梳理及概要設(shè)計(jì)

重新構(gòu)建一個(gè)線上項(xiàng)目的基本原則是，功能上要完全兼容，最好能夠?qū)崿F(xiàn)線上服務(wù)的無縫升級(jí)替換。

原版核心模塊的設(shè)計(jì)

Chopper 的核心功能是將接收到的 HTTP 請(qǐng)求分流到特定 Kafka 集群及其 Topic 中。

一、HTTP 接口部分

只開放了唯一一個(gè)對(duì)外的 API ，功能很簡(jiǎn)單：

請(qǐng)求方式：POST 請(qǐng)求路徑：/log/repo/{repo_name}請(qǐng)求體: 多行日志，滿足 JSONL 格式（即每行一條 JSON ，多行按換行符 分隔）。相應(yīng)狀態(tài)碼：- 200：投遞成功。- 5xx：投遞失敗需要重試。參數(shù)解釋: - repo_name: 對(duì)應(yīng) repo 配置名稱。

二、業(yè)務(wù)配置部分

每一類業(yè)務(wù)抽象為一個(gè) repo 配置。Repo 配置由三部分構(gòu)成：constraint、processor、kafka。 constraint 是一個(gè)對(duì)象，可以配置對(duì)日志字段的一些約束條件，不滿足條件的日志會(huì)被丟棄。 processor 是一個(gè)列表，可以組合多個(gè)處理模塊，程序?qū)错樞蛞来螌?duì)請(qǐng)求中的每條日志進(jìn)行處理。實(shí)現(xiàn)了如下幾種 processor 類型：

decoder , 配置原始數(shù)據(jù)按哪種格式反序列化到 Lua table ，但只實(shí)現(xiàn)了 JSON decoder。

splitter，配置分隔日志字段的字符。

assigner，配置一組字段名映射關(guān)系，需要與 spliter 配合。

executer， 配置額外的 lua 腳本名稱，通過動(dòng)態(tài)加載其他 lua 腳本實(shí)現(xiàn)更靈活的處理邏輯。

kafka 是一個(gè)對(duì)象，可以配置當(dāng)前業(yè)務(wù)相關(guān)聯(lián)的 Kafka 集群名，默認(rèn)投遞的 Topic ，以及生產(chǎn)者客戶端的工作模式（同步或者異步）。 新版本的改動(dòng) HTTP 接口沿用原先的設(shè)計(jì)，在業(yè)務(wù)配置部分做了一些改動(dòng)：

processor 改名為 executers ，實(shí)現(xiàn)幾個(gè)通用功能的日志處理模塊，方便組合使用。

kafka 配置中關(guān)聯(lián)的不再是集群名，而是 Kafka 生產(chǎn)者客戶端的配置標(biāo)簽。

原先保存 kafka 集群連接配置信息的配置塊，改為保存 kafka 生產(chǎn)者客戶端的配置塊，統(tǒng)一在一個(gè)配置塊區(qū)域初始化所有用到的 kafka 生產(chǎn)者客戶端。

一點(diǎn)妥協(xié)（做減法）

為了縮短新項(xiàng)目的開發(fā)周期，對(duì)原始項(xiàng)目的一些不太重要的特性我們做了一些取舍。

取消動(dòng)態(tài)腳本功能

Go 是靜態(tài)語言沒有 Lua 動(dòng)態(tài)語言那么靈活，要加載執(zhí)行動(dòng)態(tài)腳本有一定的實(shí)現(xiàn)難度，且日志處理性能沒有保障。 線上只有極少數(shù)業(yè)務(wù)在 processor 中配置了 executor，且這些 executor 的 Lua 腳本實(shí)現(xiàn)相近，完全可以抽取出通用的代碼。

不支持外部配置中心

為了讓發(fā)布和回退有記錄可回溯，從 Consul 等配置中心熱加載服務(wù)配置的功能我們也去掉了。利用好容器平臺(tái)的金絲雀發(fā)布功能，就能將服務(wù)更新的影響降到最低。

不支持復(fù)雜的路由重寫

OpenResty 項(xiàng)目?jī)?nèi)置 Nginx 可以利用 Nginx 強(qiáng)大的配置實(shí)現(xiàn)豐富的路由 rewrite 功能，就具體使用場(chǎng)景而言，我們只需要簡(jiǎn)單的路由映射即可。況且更復(fù)雜的需求也可以由上一級(jí)網(wǎng)關(guān)完成。

選擇合適的開源庫

Web 框架的選擇

最終我們選擇了 Gin 。原因是用得多比較熟，而且文檔看著舒服。

Kafka 生產(chǎn)者客戶端的選擇

社區(qū)中熱度最高的幾款 Go Kafka 客戶端庫：

segmentio/kafka-go

Shopify/sarama

confluentinc/confluent-kafka-go

實(shí)際上三款客戶端庫我們?cè)跉v史項(xiàng)目中都使用過，其中 kafka-go 的 API 是三者中最簡(jiǎn)潔易用的，我們的多個(gè)消費(fèi)端程序都是基于它實(shí)現(xiàn)的。

但是在 chopper-go 中僅需要用到生產(chǎn)者客戶端，我們沒有選擇 kafka-go 。

那是因?yàn)槲覀冏隽艘恍┗鶞?zhǔn)測(cè)試，發(fā)現(xiàn) kafka-go 的生產(chǎn)者客戶端存在性能風(fēng)險(xiǎn)：?jiǎn)⒂?async 模式時(shí)盡管消息發(fā)送特別快，但是內(nèi)存占用也增長(zhǎng)特別快。

最終我們選擇 sarama ，一方面是性能很穩(wěn)定，另一方面是它開放的 API 較多，但是用起來確實(shí)有點(diǎn)費(fèi)勁。

測(cè)試框架的選擇

程序的可靠性，一定需要測(cè)試來保證。除了編寫小模塊中編寫單元從測(cè)試，我們對(duì)整個(gè)日志網(wǎng)關(guān)服務(wù)還要做集成測(cè)試。集成測(cè)試涉及到一些外部服務(wù)依賴，此項(xiàng)目中主要的外部依賴是 Kafka 和 Zookeeper 。

利用 Docker 可以很方便的拉起測(cè)試環(huán)境，我們注意到了兩款可以用來在 Go test 中編寫集成測(cè)試的庫：

ory/dockertest

testcontainers/testcontainers-go

使用下來，我們最終選擇了 testcontainers-go，簡(jiǎn)單介紹下原因： 在編寫集成測(cè)試時(shí)，我們需要有個(gè)等待機(jī)制來確保依賴服務(wù)的容器是否準(zhǔn)備就緒，并以此控制測(cè)試流程，以及測(cè)試結(jié)束后需要把測(cè)試開啟的臨時(shí)容器都清理干凈。 testcontainers-go 的設(shè)計(jì)要優(yōu)于 dockertest 。

testcontainers-go 提供一個(gè) wait 子包，可以配置多種等待策略來確保依賴服務(wù)就緒，以及測(cè)試結(jié)束時(shí)它會(huì)調(diào)用一個(gè)特殊的名為 Ryuk 的容器來確保測(cè)試容器都被關(guān)閉。

相對(duì)而言，dockertest 要簡(jiǎn)陋不少。 需要注意的是，在 CI 環(huán)境運(yùn)行集成測(cè)試都需要確保 ci-runner 支持 DinD ，否則運(yùn)行 go test 會(huì)失敗。

項(xiàng)目開發(fā)

項(xiàng)目開發(fā)過程中基本按照需求來實(shí)現(xiàn)沒有太多難點(diǎn)。這里分享踩到的幾個(gè)坑。

循環(huán)中變量的引用問題

在測(cè)試中發(fā)現(xiàn)，Kafka 生產(chǎn)者沒有按期望把消息投遞到指定的 Kafka 集群。 經(jīng)過排查到如下代碼：

func New(cfg Config) (*Manager, error) {
        var newProducers = make(NewProducerFuncs)
        for name, kCfg := range cfg.Mapping {
                newProducers[name] = func() (kafka.Producer, error) { return kafka.New(kCfg) }
        }
        // 略
}

其作用是將配置每個(gè) Kafka 生產(chǎn)者配置先保存為一個(gè)函數(shù)閉包，待后續(xù)初始化 repo 的時(shí)候再初始化生產(chǎn)者客戶端。

經(jīng)驗(yàn)豐富的同學(xué)可以發(fā)現(xiàn)，for 循環(huán)的 kCfg 變量其實(shí)是指向迭代對(duì)象的地址，整個(gè)循環(huán)下來所有的函數(shù)閉包中用到的 kCfg 都指向 cfg.Mapping 的最后一個(gè)迭代值。 解決辦法很簡(jiǎn)單，先做一遍變量拷貝即可：

func New(cfg Config) (*Manager, error) {
        var newProducers = make(NewProducerFuncs)
        for name, kCfg := range cfg.Mapping {
                newProducers[name] = func() (kafka.Producer, error) { return kafka.New(kCfg) }
        }
        // 略
}

Sarama 客戶端的一點(diǎn)坑

對(duì)于重要的日志數(shù)據(jù)，我們希望在 HTTP 請(qǐng)求返回時(shí)明確反饋是否成功寫入 Kafka 。那么最好將 Kafka 生產(chǎn)者客戶端配置為同步模式。

而同步模式的生產(chǎn)者要提高吞吐量，批量發(fā)送是必不可少的。 批量發(fā)送的配置位于 sarama.Config.Producer.Flush

cfg := sarama.NewConfig()
// 單次請(qǐng)求中消息數(shù)量的絕對(duì)上限
cfg.Producer.Flush.MaxMessages = batchMaxMsgs
// 能夠觸發(fā)請(qǐng)求發(fā)出的消息數(shù)量閾值
cfg.Producer.Flush.Messages = batchMsgs
// 能夠觸發(fā)請(qǐng)求發(fā)出的消息字節(jié)大小閾值
cfg.Producer.Flush.Bytes = batchBytes
// 批量請(qǐng)求的觸發(fā)間隔時(shí)間
cfg.Producer.Flush.Frequency = batchTimeout

實(shí)踐中發(fā)現(xiàn)，如果配置了 Flush.Bytes 而沒有配置Flush.Frequency 就存在問題。如果消息大小始終未達(dá)閾值就不會(huì)觸發(fā)批量請(qǐng)求，故 HTTP 請(qǐng)求就會(huì)阻塞直到客戶端請(qǐng)求超時(shí)。 所以在配置參數(shù)的讀取上，我們把這兩個(gè)配置項(xiàng)做了關(guān)聯(lián)，只有配置了 Flush.Frequency 才能讓 Flush.Bytes 的配置生效。

項(xiàng)目上線

容器平臺(tái)上的灰度技巧

原本圖方便我們的路由轉(zhuǎn)發(fā)規(guī)則配置的是全部路由直接轉(zhuǎn)給同一組 Chopper 實(shí)例。 前面介紹了，每一個(gè)業(yè)務(wù)對(duì)應(yīng)一個(gè) repo，也就對(duì)應(yīng)一個(gè)獨(dú)立的請(qǐng)求路徑。如果要灰度新的服務(wù)，需要對(duì)不同業(yè)務(wù)單獨(dú)灰度，所以我們需要將不同業(yè)務(wù)的流量去分開。

好在容器平臺(tái)的 k8s-ingress 使用的是 APISIX 作為接入網(wǎng)關(guān)，其路由匹配的優(yōu)先級(jí)是：絕對(duì)匹配 > 前綴匹配。 只需要針對(duì)特定業(yè)務(wù)增加一條絕對(duì)匹配規(guī)則，就可以分離出特定業(yè)務(wù)的流量。

舉個(gè)例子： 原本的轉(zhuǎn)發(fā)規(guī)則是：/* -> workers-0 我們新建一條轉(zhuǎn)發(fā)規(guī)則：/log/repo/cdn-access -> workers-1 workers-0 和 workers-1 兩組服務(wù)的配置完全相同。 然后我們對(duì) workers-1 這組服務(wù)灰度發(fā)布新版程序。

逐步擴(kuò)大

每灰度一條路由，我們可以從監(jiān)控 Dashboard 上觀察 HTTP 請(qǐng)求是否有異常，觀察 Kafka 對(duì)應(yīng)的 topic 的寫入速率是否有異常抖動(dòng)。

一旦觀測(cè)到異常，立即停止灰度，然后檢查程序運(yùn)行日志，修正問題后重新開始灰度。 如果無異常，則逐步擴(kuò)大灰度比例，直到完成服務(wù)更新。

總結(jié)起來就是灰度、觀測(cè)、回退、修改循環(huán)推進(jìn)，確保升級(jí)對(duì)每個(gè)業(yè)務(wù)都無感知。

完成發(fā)布

對(duì)比服務(wù)端資源占用情況

舊版 chopper (4C8G x 20) 灰度比例 10% -> 50%



chopper-go (4C4G x 20) 10% -> 50%



50% -> 100%



結(jié)論：新版日志網(wǎng)關(guān)的內(nèi)存和 CPU 的資源使用都有顯著降低。

服務(wù)端程序的資源占用情況

舊版 chopper 的 Kafka 客戶端不支持消息壓縮，chopper-go 發(fā)布中就配置了 Kafka 生產(chǎn)者消息的功能。 壓縮算法選擇 lz4 ，觀察兩組消費(fèi)服務(wù)的資源實(shí)用率的變化： 消費(fèi)服務(wù)0

內(nèi)存使用率 27% -> 40%

網(wǎng)絡(luò)流入 253Mbps -> 180Mbps

消費(fèi)服務(wù)1

內(nèi)存使用率 28% -> 39%

網(wǎng)絡(luò)流入 380Mbps -> 267Mbps

結(jié)論：開啟消息壓縮功能后，消費(fèi)實(shí)例的內(nèi)存使用率普遍有增長(zhǎng)，但內(nèi)網(wǎng)傳輸帶寬占用降低約 30% 。

更新計(jì)劃

重構(gòu)后的流式日志網(wǎng)關(guān)，尚有許多可優(yōu)化空間，例如：

采用更節(jié)省帶寬的日志傳輸格式；

進(jìn)一步細(xì)化 Kafka topic 的分流粒度；

日志消息處理階段多級(jí)處理執(zhí)行器之間增加緩存提高字段訪問速度等等。

在豐富開源生態(tài)的加持下，該項(xiàng)目的優(yōu)化迭代也將有條不紊地進(jìn)行。

審核編輯：劉清