记一次偶发的bug排查——redis-py-cluster库的bug

排查流水账：

通过平台监控，发现很多偶发的查看推荐列表的接口时延大于0.5s
写单元测试，不能重现。在测试环境不能重现。只有在正式环境可以偶发重现。
通过日志埋点，等待重现
不断地加日志埋点后发现耗时在redis的hmget操作

这时猜想原因

hmget命令才会有，会不会是hmget命令的问题
查看redis的慢查询日志，发现没有慢查询。排除是Redis执行慢的原因
查看当时的负载情况，负载很低，并发也不多。所以排除是Redis的命令等待原因
多协程下，20条hmget，不会全部都卡，只会卡几条，后面又会正常
正常hmget的用时是0.01s左右，卡的时候需要0.3-0.9s
自己写个脚本，不断地用多协程去执行hmget，不能重现。

猜想解决方案：

修改多协程池的数量，从20改为3
获取用户信息改为串行

继续往hmget命令里面的代码加埋点日志

由于是修改第三方库，所以要格外小心
通过阅读源码，发现hmget的底层流程是rediscluster模块里面的client文件里面的execute_command函数
修改后，测试环境之下单元测试没问题后，部署到正式环境

最后定位到是在里面的self.connection_pool.nodes.initialize()这行代码耗时
只有当refresh_table_asap=true才会执行这行代码，所以解决思路

为什么refresh_table_asap会等于true

发现只有当连接redis的时候报错ClusterDownError或者MovedError才会设置refresh_table_asap=True
通过日志埋点。发现是MovedError异常导致的。
继续增加日志埋点，发现整个触发的流程是：

触发异常ConnectionError
设置try_random_node=True
下一次随机拿一个节点，这时候可能拿到slot不对的节点
连接节点后，会报异常MovedError，并把目标节点的信息返回来，同时设置refresh_table_asap=True。
这时会把slot对应的节点设置为返回来的节点信息
重新连接节点，执行命令成功
但是这时候已经设置了refresh_table_asap=True，执行下一个命令的时候，就会执行self.connection_pool.nodes.initialize()
由于使用了多协程，而且self.connection_pool.nodes.initialize()命令没有加锁，所以会导致这个耗时加剧

通过print traceback，看看为什么会触发ConnectionError异常，发现是redis服务端断开了连接。
这时候回想到redis有机制，超过一定时间没有命令过来，就会关闭连接。在redis的timeout 配置，一般是300s。所以这样解释了为什么这个是偶发的。
写单元测试，建立连接后，等待350s再执行命令，稳定重现bug。

为什么initialize耗时这么慢

通过单元测试，发现initialize命令并不慢，大于0.04s左右就能完成，但是多协程下是0.5s左右。
所以考虑是多协程下，因为没有锁，所以多个协程都执行了这条命令，导致最终的用时是原来的10倍

修改测试环境redis的timeout=5s，写个测试用例，在测试环境可以稳定重现。
所以定位到rediscluster有问题，解决思路

不要在多协程执行redis命令（感觉不好）
升级库，看能不能解决。查看这个库的git地址（https://github.com/Grokzen/redis-py-cluster）的最新版本，问题依然存在。
catchConnectionError异常的时候，区分是否服务端断开连接，如果是，不设置try_random_node=True，重试
init的时候加锁
参考redis.py的做法，在catch服务端断开连接异常后，重新连接后重试

最后选用了思路5。

execute_command函数（包含埋点日志，去除不必要的代码段），在rediscluster库的client.py

@clusterdown_wrapper

def execute_command(self, *args, **kwargs):

    """

    Send a command to a node in the cluster

    """

    import logging

    log=logging.getLogger('service.log')

    log.error(u'redis execute_command 1 %s ' % str(args))

    # If set externally we must update it before calling any commands

    if self.refresh_table_asap:  #执行self.connection_pool.nodes.initialize()的代码段

        log.error(u'redis execute_command 2 %s ' % str(args))

        self.connection_pool.nodes.initialize()

        log.error(u'redis execute_command 3 %s ' % str(args))

        self.refresh_table_asap = False

    log.error(u'redis execute_command 4 %s ' % str(args))

    redirect_addr = None

    asking = False

    try_random_node = False

    log.error(u'redis execute_command 7 %s ' % str(args))

    slot = self._determine_slot(*args)

    log.error(u'redis execute_command 8 %s ' % str(args))

    ttl = int(self.RedisClusterRequestTTL)

    while ttl > 0:

        ttl -= 1

        if asking:

            node = self.connection_pool.nodes.nodes[redirect_addr]

            r = self.connection_pool.get_connection_by_node(node)

        elif try_random_node:

            r = self.connection_pool.get_random_connection()

            try_random_node = False

        else:

            if self.refresh_table_asap:

                # MOVED

                node = self.connection_pool.get_master_node_by_slot(slot)

            else:

                node = self.connection_pool.get_node_by_slot(slot)

            r = self.connection_pool.get_connection_by_node(node)

        try:

              r.send_command(*args)

              log.error(u'redis execute_command 10 %s ' % str(args))

              ret= self.parse_response(r, command, **kwargs)

              log.error(u'redis execute_command 11 %s ' % str(args))

              return ret

        except (RedisClusterException, BusyLoadingError):

            raise

        except (ConnectionError, TimeoutError):

            try_random_node = True

            log.error(u'redis execute_command 14 %s ' % str(args))

            if ttl < self.RedisClusterRequestTTL / 2:

                log.error(u'redis execute_command 15 %s ' % str(args))

                time.sleep(0.1)

        except ClusterDownError as e:

            log.error(u'redis execute_command 17 %s ' % str(args))

            self.connection_pool.disconnect()

            self.connection_pool.reset()

            self.refresh_table_asap = True

            raise e

        except MovedError as e:

            # Reinitialize on ever x number of MovedError.

            # This counter will increase faster when the same client object

            # is shared between multiple threads. To reduce the frequency you

            # can set the variable 'reinitialize_steps' in the constructor.

            import traceback

            print traceback.format_exc()

            log.error(u'redis execute_command 16 %s ' % str(args))

            self.refresh_table_asap = True

            self.connection_pool.nodes.increment_reinitialize_counter()

            node = self.connection_pool.nodes.set_node(e.host, e.port, server_type='master')

            self.connection_pool.nodes.slots[e.slot_id][0] = node

优化：

把

              r.send_command(*args)

              ret= self.parse_response(r, command, **kwargs)

              return ret

改为

try:

    r.send_command(*args)

    return self.parse_response(r, command, **kwargs)

except ConnectionError as e:

    from redis.connection import SERVER_CLOSED_CONNECTION_ERROR

    if SERVER_CLOSED_CONNECTION_ERROR in e.message:

        r.disconnect()

        r.send_command(*args)

        return self.parse_response(r, command, **kwargs)

    else:

        raise

　未经许可，请不要转载

hmget

记一次偶发的bug排查——redis-py-cluster库的bug