如果你想了解更多关于 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 rootadd_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"}
}

1. 查询外部数据库以获取将在本地缓存的数据集。
2. 定义用于构建本地数据库结构的列、类型和索引。 列名和类型应与外部数据库相匹配。
3. 对本地数据库执行查找查询以丰富事件。
4. 指定将存储查找数据的事件字段。 如果查找返回多列，则数据将作为 JSON 对象存储在字段中。
5. 从 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 入门教程 （二）” 以了解更多。