7x31
/
sniffer-agent
mirror of https://github.com/zr-hebo/sniffer-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix bug of lost user info and use db stmt

This commit is contained in:
hebo 2019-08-30 12:45:04 +08:00
parent a09af431ee
commit 10b08dec52
1 changed files with 49 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
session-dealer/mysql/session.go
View File

@ -6,8 +6,8 @@ import (
"sync" "sync"
"time" "time"
"github.com/siddontang/go/hack"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/pingcap/tidb/util/hack"
"github.com/zr-hebo/sniffer-agent/model" "github.com/zr-hebo/sniffer-agent/model"
) )
@ -269,8 +269,7 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
var mqp *model.PooledMysqlQueryPiece var mqp *model.PooledMysqlQueryPiece
var querySQLInBytes []byte var querySQLInBytes []byte
switch ms.cachedStmtBytes[0] { if ms.cachedStmtBytes[0] > 32 {
case ComAuth:
var userName, dbName string var userName, dbName string
var err error var err error
userName, dbName, err = parseAuthInfo(ms.cachedStmtBytes) userName, dbName, err = parseAuthInfo(ms.cachedStmtBytes)
@ -280,51 +279,56 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
ms.visitUser = &userName ms.visitUser = &userName
ms.visitDB = &dbName ms.visitDB = &dbName
case ComInitDB: } else {
newDBName := string(ms.cachedStmtBytes[1:]) switch ms.cachedStmtBytes[0] {
useSQL := fmt.Sprintf("use %s", newDBName) case ComInitDB:
mqp = ms.composeQueryPiece() newDBName := string(ms.cachedStmtBytes[1:])
mqp.QuerySQL = &useSQL useSQL := fmt.Sprintf("use %s", newDBName)
// update session database querySQLInBytes = hack.Slice(useSQL)
ms.visitDB = &newDBName mqp = ms.composeQueryPiece()
mqp.QuerySQL = &useSQL
// update session database
ms.visitDB = &newDBName
case ComDropDB: case ComDropDB:
dbName := string(ms.cachedStmtBytes[1:]) dbName := string(ms.cachedStmtBytes[1:])
dropSQL := fmt.Sprintf("drop database %s", dbName) dropSQL := fmt.Sprintf("drop database %s", dbName)
mqp = ms.composeQueryPiece() mqp = ms.composeQueryPiece()
mqp.QuerySQL = &dropSQL mqp.QuerySQL = &dropSQL
case ComCreateDB, ComQuery: case ComCreateDB, ComQuery:
mqp = ms.composeQueryPiece() mqp = ms.composeQueryPiece()
querySQLInBytes = ms.cachedStmtBytes[1:] querySQLInBytes = ms.cachedStmtBytes[1:]
querySQL := hack.String(querySQLInBytes) querySQL := hack.String(querySQLInBytes)
mqp.QuerySQL = &querySQL mqp.QuerySQL = &querySQL
case ComStmtPrepare: case ComStmtPrepare:
mqp = ms.composeQueryPiece() mqp = ms.composeQueryPiece()
querySQLInBytes = ms.cachedStmtBytes[1:] querySQLInBytes = ms.cachedStmtBytes[1:]
querySQL := hack.String(querySQLInBytes) querySQL := hack.String(querySQLInBytes)
mqp.QuerySQL = &querySQL mqp.QuerySQL = &querySQL
ms.cachedPrepareStmt[ms.prepareInfo.prepareStmtID] = querySQLInBytes ms.cachedPrepareStmt[ms.prepareInfo.prepareStmtID] = querySQLInBytes
log.Debugf("prepare statement %s, get id:%d", querySQL, ms.prepareInfo.prepareStmtID) log.Debugf("prepare statement %s, get id:%d", querySQL, ms.prepareInfo.prepareStmtID)
case ComStmtExecute: case ComStmtExecute:
prepareStmtID := bytesToInt(ms.cachedStmtBytes[1:5]) prepareStmtID := bytesToInt(ms.cachedStmtBytes[1:5])
mqp = ms.composeQueryPiece() mqp = ms.composeQueryPiece()
querySQLInBytes = ms.cachedPrepareStmt[prepareStmtID] querySQLInBytes = ms.cachedPrepareStmt[prepareStmtID]
querySQL := hack.String(querySQLInBytes) querySQL := hack.String(querySQLInBytes)
mqp.QuerySQL = &querySQL mqp.QuerySQL = &querySQL
log.Debugf("execute prepare statement:%d", prepareStmtID) log.Debugf("execute prepare statement:%d", prepareStmtID)
case ComStmtClose: case ComStmtClose:
prepareStmtID := bytesToInt(ms.cachedStmtBytes[1:5]) prepareStmtID := bytesToInt(ms.cachedStmtBytes[1:5])
delete(ms.cachedPrepareStmt, prepareStmtID) delete(ms.cachedPrepareStmt, prepareStmtID)
log.Debugf("remove prepare statement:%d", prepareStmtID) log.Debugf("remove prepare statement:%d", prepareStmtID)
default: default:
return return
}
} }
if strictMode && mqp != nil && mqp.VisitUser == nil { if strictMode && mqp != nil && mqp.VisitUser == nil {
user, db, err := querySessionInfo(ms.serverPort, mqp.SessionID) user, db, err := querySessionInfo(ms.serverPort, mqp.SessionID)
if err != nil { if err != nil {
@ -341,10 +345,12 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
func filterQueryPieceBySQL(mqp *model.PooledMysqlQueryPiece, querySQL []byte) (model.QueryPiece) { func filterQueryPieceBySQL(mqp *model.PooledMysqlQueryPiece, querySQL []byte) (model.QueryPiece) {
if mqp == nil || querySQL == nil { if mqp == nil || querySQL == nil {
return nil return nil
} else if (uselessSQLPattern.Match(querySQL)) {
return nil
} }
/*
else if (uselessSQLPattern.Match(querySQL)) {
return nil
}
*/
if ddlPatern.Match(querySQL) { if ddlPatern.Match(querySQL) {
mqp.SetNeedSyncSend(true) mqp.SetNeedSyncSend(true)