6.3. 访问事件数据和字段

8.9

8.9

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 中也被称为 打印格式 。这种格式允许您在其他字符串中嵌入字段值。例如，statsd 输出插件有一个 increment 配置项可以使你通过状态码对 Apache 日志进行计数。

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

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

比在大括号中指定字段名更好的做法是：直接使用 %{{FORMAT}} 的语法（FORMAT 与 java 时间格式相同）。

举个例子，如果你想在文件输出插件中记录日志，日志的格式基于事件的 UTC 日期 和 小时 部分，另外还包括 类型 字段，你可以这么定义：

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

注意：打印格式将继续支持 废弃的 joda 时间格式 字符串，并使用% {+FORMAT }语法。这些格式不能直接相互转换，我们建议您开始使用更现代的JavaTime格式。

注意：Logstash 时间戳基于 UTC 时间轴，因此使用打印格式化程序将产生可能与计算机本地时区不一致的结果。

条件语句

在某些情况下你只想在特定条件下才过滤或者输出一个事件，这时你可以使用条件语句。

Logstash 中的条件语句和编程语言中的条件语句，无论是在外观还是行为上都是一样的。条件语句支持 if，else if 和 else 语句并且支持嵌套。

条件语句这样使用：

if 表达式 {
  ...
} else if 表达式 {
  ...
} else {
  ...
}

什么是表达式？就是比较测试，布尔逻辑等等。

你可以使用以下比较操作符：

• 等式：==，!=，<，>，<=，>=
• 正则表达式：=~，!~（检查左边的字符串和右边的模式是否匹配）
• 包含：in，not in

支持的一元操作符：

• !

表达式可以很长也可以很复杂。表达式可以包含其他表达式，你可以使用 ! 对表达式结果取反，你也可以使用括号(...)对表达式进行分组。

举个例子，以下这个条件语句判断action字段值是否为login，如果是的话使用 mutate 过滤插件将 secret 字段移除掉。

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

你可以在一个单一条件中指定多个表达式：

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 => "字段在字段中" }
  }
  if [foo] in "foo" {
    mutate { add_tag => "字段是字符串的子字符串" }
  }
  if "hello" in [greeting] {
    mutate { add_tag => "字符串在字段中" }
  }
  if [foo] in ["hello", "world", "foo"] {
    mutate { add_tag => "字段在列表中" }
  }
  if [missing] in [alsomissing] {
    mutate { add_tag => "字段不应该存在" }
  }
  if !("foo" in ["hello", "world"]) {
    mutate { add_tag => "字段应该存在" }
  }
}

使用 not in 条件语句也是如此。例如，你可以使用 not in 指定在 grok 成功时，将事件仅路由向 Elasticsearch：

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

你可以检查指定字段的存在性，但是目前还没有办法区分字段是不存在还是字段存在简单的错误。表达式 if [foo] 在以下这些情况会返回 false：

• [foo] 字段不存在与事件中
• [foo] 字段存在于事件中，但是值为 false
• [foo] 字段存在于事件中，但是值为 null

想了解更复杂的用例，请参考 使用条件语句。

注意：格式化打印 日期/时间 格式在条件语句中目前还不支持。不过有一个 @元数据 字段的解决方案目前可行，想了解更多细节和案例的话请参考 条件语句中打印 日期/时间 格式。

@元数据 字段

在 Logstash 中，有一个特殊的字段被称作 @元数据。在输出时，@元数据不是任何事件的一部分，这使得它非常适合用于条件语句，或者使用字段引用和 sprintf 格式扩展和构建事件字段。

配置文件定义来自标准输入的事件。无论你输入什么内容都会成为事件中的 message 字段。过滤区块中的 mutate 事件添加一些字段，其中一些字段嵌套在了 @元数据 中。

input { stdin { } }

filter {
  mutate { add_field => { "show" => "这条数据将会出现在输出中" } }
  mutate { add_field => { "[@metadata][test]" => "Hello" } }
  mutate { add_field => { "[@metadata][no_show]" => "这条数据将不会出现在输出中" } }
}

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" => "这条数据将会出现在输出中",
       "message" => "asdf"
}

输入的「asdf」会变成 message 字段的内容，并且条件语句成功地计算了嵌套在@元数据字段中的 test 字段的内容。但是输出不会展示@元数据的字段，或者它的内容。

如果你添加了 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" => "这条数据将不会出现在输出中"
     },
     "@version" => "1",
     "host" => "example.com",
     "show" => "这条数据将会出现在输出中",
     "message" => "asdf"
}

现在你可以看到 @元数据的字段和它的子字段了。

重点：只有 rubydebug 译码器插件允许展示 @元数据 字段的内容。

在需要临时字段但又不希望它出现在最终输出中的任何时候，都可以使用@元数据字段。

这个新字段最常见的用例之一可能就是使用 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 过滤器中进行转换之后，不需要再删除 timestamp 字段了。

另一个应用场景是 CouchDB Changes输入插件。这个插件自动将 CouchDB 文档字段元数据捕获到输入插件本身内的@metadata 字段中。当事件通过被 Elasticsearch 索引传递时，Elasticsearch 输出插件允许指定 action（delete，update，insert 等）和 document_id，例如：

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

在条件语句中格式化打印 日期/时间 格式

在条件语句中格式化打印 日期/时间 格式目前还不支持，但是有一个解决方案。将日期计算放在字段中，这样就可以在条件中使用字段进行引用了。

例子

直接通过使用格式化打印时间格式添加一个基于摄入时间的字段将不能正常运行：

# 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"
      }
    }
  }
}