SERP 数据血缘 + 审计日志:出问题 5 分钟定位

客户说"这个排名数据不对",你从哪开始查?如果没有血缘和审计日志,排查要靠猜。加上之后,5 分钟能定位到是哪次请求、哪个参数、哪个数据源。

1. 记录什么

每个 SERP 请求记 4 类信息:

• 请求信息:query、hl/gl、参数、request_id
• 响应信息:status、elapsed_ms、credits_charged
• 来源信息:哪个任务/客户触发的
• 结果信息:返回了哪些结果

2. 血缘表结构

CREATE TABLE serp_lineage (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    request_id VARCHAR(64),
    trace_id VARCHAR(64),
    query TEXT,
    params TEXT,
    search_type VARCHAR(32),
    status INT,
    elapsed_ms INT,
    credits_charged INT,
    source_task VARCHAR(128),
    created_at TIMESTAMP,
    INDEX idx_trace (trace_id),
    INDEX idx_request (request_id),
    INDEX idx_created (created_at)
);

3. 记录代码

def search_with_audit(query, source_task=None, trace_id=None):
    import uuid
    trace_id = trace_id or uuid.uuid4().hex

    r = requests.post(
        "https://api.serpbase.dev/google/search",
        headers={"X-API-Key": os.environ["SERPBASE_API_KEY"]},
        json={"q": query, "hl": "zh-CN", "gl": "cn"},
        timeout=5,
    )
    data = r.json()

    # 记血缘
    db.execute(
        "INSERT INTO serp_lineage (request_id, trace_id, query, params,"
        " search_type, status, elapsed_ms, credits_charged, source_task)"
        " VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)",
        (
            data.get("request_id"),
            trace_id,
            query,
            json.dumps({"hl": "zh-CN", "gl": "cn"}),
            data.get("search_type"),
            data.get("status"),
            data.get("elapsed_ms"),
            data.get("credits_charged"),
            source_task,
        ),
    )
    return data, trace_id

4. 排查:按 request_id 溯源

def trace_by_request(request_id):
    """按 request_id 查这次请求的全部信息"""
    return db.query(
        "SELECT * FROM serp_lineage WHERE request_id = %s",
        [request_id],
    )

客户给的 request_id,直接查到 query、参数、耗时、扣费,定位是参数问题还是数据问题。

5. 排查:按 URL 反查

客户说"这个页面排名不对",按 URL 反查所有相关请求:

SELECT request_id, query, params, created_at
FROM serp_lineage
WHERE request_id IN (
    SELECT request_id FROM serp_results
    WHERE link = 'https://example.com/page'
);

6. 审计日志

记录谁在什么时候查了什么:

def audit_log(api_key, query, action):
    db.execute(
        "INSERT INTO serp_audit (api_key, query, action, created_at)"
        " VALUES (%s,%s,%s,NOW())",
        [api_key, query, action],
    )

7. 保留策略

def cleanup(retention_days=90):
    """血缘保留 90 天,审计保留 180 天"""
    db.execute(
        "DELETE FROM serp_lineage WHERE created_at < NOW() - INTERVAL %s DAY",
        [retention_days],
    )
    db.execute(
        "DELETE FROM serp_audit WHERE created_at < NOW() - INTERVAL 180 DAY",
        [],
    )

8. 监控血缘健康

from prometheus_client import Counter

lineage_missing = Counter("serp_lineage_missing", "Responses without lineage")

def monitored_search(query, **kw):
    data, _ = search_with_audit(query, **kw)
    if not data.get("request_id"):
        lineage_missing.inc()  # 缺 request_id,预警
    return data

9. 30 天实测

10. 常见坑

坑 1:request_id 没入库,排查时没有抓手。

坑 2:血缘表和业务表分开存,join 慢。加 request_id 索引。

坑 3:trace_id 传播断链(异步任务没传 context),跨系统追不到。

11. 总结

血缘 + 审计双表:血缘管"这次请求是啥",审计管"谁查的"。客户投诉从"猜"变"查",5 分钟定位。完整字段参考在 SerpBase 文档(serpbase.dev/docs)。