搜索

MOSN热升级逻辑浅析


发布时间: 2023-02-06 20:29:00    浏览次数:16 次

分享时间:2023-02-06 20:29:00
数据来源:网络
提取密码:在线浏览
文件类型:文章

背景

MOSN在热升级上面,也曾做过自己的探索;业界虽然有Nginx Envoy 也都实现了热升级方式,那么这里有什么异同呢?

Nginx: 通过Fork的方式直接继承父进程的监听信息和链接信息等,仅仅只用于重启

Envoy: Envoy对端口监听(Listener)进行了迁移,但是对建立的连接(connection)则通过命令的方式,进行主动断开重连

MOSN: 鉴于低版本Bolt 等协议,不支持主动断连,且是长连接,导致MOSN在进行热升级的时候,不仅仅进行了端口监听的迁移,还有connection的迁移,保证了热升级过程中链接不中断,客户端服务端无感的升级体验

升级流程

触发情况

这里赘述一下,MOSN的热升级方式,是通过一个Operator实现的

  1. Operator在Pod中增加新的MOSN容器
  2. New Mosn启动时候,观测到有Old Mosn存在,开启热升级的逻辑
  3. New Mosn启动成功后,Old Mosn进行退出
  4. Operator销毁Old Mosn的Container

至此,一个完整的流程热升级流程结束

整体设计

MOSN热升级中核心的数据

  1. 配置数据(这样New Mosn才知道代理的是什么应用,有哪些配置信息等等)
  2. 监听端口
  3. 连接

image-20230204160716742

MOSN热升级的交互流程设计如上,接下来对各个模块迁移逻辑进行一下跟踪

源码解析

代码逻辑仅保留核心流程

socket监听情况

func (stm *StageManager) Run() {
    // 1: parser params
    stm.runParamsParsedStage()
    // 2: init
    stm.runInitStage()
    // 3: pre start
    stm.runPreStartStage()
    // 4: run
    stm.runStartStage()
    // 5: after start
    stm.runAfterStartStage()

    stm.SetState(Running)
}

MOSN启动分为上述几个流程,其中 热升级逻辑主要分布在 InitStageStartStage

InitStage: 迁移配置信息 和 迁移Listener

StartStage: 创建 reconfig.sock 的监听 和 迁移 connection

在MOSN中有4个socket监听

reconfig.sock: 由 old mosn 监听,用于 new mosn 感知 old mosn存在

listen.sock: 由new mosn 监听,用于old mosn传递 listener数据给new mosn

conn.sock: 由 new mosn监听,用于 old mosn 传递 connection 的 fd和connection读取到的数据 (没有则不传)给 new mosn

mosnconfig.sock: 由 new mosn 监听,用于 old mosn 传递配置信息给 new mosn

Old Mosn 迁移主流程

func ReconfigureHandler() error {
    // dump lastest config, and stop DumpConfigHandler()
    configmanager.DumpLock()
    configmanager.DumpConfig()
    // if reconfigure failed, enable DumpConfigHandler()
    defer configmanager.DumpUnlock()

    // transfer listen fd
    var listenSockConn net.Conn
    var err error
    var n int
    var buf [1]byte
    if listenSockConn, err = sendInheritListeners(); err != nil {
        return err
    }

    if enableInheritOldMosnconfig {
        if err = SendInheritConfig(); err != nil {
            listenSockConn.Close()
            log.DefaultLogger.Alertf(types.ErrorKeyReconfigure, "[old mosn] [SendInheritConfig] new mosn start failed")
            return err
        }
    }

    // Wait new mosn parse configuration
    listenSockConn.SetReadDeadline(time.Now().Add(10 * time.Minute))
    n, err = listenSockConn.Read(buf[:])
    if n != 1 {
        log.DefaultLogger.Alertf(types.ErrorKeyReconfigure, "[old mosn] [read ack] new mosn start failed")
        return err
    }

    // ack new mosn
    if _, err := listenSockConn.Write([]byte{0}); err != nil {
        log.DefaultLogger.Alertf(types.ErrorKeyReconfigure, "[old mosn] [write ack] new mosn start failed")
        return err
    }

    // stop other services
    store.StopService()

    // Wait for new mosn start
    time.Sleep(3 * time.Second)

    // Stop accepting new connections & graceful close the existing connections if they supports graceful close.
    shutdownServers()

    // Wait for all connections to be finished
    WaitConnectionsDone(GracefulTimeout)

    log.DefaultLogger.Infof("[server] [reconfigure] process %d gracefully shutdown", os.Getpid())

    // will stop the current old mosn in stage manager
    return nil
}
  • old mosn 首先将 listener 迁移到new mosn
  • 然后 迁移 配置信息,这一步是可选的
  • 等待 new mosn ack完成后
  • 调用 WaitConnectionsDone 将 connection close掉,迁移 connection

配置迁移

配置迁移是用过 mosnconfig.sock 来完成的,new mosn 监听,old mosn 连接上去并传输数据

new mosn监听是在 GetInheritConfig 这个方法中实现的

func GetInheritConfig() (*v2.MOSNConfig, error) {
    ......

    l, err := net.Listen("unix", types.TransferMosnconfigDomainSocket)
    ......
    defer l.Close()

    ul := l.(*net.UnixListener)
    ul.SetDeadline(time.Now().Add(time.Second * 10))
    uc, err := ul.AcceptUnix()
    ......
    defer uc.Close()
    log.StartLogger.Infof("[server] Get GetInheritConfig Accept")
    configData := make([]byte, 0)
    buf := make([]byte, 1024)
    for {
        n, err := uc.Read(buf)
        configData = append(configData, buf[:n]...)
        .......
    }

    // log.StartLogger.Infof("[server] inherit mosn config data: %v", string(configData))

    oldConfig := &v2.MOSNConfig{}
    err = json.Unmarshal(configData, oldConfig)
    if err != nil {
        return nil, err
    }

    return oldConfig, nil
}

New mosn 建立好socket监听后,卡在 Accept 函数中,等待 old mosn 建立链接

然后 接收old mosn 传递过来的数据即可

old mosn是在 SendInheritConfig 中与 new mosn建立链接并传递配置信息的

func SendInheritConfig() error {
    var unixConn net.Conn
    var err error
    // retry 10 time
    for i := 0; i < 10; i++ {
        unixConn, err = net.DialTimeout("unix", types.TransferMosnconfigDomainSocket, 1*time.Second)
        ......
    }
    ......
    configData, err := configmanager.InheritMosnconfig()
    ......

    uc := unixConn.(*net.UnixConn)
    defer uc.Close()

    n, err := uc.Write(configData)
    ......

    return nil
}

至此,配置数据传递也就完成了

监听端口迁移

listener迁移是用过 listen.sock 来完成的,new mosn 监听,old mosn 连接上去并传输数据,跟配置传输的逻辑差不多

new mosn监听是在 GetInheritListeners 这个方法中实现的,并获取所有的listener

func GetInheritListeners() ([]net.Listener, []net.PacketConn, net.Conn, error) {

    l, err := net.Listen("unix", types.TransferListenDomainSocket)
    ......
    defer l.Close()


    ul := l.(*net.UnixListener)
    ul.SetDeadline(time.Now().Add(time.Second * 10))
    // 这里卡主,等待 old mosn连接
    uc, err := ul.AcceptUnix()

    buf := make([]byte, 1)
    oob := make([]byte, 1024)
  // 接收 old mosn传递过来的数据
    _, oobn, _, _, err := uc.ReadMsgUnix(buf, oob)
    
    scms, err := unix.ParseSocketControlMessage(oob[0:oobn])
    
  // 解析出来fd
    gotFds, err := unix.ParseUnixRights(&scms[0])

    var listeners []net.Listener
    var packetConn []net.PacketConn
    for i := 0; i < len(gotFds); i++ {
        fd := uintptr(gotFds[i])
        file := os.NewFile(fd, "")
        .....
        defer file.Close()

    // 通过fd 恢复 listener,本质是对fd的监听
        fileListener, err := net.FileListener(file)
        if err != nil {
            pc, err := net.FilePacketConn(file)
            if err == nil {
                packetConn = append(packetConn, pc)
            } else {

                log.StartLogger.Errorf("[server] recover listener from fd %d failed: %s", fd, err)
                return nil, nil, nil, err
            }
        } else {
            // for tcp or unix listener
            listeners = append(listeners, fileListener)
        }
    }

    return listeners, packetConn, uc, nil
}

通过上述方法,new mosn建立 socket监听,并等待 old mosn的连接,old mosn连接上后,等待 old mosn 传递 所有listener 的fd,然后 new mosn 进行恢复即可

但是 同一个 fd,有两个监听,不就乱了吗,所以 当 old mosn 传递过来fd后,会主动 stop accept,不再进行监听

new mosn 的listener 传递逻辑在sendInheritListeners 里面

func sendInheritListeners() (net.Conn, error) {
  // 列出来所有的 listener,返回格式 os.File
    lf := ListListenersFile()
    ......

    lsf, lerr := admin.ListServiceListenersFile()
    ......

    var files []*os.File
    files = append(files, lf...)
    files = append(files, lsf...)

    ......
    fds := make([]int, len(files))
    for i, f := range files {
    // 获取 file 的 fd
        fds[i] = int(f.Fd())
        defer f.Close()
    }

    var unixConn net.Conn
    var err error
    // retry 10 time
    for i := 0; i < 10; i++ {
        unixConn, err = net.DialTimeout("unix", types.TransferListenDomainSocket, 1*time.Second)
        .......
    }
    ......

    uc := unixConn.(*net.UnixConn)
    buf := make([]byte, 1)
  // 将 fd 转成 socket message
    rights := syscall.UnixRights(fds...)
    n, oobn, err := uc.WriteMsgUnix(buf, rights, nil)
    ......

    return uc, nil
}

Old mosn 通过 ListListenersFile 将所有的listener罗列出来,这个主要得益于MOSN良好的设计模式;mosn维护了一个全局的servers,而server的结构如下

type server struct {
    serverName string
    stopChan   chan struct{}
    handler    types.ConnectionHandler
}

type ConnectionHandler interface {
    ......

    // ListListenersFD reports all listeners' fd
    ListListenersFile(lctx context.Context) []*os.File
}

每个server对自己的listener描述清晰

继续回到 old mosn传递listener的过程

  • old mosn 罗列出来所有的listener,并获取file
  • old mosn 获取所有 file的fd,并将fd通过UnixRights转成 socket message,供传递
  • new mosn 接收到 socket message,转成fd,并通过文件建立listener

然后 old mosn 会关闭掉所有的listener,停止accept 新的链接

func shutdownServers() {
    for _, server := range servers {
        server.Shutdown()
    }
}

func (ch *connHandler) GracefulStopListeners() error {
    var failed bool
    listeners := ch.listeners
    wg := sync.WaitGroup{}
    wg.Add(len(listeners))
    for _, l := range listeners {
        al := l
        log.DefaultLogger.Infof("graceful shutdown listener %v", al.listener.Name())
        // Shutdown listener in parallel
        utils.GoWithRecover(func() {
            defer wg.Done()
            if err := al.listener.Shutdown(); err != nil {
                log.DefaultLogger.Errorf("failed to shutdown listener %v: %v", al.listener.Name(), err)
                failed = true
            }
        }, nil)
    }
    wg.Wait()

    return nil
}

func (l *listener) Shutdown() error {
    changed, err := l.stopAccept()
    if changed {
        l.cb.OnShutdown()
    }
    return err
}

至此,只有new mosn 能建立新的链接,old mosn不再建立新的链接了

listener传递完成后,新的链接都建立到 new mosn上去了,剩余的就是存量长链接了

长链接迁移

长链接迁移是用过 conn.sock 来完成的,同上,也是由new mosn 监听,old mosn 连接上去并传输数据;不过,这里并没有传递配置和listener那样简单,需要考虑很多边际问题

在这里,链接分为两部分

  1. client/server -> mosn的链接
  2. mosn -> client/server 的链接

我们只需要考虑 client/server -> mosn的链接 的情况即可,mosn -> client/server 的链接 这种情况,由于mosn是主动连接方,断开并不会对下游造成任何影响

先来看下 长链接迁移的流程

长连接迁移过程

  • Client 发送请求到 MOSN
  • MOSN 通过 domain socket(conn.sock) 把 TCP1 的 FD 和连接的状态数据发送给 New MOSN
  • New MOSN 接受 FD 和请求数据创建新的 Conection 结构,然后把 Connection id 传给 MOSN,New MOSN 此时就拥有了TCP1 的一个拷贝。Old MOSN 停止读取 TCP1 的请求,New MOSN 开始读取 TCP1的请求,TCP1的迁移就完成了
  • New MOSN 通过 LB 选取一个新的 Server,建立 TCP3 连接,转发请求到 Server
  • Server 回复响应到 New MOSN
  • New MOSN 通过 MOSN 传递来的 TCP1 的拷贝,回复响应到 Client

接下来看下代码

注: 前面 WaitConnectionsDone 已经将 connection.stopChan close掉了

链接迁移是在 startReadLoop 中完成的

func (c *connection) startReadLoop() {
    var transferTime time.Time
    for {
        ......
        select {
        case <-c.stopChan:
      // 首先设置 transfer 时间
            if transferTime.IsZero() {
                if c.transferCallbacks != nil && c.transferCallbacks() {
                    randTime := time.Duration(rand.Intn(int(TransferTimeout.Nanoseconds())))
                    transferTime = time.Now().Add(TransferTimeout).Add(randTime)
                    log.DefaultLogger.Infof("[network] [read loop] transferTime: Wait %d Second", (TransferTimeout+randTime)/1e9)
                } else {
                    // set a long time, not transfer connection, wait mosn exit.
                    transferTime = time.Now().Add(10 * TransferTimeout)
                    log.DefaultLogger.Infof("[network] [read loop] not support transfer connection, Connection = %d, Local Address = %+v, Remote Address = %+v",
                        c.id, c.rawConnection.LocalAddr(), c.RemoteAddr())
                }
            } else {
                if transferTime.Before(time.Now()) {
                    c.transfer()
                    return
                }
            }
        default:
        }
    .......
  }
  • 在 old mosn 调用 WaitConnectionsDone 将 connection.stopChan close之后,在 startReadLoop 循环中 首先设置以下 随机 transfer 时间
  • 在 transfer时间到了之后,开始 迁移 链接
func (c *connection) transfer() {
    c.notifyTransfer()
    id, _ := transferRead(c)
    c.transferWrite(id)
}

func transferRead(c *connection) (uint64, error) {
    ......
    unixConn, err := net.Dial("unix", types.TransferConnDomainSocket)
    ......
    defer unixConn.Close()

    file, tlsConn, err := transferGetFile(c)
    ......

    uc := unixConn.(*net.UnixConn)
    // send type and TCP FD
    err = transferSendType(uc, file)
    ......
    // send header + buffer + TLS
    err = transferReadSendData(uc, tlsConn, c.readBuffer)
    ......
    // recv ID
    id := transferRecvID(uc)
    log.DefaultLogger.Infof("[network] [transfer] [read] TransferRead NewConn Id = %d, oldId = %d, %p, addrass = %s", id, c.id, c, c.RemoteAddr().String())

    return id, nil
}
  • 每一个connection迁移的时候,会首先构建一个 unix connection 用于 old mosn 和 new mosn交互
  • 首先将 connection 的 fd 通过 scoket传递给new mosn
  • 然后 将 connection tls 和 读取的buf数据再传递给 new mosn 处理
  • 最后 记录下来 new mosn 根据 fd 创建的 新的connection的id

这里mosn构造了一个简单的socket协议,用于 传递 connection 的tls和buf数据

/**
 *  transfer read protocol
 *  header (8 bytes) + (readBuffer data) + TLS
 *
 * 0                       4                       8
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 * |      data length      |     TLS length        |
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 * |                     data                      |
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 * |                     TLS                       |
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 *

接下来继续看下,new mosn 收到 connection后的处理, new mosn 处理是在 transferHandler 中完成的

func transferHandler(c net.Conn, handler types.ConnectionHandler, transferMap *sync.Map) {
    ......

    uc, ok := c.(*net.UnixConn)
    ......
    // recv type
    conn, err := transferRecvType(uc)
    ......

    if conn != nil {
        // transfer read
        // recv header + buffer
        dataBuf, tlsBuf, err := transferReadRecvData(uc)
        ......
        connection := transferNewConn(conn, dataBuf, tlsBuf, handler, transferMap)
        if connection != nil {
            transferSendID(uc, connection.id)
        } else {
            transferSendID(uc, transferErr)
        }
    }
  ......
}

new mosn 收到connection迁移请求后,根据传递过来的fd,首先转换成connection,然后 根据 tls 数据,构建为 tls conn,这些完成后,根据 conn的监听信息和handler类型,找到对应的listener,并将这个connection加进去,然后就可以开始处理 传递过来的buf数据了

最后,new mosn 将新的connection的id,传给old mosn,用于传递 写请求

至此,old mosn connection的 readLoop 也退出了,不再读取新的数据,数据也都由new mosn来读取了

接下来就是写请求了

old mosn 如果继续往连接里面写数据,可能会和new mosn冲突,导致数据操作,所以 old mosn的写请求,是直接转给 new mosn来处理的

Old mosn 迁移写请求:

func transferWrite(c *connection, id uint64) error {
    ......
    unixConn, err := net.Dial("unix", types.TransferConnDomainSocket)
    ......
    defer unixConn.Close()

    uc := unixConn.(*net.UnixConn)
    err = transferSendType(uc, nil)
    ......
    // build net.Buffers to IoBuffer
    buf := transferBuildIoBuffer(c)
    // send header + buffer
    err = transferWriteSendData(uc, int(id), buf)
    ......
    return nil
}

Mosn 也为写请求,构建了一个简单的socket协议

*  transfer write protocol
 *  header (8 bytes) + (writeBuffer data)
 *
 * 0                       4                       8
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 * |      data length      |    connection  ID     |
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 * |                     data                      |
 * +-----+-----+-----+-----+-----+-----+-----+-----+
 *

这里会将,迁移 读请求时获取到的 connection id 记录下来,并传递给 new mosn,让 new mosn 根据 id 找到对应的链接

new mosn 接收写请求处理逻辑,也是在 transferHandler 中完成的:

func transferHandler(c net.Conn, handler types.ConnectionHandler, transferMap *sync.Map) {
    ......
        // transfer write
        // recv header + buffer
        id, buf, err := transferWriteRecvData(uc)
        ......
        connection := transferFindConnection(transferMap, uint64(id))
        ......
        err = transferWriteBuffer(connection, buf)
        ......
    
}

New mosn 从 old mosn的socket请求中,解析出来 connection id 和 buf 数据

根据 id 找到 new mosn中的 connection,然后将buf写入即完成

至此,读写请求都迁移完成,整个长连接的迁移也就完成了

总结

MOSN对热升级的处理,是做到极致的,深度是远远高于市场上其他产品的;同时,深度也往往伴随着风险,做到链接迁移层面,可能也并不是MOSN的本意,而是历史原因的驱动

在源码逻辑层面,个人也有一点简单的看法

  1. 整个热升级模块,更像是函数驱动,而缺少设计,从 conn.sock listen.sock 等多个sock文件就可以看出,如果设计好点的话,完全可以通过构建socket协议,而规避掉多个socket文件,同时,逻辑也会更清晰简洁一点,而非散落在各个方法里面
  2. 有些边界性问题还是没有处理的很好;例如,listener迁移的时候,如果有listener close掉了,但是配置还存在MOSN中,这里就会panic了;eg: reconfig.sock 在处理最后会删除,如果走到了这一步,然后回滚了,后续也无法进行热升级

最后,MOSN还是很优秀的,respect!!!

免责声明 MOSN热升级逻辑浅析,资源类别:文本, 浏览次数:16 次, 文件大小:-- , 由本站蜘蛛搜索收录2023-02-06 08:29:00。此页面由程序自动采集,只作交流和学习使用,本站不储存任何资源文件,如有侵权内容请联系我们举报删除, 感谢您对本站的支持。 原文链接:https://segmentfault.com/a/1190000043391762