收集mysql、MSSQL和mongodb指定表的log

背景说明:

需要采集mysql、sql Server和mongodb指定表的相关操作记录并加以分析，从而形成对数据变化的监测；

实现原理:

mysql则通过解析binlog并过滤指定的表Mssq存储过程，并将结果保存到表中。

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 && 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)

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) 创建计划任务，形成自动监控。

（编辑：广州站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!