总体架构简介
SQL审计架构.png
1. mysql-sniffer监听mysql实例端口，解析出客户端ip，用户名，数据库，执行时间，返回行数，sql文本
2. 写入到指定日志文件，可按固定时间进行切割，控制sql文本大小
3. logstash按照正则表达式监听解析logfile
4. logstash将监听到的文件变更内容写入到es，按天进行索引分割
5. 可通过查询统计es进行慢sql或者sql审计
MySQL Sniffer
简介
360开源的一款工具，基于mysql端口，解析出需要的客户端ip，数据库，数据库用户，执行时间，返回行数，sql文本等，可再后台运行，将结果输出到指定日志目录中，同时可按照指定时间进行日志切割。
项目地址
安装(centos)
yum -y install glib2-devel libpcap-devel libnet-devel cmake
git clone https://github.com/Qihoo360/mysql-sniffer.git
cd mysql-sniffer
mkdir proj
cd proj
cmake ../
make
cd bin/
启动及参数说明
mysql-sniffer -d -s 60 -t 2000 -i eth0 -p 3306 -l /var/log/mysql-sniffer/
-d 后台运行
-s 每隔多少分钟进行日志切割
-t 截取sql字符长度
-i 网卡
-p 实例端口
-l 指定输出目录，默认以端口.log命名
-n 指定数据库包个数
输出日志
2017-10-17 19:22:51 atlas 192.168.3.46 mysql_metadb 0ms 1 select @@version_comment limit 1
2017-10-17 19:22:51 atlas 192.168.3.46 mysql_metadb 0ms 5 select * from backup_info
2017-10-17 19:22:52 atlas 192.168.3.46 mysql_metadb 0ms 1 select @@version_comment limit 1
2017-10-17 19:22:52 atlas 192.168.3.46 mysql_metadb 0ms 5 select * from backup_info
Logstash
简介
logstash就是一根具备实时数据传输能力的管道，负责将数据信息从管道的输入端传输到管道的输出端；与此同时这根管道还可以让你根据自己的需求在中间加上滤网，Logstash提供里很多功能强大的滤网以满足你的各种应用场景。
项目地址
配置文件
自定义grok文件/usr/local/logstash/config/patterns/mysql
DATECHI %{MONTHNUM}[/-]%{MONTHDAY}[/-]%{YEAR}\s%{HOUR}[/:]%{MINUTE}[/:]%{SECOND}
SQLLOG %{DATECHI:logdate}\t\s+%{WORD:username}\t\s+%{IPV4:clientip}\t\s+%{WORD:dbname}\t\s+%{INT:excutetime}ms\t\s+%{INT:fetchrows}\t\s+%{GREEDYDATA:sqltext}
/usr/local/logstash/config/mysql_sql.yml
input {
file {
type => "mysql_sql_file"
path => "/var/log/mysql-sniffer/3306.log"
sincedb_path => "/usr/local/logstash/logs/sincedb.pos"
start_position => beginning
}
}
filter {
if [type] == "mysql_sql_file" {
grok {
patterns_dir => "/usr/local/logstash/config/patterns/mysql"
match => {
"message" => "%{SQLLOG}"
}
add_field => [ "port", "3306" ]
add_field => [ "dbip", "192.168.3.45" ]
}
}
}
output {
elasticsearch {
hosts => "192.168.1.31"
user => "root"
password => "JreOJVK0O0S2zU5SIMAGn33"
index => "mysql_general_log-%{+YYYY.MM.dd}"
}
}
启动
/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/mysql_sql.yml &
kibana查询结果
image.png
定时从es查询慢sql，用于zabbix监控
class ElasticSearchClass:
def __init__(self, host, port, user, passwrod):
self.host = host
self.port = port
self.user = user
self.password = passwrod
self.connect()
def connect(self):
self.es = Elasticsearch(hosts=[{'host': self.host, 'port': self.port}],
http_auth=(self.user, self.password))
def count(self,db_host, db_port):
query_json = {
'query':{
'filtered':{
'filter':{
'bool':{
'must':[
{
'range':{
'excutiontime':{
'gt': 500
}
}
},
{
'range':{
'@timestamp':{
'from': 'now-1m',
'to': 'now'
}
}
},
{
'term':{
'dbip': db_host
}
},
{
'term':{
'port': db_port
}
}
]
}
}
}
}
}
logdate = str(time.strftime("%Y.%m.%d", time.localtime()))
idx_name = 'mysql_general_log-' + logdate
slowsql_count = self.es.count(index=idx_name, body=query_json)['count']
items = {'slq': slowsql_count}
return items
