使用 Python 查询与写入 LokiHTTP API 客户端实战指南【免费下载链接】lokiLike Prometheus, but for logs.项目地址: https://gitcode.com/GitHub_Trending/lok/lokiLoki 的 HTTP API/loki/api/v1/*是它与外部系统交互的通用接口本文基于 Loki 官方 Python 客户端示例完整演示如何用requests与httpx完成日志范围查询、即时指标查询、日志推送、标签枚举等高频操作并覆盖多租户认证、Grafana Cloud 接入、TLS 校验与错误重试等生产环境细节。读完本文你将掌握一套可直接复制运行的 Python 代码骨架并能结合 Loki HTTP API 参考 深入每个端点的全部参数与响应格式。前置条件安装 HTTP 客户端库先安装同步客户端 requestspip install requests如果你需要异步能力httpx 提供了几乎一致的 APIpip install httpx两种库在本文示例中的调用方式完全对称唯一的细微差别是httpx的verify参数还额外接受ssl.SSLContext对象。下述所有函数都统一接收headers、auth、verify三个可选参数方便你在不同部署环境下复用同一套代码。认证从本地实例到多租户与 Grafana Cloud本文示例默认连接无认证的本地 Loki 实例如http://localhost:3100。如果你的集群启用了多租户或接入了 Grafana Cloud需要按下述方式补充认证信息。多租户模式X-Scope-OrgID请求头当集群开启多租户隔离后每个请求都必须携带租户 ID。直接在请求头中注入X-Scope-OrgIDheaders {X-Scope-OrgID: my-tenant} resp requests.get(url, paramsparams, headersheaders)套用到本文定义的函数上results query_range( urlhttp://localhost:3100, query{jobvarlogs}, startdatetime.now() - timedelta(hours1), enddatetime.now(), headers{X-Scope-OrgID: my-tenant}, )要跨多个租户查询用管道符|分隔租户名headers {X-Scope-OrgID: tenant1|tenant2|tenant3}Grafana CloudBasic Auth对 Grafana Cloud 的 Loki 服务使用 Basic Auth 传入你的 Grafana Cloud 用户和 API Tokenresp requests.get(url, paramsparams, auth(user, API_TOKEN))套用到函数上results query_range( urlhttps://logs-prod-us-central1.grafana.net, query{jobvarlogs}, startdatetime.now() - timedelta(hours1), enddatetime.now(), auth(user, API_TOKEN), )User与URL两个值都可以在 Grafana Cloud Stack 的 Loki 日志服务详情页中找到。注意auth元组同时兼容requests与httpx的签名代码无需改动。自签名证书控制 TLS 校验本地或内网 Loki 常使用自签名 TLS 证书此时可临时跳过证书校验resp requests.get(url, paramsparams, verifyFalse)生产环境不建议关闭校验改为传入 CA 证书包路径resp requests.get(url, paramsparams, verify/path/to/ca-bundle.crt)套用到函数上results query_range( urlhttps://loki.internal:3100, query{jobvarlogs}, startdatetime.now() - timedelta(hours1), enddatetime.now(), verify/path/to/ca-bundle.crt, )范围查询GET /loki/api/v1/query_range/loki/api/v1/query_range在一个时间区间内查询日志是最常用的查询操作也用于返回日志行的时间范围查询。它由querier、query-frontend、read和all组件暴露具体见 Loki HTTP API 参考。使用 requestsimport requests from datetime import datetime, timedelta def query_range( url: str, query: str, start: datetime, end: datetime, limit: int 1000, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - list: Query Loki for log entries within a time range. resp requests.get( f{url}/loki/api/v1/query_range, params{ query: query, start: str(int(start.timestamp() * 1e9)), # nanoseconds end: str(int(end.timestamp() * 1e9)), limit: limit, direction: backward, }, headersheaders, authauth, verifyverify, ) resp.raise_for_status() return resp.json()[data][result] results query_range( urlhttp://localhost:3100, query{jobvarlogs} | error, startdatetime.now() - timedelta(hours1), enddatetime.now(), ) for stream in results: print(fLabels: {stream[stream]}) for ts, line in stream[values]: print(f {datetime.fromtimestamp(int(ts) / 1e9)}: {line})使用 httpximport httpx from datetime import datetime, timedelta def query_range( url: str, query: str, start: datetime, end: datetime, limit: int 1000, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle; httpx also accepts ssl.SSLContext ) - list: Query Loki for log entries within a time range. resp httpx.get( f{url}/loki/api/v1/query_range, params{ query: query, start: str(int(start.timestamp() * 1e9)), # nanoseconds end: str(int(end.timestamp() * 1e9)), limit: limit, direction: backward, }, headersheaders, authauth, verifyverify, ) resp.raise_for_status() return resp.json()[data][result] results query_range( urlhttp://localhost:3100, query{jobvarlogs} | error, startdatetime.now() - timedelta(hours1), enddatetime.now(), ) for stream in results: print(fLabels: {stream[stream]}) for ts, line in stream[values]: print(f {datetime.fromtimestamp(int(ts) / 1e9)}: {line})参数背后的实现细节从 pkg/loghttp/params.go 的源码可以确认这些参数在服务端的解析规则start/end由parseTimestampparams.go#L175-L198解析支持纳秒整数时间戳、含小数点的浮点时间戳甚至是RFC3339Nano格式的字符串位数不超过 10 位时按秒处理。官方示例统一用纳秒字符串最稳妥。direction由parseDirectionparams.go#L202-L212解析默认值为backwarddefaultDirection常量见 params.go#L24即最新日志在前。limit默认值在 params.go#L21 定义为defaultQueryLimit 100与文档“Common problems”一节描述的默认 100 条一致。本文示例显式传limit1000覆盖该默认值。since除start/end外服务端还支持since参数如since2h缺省start时会按end减去since计算默认 1 小时defaultSince见 params.go#L23。step范围查询的步长可省略服务端会按时间跨度动态计算默认步长max(floor((end-start)/250), 1)秒defaultQueryRangeStep见 params.go#L149-L151。响应体结构定义在 pkg/loghttp/query.go 的QueryResponsequery.go#L55-L60包含status、可选的warnings以及data字段data.result是一个数组日志查询中每个元素含stream标签集合与values[纳秒时间戳, 日志行]列表这正是示例中遍历结构的依据。即时查询GET /loki/api/v1/query/loki/api/v1/query在单个时间点默认当前时刻评估查询适用于rate()、count_over_time()、bytes_over_time()等聚合构成的即时指标查询。注意返回日志行的流选择器如{jobmyapp}不支持即时查询必须改用query_range。import requests from datetime import datetime def query_instant( url: str, query: str, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - list: Run an instant metric query against Loki. resp requests.get( f{url}/loki/api/v1/query, params{ query: query, time: str(int(datetime.now().timestamp() * 1e9)), }, headersheaders, authauth, verifyverify, ) resp.raise_for_status() return resp.json()[data][result] results query_instant( urlhttp://localhost:3100, querysum(rate({jobvarlogs}[10m])) by (level), ) for entry in results: print(f{entry[metric]}: {entry[value][1]})time参数同样由parseTimestamp解析见 params.go#L80-L82缺省取当前时间因此最小可用的即时查询甚至可以只传query一个参数。指标查询的响应中每个结果元素含metric分组标签与value[时间戳, 值]二元组与示例中的打印逻辑一一对应。推送日志POST /loki/api/v1/push/loki/api/v1/push是向 Loki 写入日志的端点在微服务模式下由distributor暴露。默认的 POST body 是 Snappy 压缩的 Protobuf 消息但当Content-Type为application/json时可以直接发送 JSON格式为{ streams: [ { stream: { label: value }, values: [ [ unix epoch in nanoseconds, log line ], [ unix epoch in nanoseconds, log line ] ] } ] }对应到 Python 实现import json import time import requests def push_logs( url: str, labels: dict[str, str], entries: list[tuple[str, str]], headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - None: Push log entries to Loki. Args: url: Loki base URL. labels: Stream labels, for example {job: myapp, env: dev}. entries: List of (timestamp_ns, log_line) tuples. Use str(int(time.time() * 1e9)) to get a nanosecond timestamp. payload { streams: [ { stream: labels, values: [list(e) for e in entries], } ] } req_headers {**(headers or {}), Content-Type: application/json} resp requests.post( f{url}/loki/api/v1/push, headersreq_headers, datajson.dumps(payload), authauth, verifyverify, ) resp.raise_for_status() now_ns str(int(time.time() * 1e9)) push_logs( urlhttp://localhost:3100, labels{job: myapp, env: dev}, entries[ (now_ns, application started), (now_ns, listening on port 8080), ], )JSON Push 的几个关键约束结合 HTTP API 文档的 Ingest 章节时间戳必须传字符串而非数字否则端点会返回 400 错误示例用str(int(time.time() * 1e9))构造纳秒字符串。每行日志必须是[纳秒时间戳, 日志行]二元组写入时转成 JSON 数组list(e)。可以用Content-Encoding: gzip请求头发送 gzip 压缩的 JSON 体降低大流量推送的网络开销。可选地在每个日志行的数组末尾追加一个结构化元数据JSON 对象键和值都必须是字符串不允许嵌套例如[纳秒时间戳, 日志行, {trace_id: 0242ac120002, user_id: superUser123}]配合结构化元数据文档使用。服务端对 JSON 请求体的解析在 pkg/loghttp/query.go 的PushRequest/LogProtoStream中实现query.go#L100-L120JSON 流会被转换为内部logproto.Stream的标签串格式因此“stream”里的标签键值必须是合法标签。查询标签与标签值GET /loki/api/v1/labels与GET /loki/api/v1/label/name/values/loki/api/v1/labels返回所有已知标签名列表/loki/api/v1/label/name/values返回某个标签的全部取值。两者常用于驱动探索式分析或构建日志筛选 UI。import requests def get_labels( url: str, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - list[str]: List all known label names. resp requests.get( f{url}/loki/api/v1/labels, headersheaders, authauth, verifyverify, ) resp.raise_for_status() return resp.json()[data] def get_label_values( url: str, label: str, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - list[str]: List values for a specific label. resp requests.get( f{url}/loki/api/v1/label/{label}/values, headersheaders, authauth, verifyverify, ) resp.raise_for_status() return resp.json()[data] labels get_labels(http://localhost:3100) print(fLabels: {labels}) for label in labels: values get_label_values(http://localhost:3100, label) print(f {label}: {values})注意响应中的data字段是一个字符串数组标签名或标签值列表与查询接口的data.result结构不同直接索引返回即可。错误处理HTTP 状态码与指数退避重试Loki 返回标准的 HTTP 状态码常见错误包括状态码含义常见原因400Bad RequestLogQL 语法无效429Too Many Requests触发速率限制5xxServer ErrorLoki 不可用或过载生产客户端应通过raise_for_status()捕获错误并结合响应体排查详情。针对 429 这类可恢复错误可加入指数退避重试import time import requests def query_with_retry( url: str, query: str, max_retries: int 3, backoff: float 1.0, headers: dict[str, str] | None None, auth: tuple[str, str] | None None, verify: bool | str True, # False to skip TLS, or path to CA bundle ) - dict: Query Loki with simple retry logic for rate limits. for attempt in range(max_retries): resp requests.get( f{url}/loki/api/v1/query, params{query: query}, headersheaders, authauth, verifyverify, ) if resp.status_code 429: wait backoff * (2 ** attempt) print(fRate limited, retrying in {wait}s...) time.sleep(wait) continue resp.raise_for_status() return resp.json() raise Exception(fQuery failed after {max_retries} retries)重试等待时间按backoff * 2^attempt指数增长1s、2s、4s……在不超过max_retries的前提下自动降频避免在限流期间继续冲击服务端。关于请求级与租户级限流配置可进一步参考请求校验与速率限制文档。常见问题速查时间戳必须是纳秒。Loki 期望 Unix 纳秒时间戳而不是秒或毫秒。用time.time() * 1e9换算后转成字符串查询端别忘了把响应中的纳秒除以1e9再转datetime展示。至少需要一个标签匹配器。没有流选择器就无法查询。{jobmyapp}合法空选择器不合法。direction参数决定结果排序。backward默认先返回最新条目forward先返回最旧条目。范围查询时分页依赖这个语义见下一条。即时查询端点只支持指标查询。在/query上执行{jobmyapp}这类日志流选择器会返回 400日志查询请走/query_rangerate()/count_over_time()等聚合走/query。用limit控制结果规模并实现分页。查询类接口的limit默认值为 100见 params.go#L21。对于大时间范围可调高limit或根据返回的最后一条时间戳把start前移后继续拉取实现游标式分页query_range还支持按step控制指标计算的分辨率。延伸阅读完整的端点清单、全部查询参数与响应格式参见 Loki HTTP API 参考。多租户机制的配置与原理参见多租户文档。本文示例查询中使用的 LogQL 语法与聚合函数参见 LogQL 查询文档。服务端对query_range/query/labels各参数的解析实现可深入 pkg/loghttp/params.go 与 pkg/loghttp/query.go 阅读源码。【免费下载链接】lokiLike Prometheus, but for logs.项目地址: https://gitcode.com/GitHub_Trending/lok/loki创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考