Logstash translate 过滤器

用于根据字典或查找文件过滤传入数据中的特定字段。 然后，如果输入字段与字典查找文件中的条目匹配，则它将执行操作，例如，将字段添加到数据或发送电子邮件。这个和我们之前介绍的数据丰富是一样的。

例子：/tmp/blacklisted_ip.yaml

"216.46.173.126": "true"
"180.179.174.219": "true"
"204.77.168.241": "true"
"65.39.197.164": "true"
"80.91.33.133": "true"
"84.208.15.12": "true"
"74.125.60.158": "true"

我们将在 remote_ip 字段中检查这些特定 IP 的传入日志数据。 如果我们在 remote_ip 字段中看到其中一个 IP，则将在文档中添加一个值为 true”的新字段 blacklisted_ip。 这是我们的配置：

下载日志文件

wget https://raw.githubusercontent.com/elastic/examples/master/Common%20Data%20Formats/nginx_json_logs/nginx_json_logs

logstash_translate.conf

input {
  stdin {
    codec => json
  }
}
 
filter {
  date {
    match => ["time", "dd/MMM/YYYY:HH:mm:ss Z" ]
    locale => en
  }
 
  geoip {
    source => "remote_ip"
    target => "geoip"
  }
 
  translate {
    field => "remote_ip"
    destination => "blacklisted_ip"
    dictionary_path => '/tmp/blacklisted_ip.yaml'
  }
 
  grok {
    match => [ "request" , "%{WORD:request_action} %{DATA:request1} HTTP/%{NUMBER:http_version}" ]
  }
}
 
output {
  stdout {
    codec => plain {
      charset => "ISO-8859-1"
    }
  }
 
  elasticsearch {
    index => "logstash-nginx"
    hosts => ["127.0.0.1:9200"]
  }
}

查看索引

GET _cat/indices

我们为它创建一个index pattern，查看内容

在上面，我们使用了 yaml 文件作为 dictionary_path。根据文档，我们实际上也可以使用 csv 格式的文档：

blacklisted_ip.csv

"216.46.173.126","true"
"180.179.174.219","true"
"204.77.168.241","true"
"65.39.197.164","true"
"80.91.33.133","true"
"84.208.15.12","true"
"74.125.60.158","true"

logstash_translate_csv.conf

input {
  stdin {
    codec => json
  }
}
 
filter {
  date {
    match => ["time", "dd/MMM/YYYY:HH:mm:ss Z" ]
    locale => en
  }
 
  geoip {
    source => "remote_ip"
    target => "geoip"
  }
 
  translate {
    field => "remote_ip"
    destination => "blacklisted_ip"
    dictionary_path => '/Users/liuxg/data/translate/blacklisted_ip.csv'
  }
 
  grok {
    match => [ "request" , "%{WORD:request_action} %{DATA:request1} HTTP/%{NUMBER:http_version}" ]
  }
}
 
output {
  stdout {
    codec => rubydebug
  }
 
  elasticsearch {
    index => "logstash-nginx"
    hosts => ["127.0.0.1:9200"]
  }
}

丰富多个字段

blacklisted_ip.yaml

"216.46.173.126": "true,good"
"180.179.174.219": "true,bad"
"204.77.168.241": "true,great"
"65.39.197.164": "true,teriffic"
"80.91.33.133": "true,fabulous"
"84.208.15.12": "true,excellent"
"74.125.60.158": "true,wonderful"

logstash_translate.conf 

input {
  stdin {
    codec => json
  }
}
 
filter {
  date {
    match => ["time", "dd/MMM/YYYY:HH:mm:ss Z" ]
    locale => en
  }
 
  geoip {
    source => "remote_ip"
    target => "geoip"
  }
 
  translate {
    field => "remote_ip"
    destination => "csv_data"
    dictionary_path => '/Users/liuxg/data/translate/blacklisted_ip.yaml'
  }
 
  if ("" in [csv_data]) {
    csv {
      source => csv_data
      separator => ","
      columns => [ "blacklisted_ip", "comments"]
    }
  }
 
  grok {
    match => [ "request" , "%{WORD:request_action} %{DATA:request1} HTTP/%{NUMBER:http_version}" ]
  }
}
 
output {
  stdout {
    codec => rubydebug
  }
 
  elasticsearch {
    index => "logstash-nginx"
  }
}

在kibana中查看会发现多了一个comments