前几天项目中遇到了一个长链接假死问题,服务端和client端采用的是h2c长连接。服务端作为sidecar部署在k8s的pod里面,当滚动升级pod的时候,client端和老的pod的连接一直存在,即使老的pod已经被删除了。(client和网关是一个东西)

发现问题

突然有一天前端同事说调用全部503(内部服务不可用),赶紧去环境上查看log,发现网关发送到后段的请求全部超时。用netstat查看连接状态也是没问题的。

去新建的pod上查看,发现并没有连接,因此网关端的连接并不是和新建的pod建立的。

解决问题

1.查看网关端代码

首先看一下网关测的代码:

    func NewH2cClient() *http.Client {
    client := &http.Client{
        Transport: &http2.Transport{
            AllowHTTP: true,
            DialTLS:   dialH2cTimeout,
        },
        Timeout: 10 * time.Second,
        }
        return client
    }

    func dialH2cTimeout(network, addr string, cfg *tls.Config) (net.Conn, error) {
       tylog.Infof("connect to remote address: %+v:", addr)
       return net.Dial(network, addr)
    }

client端的实现很简单,复用http.Client对象。在这里设置了TimeOut为10s,这个是发送超时时间。通过log也可以看到10s超时后返回了error,但连接并没有终端,查阅了一些net/http的文档服务端对于超时的控制还是比较细腻的。

但client端的控制就比较简陋(其实应该是net/http封装的东西太多,导致用户可直接操作的东西太少)

从图中可以看出,client端不像服务端有读写超时设置,client端就一个Timeout,是从建立链接到接收到body的时间段。因此我们在client端可以做的东西貌似并不多。

2.回归问题

查阅了部分资料后,回到问题的所在点:

为什么服务端已经被删除,但长链接还存在??

因此我们需要看的就是client端的conn什么时候应该断开。http包中的读和写是分开的,分别对应的是两个goroutine,我们主要应该看的是读操作。

      func (cc *ClientConn) readLoop() {
         rl := &clientConnReadLoop{cc: cc}
         defer rl.cleanup() //这个会清理链接
         cc.readerErr = rl.run()
         if ce, ok := cc.readerErr.(ConnectionError); ok {
              cc.wmu.Lock()
              cc.fr.WriteGoAway(0, ErrCode(ce), nil)
              cc.wmu.Unlock()
         }
     }

上述代码是read的入口,我们看一下run这个函数的实现

func (rl *clientConnReadLoop) run() error {
    cc := rl.cc
    rl.closeWhenIdle = cc.t.disableKeepAlives() || cc.singleUse
    gotReply := false // ever saw a HEADERS reply
    gotSettings := false
    for {
        f, err := cc.fr.ReadFrame()
        if err != nil {
            cc.vlogf("http2: Transport readFrame error on conn %p: (%T) %v", cc, err, err)
        }
        if se, ok := err.(StreamError); ok {
            if cs := cc.streamByID(se.StreamID, false); cs != nil {
                cs.cc.writeStreamReset(cs.ID, se.Code, err)
                cs.cc.forgetStreamID(cs.ID)
                if se.Cause == nil {
                    se.Cause = cc.fr.errDetail
                }
                rl.endStreamError(cs, se)
            }
            continue
        } else if err != nil {
            return err
        }
        if VerboseLogs {
            cc.vlogf("http2: Transport received %s", summarizeFrame(f))
        }
        if !gotSettings {
            if _, ok := f.(*SettingsFrame); !ok {
                cc.logf("protocol error: received %T before a SETTINGS frame", f)
                return ConnectionError(ErrCodeProtocol)
            }
            gotSettings = true
        }
        maybeIdle := false // whether frame might transition us to idle

        switch f := f.(type) {
        case *MetaHeadersFrame:
            err = rl.processHeaders(f)
            maybeIdle = true
            gotReply = true
        case *DataFrame:
            err = rl.processData(f)
            maybeIdle = true
        case *GoAwayFrame:
            err = rl.processGoAway(f)
            maybeIdle = true
        case *RSTStreamFrame:
            err = rl.processResetStream(f)
            maybeIdle = true
        case *SettingsFrame:
            err = rl.processSettings(f)
        case *PushPromiseFrame:
            err = rl.processPushPromise(f)
        case *WindowUpdateFrame:
            err = rl.processWindowUpdate(f)
        case *PingFrame:
            err = rl.processPing(f)
        default:
            cc.logf("Transport: unhandled response frame type %T", f)
        }
        if err != nil {
            if VerboseLogs {
                cc.vlogf("http2: Transport conn %p received error from processing frame %v: %v", cc, summarizeFrame(f), err)
            }
            return err
        }
        if rl.closeWhenIdle && gotReply && maybeIdle {
            cc.closeIfIdle()
        }
    }
}

再看一下ReadFrame函数:

// ReadFrame reads a single frame. The returned Frame is only valid
// until the next call to ReadFrame.
//
// If the frame is larger than previously set with SetMaxReadFrameSize, the
// returned error is ErrFrameTooLarge. Other errors may be of type
// ConnectionError, StreamError, or anything else from the underlying
// reader.
func (fr *Framer) ReadFrame() (Frame, error) {
    fr.errDetail = nil
    if fr.lastFrame != nil {
        fr.lastFrame.invalidate()
    }
    fh, err := readFrameHeader(fr.headerBuf[:], fr.r)
    if err != nil {
        return nil, err
    }
    if fh.Length > fr.maxReadSize {
        return nil, ErrFrameTooLarge
    }
    payload := fr.getReadBuf(fh.Length)
    if _, err := io.ReadFull(fr.r, payload); err != nil {
        return nil, err
    }
    f, err := typeFrameParser(fh.Type)(fr.frameCache, fh, payload)
    if err != nil {
        if ce, ok := err.(connError); ok {
            return nil, fr.connError(ce.Code, ce.Reason)
        }
        return nil, err
    }
    if err := fr.checkFrameOrder(f); err != nil {
        return nil, err
    }
    if fr.logReads {
        fr.debugReadLoggerf("http2: Framer %p: read %v", fr, summarizeFrame(f))
    }
    if fh.Type == FrameHeaders && fr.ReadMetaHeaders != nil {
        return fr.readMetaFrame(f.(*HeadersFrame))
    }
    return f, nil
}

其实就是一直阻塞读取数据,只有在收到reset或者fin包(EOF)的时候才会退出,然后会执行前面的清理函数:

 defer rl.cleanup()

该函数会调用MarkDead函数清理conn关联的map

func (p *clientConnPool) MarkDead(cc *ClientConn) {
    p.mu.Lock()
    defer p.mu.Unlock()
    for _, key := range p.keys[cc] {
        vv, ok := p.conns[key]
        if !ok {
            continue
        }
        newList := filterOutClientConn(vv, cc)
        if len(newList) > 0 {
            p.conns[key] = newList
        } else {
            delete(p.conns, key)
        }
    }
    delete(p.keys, cc)
}

看到这基本就知道问题的所在了:

  1. 删除pod的时候服务端没有发送Fin包
  2. 服务端发送了Fin包,但网关没收到

3.验证问题所在

使用了tcpdump抓包,发现client端的确没有收到Fin包

从抓包信息看的确没有收到Fin包,服务端抓也没抓到,于是感觉像k8s升级的时候直接删掉了老的pod(类似直接delete),并没有发送kill命令。但集群权限是拿不到的,不太好认证这个猜想。

4.解决方案

因为服务端也是自己写的,因此可以设置deployment来设置pod更新时的动作,比如:

"lifecycle": {
              "preStop": {
                "httpGet": {
                  "path": "/hc.do?a=offline",
                  "scheme": "HTTP",
                  "port": 8809
                }
              }
            },

让服务监听8809端口,k8s在删除pod前会访问 http127.0.0.1:8809/hc.do?a=offline, 这时候在服务里面做优雅下线,关掉tcp链接,这个问题就不会发生了。

5.总结

net/http包虽然很容易实现http服务,但对于client端的确不够友好,至少我没找到什么方式可以直接去控制超时时间,或者管理conn连接池,当然我也可以在它基础上实现一个自己的连接池(ClientConnPool接口)但成本不低。所以很多人都会自己去实现。

对于pod删除时信号量的问题,因为没有权限,所以没法看pod滚动升级时,kublet是怎么删除老的节点的,按照我的理解应该是先给容器发kill -15, 如果没退出再kill -9, 但就目前情况来看服务端并没有捕捉到信号量(我加了捕捉信号量,但并未打印出来log),后面有机会再去深入研究一下k8s的那块代码。