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

Question

测试数据

mysql_slow.log

/usr/sbin/mysqld, Version: 5.7.30-0ubuntu0.18.04.1-log ((Ubuntu)). started with:
Tcp port: 3306  Unix socket: /var/run/mysqld/mysqld.sock
Time                 Id Command    Argument
# Time: 2020-06-03T06:03:33.675799Z
# User@Host: root[root] @ localhost []  Id:     4
# Query_time: 2.064824  Lock_time: 0.000000 Rows_sent: 1  Rows_examined: 0
SET timestamp=1591164213;
SELECT SLEEP(2);
# Time: 2020-06-03T06:04:09.582225Z
# User@Host: root[root] @ localhost []  Id:     4
# Query_time: 3.000192  Lock_time: 0.000000 Rows_sent: 1  Rows_examined: 0
SET timestamp=1591164249;
SELECT SLEEP(3);
# Time: 2020-06-03T06:05:16.786196Z
# User@Host: root[root] @ localhost []  Id:     4
# Query_time: 5.000211  Lock_time: 0.000000 Rows_sent: 1  Rows_examined: 0
SET timestamp=1591164316;
SELECT SLEEP(5);
/usr/sbin/mysqld, Version: 5.7.30-0ubuntu0.18.04.1-log ((Ubuntu)). started with:
Tcp port: 3306  Unix socket: /var/run/mysqld/mysqld.sock
Time                 Id Command    Argument
# Time: 2020-06-03T09:55:39.044321Z
# User@Host: root[root] @ localhost []  Id:     2
# Query_time: 2.723918  Lock_time: 0.000000 Rows_sent: 0  Rows_examined: 0
use recipes_database;
SET timestamp=1591178139;
CREATE TABLE recipe_ingredients (
  recipe_id int NOT NULL,
  ingredient_id INT NOT NULL,
  amount INT NOT NULL,
  PRIMARY KEY (recipe_id,ingredient_id)
);
/usr/sbin/mysqld, Version: 5.7.30-0ubuntu0.18.04.1-log ((Ubuntu)). started with:
Tcp port: 3306  Unix socket: /var/run/mysqld/mysqld.sock
Time                 Id Command    Argument
# Time: 2020-06-03T09:55:39.044321Z
# User@Host: root[root] @ localhost []  Id:     2
# Query_time: 2.723918  Lock_time: 0.000000 Rows_sent: 0  Rows_examined: 0
use recipes_database;
SET timestamp=1591178139;
CREATE TABLE recipe_ingredients (
  recipe_id int NOT NULL,
  ingredient_id INT NOT NULL,
  amount INT NOT NULL,
  PRIMARY KEY (recipe_id,ingredient_id)
);

Logstash 配置文件

logstash.conf

input {
  file {
    path => "/Users/jasonwu/WorkSpace/learn/Elasticsearch/mysql_slow.log"
    start_position => "beginning"
    sincedb_path => "/Users/jasonwu/.Trash/sincedb.trash"
    codec => multiline {
      pattern => "^# Time: %{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
    }
   }
}

filter {
  mutate {
    gsub => [
    "message", "#", "",
    "message", "\n", " "
  ]

    remove_field => ["@version", "host"]
  }

  grok {
    match => {
      "message" => ["Time\:%{SPACE}%{TIMESTAMP_ISO8601:timestamp}%{SPACE}User\@Host\:%{SPACE}%{WORD:user}\[%{NOTSPACE}\] \@ %{NOTSPACE:host} \[\]%{SPACE}Id\:%{SPACE}%{NUMBER:sql_id}%{SPACE}Query_time\:%{SPACE}%{NUMBER:query_time}%{SPACE}Lock_time\:%{SPACE}%{NUMBER:lock_time}%{SPACE}Rows_sent\:%{SPACE}%{NUMBER:rows_sent}%{SPACE}Rows_examined\:%{SPACE}%{NUMBER:rows_examined}%{SPACE}%{GREEDYDATA}; %{GREEDYDATA:command}\;%{GREEDYDATA}"]
    }
  }
}

output {
  stdout {}

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

执行 Logstash 导入 Elasticsearch

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