背景是一个大型营销系统经常出现mysql的慢查询，导致线上服务频繁出现故障，为了查看是哪些sql有问题，并且要支持各种维度的统计查询，所以使用FileBeat+LogStash+ElasticSearch+Kibana实现此需求。本文仅描述如何配置FileBeath和LogStash实现MySQL慢查询日志解析。
filebeat.inputs:
- type: log
enabled: true
# 忽略在指定的时间跨度之前被修改的文件
ignore_older: 30000h
# mysql慢查询日志目录，支持*通配符匹配多级目录
paths:
- /opt/slow-sql/*.log
# 文档类型是mysqlslow，这是filebeat内置的一套规则
document_type: mysqlslow
multiline:
pattern: "^# User@Host: "
negate: true
match: after
tail_files: false
output.logstash:
# logstash的地址，我是部署在同一台机器上的
hosts: ["127.0.0.1:5044"]
input {
# 使用filebeat推送日志
beats {
port => 5044
host => "0.0.0.0"
}
}
filter {
grok {
# 有数据库名，有schema
match => {"message" => "(?m)^# User@Host: %{USER:user}\[[^\]]+\]\s@\s*\[%{IP:clientip}\]\s*([\s\S]*)#\s+Schema: (?<schema>\w+)([\s\S]*)\s+#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*([\s\S]*)use\s(?<dbname>\w+);([\s\S]*)SET timestamp=%{NUMBER:sql_time:int}"}
# 无数据库名,有schema
match => {"message" => "(?m)^# User@Host: %{USER:user}\[[^\]]+\]\s@\s*\[%{IP:clientip}\]\s*([\s\S]*)#\s+Schema: (?<schema>\w+)([\s\S]*)\s+#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*([\s\S]*)SET timestamp=%{NUMBER:sql_time:int}"}
# 有数据库名，无schema
match => {"message" => "(?m)^# User@Host: %{USER:user}\[[^\]]+\]\s@\s*\[%{IP:clientip}\]\s*([\s\S]*)#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*([\s\S]*)use\s(?<dbname>\w+);([\s\S]*)SET timestamp=%{NUMBER:sql_time:int}"}
# 无数据库名,无schema
match => {"message" => "(?m)^# User@Host: %{USER:user}\[[^\]]+\]\s@\s*\[%{IP:clientip}\]\s*([\s\S]*)#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*([\s\S]*)SET timestamp=%{NUMBER:sql_time:int}"}
overwrite => ["message"]
}
grok{
# 匹配 source中的ip
match => {"source" => "(?m)\s*%{IP:server_ip}"}
}
# 时间戳格式化并只保留日期
ruby {
code => "
require 'time'
event.set('date_tag', Time.at(event.get('sql_time')).to_date.to_s.delete!('-'))
"
}
# 索引时间戳使用sql生成的时间，不再使用当前时间
date {
match => ["sql_time", "yyyy-MM-dd HH:mm:ss", "UNIX"]
target => "@timestamp"
locale => "cn"
}
}
output {
# 调试时使用，在控制台打印日志分割结果
stdout {
codec => rubydebug {}
}
# es配置
elasticsearch {
hosts => "localhost:9200"
# 索引名称
index => "slow-sql-%{+YYYY.MM.dd}"
}
}
#!/bin/bash
echo "
#-----------------------------------------
#@Author: linvaux
#@Email: linvaux@outlook.com
#@Desc: Auto install ELFK
#----------------------------------------
"
INFO()
{
echo -e "\033[0;32m[INFO] $* \033[0m"
}
ERROR()
{
echo -e "\033[0;31m[ERROR] $* \033[0m"
}
WARN()
{
echo -e "\033[0;33m[WARN] $* \033[0m"
}
# booster the docker-hub
booster()
{
daemon="/etc/docker/daemon.json"
if [[ -e ${daemon} ]];then
INFO Backup ${daemon} success!
mv ${daemon} ${daemon}.bak
echo "{\"registry-mirrors\" : [\"https://hub-mirror.c.163.com\"]}" > ${daemon}
else
echo "{\"registry-mirrors\" : [\"https://hub-mirror.c.163.com\"]}" > ${daemon}
fi
INFO Config docker-hub booster success!
}
check_env()
{
if [[ -z "$(which docker)" ]]
then
WARN No docker were found,try to install!
INFO Start to install docker
source /etc/os-release
if [[ "$ID" == "ubuntu" ]] || [[ "$ID" == "debain" ]]
then
apt update
apt install curl wget -y
curl -fsSL https://get.docker.com | sh
booster
systemctl daemon-reload
systemctl restart docker
if [[ -z "$(which java)" ]];then
apt install openjdk-8-jdk -y
fi
elif [[ "$ID" == "centos" ]]
then
yum update -y
yum install wget curl net-tools -y
curl -fsSL https://get.docker.com | sh
booster
systemctl daemon-reload
systemctl restart docker
if [[ -z "$(which java)" ]];then
yum install java-1.8.0-openjdk -y
fi
else
ERROR Could not support $ID platform!
exit 1
fi
fi
}
install_elasticsearch()
{
INFO Start to install elasticsearch
echo "vm.max_map_count=655360" >> /etc/sysctl.conf
sysctl -p
docker pull docker.elastic.co/elasticsearch/elasticsearch:6.5.4
docker run -d --restart=always -p 9200:9200 -p 9300:9300 --name es -h es -e cluster.name=kiki -e node.name=node1 -e http.cors.enabled=true -e http.cors.allow-origin="*" -e xpack.security.enabled=false docker.elastic.co/elasticsearch/elasticsearch:6.5.4
}
install_kibana()
{
INFO Start to install kibana
docker pull kibana:6.5.4;
docker run --restart=always -p 5601:5601 --name kibana -e ELASTICSEARCH_URL=http://127.0.0.1:9200 --network=host -d kibana:6.5.4
}
install_filebeat_and_logstash()
{
INFO Start to install filebeat and logstash
source /etc/os-release
if [[ "$ID" == "ubuntu" ]] || [[ "$ID" == "debain" ]]
then
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -
apt-get install apt-transport-https -y
echo "deb https://artifacts.elastic.co/packages/6.x/apt stable main" | sudo tee -a /etc/apt/sources.list.d/elastic-6.x.list
apt-get update && apt install filebeat logstash -y
if [[ $? -ne 0 ]]
then
ERROR Install filebeat and logstash failed!
exit 1
fi
elif [[ "$ID" == "centos" ]]
then
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
echo -e '
[elastic-6.x]
name=Elastic repository for 6.x packages
baseurl=https://artifacts.elastic.co/packages/6.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
' > /etc/yum.repos.d/elastic.io.repo
yum makecache && yum install filebeat logstash -y
if [[ $? -ne 0 ]]
then
ERROR Install filebeat and logstash failed!
exit 1
fi
else
ERROR Could not support $ID platform!
exit 1
fi
}
start_filebeat()
{
INFO Start to call filebeat
filebeat_yaml="/etc/filebeat/filebeat.yml"
if [[ -f ${filebeat_yaml} ]];then
mv ${filebeat_yaml} ${filebeat_yaml}.bak
cp ./filebeat.yml ${filebeat_yaml}
systemctl restart filebeat
if [[ $? -ne 0 ]]
then
ERROR Start filebeat failed!
exit 1
fi
else
ERROR ${filebeat_yaml} not found, please check!
exit 1
fi
}
start_logstash()
{
INFO Start to call logstash
logstash_conf_dir="/etc/logstash/conf.d/"
if [[ -e ${logstash_conf_dir} ]];then
cp ./slow_sql_by_query.conf ${logstash_conf_dir}
systemctl restart logstash
if [[ $? -ne 0 ]]
then
ERROR Start logstash failed!
exit 1
fi
else
ERROR ${logstash_conf_dir} not found, please check!
exit 1
fi
}
run()
{
if [[ "root" == $(whoami) ]]
then
INFO Start to run...
check_env
install_elasticsearch
install_kibana
install_filebeat_and_logstash
# download_log
start_logstash
start_filebeat
# check_index
INFO Run success!
else
ERROR Run as root please!
fi
}
run