mcpField
L
loki

loki

github.com/grafana云与 DevOps2957.8k 次安装1 天前更新
代码仓库

关于 loki

loki 是介绍 Grafana Loki 日志聚合与 LogQL 查询的技能，面向需要检索和分析日志的运维及开发人员。内容涵盖标签筛选、日志解析、指标计算和日志采集方式。

主要功能

  1. 01使用标签选择日志流
  2. 02通过 LogQL 筛选与解析日志
  3. 03从日志计算指标和聚合结果
  4. 04配置 Grafana Alloy、Kubernetes 或 HTTP 推送

适用场景

  1. 01按应用和环境定位错误日志
  2. 02统计服务请求率与错误率
  3. 03查询慢请求及 HTTP 5xx 日志
  4. 04检测日志中的凭据泄露模式

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/grafana/skills --skill loki

安装后在 Claude Code 中输入/loki确认已加载。

详细介绍

Grafana Loki - 日志聚合

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

仅索引元数据（标签），而非完整日志内容，因此成本远低于全文搜索系统。

LogQL 快速参考

日志流选择器（每个查询都必须包含）
{app="nginx"}                        # exact match
{app!="nginx"}                       # not equal
{app=~"nginx|apache"}               # regex match
{app!~"debug.*"}                     # regex not match
{app="nginx", env="prod"}           # AND (multiple labels)
行过滤器（管道阶段 1 - 为提升性能，请放在最前面）
{app="nginx"} |= "error"            # contains string
{app="nginx"} != "info"             # does not contain
{app="nginx"} |~ "error|warn"       # regex match
{app="nginx"} !~ "health.*check"    # regex not match
{app="nginx"} |= `"status":5`       # backtick avoids escaping
解析器
# 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 (positional, _ discards)
{app="nginx"} | pattern `<ip> - - <_> "<method> <uri> <_>" <status> <bytes>`

# Regexp (named capture groups)
{app="nginx"} | regexp `(?P<method>\w+) (?P<path>\S+) HTTP/(?P<version>\S+)`

# Unpack (unwrap Promtail packed labels)
{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

指标查询

日志范围聚合
# Requests per second
rate({app="nginx"}[5m])

# Total log lines in window
count_over_time({app="nginx"}[1h])

# Bytes per second
bytes_rate({app="nginx"}[5m])

# Total bytes
bytes_over_time({app="nginx"}[1h])

# Returns 1 if no logs in range (for absence alerting)
absent_over_time({app="nginx"}[5m])
聚合
# Error rate by service
sum(rate({env="prod"} |= "error" [5m])) by (app)

# Top 5 most active services
topk(5, sum(rate({env="prod"}[5m])) by (app))

# Total errors across all services
sum(count_over_time({env="prod"} |= "error" [5m]))
非包装范围聚合（从日志中提取数值）
# Average request duration from logfmt
avg_over_time({app="api"} | logfmt | unwrap duration [5m])

# 95th percentile latency
quantile_over_time(0.95, {app="api"} | logfmt | unwrap duration [5m]) by (app)

# Sum of bytes from JSON logs
sum_over_time({app="api"} | json | unwrap bytes [5m])

# With conversion (duration string → seconds)
avg_over_time({app="api"} | logfmt | unwrap duration | duration_seconds [5m])
偏移修饰符
# Compare current rate vs 1 hour ago
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"]
      ]
    }]
  }'

架构

Push path:  Client → Distributor → Ingester (WAL) → Object Storage (chunks)
Read path:  Query → Query Frontend → Querier → Ingester + Store (chunks)

组件：

  • Distributor：验证并对传入的日志流进行哈希处理
  • Ingester：在内存中缓冲数据块，并将其刷新到对象存储
  • Querier：执行 LogQL 查询
  • Query Frontend：缓存、拆分并并行处理查询
  • Compactor：管理数据保留和去重

参考资料

评论

支持 Markdown