Cannot assign requested address 问题排查

背景

工单服务调用了我提供的自动化接口, 但是显示调用失败, 失败原因: Cannot assign requested address.

排查过程

根据提示猜测是端口用尽. 登录机器查看:

>>> netstat -nap | grep TIME_WAIT | awk '{print $5}' | sort | uniq -c | sort -rn | head -n10
43372 ip1:80
...

发现与 ip1:80 的 TIME_WAIT 连接数竟然高达 40k+, 抖动一下出现问题就很合理的.

为什么会出现这么多的 TIME_WAIT?

通过分析代码,请求方式如下.

def req():
    return requests.get("http://172.29.200.116:7070")

通过分析源码, requests.get 在收到响应后会主动断开连接.

def get(url, params=None, **kwargs):
    return request("get", url, params=params, **kwargs)

def request(method, url, **kwargs):
    with sessions.Session() as session:
        return session.request(method=method, url=url, **kwargs)

class Session():
    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

所以每次请求都会创建一个 tcp 连接, 请求结束后主动断开, 断开后会有 2MSL 时长的 TIME_WAIT 状态.

如何复用 tcp 连接

先创建一个 session 对象, 之后用 s 发送的同一个目标的请求都会复用 tcp 连接.

s = requests.session()
def req():
    return s.get("http://172.29.200.116:7070")

为什么会对同一 ip 有大量请求?

经分析, 目标 ip 是四层负载 vip, vip 后是 nginx, nginx 后才是真实服务. 很多内网服务的入口都在这里, 请求量大并不意外.

通过查看监控发现一个月内 time_wait 是一点点升高的, 和内存泄漏的走势十分相似, 推测是业务逻辑有什么资源忘记释放.

通过 client ip 查询 nginx 日志, 发现绝大多数请求都是发往一个域名 A 的, QPS 高达 600, url 对应的是一个服务发现功能, 服务发现请求频率这么高就很不正常了.

分析代码:

class Singleton(object):
    _instance = None
    def __new__(cls, *args, **kw):
        if not cls._instance:
            cls._instance = super(Singleton, cls).__new__(cls, *args, **kw)
        else:
            logging.warn("客户端实例已存在，请勿重复初始化!!!")

        return cls._instance

class BspClient(Singleton):
    def __init__(self, appkey, appsecret, host):
        # 启动一个后台线程,定时调用接口更新服务地址;
        thread = threading.Thread(target=scheduler, args=(self,))
        thread.start()

问题很明了了, 虽然用了单例模式, 但是__init__却在每次实例化 BspClient 后都执行了, 也就是每次都增加一个后台线程去定时调用服务发现接口.

修复方案:

由于每次调用返回的都是同一个实例, 可以在 init 中增加开关控制.

class BspClient(Singleton):
    _inited = False
    def __init__(self, appkey, appsecret, host):
        if _inited:
            return
        _inited = True
        thread = threading.Thread(target=scheduler, args=(self,))
        thread.start()