項目背景

分享之前，先來簡單介紹下該項目在流式日志處理鏈路中所處的位置。



流式日志網關的主要功能是提供 HTTP 接口，接收 CDN 邊緣節點上報的各類日志（訪問日志/報錯日志/計費日志等），將日志作預處理并分流到多個的 Kafka 集群和 Topic 中。

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

舊版網關項目

項目代號 Chopper ，其基于另一個內部 OpenResty 項目框架來開發的。其亮點功能有：支持從 Consul 、Redis 等其他外部系統熱加載配置及動態生效；能夠加載 Lua 腳本實現靈活的日志預處理能力。 其 Kafka 生產者客戶端基于 doujiang24/lua-resty-kafka 實現。經過實踐考驗，Chopper 的吞吐量是滿足現階段需求的。

存在的問題

1. 關鍵依賴庫的社區活躍度低 lua-resty-kafka 的社區活躍度較低，至今仍然處在實驗階段；而且它用作 Kafka 生產者客戶端目前沒有支持消息壓縮功能，而這在其他語言實現的 Kafka 客戶端中都是標準的選項。

2. 內存使用不節制 單實例部署配置 4 核 8 G，僅少量請求訪問后，內存占用就穩定在 2G 而沒有釋放。

3. 配置文件可維護性差 實際線上用到 Consul 作為配置中心，采用篇幅很長的 JSON 格式配置文件，不利于運維。另外在 Consul 修改配置沒有回退功能，是一個高風險操作。 好在目前日志網關的功能并不復雜，所以我們決定重構它。

新項目啟動

眾所周知， Go 語言擁有獨特的高并發模型、較低的上手難度和豐富的第三方生態。而且我們小組成員都有 Go 項目的開發經驗，所以我們選擇使用基于 Go 語言的技術棧來重新構建 Chopper 項目，所以新項目命名為 chopper-go 。

需求梳理及概要設計

重新構建一個線上項目的基本原則是，功能上要完全兼容，最好能夠實現線上服務的無縫升級替換。

原版核心模塊的設計

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

一、HTTP 接口部分

只開放了唯一一個對外的 API ，功能很簡單：

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

二、業務配置部分

每一類業務抽象為一個 repo 配置。Repo 配置由三部分構成：constraint、processor、kafka。 constraint 是一個對象，可以配置對日志字段的一些約束條件，不滿足條件的日志會被丟棄。 processor 是一個列表，可以組合多個處理模塊，程序將按順序依次對請求中的每條日志進行處理。實現了如下幾種 processor 類型：

decoder , 配置原始數據按哪種格式反序列化到 Lua table ，但只實現了 JSON decoder。

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

assigner，配置一組字段名映射關系，需要與 spliter 配合。

executer， 配置額外的 lua 腳本名稱，通過動態加載其他 lua 腳本實現更靈活的處理邏輯。

kafka 是一個對象，可以配置當前業務相關聯的 Kafka 集群名，默認投遞的 Topic ，以及生產者客戶端的工作模式（同步或者異步）。 新版本的改動 HTTP 接口沿用原先的設計，在業務配置部分做了一些改動：

processor 改名為 executers ，實現幾個通用功能的日志處理模塊，方便組合使用。

kafka 配置中關聯的不再是集群名，而是 Kafka 生產者客戶端的配置標簽。

原先保存 kafka 集群連接配置信息的配置塊，改為保存 kafka 生產者客戶端的配置塊，統一在一個配置塊區域初始化所有用到的 kafka 生產者客戶端。

一點妥協（做減法）

為了縮短新項目的開發周期，對原始項目的一些不太重要的特性我們做了一些取舍。

取消動態腳本功能

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

不支持外部配置中心

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

不支持復雜的路由重寫

OpenResty 項目內置 Nginx 可以利用 Nginx 強大的配置實現豐富的路由 rewrite 功能，就具體使用場景而言，我們只需要簡單的路由映射即可。況且更復雜的需求也可以由上一級網關完成。

選擇合適的開源庫

Web 框架的選擇

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

Kafka 生產者客戶端的選擇

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

segmentio/kafka-go

Shopify/sarama

confluentinc/confluent-kafka-go

實際上三款客戶端庫我們在歷史項目中都使用過，其中 kafka-go 的 API 是三者中最簡潔易用的，我們的多個消費端程序都是基于它實現的。

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

那是因為我們做了一些基準測試，發現 kafka-go 的生產者客戶端存在性能風險：啟用 async 模式時盡管消息發送特別快，但是內存占用也增長特別快。

最終我們選擇 sarama ，一方面是性能很穩定，另一方面是它開放的 API 較多，但是用起來確實有點費勁。

測試框架的選擇

程序的可靠性，一定需要測試來保證。除了編寫小模塊中編寫單元從測試，我們對整個日志網關服務還要做集成測試。集成測試涉及到一些外部服務依賴，此項目中主要的外部依賴是 Kafka 和 Zookeeper 。

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

ory/dockertest

testcontainers/testcontainers-go

使用下來，我們最終選擇了 testcontainers-go，簡單介紹下原因： 在編寫集成測試時，我們需要有個等待機制來確保依賴服務的容器是否準備就緒，并以此控制測試流程，以及測試結束后需要把測試開啟的臨時容器都清理干凈。 testcontainers-go 的設計要優于 dockertest 。

testcontainers-go 提供一個 wait 子包，可以配置多種等待策略來確保依賴服務就緒，以及測試結束時它會調用一個特殊的名為 Ryuk 的容器來確保測試容器都被關閉。

相對而言，dockertest 要簡陋不少。 需要注意的是，在 CI 環境運行集成測試都需要確保 ci-runner 支持 DinD ，否則運行 go test 會失敗。

項目開發

項目開發過程中基本按照需求來實現沒有太多難點。這里分享踩到的幾個坑。

循環中變量的引用問題

在測試中發現，Kafka 生產者沒有按期望把消息投遞到指定的 Kafka 集群。 經過排查到如下代碼：

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) }
        }
        // 略
}

其作用是將配置每個 Kafka 生產者配置先保存為一個函數閉包，待后續初始化 repo 的時候再初始化生產者客戶端。

經驗豐富的同學可以發現，for 循環的 kCfg 變量其實是指向迭代對象的地址，整個循環下來所有的函數閉包中用到的 kCfg 都指向 cfg.Mapping 的最后一個迭代值。 解決辦法很簡單，先做一遍變量拷貝即可：

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 客戶端的一點坑

對于重要的日志數據，我們希望在 HTTP 請求返回時明確反饋是否成功寫入 Kafka 。那么最好將 Kafka 生產者客戶端配置為同步模式。

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

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

實踐中發現，如果配置了 Flush.Bytes 而沒有配置Flush.Frequency 就存在問題。如果消息大小始終未達閾值就不會觸發批量請求，故 HTTP 請求就會阻塞直到客戶端請求超時。 所以在配置參數的讀取上，我們把這兩個配置項做了關聯，只有配置了 Flush.Frequency 才能讓 Flush.Bytes 的配置生效。

項目上線

容器平臺上的灰度技巧

原本圖方便我們的路由轉發規則配置的是全部路由直接轉給同一組 Chopper 實例。 前面介紹了，每一個業務對應一個 repo，也就對應一個獨立的請求路徑。如果要灰度新的服務，需要對不同業務單獨灰度，所以我們需要將不同業務的流量去分開。

好在容器平臺的 k8s-ingress 使用的是 APISIX 作為接入網關，其路由匹配的優先級是：絕對匹配 > 前綴匹配。 只需要針對特定業務增加一條絕對匹配規則，就可以分離出特定業務的流量。

舉個例子： 原本的轉發規則是：/* -> workers-0 我們新建一條轉發規則：/log/repo/cdn-access -> workers-1 workers-0 和 workers-1 兩組服務的配置完全相同。 然后我們對 workers-1 這組服務灰度發布新版程序。

逐步擴大

每灰度一條路由，我們可以從監控 Dashboard 上觀察 HTTP 請求是否有異常，觀察 Kafka 對應的 topic 的寫入速率是否有異常抖動。

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

總結起來就是灰度、觀測、回退、修改循環推進，確保升級對每個業務都無感知。

完成發布

對比服務端資源占用情況

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



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



50% -> 100%



結論：新版日志網關的內存和 CPU 的資源使用都有顯著降低。

服務端程序的資源占用情況

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

內存使用率 27% -> 40%

網絡流入 253Mbps -> 180Mbps

消費服務1

內存使用率 28% -> 39%

網絡流入 380Mbps -> 267Mbps

結論：開啟消息壓縮功能后，消費實例的內存使用率普遍有增長，但內網傳輸帶寬占用降低約 30% 。

更新計劃

重構后的流式日志網關，尚有許多可優化空間，例如：

采用更節省帶寬的日志傳輸格式；

進一步細化 Kafka topic 的分流粒度；

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

在豐富開源生態的加持下，該項目的優化迭代也將有條不紊地進行。

審核編輯：劉清