使用Python查询Loki日志时遇400语法错误，求问题排查

核心错误原因 #

你调用了错误的Loki API接口：

• /loki/api/v1/query是即时查询接口，仅用于获取某个时间点的聚合结果，不接受range参数，传入该参数会直接触发语法解析错误。
• 你需要查询一段时间范围内的日志，应该使用范围查询接口 /loki/api/v1/query_range。

具体修改步骤 #

1. 替换API接口路径
   将请求URL从{loki_url}/loki/api/v1/query改为{loki_url}/loki/api/v1/query_range。

2. 清理冗余代码
   移除重复导入的import requests语句。

3. 调整结果遍历逻辑
   范围查询的返回结果结构与即时查询不同，需要遍历每个日志流中的所有日志条目。

修改后的完整代码 #

import requests
import time
import urllib3
import ssl
import urllib.parse


class CustomHttpAdapter(requests.adapters.HTTPAdapter):
    # "Transport adapter" that allows us to use custom ssl_context.

    def __init__(self, ssl_context=None, **kwargs):
        self.ssl_context = ssl_context
        super().__init__(**kwargs)

    def init_poolmanager(self, connections, maxsize, block=False):
        self.poolmanager = urllib3.poolmanager.PoolManager(
            num_pools=connections, maxsize=maxsize,
            block=block, ssl_context=self.ssl_context)


def get_legacy_session():
    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    ctx.options |= 0x4  # OP_LEGACY_SERVER_CONNECT
    session = requests.session()
    session.mount('https://', CustomHttpAdapter(ctx))
    return session


loki_url = "https://url-for-loki"
query = '{instance="some-instance"}'
range_seconds = 60
limit = 10000
step = 60

headers = {
    "Content-Type": "application/json",
    "X-Scope-OrgID": "tenant-id",
}

while True:
    payload = {
        "query": query,
        "range": {
            "from": f"now-{range_seconds}s",
            "to": "now"
        },
        "limit": limit,
        "step": step,
    }

    print(payload)
    # 修改为范围查询接口
    response = get_legacy_session().get(
        f"{loki_url}/loki/api/v1/query_range", json=payload, headers=headers)

    print("text:-", response.text)

    if response.status_code != 200:
        print(f"Error: {response.status_code}")
        exit()

    # 遍历范围查询返回的所有日志条目
    for log_stream in response.json()["data"]["result"]:
        for _, log_content in log_stream["values"]:
            print(log_content)

    time.sleep(60)

额外说明 #

• step参数控制返回数据的采样间隔，若需要获取全部日志条目，可将其设置为1s（日志量较大时需注意性能）。
• 确保{instance="some-instance"}是Loki中存在的有效日志流标签，若需解析结构化日志，可添加格式化语法，比如{instance="some-instance"} | logfmt（针对logfmt格式日志）。

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