Logstash替换字符串，解析json数据，修改数据类型，获取日志时间

在某些情况下，有些日志文本文件类json，但它的是单引号，具体格式如下，我们需要根据下列日志数据，获取正确的字段和字段类型

1{'usdCnyRate': '6.728', 'futureIndex': '463.36', 'timestamp': '1532933162361'}
2{'usdCnyRate': '6.728', 'futureIndex': '463.378', 'timestamp': '1532933222335'}
3{'usdCnyRate': '6.728', 'futureIndex': '463.38', 'timestamp': '1532933348347'}
4{'usdCnyRate': '6.728', 'futureIndex': '463.252', 'timestamp': '1532933366866'}
5{'usdCnyRate': '6.728', 'futureIndex': '463.31', 'timestamp': '1532933372350'}
6{'usdCnyRate': '6.728', 'futureIndex': '463.046', 'timestamp': '1532933426899'}
7{'usdCnyRate': '6.728', 'futureIndex': '462.806', 'timestamp': '1532933432346'}
8{'usdCnyRate': '6.728', 'futureIndex': '462.956', 'timestamp': '1532933438353'}
9{'usdCnyRate': '6.728', 'futureIndex': '462.954', 'timestamp': '1532933456796'}
10{'usdCnyRate': '6.728', 'futureIndex': '462.856', 'timestamp': '1532933492411'}
11{'usdCnyRate': '6.728', 'futureIndex': '462.776', 'timestamp': '1532933564378'}
12{'usdCnyRate': '6.728', 'futureIndex': '462.628', 'timestamp': '1532933576849'}
13{'usdCnyRate': '6.728', 'futureIndex': '462.612', 'timestamp': '1532933588338'}
14{'usdCnyRate': '6.728', 'futureIndex': '462.718', 'timestamp': '1532933636808'}

此时我们如果当json直接用logstash Json filter plugin来解析会如下报错

[WARN ] 2018-07-31 10:20:12.708 [Ruby-0-Thread-5@[main]>worker1: :1] json - Error parsing json {:source=>"message", :raw=>"{'usdCnyRate': '6.728', 'futureIndex': '462.134', 'timestamp': '1532933714371'}", :exception=>#<LogStash::Json::ParserError: Unexpected character (''' (code 39)): was expecting double-quote to start field name at [Source: (byte[])"{'usdCnyRate': '6.728', 'futureIndex': '462.134', 'timestamp': '1532933714371'}"; line: 1, column: 3]>}

此处我认为简单的做法是替换单引号为双引号，替换过程应用了logstash mutate gsub 一定要看清楚我10-12行的写法，作用为替换字符串，14-15行为解析json。我们还需要将usdCnyRate和futureIndex转为float类型（18-21行），将timestamp转为时间类型，并重新定义一个logdate来存储（23-25行）此处用到 logstash date filter plugin

1input{
2    file {
3        path => "/usr/share/logstash/wb.cond/test.log"
4        start_position => "beginning"
5        sincedb_path => "/dev/null"
6    }
7}
8filter{
9    mutate {
10        gsub =>[
11            "message", "'", '"'
12        ]
13    }
14    json {
15        source => "message"
16    }
17    mutate {
18        convert => {
19            "usdCnyRate" => "float"
20            "futureIndex" => "float"
21        }
22    }
23    date {
24        match => [ "timestamp", "UNIX_MS" ]
25        target => "logdate"
26    }
27}
28output{
29    stdout{
30        codec=>rubydebug
31    }
32}

利用上述配置文件，我们能正确解析出日志文件的字段和类型

1{
2        "message" => "{\"usdCnyRate\": \"6.728\", \"futureIndex\": \"463.378\", \"timestamp\": \"1532933222335\"}",
3     "@timestamp" => 2018-07-31T10:48:48.600Z,
4           "host" => "logstashvm0",
5           "path" => "/usr/share/logstash/wb.cond/test.log",
6       "@version" => "1",
7        "logdate" => 2018-07-30T06:47:02.335Z,
8     "usdCnyRate" => 6.728,
9      "timestamp" => "1532933222335",
10    "futureIndex" => 463.378
11}
12{
13        "message" => "{\"usdCnyRate\": \"6.728\", \"futureIndex\": \"463.252\", \"timestamp\": \"1532933366866\"}",
14     "@timestamp" => 2018-07-31T10:48:48.602Z,
15           "host" => "logstashvm0",
16           "path" => "/usr/share/logstash/wb.cond/test.log",
17       "@version" => "1",
18        "logdate" => 2018-07-30T06:49:26.866Z,
19     "usdCnyRate" => 6.728,
20      "timestamp" => "1532933366866",
21    "futureIndex" => 463.252
22}
23{
24        "message" => "{\"usdCnyRate\": \"6.728\", \"futureIndex\": \"463.31\", \"timestamp\": \"1532933372350\"}",
25     "@timestamp" => 2018-07-31T10:48:48.602Z,
26           "host" => "logstashvm0",
27           "path" => "/usr/share/logstash/wb.cond/test.log",
28       "@version" => "1",
29        "logdate" => 2018-07-30T06:49:32.350Z,
30     "usdCnyRate" => 6.728,
31      "timestamp" => "1532933372350",
32    "futureIndex" => 463.31
33}