Logstash是一个具有3个阶段的处理管道：输入 → 过滤器 → 输出。输入生成事件，过滤器修改它们，输出将它们发送到其他指定的地方。其所有动作都有属性。例如，apache访问日志会有状态代码（200,404），请求路径（"/", "index.html"），HTTP动词（GET，POST），客户端IP地址等等。Logstash调用这些属性"字段"。Logstash中的一些配置选项需要存在字段才能运行。因为输入生成事件，所以在输入块中没有要评估的字段 - 它们还不存在！并且由于它们依赖于事件和字段，所以以下配置选项仅适用于过滤和输出块。
提示: 下面描述的字段引用，sprintf格式和条件在输入块中不起作用。
1.1 字段参考
通过名称引用字段通常很有用。为此，您可以使用Logstash的字段引用语法。访问字段的基本语法是[fieldname]。如果您指的是 顶级字段，则可以省略[]并简单地使用fieldname。要引用嵌套字段，请指定该字段的完整路径：[顶级字段][嵌套字段]。
例如，以下事件有五个顶级字段（代理，IP，请求，响应，ua）和三个嵌套字段（状态，字节，操作系统）。
|
1 2 3 4 5 6 7 8 9 10 11 12 |
{ "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，您只需指定字段名称即可。
1.2 sprintf格式
字段引用格式也用于Logstash调用sprintf格式。此格式使您可以从其他字符串中引用字段值。例如，statsd输出具有增量设置，使您可以按状态代码保留apache日志的计数：
|
1 2 3 4 5 |
output { statsd { increment => "apache.%{[response][status]}" } } |
同样，您可以将@timestamp字段中的时间戳转换为字符串。不要在花括号内指定字段名称，而是使用+FORMAT语法(更多信息请查看：http://joda-time.sourceforge.net/apidocs/org/joda/time/format/DateTimeFormat.html)。例如，如果要根据事件的日期和小时以及type字段使用文件输出写入日志：
|
1 2 3 4 5 |
output { file { path => "/var/log/%{type}.%{+yyyy.MM.dd.HH}" } } |
1.3 条件判断
Logstash的配置中使用条件判断可以让您实现只在特定的前提下过滤或输出事件，条件查看和使用与编程语言中的条件相同。条件语句支持if，else if以及else报表和可以被嵌套。
条件判断语法是：
|
1 2 3 4 5 6 7 |
if EXPRESSION { ... } else if EXPRESSION { ... } else { ... } |
您可以使用以下比较运算符：
|
1 2 3 |
平等： ==， !=， <， >， <=，>= regexp： =~，!~（检查右边的模式与左边的字符串值） 包含： in，not in |
支持的布尔运算符是：
|
1 |
and，or，nand，xor |
支持的一元运算符是：
|
1 |
！ |
表达式可能很长且很复杂。表达式可以包含其他表达式，您可以!使用括号来否定表达式，并且可以使用括号对它们进行分组(...)。
示例1：如果字段action的值为"login"，则使用mutate过滤器删除secret字段：
|
1 2 3 4 5 |
filter { if [action] == "login" { mutate { remove_field => "secret" } } } |
示例2：在单个条件中指定多个表达式：
|
1 2 3 4 5 6 7 8 |
output { # Send production errors to pagerduty if [loglevel] == "ERROR" and [deployment] == "production" { pagerduty { ... } } } |
示例3：使用in运算符来测试字段是否包含特定字符串，键或（对于列表）元素：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
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" } } } |
示例4：使用not in仅在grok成功时将事件路由到Elasticsearch：
|
1 2 3 4 5 |
output { if "_grokparsefailure" not in [tags] { elasticsearch { ... } } } |
您可以检查特定字段是否存在，但目前无法区分不存在的字段与完全错误的字段。表达式if [foo]返回false时：
[foo] 在事件中不存在，
[foo] 存在于事件中，但是是假的，或者
[foo] 存在于事件中，但为null
有关更复杂的示例，请参阅：使用条件判断（https://www.elastic.co/guide/en/logstash/current/config-examples.html#using-conditionals）
1.4 @metadata字段
在Logstash 1.5及更高版本中，有一个名为@metadata的特殊字段。在输出时，内容不会成为您的任何事件的一部分，这使得它非常适合用于条件，或者使用字段引用和sprintf格式扩展和构建事件字段。
示例1：以下配置文件将从STDIN生成事件。无论什么类型都将成为message事件的领域。mutate过滤器块中的事件将添加一些字段，一些字段嵌套在@metadata字段中。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
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 } } } |
运行结果：
|
1 2 3 4 5 6 7 8 9 10 |
$ 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字段内容，条件成功判断了嵌套在@metadata字段中的test字段内容。但是输出并没有显示字段@metadata的内容。
示例2：自定义该rubydebug编解码器可显示的内容，如果您在上面示例1配置的最后添加一个配置标志字段，metadata => true：
|
1 |
stdout { codec => rubydebug { metadata => true } } |
这时候，包括@metadata在内的所有字段就会都显示出来。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
$ 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" } |
提示： 只有rubydebug编解码器允许您显示该@metadata字段的内容。
示例3：使用过滤插件date，定义临时时间戳
|
1 2 3 4 5 6 7 8 9 10 |
input { stdin { } } filter { grok { match => [ "message", "%{HTTPDATE:[@metadata][timestamp]}" ] } date { match => [ "[@metadata][timestamp]", "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { stdout { codec => rubydebug } } |
此配置将提取的日期放入过滤器的[@metadata][timestamp]字段中。当我们为这个配置提供一个示例日期字符串，可以发现输出中没有额外的字段。
|
1 2 3 4 5 6 7 8 9 |
$ 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" } |
示例4：另一个用例是CouchDB Changes输入插件（参见 https://github.com/logstash-plugins/logstash-input-couchdb_changes）。此插件自动将CouchDB文档字段元数据捕获到@metadata输入插件本身的 字段中。当事件通过Elasticsearch索引时，Elasticsearch输出插件允许您指定action（删除，更新，插入等）和document_id，如下所示：
|
1 2 3 4 5 6 7 8 9 |
output { elasticsearch { action => "%{[@metadata][action]}" document_id => "%{[@metadata][_id]}" hosts => ["example.com"] index => "index_name" protocol => "http" } } |
文章内容来至于elastic官方文档，链接如下所示：
|
1 |
https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html |
