logstash grok filter 插件实战

最近遇到一个搜索的问题，发现文章中如果含有品牌信息，存储到数据库中，会用某种结构化的标签存储品牌信息，比如数据库表和ES索引：

+----------+----------------------+
| brand_id | content              |
+----------+----------------------+
|        1 | 111,222              |
|      184 | [##BRAND$$16$$3FE##] |
+----------+----------------------+

3FE 是一个品牌，可加上 [##BRAND$$16$$##] 自定义标签后，中文索引分词英文品牌会导致搜索不到（分词为 $3FE$ ）,有些同学说，那你针对英文用english分词器，但这是另外一个层面的事情。

想到的解决方案就是logstash处理的时候，匹配 [##BRAND$$16$$3FE##] 然后过滤掉结构化自定义标签，最终将3FE索引到ES中。

虽然最终因为其他原因没有采取这个方案，但对于后续实践非常重要，logstash自动化将数据索引到ES，更准确、更方便，类似于MySQL的Binlog机制。

看看具体如何解决的，先贴代码：

statement => "select brand_id,content  from test2 "

if [brand_id]>0 {
    grok { match => [ "content", "(##BRAND\$\$[0-9]{1,5}\$\$(?<brand_search>.*?)##)" ]  tag_on_failure => [ ]  }  
    } else {
        mutate {
            add_field => { "brand_search" => "" }
        }
    }
}
else {
    mutate {
        add_field => { "brand_search" => "" }
    }
}

结果如下：

{
    "_index" : "test3",
    "_type" : "_doc",
    "_id" : "184",
    "_score" : 1.0,
    "_source" : {
      "@version" : "1",
      "@timestamp" : "2020-03-06T11:50:00.173Z",
      "content" : "[##BRAND$$16$$3FE##]",
      "brand_id" : 184,
      "brand_search" : "3FE"
    }
}

tag_on_failure什么意思呢，默认匹配不成功，会增加一个tag，如下：

"@version" => "1",
      "tags" => [
    [0] "_grokparsefailure"
]

而这个tag也会索引到ES中，这是我们不想要的，而通过tag_on_failure可以不生成这个tag，也就是不会索引到ES中。

为实现这个功能也可以采取下面的方法：

grok { match => [ "content", "(##BRAND\$\$[0-9]{1,5}\$\$(?<brand_search>.*?)##)" ]  tag_on_failure => [ ]  }  
if "_grokparsefailure" in [tags] {
    mutate {
        remove_tag => [ "_grokparsefailure" ] #并不是删除tag这个field
        remove_field =>["tags"]
    }
}

有的时候匹配不成功，可以直接删除这个event，也就是整条记录不索引到ES，注意grok和if是并列的：

grok { match => [ "content", "(##BRAND\$\$[0-9]{1,5}\$\$(?<brand_search>.*?)##)" ]  } 
if "_grokparsefailure" in [tags] {
   drop {}
}

grok filter插件在调试的时候可以使用grokdebug工具，上面的正则也有缺陷，如果有两个品牌，只能匹配到一个品牌信息。