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

 排查流水账：

1. 通过平台监控，发现很多偶发的查看推荐列表的接口时延大于0.5s
2. 写单元测试，不能重现。在测试环境不能重现。只有在正式环境可以偶发重现。
3. 通过日志埋点，等待重现
4. 不断地加日志埋点后发现耗时在redis的hmget操作
1. 这时猜想原因
1. hmget命令才会有，会不会是hmget命令的问题
2. 查看redis的慢查询日志，发现没有慢查询。排除是Redis执行慢的原因
3. 查看当时的负载情况，负载很低，并发也不多。所以排除是Redis的命令等待原因
4. 多协程下，20条hmget，不会全部都卡，只会卡几条，后面又会正常
5. 正常hmget的用时是0.01s左右，卡的时候需要0.3-0.9s
6. 自己写个脚本，不断地用多协程去执行hmget，不能重现。
2. 猜想解决方案：
1. 修改多协程池的数量，从20改为3
2. 获取用户信息改为串行
5. 继续往hmget命令里面的代码加埋点日志
1. 由于是修改第三方库，所以要格外小心
2. 通过阅读源码，发现hmget的底层流程是rediscluster模块里面的client文件里面的execute_command函数
3. 修改后，测试环境之下单元测试没问题后，部署到正式环境
6. 最后定位到是在里面的self.connection_pool.nodes.initialize()这行代码耗时
7. 只有当refresh_table_asap=true才会执行这行代码，所以解决思路
1. 为什么refresh_table_asap会等于true
1. 发现只有当连接redis的时候报错ClusterDownError或者MovedError才会设置refresh_table_asap=True
2. 通过日志埋点。发现是MovedError异常导致的。
3. 继续增加日志埋点，发现整个触发的流程是：
1. 触发异常ConnectionError
2. 设置try_random_node=True
3. 下一次随机拿一个节点，这时候可能拿到slot不对的节点
4. 连接节点后，会报异常MovedError，并把目标节点的信息返回来，同时设置refresh_table_asap=True。
5. 这时会把slot对应的节点设置为返回来的节点信息
6. 重新连接节点，执行命令成功
7. 但是这时候已经设置了refresh_table_asap=True，执行下一个命令的时候，就会执行self.connection_pool.nodes.initialize()
8. 由于使用了多协程，而且self.connection_pool.nodes.initialize()命令没有加锁，所以会导致这个耗时加剧
4. 通过print traceback，看看为什么会触发ConnectionError异常，发现是redis服务端断开了连接。
5. 这时候回想到redis有机制，超过一定时间没有命令过来，就会关闭连接。在redis的timeout 配置，一般是300s。所以这样解释了为什么这个是偶发的。
6. 写单元测试，建立连接后，等待350s再执行命令，稳定重现bug。
2. 为什么initialize耗时这么慢
1. 通过单元测试，发现initialize命令并不慢，大于0.04s左右就能完成，但是多协程下是0.5s左右。
2. 所以考虑是多协程下，因为没有锁，所以多个协程都执行了这条命令，导致最终的用时是原来的10倍
8. 修改测试环境redis的timeout=5s，写个测试用例，在测试环境可以稳定重现。
9. 所以定位到rediscluster有问题，解决思路
1. 不要在多协程执行redis命令（感觉不好）
2. 升级库，看能不能解决。查看这个库的git地址（https://github.com/Grokzen/redis-py-cluster）的最新版本，问题依然存在。
3. catchConnectionError异常的时候，区分是否服务端断开连接，如果是，不设置try_random_node=True，重试
4. init的时候加锁
5. 参考redis.py的做法，在catch服务端断开连接异常后，重新连接后重试
10. 最后选用了思路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
 

　未经许可，请不要转载