菜单
文档面包屑箭头 Grafana Loki面包屑箭头 发送数据面包屑箭头 Logstash 插件
开源 RSS

Logstash 插件

Grafana Loki 有一个名为 logstash-output-loki 的 Logstash 输出插件，可以将日志发送到 Loki 实例或 Grafana Cloud。

警告

Grafana Labs 不建议在新部署中使用 Logstash 插件。即使作为使用现有 Beats/Logstash 基础设施快速测试 Loki 的一种机制，我们也非常不鼓励使用此插件。

我们多年来的经验发现，使用 Logstash 和此插件存在许多重大挑战

  • 正确配置标签非常困难。概念上，Elasticsearch 与 Loki 是非常不同的数据库，用户几乎总是会向 Loki 发送过多高基数的标签，这使得 Loki 入门变得不必要地复杂和令人困惑，不如使用其他客户端。
  • Logstash 和上游 Beats 组件实现了退避和流控制，我们发现这很难观察，导致日志摄入 Loki 出现延迟，这非常难以解决。
  • Grafana Labs 在配置 Logstash 或理解其配置语言方面没有专业知识，因此我们无法为其提供支持。
  • 故障排除和调试非常困难。我们的经验表明，在几乎所有假定这是将日志发送到 Loki 的快速路径的情况下，情况并非如此，最终花费的时间远远超出预期。

请强烈考虑使用任何其他机制将日志发送到 Loki。我们推荐使用 Grafana Alloy。这是我们构建的工具，我们可以在此提供最佳体验和最多支持。

安装

本地

如果您需要手动安装 Logstash 输出插件，只需使用以下命令即可

bash 
$ bin/logstash-plugin install logstash-output-loki

这将下载输出插件的最新 gem 并将其安装到 logstash 中。

Docker

我们还在 docker hub 上提供了 Docker 镜像。该镜像包含 logstash 和 Loki 输出插件，均已预装。

例如，如果您想使用 loki.conf 作为管道配置在 Docker 中运行 logstash，可以使用以下命令

bash 
docker run -v `pwd`/loki-test.conf:/home/logstash/ --rm grafana/logstash-output-loki:1.0.1 -f loki-test.conf

Kubernetes

我们还在 loki-stack 伞形 chart 中提供了使用 Filebeat 抓取日志并通过 logstash 将其转发到 Loki 的默认 Helm 值。您可以使用以下命令从 Promtail 切换到 logstash

bash 
helm upgrade --install loki loki/loki-stack \
    --set filebeat.enabled=true,logstash.enabled=true,promtail.enabled=false \
    --set loki.fullnameOverride=loki,logstash.fullnameOverride=logstash-loki

这将自动抓取集群中所有 Pod 的日志，并将其连同 Kubernetes 元数据作为标签发送到 Loki。您可以将 values.yaml 文件作为自己配置的起点。

使用和配置

要配置 Logstash 将日志转发到 Loki，只需将 loki 输出添加到您的 Logstash 配置文件中，如下所述

conf 
output {
  loki {
    [url => "" | default = none | required=true]

    [tenant_id => string | default = nil | required=false]

    [message_field => string | default = "message" | required=false]

    [include_fields => array | default = [] | required=false]

    [metadata_fields => array | default = [] | required=false]

    [batch_wait => number | default = 1(s) | required=false]

    [batch_size => number | default = 102400(bytes) | required=false]

    [min_delay => number | default = 1(s) | required=false]

    [max_delay => number | default = 300(s) | required=false]

    [retries => number | default = 10 | required=false]

    [username => string | default = nil | required=false]

    [password => secret | default = nil | required=false]

    [cert => path | default = nil | required=false]

    [key => path | default = nil| required=false]

    [ca_cert => path | default = nil | required=false]

    [insecure_skip_verify => boolean | default = false | required=false]
  }
}

默认情况下，Loki 将从接收到的事件字段创建条目。Logstash 事件如下所示。

conf 
{
  "@timestamp" => 2017-04-26T19:33:39.257Z,
  "src"        => "localhost",
  "@version"   => "1",
  "host"       => "localhost.localdomain",
  "pid"        => "1",
  "message"    => "Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool...",
  "type"       => "stdin",
  "prog"       => "systemd",
}

包含 message 和 @timestamp 字段，它们分别用于形成 Loki 条目日志行和时间戳。

您可以通过配置属性 message_field 为日志行使用不同的属性。如果您还需要更改时间戳值，请使用 Logstash 的 date 过滤器更改 @timestamp 字段。

所有其他字段（嵌套字段除外）将构成附加到日志行的标签集（键值对）。这意味着您需要负责修改和删除高基数标签，例如客户端 IP。通常可以使用 mutate 过滤器来完成此操作。

例如，以下配置

conf 
input {
  ...
}

filter {
  mutate {
    add_field => {
      "cluster" => "us-central1"
      "job" => "logstash"
    }
    replace => { "type" => "stream"}
    remove_field => ["src"]
  }
}
output {
  loki {
    url => "http://myloki.domain:3100/loki/api/v1/push"
  }
}

将添加 cluster 和 job 静态标签，移除 src 字段，并将 type 命名为 stream。

如果您想包含嵌套字段或元数据字段（以 @ 开头），您需要对其重命名。

例如，当将 Filebeat 与 add_kubernetes_metadata 处理器一起使用时，它会将 Kubernetes 元数据附加到您的事件中，如下所示

json 
{
  "kubernetes" : {
    "labels" : {
      "app" : "MY-APP",
      "pod-template-hash" : "959f54cd",
      "serving" : "true",
      "version" : "1.0",
      "visualize" : "true"
    },
    "pod" : {
      "uid" : "e20173cb-3c5f-11ea-836e-02c1ee65b375",
      "name" : "MY-APP-959f54cd-lhd5p"
    },
    "node" : {
      "name" : "ip-xxx-xx-xx-xxx.ec2.internal"
    },
    "container" : {
      "name" : "istio"
    },
    "namespace" : "production",
    "replicaset" : {
      "name" : "MY-APP-959f54cd"
    }
  },
  "message": "Failed to parse configuration",
  "@timestamp": "2017-04-26T19:33:39.257Z",
}

下面的过滤器展示了如何将这些 Kubernetes 字段提取为标签（container_name、namespace、pod 和 host）

conf 
filter {
  if [kubernetes] {
    mutate {
      add_field => {
        "container_name" => "%{[kubernetes][container][name]}"
        "namespace" => "%{[kubernetes][namespace]}"
        "pod" => "%{[kubernetes][pod][name]}"
      }
      replace => { "host" => "%{[kubernetes][node][name]}"}
    }
  }
  mutate {
    remove_field => ["tags"]
  }
}

版本说明

关于版本的重要说明

  • 此插件的 1.1.0 及更高版本，您还可以通过 include_fields 配置指定一个标签允许列表。
  • 此插件的 1.2.0 及更高版本，您还可以通过 metadata_fields 配置指定结构化元数据。

配置属性

url

用于发送日志的 Loki 服务器 URL。发送数据时，还需要提供推送路径，例如 https://:3100/loki/api/v1/push。

如果您想发送到 GrafanaCloud，可以使用 https://logs-prod-us-central1.grafana.net/loki/api/v1/push。

用户名 / 密码

如果 Loki 服务器需要基本认证，请指定用户名和密码。如果使用 GrafanaLab 托管的 Loki，用户名需要设置为您的实例/用户 ID，密码应为 Grafana.com api 密钥。

message_field

用于日志行的消息字段。您可以使用 logstash 键访问器语言获取嵌套属性，例如：[log][message]。

include_fields

将被映射到标签并发送到 Loki 的字段数组。配置此列表后，将**仅**发送这些字段，所有其他字段都将被忽略。

metadata_fields

将被映射到结构化元数据并为每行日志发送到 Loki 的字段数组

batch_wait

将一批记录推送到 Loki 之前等待的间隔时间（秒）。这意味着即使在 batch_wait 后未达到批次大小，也会发送部分批次，以确保数据的新鲜度。

batch_size

在推送到 Loki 之前累积的最大批次大小。默认为 102400 字节

退避配置

min_delay => 1(1秒)

重试之间的初始退避时间

max_delay => 300(5分钟)

重试之间的最大退避时间

retries => 10

最大重试次数。设置为 0 将无限重试。

tenant_id

Loki 是一个多租户日志存储平台，所有发送的请求都必须包含租户。对于某些安装，租户将由认证代理自动设置。否则，您可以定义要传递的租户。租户可以是任何字符串值。

客户端证书验证

如果 Loki 前面配置了带有客户端证书验证的反向代理，请使用 cert 和 key 指定客户端证书和私钥对。如果服务器使用自定义证书颁发机构，也可以指定 ca_cert。

insecure_skip_verify

一个禁用服务器证书验证的标志。默认设置为 false。

完整配置示例

conf 
input {
  beats {
    port => 5044
  }
}

filter {
  if [kubernetes] {
    mutate {
      add_field => {
        "container_name" => "%{[kubernetes][container][name]}"
        "namespace" => "%{[kubernetes][namespace]}"
        "pod" => "%{[kubernetes][pod][name]}"
      }
      replace => { "host" => "%{[kubernetes][node][name]}"}
    }
  }
  mutate {
    remove_field => ["tags"]  # Note: with include_fields defined below this wouldn't be necessary
  }
}

output {
  loki {
    url => "https://logs-prod-us-central1.grafana.net/loki/api/v1/push"
    username => "3241"
    password => "REDACTED"
    batch_size => 112640 #112.64 kilobytes
    retries => 5
    min_delay => 3
    max_delay => 500
    message_field => "message"
    include_fields => ["container_name","namespace","pod","host"]
    metadata_fields => ["pod"]
  }
  # stdout { codec => rubydebug }
}