有幸參加了?Flomesh^[1]?組織的workshop，了解了他們的 Pipy 網(wǎng)絡(luò)代理，以及圍繞 Pipy 構(gòu)建起來的生態(tài)。Pipy 在生態(tài)中，不止是代理的角色，還是 Flomesh 服務(wù)網(wǎng)格中的數(shù)據(jù)平面。

整理一下，做個(gè)記錄，順便瞄一下 Pipy 的部分源碼。

介紹

下面是摘自 Github 上關(guān)于 Pipy 的介紹：

Pipy 是一個(gè)輕量級、高性能、高穩(wěn)定、可編程的網(wǎng)絡(luò)代理。Pipy 核心框架使用 C 開發(fā)，網(wǎng)絡(luò) IO 采用 ASIO 庫。Pipy 的可執(zhí)行文件僅有 5M 左右，運(yùn)行期的內(nèi)存占用 10M 左右，因此 Pipy 非常適合做 Sidecar proxy。

Pipy 內(nèi)置了自研的 pjs 作為腳本擴(kuò)展，使得Pipy 可以用 JS 腳本根據(jù)特定需求快速定制邏輯與功能。

Pipy 采用了模塊化、鏈?zhǔn)降奶幚砑軜?gòu)，用順序執(zhí)行的模塊來對網(wǎng)絡(luò)數(shù)據(jù)塊進(jìn)行處理。這種簡單的架構(gòu)使得 Pipy 底層簡單可靠，同時(shí)具備了動態(tài)編排流量的能力，兼顧了簡單和靈活。通過使用 REUSE_PORT 的機(jī)制（主流 Linux 和 BSD 版本都支持該功能），Pipy 可以以多進(jìn)程模式運(yùn)行，使得 Pipy 不僅適用于 Sidecar 模式，也適用于大規(guī)模的流量處理場景。在實(shí)踐中，Pipy 獨(dú)立部署的時(shí)候用作“軟負(fù)載”，可以在低延遲的情況下，實(shí)現(xiàn)媲美硬件的負(fù)載均衡吞吐能力，同時(shí)具有靈活的擴(kuò)展性。

Pipy 的核心是消息流處理器：

Pipy 流量處理的流程：

核心概念

?流（Stream）?管道（Pipeline）?模塊（Module）?會話（Session）?上下文（Context）

以下是個(gè)人淺見：

Pipy 使用?pjs?引擎將 JavaScript格式的配置，解析成其抽象的?Configuration?對象。每個(gè)?Configuration?中包含了多個(gè)?Pipeline，每個(gè)?Configuration?中又會用到多個(gè)?Filter。這些都屬于 Pipy 的靜態(tài)配置部分。（后面會提到 Pipeline 的三種不同類型）

而屬于運(yùn)行時(shí)的就是流、會話和上下文了，在 Pipy 中，數(shù)據(jù)流是由對象（Pipy 的抽象）組成的。而這些對象抵達(dá) Pipy，被抽象成不同的事件。而事件觸發(fā)不同的過濾器的執(zhí)行。

我個(gè)人更喜歡將其核心理解為：對數(shù)據(jù)流的事件處理引擎。

理解歸理解，實(shí)踐出真知。“大膽假設(shè)，小心求證！”

本地編譯

從編譯 Pipy 開始。

環(huán)境準(zhǔn)備

#安裝 nodejs$ nvm install lts/erbium #安裝 cmake$ brew install cmake

編譯 Pipy

從?https://github.com/flomesh-io/pipy.git?克隆代碼。

Pipy 的編譯包括了兩個(gè)部分，GUI 和 Pipy 本體。

GUI 是 Pipy 提供的一個(gè)用于開發(fā)模式下進(jìn)行配置的界面，首先編譯Pipy GUI。

# pipy root folder$ cd gui$ npm install$ npm run build接著編譯 Pipy 的本體

# pipy root folder$ mkdir build$ cd build$ cmake -DCMAKE_BUILD_TYPE=Release -DPIPY_GUI=ON ..$ make完成后檢查根目錄下的?bin?目錄，可以看到 pipy 的可執(zhí)行文件，大小只有 11M。

$ bin/pipy --helpUsage: pipy [options] <script filename>Options: -h, -help, --help Show help information -v, -version, --version Show version information --list-filters List all filters --help-filters Show detailed usage information for all filters --log-level=<debug|info|warn|error> Set the level of log output --verify Verify configuration only --reuse-port Enable kernel load balancing for all listening ports --gui-port=<port> Enable web GUI on the specified port

Demo：Hello Pipy

開發(fā)模式下可以讓 Pipy 攜帶 GUI 啟動，通過 GUI 進(jìn)行配置。

#指定 gui 的端口為 6060，從 test 目錄中加載配置$ bin/pipy --gui-port=6060 test/2021-05-30 22:48:41 [info] [gui] Starting GUI service...2021-05-30 22:48:41 [info] [listener] Listening on 0.0.0.0:6060瀏覽器中打開

配置界面

展開?002-hello?子目錄點(diǎn)選?pipy?并點(diǎn)擊運(yùn)行按鈕：

$ curl -i localhost:6080HTTP/1.1 200 OKConnection: keep-aliveContent-Length: 7Hello!

Pipy 過濾器

通過 pipe 的命令可以輸出其支持的過濾器列表，一共 31 個(gè)。通過將一系列過濾器進(jìn)行組裝，可以實(shí)現(xiàn)復(fù)雜的流處理。

比如?007-logging?的配置實(shí)現(xiàn)了日志的功能：記錄請求和響應(yīng)的數(shù)據(jù)，并批量發(fā)送到 ElasticSearch。這里就用到了?fork、connect、onSessionStart、encodeHttpRequest、decodeHttpRequest、onMessageStart、onMessage、decodeHttpResponse、replaceMessage、link、mux、task?等十多種過濾器。

$ bin/pipy --list-filtersconnect (target[, options]) Sends data to a remote endpoint and receives data from itdemux (target) Sends messages to a different pipline with each one in its own session and contextdecodeDubbo () Deframes a Dubbo messagedecodeHttpRequest () Deframes an HTTP request messagedecodeHttpResponse () Deframes an HTTP response messagedummy () Eats up all eventsdump ([tag]) Outputs events to the standard outputencodeDubbo ([head]) Frames a Dubbo messageencodeHttpRequest ([head]) Frames an HTTP request messageencodeHttpResponse ([head]) Frames an HTTP response messageexec (command) Spawns a child process and connects to its input/outputfork (target[, sessionData]) Sends copies of events to other pipeline sessionslink (target[, when[, target2[, when2, ...]]]) Sends events to a different pipelinemux (target[, selector]) Sends messages from different sessions to a shared pipeline sessiononSessionStart (callback) Handles the initial event in a sessiononData (callback) Handles a Data eventonMessageStart (callback) Handles a MessageStart eventonMessageEnd (callback) Handles a MessageEnd eventonSessionEnd (callback) Handles a SessionEnd eventonMessageBody (callback) Handles a complete message bodyonMessage (callback) Handles a complete message including the head and the bodyprint () Outputs raw data to the standard outputreplaceSessionStart (callback) Replaces the initial event in a sessionreplaceData ([replacement]) Replaces a Data eventreplaceMessageStart ([replacement]) Replaces a MessageStart eventreplaceMessageEnd ([replacement]) Replaces a MessageEnd eventreplaceSessionEnd ([replacement]) Replaces a SessionEnd eventreplaceMessageBody ([replacement]) Replaces an entire message bodyreplaceMessage ([replacement]) Replaces a complete message including the head and the bodytap (quota[, account]) Throttles message rate or data rateuse (module, pipeline[, argv...]) Sends events to a pipeline in a different modulewait (condition) Buffers up events until a condition is fulfilled

原理

“Talk is cheap, show me the code.”

配置加載

個(gè)人比較喜歡看源碼來理解實(shí)現(xiàn)，即使是 C 。從瀏覽器請求入手發(fā)現(xiàn)運(yùn)行時(shí)向/api/program?發(fā)送了?POST?請求，請求的內(nèi)容是配置文件的地址。

檢查源碼后，找到邏輯的實(shí)現(xiàn)在?src/gui.cpp:189：

1.創(chuàng)建新的 worker2.加載配置，將 JavaScrip 代碼解析成?Configuration?對象3.啟動 worker，執(zhí)行Configuration::apply()4.卸載舊的 worker

從?src/api/configuration.cpp:267?處看：pipeline、listen?和?task?配置實(shí)際在 Pipy 的配置中都是被抽象為?Pipeline?對象，只是在類型上有差異分別為：NAMED、LISTEN?和?TASK。比如?listen?中可以通過?fork?過濾器將事件的副本發(fā)送到指定的?pipeline?中。

基于數(shù)據(jù)流事件的處理

src/inbound.cpp:171

結(jié)語

Pipy 雖?。ㄖ挥?11M），但以其可編程的特性提供了靈活的配置能力，潛力無限。

Pipy 像處理 HTTP 一樣處理任意的七層協(xié)議。內(nèi)部版本支持Dubbo、Redis、Socks 等，目前正在遷移到開源版本。

期待即將開源的 Portal，以及服務(wù)網(wǎng)格 Flomesh。持續(xù)關(guān)注，后面考慮再寫幾篇。

“未來可期！”

引用鏈接

[1]?Flomesh:?https://flomesh.cn/