访问事件数据和字段

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

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

Logstash 中的某些配置选项需要存在字段才能起作用。因为输入生成事件，所以在输入块中没有要评估的字段——它们尚不存在！

字段引用编辑

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

访问字段的基本语法是 [字段名]。如果您指的是顶级字段，则可以省略 [] 并简单地使用 字段名。要引用嵌套字段，请指定该字段的完整路径：[顶级字段][嵌套字段]。

例如，此事件具有五个顶级字段（代理、ip、请求、响应、ua）和三个嵌套字段（状态、字节、操作系统）。

{
  "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），您可以简单地指定字段名称。

有关更详细的信息，请参阅字段引用深度解析。

sprintf 格式编辑

字段引用格式也用于 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 {
  ...
}

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

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

支持的布尔运算符为

支持的一元运算符为

表达式可以很长且很复杂。表达式可以包含其他表达式，您可以使用 ! 否定表达式，并且可以使用括号 (...) 对其进行分组。

例如，如果字段 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 => "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

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

@metadata 字段编辑

在 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 过滤器中转换后删除“timestamp”字段。

另一个用例是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 日期/时间格式，但可以使用解决方法。将日期计算放入字段中，以便您可以在条件中使用字段引用。

示例

直接使用 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"
      }
    }
  }
}