如果你想了解更多关于 lookup 的内容，请参阅文章 “Elastic：开发者上手指南” 中的 “丰富数据及 lookup” 章节。在今天的文章中，我来总结在 Logstash 中一些常用的 lookups。如下的这些插件可以帮助你使用附加信息丰富数据，例如 GeoIP 和用户代理信息：
- dns filter
- elasticsearch filter
- geoip filter
- http filter
- jdbc_static filter
- jdbc_streaming filter
- memcached filter
- translate filter
- useragent filter
Lookup 插件
dns filter
dns 过滤器插件执行标准或反向 DNS 查找。
以下配置对 source_host 字段中的地址进行反向查找，并将其替换为域名：
filter {
dns {
reverse => [ "source_host" ]
action => "replace"
}
}我们可以使用如下的例子来进行测试：
logstash.conf
input {
generator {
message => '8.8.8.8'
count => 1
}
}
filter {
dns {
reverse => [ "message" ]
action => "replace"
}
}
output {
stdout {
codec => rubydebug
}
}使用 Logstash 运行上面的配置文件：
./bin/logstash -f logstash.conf
elasticsearch filter
elasticsearch filter 将 Elasticsearch 中以前的日志事件的字段复制到当前事件。
以下配置显示了如何使用此过滤器的完整示例。 每当 Logstash 收到 “end” 事件时，它都会使用此 Elasticsearch filter 根据某些操作标识符找到匹配的 “start” 事件。 然后它将 @timestamp 字段从 “start” 事件复制到 “end” 事件的新字段中。 最后，结合使用 date filter 和 ruby 筛选器，示例中的代码计算两个事件之间的持续时间（以小时为单位）。
if [type] == "end" {
elasticsearch {
hosts => ["es-server"]
query => "type:start AND operation:%{[opid]}"
fields => { "@timestamp" => "started" }
}
date {
match => ["[started]", "ISO8601"]
target => "[started]"
}
ruby {
code => 'event.set("duration_hrs", (event.get("@timestamp") - event.get("started")) / 3600) rescue nil'
}
}更为详细的介绍，请阅读文章 “Logstash：运用 Elasticsearch filter 来丰富地理数据”。
geoip filter
geoip 过滤器添加有关 IP 地址位置的地理信息。 例如：
filter {
geoip {
source => "clientip"
}
}在 Elastic Stack 8.x. 上，我们做如下的一个例子：
logstash.conf
input {
generator {
message => '8.8.8.8'
count => 1
}
}
filter {
mutate {
rename => {"message" => "[clientip][ip]"}
}
geoip {
source => "[clientip][ip]"
target => "geoip"
}
}
output {
stdout {
codec => rubydebug
}
}运行上面的 Logstash 配置文件：
./bin/logstash -f logstash.conf
详细阅读文章 “Logstash：Logstash 入门教程 （二）”。
http filter
http 过滤器与外部 Web 服务/REST API 集成，并启用针对任何 HTTP 服务或端点的查找扩充。 此插件非常适合许多丰富用例，例如社交 API、情感 API、安全源 API 和业务服务 API。
更为详细的介绍，请阅读文章 “Logstash：HTTP 过滤器介绍”。
jdbc_static filter
jdbc_static 过滤器使用从远程数据库预加载的数据来丰富事件。
以下示例从远程数据库获取数据，将其缓存在本地数据库中，并使用查找来使用本地数据库中缓存的数据丰富事件。
filter {
jdbc_static {
loaders => [ (1)
{
id => "remote-servers"
query => "select ip, descr from ref.local_ips order by ip"
local_table => "servers"
},
{
id => "remote-users"
query => "select firstname, lastname, userid from ref.local_users order by userid"
local_table => "users"
}
]
local_db_objects => [ (2)
{
name => "servers"
index_columns => ["ip"]
columns => [
["ip", "varchar(15)"],
["descr", "varchar(255)"]
]
},
{
name => "users"
index_columns => ["userid"]
columns => [
["firstname", "varchar(255)"],
["lastname", "varchar(255)"],
["userid", "int"]
]
}
]
local_lookups => [ (3)
{
id => "local-servers"
query => "select descr as description from servers WHERE ip = :ip"
parameters => {ip => "[from_ip]"}
target => "server"
},
{
id => "local-users"
query => "select firstname, lastname from users WHERE userid = :id"
parameters => {id => "[loggedin_userid]"}
target => "user" (4)
}
]
# using add_field here to add & rename values to the event root
add_field => { server_name => "%{[server][0][description]}" }
add_field => { user_firstname => "%{[user][0][firstname]}" } (5)
add_field => { user_lastname => "%{[user][0][lastname]}" }
remove_field => ["server", "user"]
jdbc_user => "logstash"
jdbc_password => "example"
jdbc_driver_class => "org.postgresql.Driver"
jdbc_driver_library => "/tmp/logstash/vendor/postgresql-42.1.4.jar"
jdbc_connection_string => "jdbc:postgresql://remotedb:5432/ls_test_2"
}
}- 查询外部数据库以获取将在本地缓存的数据集。
- 定义用于构建本地数据库结构的列、类型和索引。 列名和类型应与外部数据库相匹配。
- 对本地数据库执行查找查询以丰富事件。
- 指定将存储查找数据的事件字段。 如果查找返回多列，则数据将作为 JSON 对象存储在字段中。
- 从 JSON 对象中获取数据并将其存储在顶级事件字段中，以便在 Kibana 中进行更轻松的分析。
更为详细的例子，请参考文章 “Logstash：Jdbc static filter plugin 介绍”。
jdbc_streaming filter
jdbc_streaming 过滤器使用数据库数据丰富事件。
以下示例执行 SQL 查询并将结果集存储在名为 country_details 的字段中：
filter {
jdbc_streaming {
jdbc_driver_library => "/path/to/mysql-connector-java-5.1.34-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://localhost:3306/mydatabase"
jdbc_user => "me"
jdbc_password => "secret"
statement => "select * from WORLD.COUNTRY WHERE Code = :code"
parameters => { "code" => "country_code"}
target => "country_details"
}
}更为详细的例子可一些参阅文章 “Logstash：运用 jdbc_streaming 来丰富我们的数据”。
memcached filter
Memcached 过滤器启用针对 Memcached 对象缓存系统的键/值查找扩充。 它支持读取 (GET) 和写入 (SET) 操作。 它是安全分析用例的显着补充。
更为详细的描述可以在文章中 “Logstash：运用 memcache 过滤器进行大规模的数据丰富” 看到。
translate filter
Translate 过滤器根据 hash 或文件中指定的替换值替换字段内容。 目前支持这些文件类型：YAML、JSON 和 CSV。
以下示例采用 response_code 字段的值，根据字典中指定的值将其转换为描述，然后从事件中删除 response_code 字段：
filter {
translate {
field => "response_code"
destination => "http_response"
dictionary => {
"200" => "OK"
"403" => "Forbidden"
"404" => "Not Found"
"408" => "Request Timeout"
}
remove_field => "response_code"
}
}详细阅读文章 “Logstash：Logstash translate 过滤器简介”。
useragent filter
useragent 过滤器将用户代理字符串解析为字段。
以下示例获取代理字段中的用户代理字符串，将其解析为用户代理字段，并将用户代理字段添加到名为 user_agent 的新字段中。 它还删除了原始代理字段
filter {
useragent {
source => "agent"
target => "user_agent"
remove_field => "agent"
}
}应用过滤器后，事件将丰富用户代理字段。 例如：
"user_agent": {
"os": "Mac OS X 10.12",
"major": "50",
"minor": "0",
"os_minor": "12",
"os_major": "10",
"name": "Firefox",
"os_name": "Mac OS X",
"device": "Other"
}详细阅读文章 “Logstash：Logstash 入门教程 （二）” 以了解更多。
