如何通过Tempo/Loki从Grafana获取日志与追踪数据？

完全可以通过Loki（处理日志）和Tempo（处理追踪数据）的官方HTTP API便捷获取数据，以下是具体的Python实现方案：

一、获取Loki日志数据 #

Loki提供两类核心查询API：query（查询指定时间点的日志）和query_range（查询时间范围内的日志），Python中用requests库即可调用。

示例代码 #

import requests
import time

# Loki服务地址（根据实际部署修改）
LOKI_URL = "http://your-loki-instance:3100"

def get_loki_logs(query, start_time, end_time):
    # 转换为Unix时间戳（秒）
    start_ts = int(time.mktime(start_time.timetuple()))
    end_ts = int(time.mktime(end_time.timetuple()))
    
    # 构造请求参数
    params = {
        "query": query,
        "start": start_ts,
        "end": end_ts
    }
    
    # 如果Loki开启了认证，添加headers（比如API Key）
    headers = {
        # "Authorization": "Bearer YOUR_API_TOKEN"
    }
    
    # 调用时间范围查询接口
    response = requests.get(f"{LOKI_URL}/loki/api/v1/query_range", params=params, headers=headers)
    response.raise_for_status()
    
    # 解析返回的日志数据
    logs_data = response.json()
    for stream in logs_data["data"]["result"]:
        labels = stream["stream"]
        entries = stream["values"]
        print(f"日志标签: {labels}")
        for entry in entries:
            timestamp = entry[0]
            log_line = entry[1]
            print(f"[{timestamp}] {log_line}")

# 调用示例：查询过去1小时内job为"backend"且包含"error"的日志
if __name__ == "__main__":
    import datetime
    end_time = datetime.datetime.now()
    start_time = end_time - datetime.timedelta(hours=1)
    get_loki_logs('{job="backend"} |= "error"', start_time, end_time)

二、获取Tempo追踪数据 #

Tempo的API支持两类操作：通过标签搜索追踪ID，以及根据追踪ID获取完整的追踪链路数据。

示例代码 #

import requests

# Tempo服务地址（根据实际部署修改）
TEMPO_URL = "http://your-tempo-instance:3200"

def search_tempo_traces(tags, start_time, end_time):
    # 构造搜索请求体
    payload = {
        "query": tags,
        "start": start_time,
        "end": end_time
    }
    
    # 认证headers（如果需要）
    headers = {
        # "Authorization": "Bearer YOUR_API_TOKEN"
    }
    
    # 调用搜索接口获取追踪ID列表
    response = requests.post(f"{TEMPO_URL}/api/search", json=payload, headers=headers)
    response.raise_for_status()
    return response.json()["traces"]

def get_tempo_trace(trace_id):
    # 根据追踪ID获取完整链路数据
    response = requests.get(f"{TEMPO_URL}/api/traces/{trace_id}")
    response.raise_for_status()
    return response.json()

# 调用示例：搜索过去2小时内service为"payment"的追踪，并获取第一个追踪的详细数据
if __name__ == "__main__":
    import datetime
    end_ts = int(datetime.datetime.now().timestamp() * 1000)  # Tempo使用毫秒时间戳
    start_ts = end_ts - 2 * 3600 * 1000
    
    trace_ids = search_tempo_traces('service="payment"', start_ts, end_ts)
    if trace_ids:
        first_trace = get_tempo_trace(trace_ids[0]["traceID"])
        print("追踪链路详情:")
        for span in first_trace["data"][0]["spans"]:
            print(f"Span名称: {span['name']}, 耗时: {span['duration']}ms, 标签: {span['attributes']}")

注意事项 #

• 认证配置：如果Loki/Tempo实例开启了权限控制（比如Grafana API Key、Basic Auth），需在请求中添加对应认证信息。
• 查询语法：Loki使用LogQL语法，Tempo的标签查询遵循PromQL风格的标签匹配规则。
• 数据解析：可根据业务需求提取返回JSON中的字段，比如日志内容、追踪Span属性、耗时等。

内容的提问来源于stack exchange，提问作者Imen Benamar