Logstash Grok 导入 Elasticsearch 慢查询日志文件

Question

测试数据

elasticsearch.log

[2018-03-13T00:01:15,437][TRACE][index.search.slowlog.query] [node23] [inv_07][4] took[2.3ms], took_millis[2], types[], stats[], search_type[QUERY_THEN_FETCH], total_shards[105], source[{"size":1000,"query":{"has_parent":{"query":{"bool":{"must":[{"terms":{"id_receipt":[234707456,234707458],"boost":1.0}},{"term":{"receipt_key":{"value":6799,"boost":1.0}}},{"term":{"code_receipt":{"value":"TKMS","boost":1.0}}}],"disable_coord":false,"adjust_pure_negative":true,"boost":1.0}},"parent_type":"receipts","score":false,"ignore_unmapped":false,"boost":1.0}},"version":true,"_source":false,"sort":[{"_doc":{"order":"asc"}}]}],
[2018-03-13T00:01:16,609][TRACE][index.search.slowlog.query] [node23] [inv_06][1] took[77micros], took_millis[0], types[], stats[], search_type[QUERY_THEN_FETCH], total_shards[105], source[{"size":1000,"query":{"has_parent":{"query":{"bool":{"must":[{"terms":{"id_receipt":[234707456,234707458],"boost":1.0}},{"term":{"receipt_key":{"value":6799,"boost":1.0}}},{"term":{"code_receipt":{"value":"TKMS","boost":1.0}}}],"disable_coord":false,"adjust_pure_negative":true,"boost":1.0}},"parent_type":"receipts","score":false,"ignore_unmapped":false,"boost":1.0}},"version":true,"_source":false,"sort":[{"_doc":{"order":"asc"}}]}],
[2018-03-13T00:01:17,609][TRACE][index.search.slowlog.query] [node23] [inv_02][2] took[160.2micros], took_millis[0], types[], stats[], search_type[QUERY_THEN_FETCH], total_shards[105], source[{"size":1000,"query":{"has_parent":{"query":{"bool":{"must":[{"terms":{"id_receipt":[234707456,234707458],"boost":1.0}},{"term":{"receipt_key":{"value":6799,"boost":1.0}}},{"term":{"code_receipt":{"value":"TKMS","boost":1.0}}}],"disable_coord":false,"adjust_pure_negative":true,"boost":1.0}},"parent_type":"receipts","score":false,"ignore_unmapped":false,"boost":1.0}},"version":true,"_source":false,"sort":[{"_doc":{"order":"asc"}}]}],

Logstash 配置文件

logstash.conf

input {
  file {
    path => "/Users/jasonwu/WorkSpace/learn/Elasticsearch/es_slow.log"
    start_position => "beginning"
    sincedb_path => "/Users/jasonwu/.Trash/sincedb.trash"
  }
}

filter {
  grok {
    match => {
      "message" => ["\[%{TIMESTAMP_ISO8601:timestamp}\]\[%{LOGLEVEL:level}\]\[%{HOSTNAME:type}\]%{SPACE}\[%{HOSTNAME:[node_name]}\]%{SPACE}\[%{WORD:[index_name]}\]%{NOTSPACE}%{SPACE}took\[%{NUMBER:took_micro}%{NOTSPACE}\]%{NOTSPACE}%{SPACE}%{NOTSPACE}%{NOTSPACE}%{SPACE}%{NOTSPACE}%{NOTSPACE}%{SPACE}%{NOTSPACE}%{NOTSPACE}%{SPACE}search_type\[%{WORD:search_type}\]%{NOTSPACE}%{SPACE}total_shards\[%{NUMBER:total_shards}\]%{NOTSPACE}%{SPACE}source%{GREEDYDATA:query}\Z"]
    }
  }

  mutate {
    remove_field => ["@version", "@timestamp", "host", "path", "message", "logTook"]
  }
}

output {
  stdout {}

  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logs"
  }
}

执行 Logstash 导入 Elasticsearch

$ bin/logstash -f /Users/jasonwu/WorkSpace/learn/Elasticsearch/logstash.conf