【blog】以一次RPC请求为例探索MOSN的工作流程 (#250)

* 以一次RPC请求为例探索MOSN的工作流程 Co-authored-by: 呈铭 <[email protected]>
mosn · Mar 11, 2024 · ebdaa86 · ebdaa86
1 parent ab7ab1b
commit ebdaa86
Show file tree

Hide file tree

Showing 4 changed files with 296 additions and 0 deletions.
diff --git a/content/zh/blog/posts/mosn_workflow/img.png b/content/zh/blog/posts/mosn_workflow/img.png
diff --git a/content/zh/blog/posts/mosn_workflow/img_1.png b/content/zh/blog/posts/mosn_workflow/img_1.png
diff --git a/content/zh/blog/posts/mosn_workflow/img_2.png b/content/zh/blog/posts/mosn_workflow/img_2.png
diff --git a/content/zh/blog/posts/mosn_workflow/index.md b/content/zh/blog/posts/mosn_workflow/index.md
@@ -0,0 +1,296 @@
+---
+title: "以一次 RPC 请求为例探索 MOSN的 工作流程"
+linkTitle: "以一次 RPC 请求为例探索 MOSN 的工作流程"
+date: 2024-03-11
+author: "[wangchengming666](https://github.com/wangchengming666)"
+description: "本文以工作中非常常见的一个思路为出发点，详细描述了 MOSN 内部网络转发的详细流程，可以帮助小伙伴加深对 MOSN 的理解。"
+aliases: "/zh/blog/posts/mosn-workflow"
+---
+
+## 1. 前言
+
+MOSN（Modular Open Smart Network）是一款主要使用 Go 语言开发的云原生网络代理平台，由蚂蚁集团开源并经过双11大促几十万容器的生产级验证。
+MOSN 为服务提供多协议、模块化、智能化、安全的代理能力，融合了大量云原生通用组件，同时也可以集成 Envoy 作为网络库，具备高性能、易扩展的特点。MOSN 可以和 Istio 集成构建 Service Mesh，也可以作为独立的四、七层负载均衡，API Gateway、云原生 Ingress 等使用。
+
+MOSN 作为数据面，整体 NET/IO、Protocol、Stream、Proxy 四个层次组成，其中
+- NET/IO 用于底层的字节流传输
+- Protocol 用于协议的 decode/encode
+- Stream 用于封装请求和响应，在一个 conn 上做连接复用
+- Proxy 做 downstream 和 upstream 之间 stream 的转发
+
+那么 MOSN 是如何工作的呢？下图展示的是使用 Sidecar 方式部署运行 MOSN 的示意图，您可以在配置文件中设置 MOSN 的上游和下游协议，协议可以在 HTTP、HTTP2.0、以及SOFA RPC 等中选择。
+![img.png](img.png)
+以上内容来自官网 https://mosn.io/
+
+## 2. RPC 场景下 MOSN 的工作机制
+RPC 场景下 MOSN 的工作机制示意图如下
+![img_1.png](img_1.png)
+
+我们简单理解一下上面这张图的意义：
+1. Server 端 MOSN 会将自身 ingress 的协议端口写入到注册中心
+2. Client 端 MOSN 会从注册中心订阅地址列表，第一次订阅也会返回全量地址列表，端口号是 Server 端 ingress 绑定的端口号
+3. 注册中心会实时推送地址列表变更到 Client 端（全量）
+4. Client 端发起rpc 调用时，请求会被 SDK 打到本地 Client 端 MOSN 的 egress 端口上
+5. Client 端 MOSN 将 RPC 请求通过网络转发，将流量通过负载均衡转发到某一台 Server 端 MOSN 的 ingress 端口处理
+6. 最终到了 Server 端 ingress listener，会转发给本地 Server 应用
+7. 最终会根据原来的 TCP 链路返回
+
+## 3. 全局视野下的 MOSN 工作流程
+
+![img_2.png](img_2.png)
+
+为了方便大家理解，我将以上时序图内容进行拆分，我们一一攻破。
+
+### 3.1 建立连接
+MOSN 在启动期间，会暴露本地 egress 端口接收 Client 的请求。MOSN 会开启 2 个协程，分别死循环去对 TCP 进行读取和写处理。MOSN 会通过读协程获取到请求字节流，进入 MOSN 的协议层处理。
+
+```
+// 代码路径 mosn.io/mosn/pkg/network/connection.go
+func (c *connection) Start(lctx context.Context) {
+ // udp downstream connection do not use read/write loop
+ if c.network == "udp" && c.rawConnection.RemoteAddr() == nil {
+ return
+ }
+ c.startOnce.Do(func() {
+ // UseNetpollMode = false
+ if UseNetpollMode {
+ c.attachEventLoop(lctx)
+ } else {
+ // 启动读/写循环
+ c.startRWLoop(lctx)
+ }
+ })
+}
+
+func (c *connection) startRWLoop(lctx context.Context) {
+ // 标记读循环已经启动
+ c.internalLoopStarted = true
+
+ utils.GoWithRecover(func() {
+ // 开始读操作
+ c.startReadLoop()
+ }, func(r interface{}) {
+ c.Close(api.NoFlush, api.LocalClose)
+ })
+ // 省略。。。
+}
+```
+### 3.2 Protocol 处理
+Protocol 作为多协议引擎层，对数据包进行检测，并使用对应协议做 decode/encode 处理。MOSN 会循环解码，一旦收到完整的报文就会创建与其关联的 xstream，用于保持 tcp 连接用于后续响应。
+```
+// 代码路径 mosn.io/mosn/pkg/stream/xprotocol/conn.go
+func (sc *streamConn) Dispatch(buf types.IoBuffer) {
+ // decode frames
+ for {
+ // 协议 decode，比如 dubbo、bolt 协议等
+ frame, err := sc.protocol.Decode(streamCtx, buf)
+
+ if frame != nil {
+ // 创建和请求 frame 关联的 xstream，用于保持 tcp 连接用于后续响应
+ sc.handleFrame(streamCtx, xframe)
+ }
+ }
+}
+
+func (sc *streamConn) handleFrame(ctx context.Context, frame api.XFrame) {
+ switch frame.GetStreamType() {
+ case api.Request:
+ // 创建和请求 frame 关联的 xstream，用于保持 tcp 连接用于后续响应，之后进入 proxy 层
+ sc.handleRequest(ctx, frame, false)
+ }
+}
+
+func (sc *streamConn) handleRequest(ctx context.Context, frame api.XFrame, oneway bool) {
+ // 创建和请求 frame 关联的 xstream
+ serverStream := sc.newServerStream(ctx, frame)
+ // 进入 proxy 层并创建 downstream
+ serverStream.receiver = sc.serverCallbacks.NewStreamDetect(serverStream.ctx, sender, span)
+ serverStream.receiver.OnReceive(serverStream.ctx, frame.GetHeader(), frame.GetData(), nil)
+}
+```
+### 3.3 Proxy 层处理
+proxy 层负责 filter 请求/响应链、路由匹配、负载均衡最终将请求转发到集群的某台机器上。
+#### 3.3.1 downStream 部分
+```
+// 代码路径 mosn.io/mosn/pkg/proxy/downstream.go
+func (s *downStream) OnReceive(ctx context.Context, headers types.HeaderMap, data types.IoBuffer, trailers types.HeaderMap) {
+ s.downstreamReqHeaders = headers
+ // filter 请求/响应链、路由匹配、负载均衡
+ phase = s.receive(s.context, id, phase)
+}
+
+func (s *downStream) receive(ctx context.Context, id uint32, phase types.Phase) types.Phase {
+ for i := 0; i <= int(types.End-types.InitPhase); i++ {
+ s.phase = phase
+
+ switch phase {
+
+ // downstream filter 相关逻辑
+ case types.DownFilter:
+ s.printPhaseInfo(phase, id)
+ s.tracks.StartTrack(track.StreamFilterBeforeRoute)
+
+ s.streamFilterChain.RunReceiverFilter(s.context, api.BeforeRoute,
+ s.downstreamReqHeaders, s.downstreamReqDataBuf, s.downstreamReqTrailers, s.receiverFilterStatusHandler)
+ s.tracks.EndTrack(track.StreamFilterBeforeRoute)
+
+ if p, err := s.processError(id); err != nil {
+ return p
+ }
+ phase++
+
+ // route 相关逻辑
+ case types.MatchRoute:
+ s.printPhaseInfo(phase, id)
+
+ s.tracks.StartTrack(track.MatchRoute)
+ s.matchRoute()
+ s.tracks.EndTrack(track.MatchRoute)
+
+ if p, err := s.processError(id); err != nil {
+ return p
+ }
+ phase++
+ 
+ // 在集群中选择一个机器、包含cluster和loadblance
+ case types.ChooseHost:
+ s.printPhaseInfo(phase, id)
+
+ s.tracks.StartTrack(track.LoadBalanceChooseHost)
+ // 这里很重要，在选中一个机器之后，这里upstreamRequest对象有两个作用
+ // 1. 这里通过持有downstream保持着对客户端app的tcp引用，用来接收请求
+ // 2. 转发服务端tcp引用，转发客户端app请求以及响应服务端response时的通知
+ s.chooseHost(s.downstreamReqDataBuf == nil && s.downstreamReqTrailers == nil)
+ s.tracks.EndTrack(track.LoadBalanceChooseHost)
+
+ if p, err := s.processError(id); err != nil {
+ return p
+ }
+ phase++
+ }
+ }
+}
+```
+#### 3.3.2 upStream 部分
+至此已经选中一台服务端的机器，开始准备转发。
+```
+// 代码路径 mosn.io/mosn/pkg/proxy/upstream.go
+func (r *upstreamRequest) appendHeaders(endStream bool) {
+
+ if r.downStream.oneway {
+ _, streamSender, failReason = r.connPool.NewStream(r.downStream.context, nil)
+ } else {
+ // 会使用 ChooseHost 中选中的机器 host 创建 sender，xstream 是客户端的流对象
+ _, streamSender, failReason = r.connPool.NewStream(r.downStream.context, r)
+ }
+}
+```
+接下来会到达 conn.go 的 handleFrame 的 handleResponse 方法，此时 handleResponse 方法继续调用 downStream 的 receiveData 方法接收数据。
+```
+//代码路径 mosn.io/mosn/pkg/stream/xprotocol/conn.go
+func (sc *streamConn) handleFrame(ctx context.Context, frame api.XFrame) {
+ switch frame.GetStreamType() {
+ case api.Response:
+ // 调用 downStream 的 receiveData 方法接收数据
+ // 因为 mosn 在转发之前修改了请求id，因此会重新 encode 请求
+ sc.handleResponse(ctx, frame)
+ }
+}
+```
+一旦准备好转发就会通过 upstreamRequest 选择的下游主机直接发送 write 请求，请求的协程此时会被阻塞。
+```
+// 代码路径 mosn.io/mosn/pkg/stream/xprotocol/stream.go
+func (s *xStream) endStream() {
+ defer func() {
+ if s.direction == stream.ServerStream {
+ s.DestroyStream()
+ }
+ }()
+
+ if log.Proxy.GetLogLevel() >= log.DEBUG {
+ log.Proxy.Debugf(s.ctx, "[stream] [xprotocol] connection %d endStream, direction = %d, requestId = %v", s.sc.netConn.ID(), s.direction, s.id)
+ }
+
+ if s.frame != nil {
+ // replace requestID
+ s.frame.SetRequestId(s.id)
+ // 因为 mosn 在转发之前修改了请求 id，因此会重新 encode 请求
+ buf, err := s.sc.protocol.Encode(s.ctx, s.frame)
+ if err != nil {
+ log.Proxy.Errorf(s.ctx, "[stream] [xprotocol] encode error:%s, requestId = %v", err.Error(), s.id)
+ s.ResetStream(types.StreamLocalReset)
+ return
+ }
+
+ tracks := track.TrackBufferByContext(s.ctx).Tracks
+
+ tracks.StartTrack(track.NetworkDataWrite)
+ // 一旦准备好转发就会通过upstreamRequest选择的下游主机直接发送 write 请求，请求的协程此时会被阻塞
+ err = s.sc.netConn.Write(buf)
+ tracks.EndTrack(track.NetworkDataWrite)
+ }
+ }
+}
+```
+### 3.4 准备将响应写回客户端
+接下来客户端 xstream 将通过读协程接收响应的字节流，proxy.go 的 OnData 方法作为 proxy 层的数据接收点。
+```
+// 代码位置 mosn.io/mosn/pkg/proxy/proxy.go
+func (p *proxy) OnData(buf buffer.IoBuffer) api.FilterStatus {
+ // 这里会做两件事
+ // 1. 调用 protocol 层进行decode
+ // 2. 完成后通知upstreamRequest对象，唤醒downstream阻塞的协程
+ p.serverStreamConn.Dispatch(buf)
+
+ return api.Stop
+}
+
+// 代码位置 mosn.io/mosn/pkg/proxy/upstream.go
+func (r *upstreamRequest) OnReceive(ctx context.Context, headers types.HeaderMap, data types.IoBuffer, trailers types.HeaderMap) {
+ // 结束当前stream
+ r.endStream()
+
+ // 唤醒
+ r.downStream.sendNotify()
+}
+```
+downstream 被唤醒处理收到的响应，重新替换回正确的请求ID，并调用 protocol 层重新编码成字节流写回客户端，最后销毁请求相关的资源，流程执行完毕。
+```
+// 比如我的 demo 是 dubbo 协议
+func encodeFrame(ctx context.Context, frame *Frame) (types.IoBuffer, error) {
+
+ // 1. fast-path, use existed raw data
+ if frame.rawData != nil {
+ // 1.1 replace requestId
+ binary.BigEndian.PutUint64(frame.rawData[IdIdx:], frame.Id)
+
+ // hack: increase the buffer count to avoid premature recycle
+ frame.data.Count(1)
+ return frame.data, nil
+ }
+
+ // alloc encode buffer
+ frameLen := int(HeaderLen + frame.DataLen)
+ buf := buffer.GetIoBuffer(frameLen)
+ // encode header
+ buf.WriteByte(frame.Magic[0])
+ buf.WriteByte(frame.Magic[1])
+ buf.WriteByte(frame.Flag)
+ buf.WriteByte(frame.Status)
+ buf.WriteUint64(frame.Id)
+ buf.WriteUint32(frame.DataLen)
+ // encode payload
+ buf.Write(frame.payload)
+ return buf, nil
+}
+```
+## 4. 总结
+本文以工作中非常常见的一个思路为出发点，详细描述了 MOSN 内部网络转发的详细流程，可以帮助小伙伴加深对 MOSN 的理解。MOSN 是一款非常优秀的开源产品，
+MOSN 支持多种网络协议（如HTTP/2, gRPC, Dubbo等）并且能够很容易地增加对新协议的支持；MOSN 提供了丰富的流量治理功能，例如限流、熔断、重试、
+负载均衡等；MOSN 在性能方面进行了大量优化，比如内存零拷贝、自适应缓冲区、连接池、协程池等，这些都有助于提升其在高并发环境下的表现。除此之外 
+MOSN 在连接管理方面，MOSN 设计了多协议连接池；在内存管理方面，MOSN 在 sync.Pool 之上封装了一层资源对的注册管理模块，可以方便的扩展各种类型的
+对象进行复用和管理。总的来说，MOSN 的设计体现了可扩展性、高性能、安全性、以及对现代云环境的适应性等多方面的考虑。对于开发者来说，深入研究MOSN的
+代码和架构，无疑可以学到很多关于高性能网络编程和云原生技术的知识。
+
+- MOSN 官网：[https://mosn.io/](https://mosn.io/)
+- MOSN Github：[https://github.com/mosn/mosn](https://github.com/mosn/mosn)