Logstash 保存不同类型的日志到各自的 ElasticSearch 索引

更多插件设置请参考官方文档：[Input Plugins](https://www.elastic.co/guide/en/logstash/current/input-plugins.html), [Output Plugins](https://www.elastic.co/guide/en/logstash/current/output-plugins.html), [Filter Plugins](https://www.elastic.co/guide/en/logstash/current/filter-plugins.html)

如果是一份日志同时保存到两台 ElasticSearch ，只需要在 *output* 中添加两个 `elasticsearch` plugin 设置就行了。

```json
output {
    elasticsearch {
        hosts => ["http://192.168.0.1:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
    }

elasticsearch {
        hosts => ["http://192.168.0.2:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
    }
}
```

如果需要根据某些条件保存到不同的索引，则需要写一些判断条件。
Logstash 的配置文件支持如下格式的语法：

```json
if EXPRESSION {
  ...
} else if EXPRESSION {
  ...
} else {
  ...
}
```

其中 *EXPRESSION* 支持如下运算符：

- 比较运算符
    - equality: `==`, `!=`, `<`, `>`, `<=`, `>=`
    - regexp: `=~`, `!~` (checks a pattern on the right against a string value on the left)
    - inclusion: `in`, `not in`

- 布尔运算符

- `and` : 与
    - `or` : 或
    - `nand` : 与非（有 false 则 true，全 true 则 false）
    - `xor` : 异或（只有在两个比较的值不同时其结果才是 true，否则结果为 false）

- 一元运算符
    - `!` : 非

另外，在判断条件中访问日志中的字段需要使用中括号 `[fieldname]` 的形式。

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

如果是在 Logstash 中调用 *sprintf format* 输出的字段时，需使用 `%{fieldname}` 的形式，多个字段时可以使用 `%{[fieldname1][fieldname2]}`。

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

下面是根据需求写的配置文件。
优先将 *type* 为 *parameter* 的日志单独输出到保存参数的索引，然后根据日志的 *level* 输出到各自的索引。
如果是程序的 Exception 则不会包含 *type* 和 *level* 字段，判断是否包含 *Exception* 字符，如果包含则输出到记录错误消息的索引。
最后的 `else` 则是输出到默认的 *info* 索引。
每个索引以月为单位，每月创建一个单独的索引，以便将来删除过期的日志信息。
需要注意的是 `elasticsearch` 插件的 *index* 字段不支持大写字母，所以这里增加了一个 `lowercase` 的 `filter`，将 *type* 和 *level* 字段转成了小写。

```json
input {
    rabbitmq {
        type => "log_type"
        durable => true
        exchange => "logstash"
        exchange_type => "topic"
        key => "service.#"
        host => "192.168.0.1"
        port => 5672
        user => "username"
        password => "password"
        queue => "log_queue"
        auto_delete => false
        tags => ["service"]
    }
}

filter {
    mutate {
        lowercase => [ "type", "level" ]
    }
}

output {
    if [type] and [type] == "parameter" {
        elasticsearch {
            hosts => ["http://192.168.0.1:9200"]
            index => "logstash-parameter-%{+YYYY.MM}"
        }
    } else if [level] and [level] != "" {
        elasticsearch {
            hosts => ["http://192.168.0.1:9200"]
            index => "logstash-%{level}-%{+YYYY.MM}"
        }
    } else if [message] and "Exception" in [message] {
        elasticsearch {
            hosts => ["http://192.168.0.1:9200"]
            index => "logstash-error-%{+YYYY.MM}"
        }
    } else {
        elasticsearch {
            hosts => ["http://192.168.0.1:9200"]
            index => "logstash-info-%{+YYYY.MM}"
        }
    }
}
```

**参考**

- [Structure of a Config File](https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html)
- [Accessing Event Data and Fields in the Configuration](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html)
- [Mutate filter plugin](https://www.elastic.co/guide/en/logstash/6.5/plugins-filters-mutate.html)
- [Elasticsearch output plugin](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html)

Logstash 保存不同类型的日志到各自的 ElasticSearch 索引

目录

智强 @佳佳回复

佳佳 @智强回复

智强 @佳佳回复

智强 @佳佳回复

佳佳 @智强回复

回复内容 *

目录

智强 @佳佳 回复

佳佳 @智强 回复

智强 @佳佳 回复

智强 @佳佳 回复

佳佳 @智强 回复

回复内容 *

智强 @佳佳回复

佳佳 @智强回复

智强 @佳佳回复

智强 @佳佳回复

佳佳 @智强回复