loki

Grafana Loki - 日志聚合

文档: https://grafana.com/docs/loki/latest/

仅索引元数据（标签），不索引完整日志内容——比全文搜索系统成本大幅降低。

LogQL 快速参考

日志流选择器（每个查询必需）

{app="nginx"}                        # 精确匹配
{app!="nginx"}                       # 不等于
{app=~"nginx|apache"}               # 正则匹配
{app!~"debug.*"}                     # 正则不匹配
{app="nginx", env="prod"}           # AND（多个标签）

行过滤器（管道阶段 1 - 优先放置以提高性能）

{app="nginx"} |= "error"            # 包含字符串
{app="nginx"} != "info"             # 不包含
{app="nginx"} |~ "error|warn"       # 正则匹配
{app="nginx"} !~ "health.*check"    # 正则不匹配
{app="nginx"} |= `"status":5`       # 反引号避免转义

解析器

# JSON
{app="api"} | json
{app="api"} | json status="http_status", path="request.path"

# Logfmt
{app="api"} | logfmt
{app="api"} | logfmt --strict
{app="api"} | logfmt --keep-empty

# Pattern（位置匹配，_ 表示丢弃）
{app="nginx"} | pattern `<ip> - - <_> "<method> <uri> <_>" <status> <bytes>`

# Regexp（命名捕获组）
{app="nginx"} | regexp `(?P<method>\w+) (?P<path>\S+) HTTP/(?P<version>\S+)`

# Unpack（解包 Promtail 打包的标签）
{app="api"} | unpack

标签过滤器（解析器之后）

{app="api"} | json | status >= 500
{app="api"} | json | status == 200 and method != "OPTIONS"
{app="api"} | logfmt | duration > 1s
{app="api"} | json | level =~ "error|warn"
{app="api"} | json | bytes > 20MB
{app="api"} | json | path != "/healthz"

行格式化

{app="api"} | json | line_format "{{.method}} {{.path}} -> {{.status}} ({{.duration}})"
{app="api"} | logfmt | line_format `{{.level | upper}}: {{.msg}}`

标签格式化

{app="api"} | logfmt | label_format new_name=old_name
{app="api"} | logfmt | label_format severity=level, svc=app
{app="api"} | logfmt | label_format msg=`{{.level}}: {{.message}}`

删除/保留标签

{app="api"} | json | drop filename, level="debug"
{app="api"} | json | keep level, status, method

去除颜色

{app="cli-tool"} | decolorize

指标查询

日志范围聚合

# 每秒请求数
rate({app="nginx"}[5m])

# 窗口内总日志行数
count_over_time({app="nginx"}[1h])

# 每秒字节数
bytes_rate({app="nginx"}[5m])

# 总字节数
bytes_over_time({app="nginx"}[1h])

# 如果范围内无日志则返回 1（用于缺失告警）
absent_over_time({app="nginx"}[5m])

聚合

# 按服务的错误率
sum(rate({env="prod"} |= "error" [5m])) by (app)

# 最活跃的 5 个服务
topk(5, sum(rate({env="prod"}[5m])) by (app))

# 所有服务的总错误数
sum(count_over_time({env="prod"} |= "error" [5m]))

解包范围聚合（从日志中提取数值）

# 从 logfmt 中平均请求持续时间
avg_over_time({app="api"} | logfmt | unwrap duration [5m])

# 95 百分位延迟
quantile_over_time(0.95, {app="api"} | logfmt | unwrap duration [5m]) by (app)

# 从 JSON 日志中求和字节数
sum_over_time({app="api"} | json | unwrap bytes [5m])

# 带转换（持续时间字符串 → 秒）
avg_over_time({app="api"} | logfmt | unwrap duration | duration_seconds [5m])

偏移修饰符

# 比较当前速率与 1 小时前的速率
rate({app="nginx"}[5m]) / rate({app="nginx"}[5m] offset 1h)

实用示例

错误率告警查询

sum(rate({env="prod"} |= "error" [5m])) by (service)
/
sum(rate({env="prod"}[5m])) by (service)
> 0.05

慢请求

{app="api"} | logfmt | duration > 1s | line_format "SLOW: {{.method}} {{.path}} {{.duration}}"

HTTP 5xx 错误及详情

{app="nginx"} | pattern `<ip> - - <_> "<method> <uri> <_>" <status> <bytes>` | status >= 500

凭证泄露检测

{namespace="prod"} |~ `https?://\w+:\w+@`

向 Loki 发送日志

通过 Grafana Alloy

loki.source.file "app" {
  targets    = [{__path__ = "/var/log/app/*.log", job = "app"}]
  forward_to = [loki.process.parse.receiver]
}

loki.process "parse" {
  forward_to = [loki.write.cloud.receiver]
  stage.json {
    expressions = { level = "level", msg = "message" }
  }
  stage.labels {
    values = { level = "" }
  }
  stage.drop {
    expression = ".*healthcheck.*"
  }
}

loki.write "cloud" {
  endpoint {
    url = "https://logs-xxx.grafana.net/loki/api/v1/push"
    basic_auth {
      username = sys.env("LOKI_USER")
      password = sys.env("GRAFANA_API_KEY")
    }
  }
  external_labels = { cluster = "prod" }
}

通过 Kubernetes（Alloy DaemonSet）

discovery.kubernetes "pods" {
  role = "pod"
}

loki.source.kubernetes "pods" {
  targets    = discovery.kubernetes.pods.targets
  forward_to = [loki.write.cloud.receiver]
}

Loki HTTP 推送 API

curl -X POST https://logs-xxx.grafana.net/loki/api/v1/push \
  -u "user:apikey" \
  -H 'Content-Type: application/json' \
  -d '{
    "streams": [{
      "stream": { "app": "myapp", "env": "prod" },
      "values": [
        ["1609459200000000000", "log line here"]
      ]
    }]
  }'

架构

推送路径:  客户端 → 分发器 → 摄取器 (WAL) → 对象存储 (chunks)
读取路径:  查询 → 查询前端 → 查询器 → 摄取器 + 存储 (chunks)

组件:

• 分发器: 验证并哈希传入的日志流
• 摄取器: 在内存中缓冲块，刷新到对象存储
• 查询器: 执行 LogQL 查询
• 查询前端: 缓存、拆分和并行化查询
• 压缩器: 管理保留和去重

参考

• LogQL 参考
• 配置
• 发送数据