访问事件数据和字段

Logstash 管道通常有三个阶段：输入 → 过滤器 → 输出。输入生成事件，过滤器修改事件，输出将事件发送到其他地方。

所有事件都有属性。例如，Apache 访问日志具有诸如状态代码（200、404）、请求路径（“/”，“index.html”）、HTTP 动词（GET、POST）、客户端 IP 地址等属性。Logstash 将这些属性称为“字段”。

Logstash 中的某些配置选项需要字段存在才能起作用。由于输入会生成事件，因此输入块中没有要评估的字段 - 它们还不存在！

当您需要按名称引用字段时，可以使用 Logstash 字段引用语法。

访问字段的基本语法是 [字段名称]。如果您引用的是顶层字段，则可以省略 []，而只需使用 字段名称。要引用嵌套字段，请指定该字段的完整路径：[顶层字段][嵌套字段]。

例如，此事件有五个顶层字段（agent、ip、request、response、ua）和三个嵌套字段（status、bytes、os）。

{
  "agent": "Mozilla/5.0 (compatible; MSIE 9.0)",
  "ip": "192.168.24.44",
  "request": "/index.html"
  "response": {
    "status": 200,
    "bytes": 52353
  },
  "ua": {
    "os": "Windows 7"
  }
}

要引用 os 字段，请指定 [ua][os]。要引用诸如 request 之类的顶层字段，只需指定字段名称即可。

有关更多详细信息，请参阅 字段引用深入探讨。

字段引用格式也用于 Logstash 所谓的 *sprintf 格式*。此格式使您能够将字段值嵌入其他字符串中。例如，statsd 输出具有一个 *increment* 设置，使您可以按状态代码保持 Apache 日志的计数

output {
  statsd {
    increment => "apache.%{[response][status]}"
  }
}

同样，您可以将 @timestamp 字段中的 UTC 时间戳转换为字符串。

不要在花括号内指定字段名称，而是使用 %{{FORMAT}} 语法，其中 FORMAT 是 java 时间格式。

例如，如果您要使用文件输出根据事件的 UTC 日期和小时以及 type 字段写入日志

output {
  file {
    path => "/var/log/%{type}.%{{yyyy.MM.dd.HH}}"
  }
}

有时，您只想在特定条件下过滤或输出事件。为此，您可以使用条件。

Logstash 中的条件看起来和行为方式与它们在编程语言中相同。条件支持 if、else if 和 else 语句，并且可以嵌套。

条件语法为

if EXPRESSION {
  ...
} else if EXPRESSION {
  ...
} else {
  ...
}

什么是表达式？比较测试、布尔逻辑等等！

您可以使用以下比较运算符

支持的布尔运算符是

支持的一元运算符是

表达式可以是长而复杂的。表达式可以包含其他表达式，您可以使用 ! 来否定表达式，并且可以使用括号 (...) 将它们分组。

例如，以下条件使用 mutate 过滤器删除字段 secret（如果字段 action 的值为 login）

filter {
  if [action] == "login" {
    mutate { remove_field => "secret" }
  }
}

如果表达式在求值时产生错误，则事件处理将停止，并且会将警告消息写入日志。例如，将整数值 100 与字符串值 "100" 进行比较无法确定地求值，因此处理会停止，并且会记录错误。

要捕获错误发生时消息的完整内容，请将日志级别设置为 debug。查看 日志记录，了解有关如何配置日志记录和可用的日志级别的更多信息。

您可以在单个条件中指定多个表达式

output {
  # Send production errors to pagerduty
  if [loglevel] == "ERROR" and [deployment] == "production" {
    pagerduty {
    ...
    }
  }
}

您可以使用 in 运算符来测试字段是否包含特定的字符串、键或列表元素。请注意，in 的语义含义可能会因目标类型而异。例如，当应用于字符串时。in 表示“是...的子字符串”。当应用于集合类型时，in 表示“集合包含确切的值”。

filter {
  if [foo] in [foobar] {
    mutate { add_tag => "field in field" }
  }
  if [foo] in "foo" {
    mutate { add_tag => "field in string" }
  }
  if "hello" in [greeting] {
    mutate { add_tag => "string in field" }
  }
  if [foo] in ["hello", "world", "foo"] {
    mutate { add_tag => "field in list" }
  }
  if [missing] in [alsomissing] {
    mutate { add_tag => "shouldnotexist" }
  }
  if !("foo" in ["hello", "world"]) {
    mutate { add_tag => "shouldexist" }
  }
}

您使用 not in 条件的方式相同。例如，您可以使用 not in 仅在 grok 成功时才将事件路由到 Elasticsearch

output {
  if "_grokparsefailure" not in [tags] {
    elasticsearch { ... }
  }
}

您可以检查特定字段是否存在，但是目前没有办法区分不存在的字段与只是 false 的字段。当以下情况时，表达式 if [foo] 返回 false

有关更复杂的示例，请参阅 使用条件。

在 Logstash 中，有一个名为 @metadata 的特殊字段。@metadata 的内容在输出时不是任何事件的一部分，这使得它非常适合用于条件，或者使用字段引用和 sprintf 格式化来扩展和构建事件字段。

此配置文件会从 STDIN 生成事件。您键入的任何内容都将成为事件中的 message 字段。过滤器块中的 mutate 事件会添加一些字段，其中一些字段嵌套在 @metadata 字段中。

input { stdin { } }

filter {
  mutate { add_field => { "show" => "This data will be in the output" } }
  mutate { add_field => { "[@metadata][test]" => "Hello" } }
  mutate { add_field => { "[@metadata][no_show]" => "This data will not be in the output" } }
}

output {
  if [@metadata][test] == "Hello" {
    stdout { codec => rubydebug }
  }
}

让我们看看会输出什么

$ bin/logstash -f ../test.conf
Pipeline main started
asdf
{
    "@timestamp" => 2016-06-30T02:42:51.496Z,
      "@version" => "1",
          "host" => "example.com",
          "show" => "This data will be in the output",
       "message" => "asdf"
}

键入的“asdf”变成了 message 字段内容，并且条件成功地评估了嵌套在 @metadata 字段中的 test 字段的内容。但是，输出中没有显示名为 @metadata 的字段或其内容。

如果您添加配置标志 metadata => true，则 rubydebug 编解码器允许您显示 @metadata 字段的内容

    stdout { codec => rubydebug { metadata => true } }

让我们看看更改后的输出是什么样的

$ bin/logstash -f ../test.conf
Pipeline main started
asdf
{
    "@timestamp" => 2016-06-30T02:46:48.565Z,
     "@metadata" => {
           "test" => "Hello",
        "no_show" => "This data will not be in the output"
    },
      "@version" => "1",
          "host" => "example.com",
          "show" => "This data will be in the output",
       "message" => "asdf"
}

现在您可以查看 @metadata 字段及其子字段。

在您需要临时字段但不希望它出现在最终输出中时，请使用 @metadata 字段。

或许此新字段最常见的用例之一是使用 date 过滤器并具有临时时间戳。

此配置文件已简化，但使用了 Apache 和 Nginx Web 服务器常用的时间戳格式。过去，在使用它覆盖 @timestamp 字段后，您必须自己删除时间戳字段。借助 @metadata 字段，这不再是必需的

input { stdin { } }

filter {
  grok { match => [ "message", "%{HTTPDATE:[@metadata][timestamp]}" ] }
  date { match => [ "[@metadata][timestamp]", "dd/MMM/yyyy:HH:mm:ss Z" ] }
}

output {
  stdout { codec => rubydebug }
}

请注意，此配置会将提取的日期放入 grok 过滤器中的 [@metadata][timestamp] 字段中。让我们为此配置提供一个示例日期字符串，看看会输出什么

$ bin/logstash -f ../test.conf
Pipeline main started
02/Mar/2014:15:36:43 +0100
{
    "@timestamp" => 2014-03-02T14:36:43.000Z,
      "@version" => "1",
          "host" => "example.com",
       "message" => "02/Mar/2014:15:36:43 +0100"
}

就这样！输出中没有多余的字段，并且配置文件更简洁，因为您不必在 date 过滤器中转换后删除“时间戳”字段。

另一个用例是 CouchDB Changes 输入插件。此插件会自动将 CouchDB 文档字段元数据捕获到输入插件本身的 @metadata 字段中。当事件传递到由 Elasticsearch 索引时，Elasticsearch 输出插件允许您指定 action（删除、更新、插入等）和 document_id，如下所示

output {
  elasticsearch {
    action => "%{[@metadata][action]}"
    document_id => "%{[@metadata][_id]}"
    hosts => ["example.com"]
    index => "index_name"
    protocol => "http"
  }
}

目前不支持条件中的 Sprintf 日期/时间格式，但有一种解决方法。将日期计算放在一个字段中，以便您可以在条件中使用字段引用。

示例

直接使用 sprintf 时间格式来添加基于摄取时间的字段将不起作用

----------
# non-working example
filter{
  if "%{+HH}:%{+mm}" < "16:30" {
    mutate {
      add_field => { "string_compare" => "%{+HH}:%{+mm} is before 16:30" }
    }
  }
}
----------

此解决方法可为您提供预期的结果

filter {
  mutate{
     add_field => {
      "[@metadata][time]" => "%{+HH}:%{+mm}"
     }
  }
  if [@metadata][time] < "16:30" {
    mutate {
      add_field => {
        "string_compare" => "%{+HH}:%{+mm} is before 16:30"
      }
    }
  }
}