問題

有時候我們想要在Logstash裏對收集到的日誌等信息進行分割，並且將分割後的字符作為新的字符來index到Elasticsearch裏。假定需求如下：

Logstash收集到的日誌字段message的值是由多個字段拼接而成的，分隔符是;,;，如下：

{
    "message": "key_1=value_1;,;key_2=value_2"
}

現在想要將message的值拆分成2個新的字段：key_1、key_2，並且將它們index到ES裏，可以借助Logstash的filter的插件來完成；這裏提供兩種解決方案。

方案一：使用mutate插件

filter {
    mutate {
        split => ["message",";,;"]
        add_field =>   {
            "temp1" => "%{[message][0]}"
        }
        add_field =>   {
            "temp2" => "%{[message][1]}"
        }
    }
    
    mutate {
        split => ["temp1","="]
        add_field =>   {
            "%{[temp1][0]}" => "%{[temp1][1]}"
        }
    }
    
    mutate {
        split => ["temp2","="]
        add_field =>   {
            "%{[temp2][0]}" => "%{[temp2][1]}"
        }
        remove_field => [ "temp1", "temp2", "message" ]
    }
}
 

看得出來，這種做法很麻煩，也不利於日後的維護。每當message裏被拼接的字段的數量增加時，就必須同步改動這裏的filter邏輯，而且添加的代碼量也是呈線性遞增的。

方案二：使用ruby插件

filter {
    ruby {
        code => "
            array1 = event.get('message').split(';,;')
            array1.each do |temp1|
                if temp1.nil? then
                    next
                end
                array2 = temp1.split('=')
                key = array2[0]
                value = array2[1]
                if key.nil? then
                    next
                end
                event.set(key, value)
            end
        "
        remove_field => [ "message" ]
    }
}
 

ruby插件可以允許你使用ruby的語法來完成各種復雜的邏輯，使用這種方案可以完美解決方案一中的不足之處，便於日後的維護。

參考鏈接

• Logstash事件字段遍歷
• Logstash詳解之——filter模塊

ELK系列(5) - Logstash怎麽分割字符串並添加新的字段到Elasticsearch