【测试工具】一次全量数据对比工具发现问题的过程与思考

image1

一次全量数据对比工具发现问题的过程与思考

image2

如果没有这次全量数据对比工具,那么也许这个历史问题会继续隐藏着,直到发生线上事故才暴露出来,毕竟人工抽样验证发现的概率只有5.8%

背景是发票系统有18500个电子发票订单被财务系统驳回了,驳回原因是财务系统上线了全电发票需求,上线后电子发票枚举被误删,无法处理电子发票。需要我们发票系统对这18500电子发票订单,重新触发提票,让发票能正常开出来。也就是,我们需要刷数。刷数是个高危操作,极易引发线上问题。

经验教训告诉我们,刷数虽然是一种处理线上问题的方法,但是也特别容易引起二次事故。对于刷数,我们需要像新需求一样对待,经过完备的需求分析、设计评审、代码评审。主要考虑以下3点:

  • 刷数范围,怎么筛选问题数据,评审过滤条件;

  • 刷数程序,怎么修复问题数据,评审代码逻辑;

  • 验证方法,怎么验证修复数据,分析测试场景;

在刷数实施前,群里报备,周知相关方及相关人员。先试刷,验证无问题后,再全刷,最后验证问题数据已经得以修复。这一套刷数流程,通过加强前期评审,能很好地预防缺陷,增强刷数成功的信心。但是人工评估可能遗漏场景,可能对真实数据情况把握不全,刷数的关键还在于最后的数据验证。

抽样验证还是全量验证,这是一个问题。抽样验证是人工随机挑几个数据进行验证,我们通常倾向于使用抽样验证,一是抽样验证是一种科学的有效的验证方法,虽然它存在一定概率的遗漏,但是很多时候是可以接受的风险;二是抽样验证也是无奈之举,找不到办法进行全量验证。我们遇到的困难是,数据存在ES中,批量把所有数据查出来很麻烦,也无法直接编写校验逻辑,全量验证似乎是不可能的。

全量验证有2个思路:

  • 如果能直连库,那么先提数,再写程序对比;

  • 如果只能WEB页面查数,那么使用Python爬虫提数,再写程序对比;

后者适用于我们的情况。ES能通过WEB页面查询数据,只要是WEB页面,即使有Cookie,也能爬取到接口数据。F12抓包到查询接口的URL、Cookie、入参后,使用Python的requests库可以爬取查询结果数据:

url = 'http://xxx'
headers = {
    'Cookie': 'xxx',
    'Content-Type': 'application/json'
}
response = request('get', url=url, headers=headers)
result = response.json()

我们使用这种方式,传入订单号,查询到了申请单数据,以便进行对比校验逻辑。而订单号,研发打在了日志里,需要下载日志文件后,进行解析,可以使用Python切片:

orders = set()
with open('some.log') as f:
    for line in f.read().splitlines():
        if 'xxx' in line:
            orders.add(line[line.index('orderId='):line.index(',已执行')].replace('orderId=', ''))
print(len(orders))
with open('orders.txt', 'w') as f:
    f.writelines(','.join(orders))

.index可以定位到关键词的索引,然后[:]切片获取指定内容。

日志解析订单号+爬虫获取申请单+编写对比校验逻辑,全量数据对比工具就完成了。可是18500单,上万级别数据,要全部对比完,至少要几个小时。是时候使用多线程了。

多线程第一步,拆解数据,将18500单拆成以100单为一组的列表:

def split_list(lst, size):
    """
    将列表 lst 拆分成每份 size 个元素的子列表,并返回一个包含所有子列表的列表。
    """
    return [lst[i:i + size] for i in range(0, len(lst), size)]

多线程第二步,队列提数,让多个线程依次从列表中取出数据,每个线程每次取不同的数据:

import threading

### 待处理的数据列表
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

### 创建锁对象
lock = threading.Lock()

### 定义线程函数
def process_data():
    global data
    while True:
        # 加锁
        lock.acquire()
        # 如果列表为空,说明所有数据已被处理完毕,退出循环
        if len(data) == 0:
            lock.release()
            break
        # 取出列表中的第一个数据
        num = data.pop(0)
        # 释放锁
        lock.release()
        # 对数据进行处理
        print("Processing data:", num)

### 创建多个线程
threads = []
for i in range(5):
    t = threading.Thread(target=process_data)
    threads.append(t)

### 启动所有线程
for t in threads:
    t.start()

### 等待所有线程结束
for t in threads:
    t.join()

print("All data processed.")

阻塞队列是通过加锁来实现的,每个线程在取数前先加锁,然后pop(0)取出列表中的第一个数据,再释放锁。上述程序修改①data②数据处理逻辑③线程数即可使用。

对比工具使用多线程后,运行时间从小时级别降到了分钟级别。当天研发本来以为要跑很久,准备第二天再来看,就先撤了。我执着了一下多线程实现,在ChatGPT帮助下,很快就把结果跑出来。赶紧打电话摇人,让研发回来看问题,研发那时刚到家,掏出钥匙把门打开。在全量对比前,我们也都做了一轮抽样验证,均没有发现任何问题。18500单全量对比,发现有1064单存在问题,能抽样发现的概率只有5.8%。

总结,分析这些问题原因:

  • 遗漏了1种数据情况,评估不到位

  • 未考虑到刷数环境影响,在预发环境刷数,上下游环境都是预发,可能跟线上版本不一样,尤其是做写操作时,格外需要注意

  • 刷数程序本身缺陷,这个缺陷隐藏在一段用了很多次刷数的历史代码里面,不是100%会导致问题

可以发现,大数据量验证,人工无法百分百保证数据准确性,抽样检查,94.2%概率发现不了问题。最稳妥的办法,还是全量对比,让每条数据,都经过对比规则的检验。