有幸加入了 Flomesh 组织的workshop，理解了他们的 Pipy 网络代理，以及围绕 Pipy 构建起来的生态。Pipy 在生态中，不止是代理的角色，还是 Flomesh 服务网格中的数据立体。

整顿一下，做个记录，顺便瞄一下 Pipy 的局部源码。

介绍

上面是摘自 Github 上对于 Pipy 的介绍：

Pipy 是一个轻量级、高性能、高稳固、可编程的网络代理。Pipy 外围框架应用 C++ 开发，网络 IO 采纳 ASIO 库。 Pipy 的可执行文件仅有 5M 左右，运行期的内存占用 10M 左右，因而 Pipy 非常适合做 Sidecar proxy。

Pipy 内置了自研的 pjs 作为脚本扩大，使得Pipy 能够用 JS 脚本依据特定需要疾速定制逻辑与性能。

Pipy 采纳了模块化、链式的解决架构，用程序执行的模块来对网络数据块进行解决。这种简略的架构使得 Pipy 底层简略牢靠，同时具备了动静编排流量的能力，兼顾了简略和灵便。通过应用 REUSE_PORT 的机制（支流 Linux 和 BSD 版本都反对该性能），Pipy 能够以多过程模式运行，使得 Pipy 不仅实用于 Sidecar 模式，也实用于大规模的流量解决场景。 在实践中，Pipy 独立部署的时候用作“软负载”，能够在低提早的状况下，实现媲美硬件的负载平衡吞吐能力，同时具备灵便的扩展性。

Pipy 的外围是音讯流处理器：

Pipy 流量解决的流程：

外围概念

  • 流（Stream）：Pipy
  • 管道（Pipeline）
  • 模块（Module）
  • 会话（Session）
  • 上下文（Context）

<u>以下是集体浅见</u>：

Pipy 应用 pjs 引擎将 JavaScript格局的配置，解析成其形象的 Configuration 对象。每个 Configuration 中蕴含了多个 Pipeline，每个 Configuration 中又会用到多个 Filter。这些都属于 Pipy 的动态配置局部。（前面会提到 Pipeline 的三种不同类型）

而属于运行时的就是流、会话和上下文了，在 Pipy 中，数据流是由对象（Pipy 的形象）组成的。而这些对象到达 Pipy，被形象成不同的<u>事件</u>。而事件触发不同的过滤器的执行。

我集体更喜爱将其外围了解为：对数据流的事件处理引擎。

了解归了解，实际出真知。“大胆假如，小心求证！”

本地编译

从编译 Pipy 开始。

环境筹备

#装置 nodejs$ nvm install lts/erbium #装置 cmake$ brew install cmake

编译 Pipy

从 https://github.com/flomesh-io/pipy.git 克隆代码。

Pipy 的编译包含了两个局部，GUI 和 Pipy 本体。

GUI 是 Pipy 提供的一个用于开发模式下进行配置的界面，首先编译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 的可执行文件，大小只有 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

开发模式下能够让 Pipy 携带 GUI 启动，通过 GUI 进行配置。

#指定 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 子目录点选 pipy 并点击运行按钮：

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

Pipy 过滤器

通过 pipe 的命令能够输入其反对的过滤器列表，一共 31 个。通过将一系列过滤器进行组装，能够实现简单的流解决。

比方 007-logging 的配置实现了日志的性能：记录申请和响应的数据，并批量发送到 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.”

配置加载

集体比拟喜爱看源码来了解实现，即便是 C++。从浏览器申请动手发现运行时向/api/program 发送了 POST 申请，申请的内容是配置文件的地址。

查看源码后，找到逻辑的实现在 src/gui.cpp:189：

  1. 创立新的 worker
  2. 加载配置，将 JavaScrip 代码解析成 Configuration 对象
  3. 启动 worker，执行Configuration::apply()
  4. 卸载旧的 worker

从 src/api/configuration.cpp:267 处看：pipeline、listen 和 task 配置理论在 Pipy 的配置中都是被形象为 Pipeline 对象，只是在类型上有差别别离为：NAMED、LISTEN 和 TASK。比方 listen 中能够通过 fork 过滤器将事件的正本发送到指定的 pipeline 中。

基于数据流事件的解决

src/inbound.cpp:171

结语

Pipy 虽小（只有 11M），但以其可编程的个性提供了灵便的配置能力，后劲有限。

Pipy 像解决 HTTP 一样解决任意的七层协定。外部版本反对Dubbo、Redis、Socks 等，目前正在迁徙到开源版本。

期待行将开源的 Portal，以及服务网格 Flomesh。继续关注，前面思考再写几篇。

“将来可期！”

文章对立公布在公众号云原生指北