背景说明:
需要采集mysql、sql Server和mongodb指定表的相关操作记录并加以分析，从而形成对数据变化的监测；
实现原理:
mysql则通过解析binlog并过滤指定的表，并将结果保存到表中。
Sql Server则是通过以profiler的形式，以sql实现相关功能，跟踪相关的SQL语句并实现过滤和筛选。
Mongodb则是通过对oplog分析并过滤，并将结果持久化操作。
Mysql实现代码:
1)记录表：
create table sqlAudit(
id bigint primary key auto_increment COMMENT '主键自增ID',
stringsql text COMMENT '跟踪到的SQL语句',
postioninfo varchar(500) COMMENT 'postion的开始、结束位置和执行时间',
addtime datetime default now() COMMENT '添加时间'
)
2)下载并安装工具:
shell> git clone https://github.com/danfengcao/binlog2sql.git && cd binlog2sql
shell> pip install -r requirements.txt
3)Shell文件getbinlog.sh的脚本:
#!/bin/bash
HOSTNAME="10.12.11.34"
PORT=30001
USERNAME="root_user"
PASSWORD="//@#Y3M2T1pwd"
select_sql="show master status;"
result=`mysql -h${HOSTNAME} -P${PORT} -u${USERNAME} -p${PASSWORD} ${DBNAME} -e "${select_sql}" | awk 'NR>1'`
#echo `mysql -h${HOSTNAME} -P${PORT} -u${USERNAME} -p${PASSWORD} ${DBNAME} -e "${select_sql}" | awk 'NR>1'`
binlog=`echo $result | awk '{ print $1 }'`
postion=`echo $result | awk '{ print $2 }'`
bin=`echo $binlog|sed 's/.*\(...\)$/\1/'`
intbin=`echo $bin | awk '{print int($0)}'`
binsmall=$[$intbin-1]
lenbin=`echo $binsmall |wc -L`
if [ $lenbin == 2 ];then
sbinlog=${binlog/$bin/"0"$binsmall}
else
sbinlog=${binlog/$bin/$binsmall}
fi
#sbinlog=${binlog/$bin/$binsmall}
echo $sbinlog
if [ ! -f "/$sbinlog" ];then
python ./binlog2sql.py -h$HOSTNAME -P$PORT -u$USERNAME -p$PASSWORD -t filemd5 --start-file=$sbinlog>/$sbinlog
suser="root_user"
spassword="//@#Y3M2T1pwd"
shost="monitor.db.ymatou.com"
sport="30001"
mysql_conn="mysql -h"$shost" -P$sport -u"$suser" -p"$spassword""
cat /$sbinlog | while read stringsql
do
#OIFS=$IFS; IFS="#"; set -- $stringsql; aa=$1;bb=$2; IFS=$OIFS
stringsql1=`echo $stringsql | awk -F"; #" '{print $1}'`
stringsql2=`echo $stringsql | awk -F"; #" '{print $2}'`
sqlstring=`echo $stringsql1 | sed #39;s/\'/\'\'/g'`
#echo $sqlstring
$mysql_conn -e "INSERT INTO dbmonitor.sqlAudit(ip,dbport,stringsql,postioninfo) values('$HOSTNAME',$PORT,'$sqlstring','$stringsql2')"
done
else
echo 'exists'
fi
说明:红色代码部分是需要采集的mysql服务器IP地址、端口及指定的表名; 部署完毕后则采集到的时候写入到sqlAudit表中
Sql Server实现代码:
1) 创建分割函数
CREATE FUNCTION dbo.Split(@String varchar(8000), @Delimiter char(1))
returns @temptable TABLE (items varchar(8000))
as
begin
declare @idx int
declare @slice varchar(8000)
select @idx = 1
if len(@String)<1 or @String is null return
while @idx!= 0
begin
set @idx =charindex(@Delimiter,@String)
if @idx!=0
set @slice =left(@String,@idx - 1)
else
set @slice = @String
if(len(@slice)>0)
insert into @temptable(items)values(@slice)
set @String =right(@String,len(@String)- @idx)
if len(@String)= 0 break
end
return
end
2) 创建采集存储过程
USE [dbmanage]
GO
/****** Object: StoredProcedure [dbo].[sp_trace_sql_durtion] Script Date: 2021/1/21 14:43:21 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
Create PROC [dbo].[sp_trace_sql_durtion]
@tablename nvarchar(500),
@FilePath nvarchar(260),
@onoff int
AS
BEGIN
if @onoff=0
begin
declare @int int
declare yb cursor for
SELECT id FROM sys.traces where path like '%'+substring(@FilePath,1,8)+'%'
open yb
fetch next from yb into @int
while @@FETCH_STATUS=0
begin
EXEC sp_trace_setstatus @int,0 --停止, 第一个参数为SELECT * FROM sys.traces中的ID列
EXEC sp_trace_setstatus @int,2 --删除
fetch next from yb into @int
end
close yb
deallocate yb
end
else
begin
DECLARE @rc int,@TraceID int,@MaxFileSize bigint;
SET @MaxFileSize = 50;
declare @date nvarchar(10)
set @date=replace(convert(nvarchar(10),getdate(),120),'-','')
--select @date
set @FilePath=@FilePath+'trc'+@date
declare @filename nvarchar(100)
declare @FilePathnew nvarchar(500)
set @filename=@FilePath+'.trc'
set @FilePathnew=@FilePath
declare @result int =0
--select @filename
execute master.[sys].[xp_fileexist] @filename,@result output
if @result =1
begin
--select 'aaa'
set @FilePathnew=@FilePath+'_a'
select @FilePathnew
end
EXEC sp_trace_create @TraceID OUTPUT,2,@FilePathnew,@MaxFileSize,NULL;
IF @rc != 0
RETURN;
DECLARE @On bit;
SET @On = 1;
EXEC sp_trace_setevent @TraceID,10,35,@On;
EXEC sp_trace_setevent @TraceID,10,1,@On;
EXEC sp_trace_setevent @TraceID,10,13,@On;
EXEC sp_trace_setevent @TraceID,10,14,@On;
EXEC sp_trace_setevent @TraceID,10,15,@On;
EXEC sp_trace_setevent @TraceID,10,11,@On;
EXEC sp_trace_setevent @TraceID,10,8,@On;
EXEC sp_trace_setevent @TraceID,10,18,@On;
EXEC sp_trace_setevent @TraceID,41,35,@On;
EXEC sp_trace_setevent @TraceID,41,1,@On;
EXEC sp_trace_setevent @TraceID,41,13,@On;
EXEC sp_trace_setevent @TraceID,41,14,@On;
EXEC sp_trace_setevent @TraceID,41,15,@On;
EXEC sp_trace_setevent @TraceID,41,11,@On;
EXEC sp_trace_setevent @TraceID,41,8,@On;
EXEC sp_trace_setevent @TraceID,41,18,@On;
--SET @Seconds = @Seconds * 1000000;
--EXEC sp_trace_setfilter @TraceID,13,0,4,@Seconds;
IF @tablename IS NOT NULL
declare @items nvarchar(500)
declare yb2 cursor for
select items from master.dbo.split(@tablename,',')
open yb2
fetch next from yb2 into @items
while @@FETCH_STATUS=0
begin
--EXEC sp_trace_setfilter @TraceID,35,0,0,@DatabaseName
set @items='%'+@items+'%'
EXEC sp_trace_setfilter @TraceID,1,1,6,@items
--select @items
fetch next from yb2 into @items
end
close yb2
deallocate yb2
EXEC sp_trace_setfilter @TraceID,1,0,7,N'%select%'
EXEC sp_trace_setstatus @TraceID,1
SELECT TraceID = @TraceID;
end
END
存储过程参数说明:
@tablename:需监控表的名称，多个表用,隔开
@FilePath:采集文件存储路径
@onoff;采集开关，0代表关闭，1代表开启
exec [sp_trace_sql_durtion2] null,'e:\autditrace\',0 #关闭监控
exec [sp_trace_sql_durtion2] 'test,test2,test3','e:\autditrace\',1 #开启监控记录表test,test2,test3的相关操作记录，并将文件记录到e:\autditrace\目录下面
3) 创建计划任务，形成自动监控。