如何合并从Prometheus查询函数获取的Python字典

我向Prometheus发起了多次不同指标的查询请求，希望将返回的结果合并为一个字典。预期结果格式如下：

{'domain': 'domain.com', 'uptime': '100', 'threats_number': '12345'}

现有查询函数 #

正常运行时间查询函数 #

# 正常运行时间
def uptime_percent():
    uptime_data = prom.custom_query(
        query="avg_over_time(host_tracker_uptime_percent{job='donodeexporter'}[10d])"
    )
    for domain_z in uptime_data:
        domain_zone = domain_z['metric']['zone']
        uptime = domain_z['value'][1]
        yield ({"domain": f'{domain_zone}', "uptime": f'{uptime}'})

Cloudflare威胁总数查询函数 #

# 按区域统计Cloudflare威胁总数
def cloudflare_threats():
    cloudflare_threats_total = prom.custom_query(
        query="sum by(zone) (increase(cloudflare_zone_threats_total{job='cloudflare'}[10d]))"
    )
    for domain_z in cloudflare_threats_total:
        domain_zone = domain_z['metric']['zone']
        threats_number = domain_z['value'][1]
        yield ({"domain": f'{domain_zone}', "threats_number": f'{threats_number}'})

解决方案 #

以域名（zone）作为唯一标识，将两个查询的结果合并，具体实现代码如下：

def merge_prom_results():
    # 把生成器结果转成以域名为键的字典
    uptime_map = {item["domain"]: item for item in uptime_percent()}
    threats_map = {item["domain"]: item for item in cloudflare_threats()}
    
    merged_list = []
    # 覆盖所有出现过的域名
    all_domains = set(uptime_map.keys()).union(threats_map.keys())
    
    for domain in all_domains:
        combined = {"domain": domain}
        # 合并正常运行时间数据
        if domain in uptime_map:
            combined["uptime"] = uptime_map[domain]["uptime"]
        # 合并威胁总数数据
        if domain in threats_map:
            combined["threats_number"] = threats_map[domain]["threats_number"]
        merged_list.append(combined)
    
    return merged_list

补充说明 #

• 上述代码会生成一个包含所有域名合并结果的列表，每个元素是符合预期格式的字典
• 如果某域名只在其中一个查询结果中存在，对应的缺失指标不会被添加；也可根据需求设置默认值，比如combined["uptime"] = uptime_map.get(domain, {}).get("uptime", "0")

内容的提问来源于stack exchange，提问作者Rostyslav Malenko