Skip to content

Commit

Permalink
*: record previous statement when commit is slow (#11908) (#12724)
Browse files Browse the repository at this point in the history
  • Loading branch information
@jackysp @sre-bot
jackysp authored and sre-bot committed Oct 16, 2019
1 parent d166d19 commit c1c80d6
Show file tree
Hide file tree
Showing 6 changed files with 96 additions and 74 deletions.
67 changes: 32 additions & 35 deletions executor/adapter.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,7 @@ func (a *recordSet) NewChunk() *chunk.Chunk {
func (a *recordSet) Close() error {
err := a.executor.Close()
a.stmt.LogSlowQuery(a.txnStartTS, a.lastErr == nil)
a.stmt.Ctx.GetSessionVars().PrevStmt = a.stmt.OriginText()
a.stmt.logAudit()
return errors.Trace(err)
}
Expand Down Expand Up @@ -393,6 +394,16 @@ func (a *ExecStmt) logAudit() {
}
}

// FormatSQL is used to format the original SQL, e.g. truncating long SQL, appending prepared arguments.
func FormatSQL(sql string, sessVars *variable.SessionVars) string {
cfg := config.GetGlobalConfig()
length := len(sql)
if maxQueryLen := atomic.LoadUint64(&cfg.Log.QueryLogMaxLen); uint64(length) > maxQueryLen {
sql = fmt.Sprintf("%.*q(len:%d)", maxQueryLen, sql, length)
}
return QueryReplacer.Replace(sql) + sessVars.GetExecuteArgumentsInfo()
}

// LogSlowQuery is used to print the slow query in the log files.
func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool) {
level := log.GetLevel()
Expand All @@ -402,12 +413,8 @@ func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool) {
if costTime < threshold && level > zapcore.DebugLevel {
return
}
sql := a.Text
if maxQueryLen := atomic.LoadUint64(&cfg.Log.QueryLogMaxLen); uint64(len(sql)) > maxQueryLen {
sql = fmt.Sprintf("%.*q(len:%d)", maxQueryLen, sql, len(a.Text))
}
sessVars := a.Ctx.GetSessionVars()
sql = QueryReplacer.Replace(sql) + sessVars.GetExecuteArgumentsInfo()
sql := FormatSQL(a.Text, sessVars)

var tableIDs, indexNames string
if len(sessVars.StmtCtx.TableIDs) > 0 {
Expand All @@ -420,38 +427,28 @@ func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool) {
copTaskInfo := sessVars.StmtCtx.CopTasksDetails()
statsInfos := a.getStatsInfo()
memMax := sessVars.StmtCtx.MemTracker.MaxConsumed()
_, digest := sessVars.StmtCtx.SQLDigest()
slowItems := &variable.SlowQueryLogItems{
TxnTS: txnTS,
SQL: sql,
Digest: digest,
TimeTotal: costTime,
TimeParse: a.Ctx.GetSessionVars().DurationParse,
TimeCompile: a.Ctx.GetSessionVars().DurationCompile,
IndexNames: indexNames,
StatsInfos: statsInfos,
CopTasks: copTaskInfo,
ExecDetail: execDetail,
MemMax: memMax,
Succ: succ,
}
if _, ok := a.StmtNode.(*ast.CommitStmt); ok {
slowItems.PrevStmt = FormatSQL(sessVars.PrevStmt, sessVars)
}
if costTime < threshold {
_, digest := sessVars.StmtCtx.SQLDigest()
logutil.SlowQueryLogger.Debug(sessVars.SlowLogFormat(&variable.SlowQueryLogItems{
TxnTS: txnTS,
SQL: sql,
Digest: digest,
TimeTotal: costTime,
TimeParse: a.Ctx.GetSessionVars().DurationParse,
TimeCompile: a.Ctx.GetSessionVars().DurationCompile,
IndexNames: indexNames,
StatsInfos: statsInfos,
CopTasks: copTaskInfo,
ExecDetail: execDetail,
MemMax: memMax,
Succ: succ,
}))
logutil.SlowQueryLogger.Debug(sessVars.SlowLogFormat(slowItems))
} else {
_, digest := sessVars.StmtCtx.SQLDigest()
logutil.SlowQueryLogger.Warn(sessVars.SlowLogFormat(&variable.SlowQueryLogItems{
TxnTS: txnTS,
SQL: sql,
Digest: digest,
TimeTotal: costTime,
TimeParse: a.Ctx.GetSessionVars().DurationParse,
TimeCompile: a.Ctx.GetSessionVars().DurationCompile,
IndexNames: indexNames,
StatsInfos: statsInfos,
CopTasks: copTaskInfo,
ExecDetail: execDetail,
MemMax: memMax,
Succ: succ,
}))
logutil.SlowQueryLogger.Warn(sessVars.SlowLogFormat(slowItems))
metrics.TotalQueryProcHistogram.Observe(costTime.Seconds())
metrics.TotalCopProcHistogram.Observe(execDetail.ProcessTime.Seconds())
metrics.TotalCopWaitHistogram.Observe(execDetail.WaitTime.Seconds())
Expand Down
78 changes: 44 additions & 34 deletions infoschema/slow_log.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ var slowQueryCols = []columnInfo{
{variable.SlowLogCopWaitMax, mysql.TypeDouble, 22, 0, nil, nil},
{variable.SlowLogMemMax, mysql.TypeLonglong, 20, 0, nil, nil},
{variable.SlowLogSucc, mysql.TypeTiny, 1, 0, nil, nil},
{variable.SlowLogPrevStmt, mysql.TypeLongBlob, types.UnspecifiedLength, 0, nil, nil},
{variable.SlowLogQuerySQLStr, mysql.TypeLongBlob, types.UnspecifiedLength, 0, nil, nil},
}

Expand Down Expand Up @@ -110,15 +111,19 @@ func ParseSlowLog(tz *time.Location, reader *bufio.Reader) ([][]types.Datum, err
// Parse slow log field.
if strings.HasPrefix(line, variable.SlowLogPrefixStr) {
line = line[len(variable.SlowLogPrefixStr):]
fieldValues := strings.Split(line, " ")
for i := 0; i < len(fieldValues)-1; i += 2 {
field := fieldValues[i]
if strings.HasSuffix(field, ":") {
field = field[:len(field)-1]
}
err = st.setFieldValue(tz, field, fieldValues[i+1])
if err != nil {
return rows, err
if strings.HasPrefix(line, variable.SlowLogPrevStmtPrefix) {
st.prevStmt = line[len(variable.SlowLogPrevStmtPrefix):]
} else {
fieldValues := strings.Split(line, " ")
for i := 0; i < len(fieldValues)-1; i += 2 {
field := fieldValues[i]
if strings.HasSuffix(field, ":") {
field = field[:len(field)-1]
}
err = st.setFieldValue(tz, field, fieldValues[i+1])
if err != nil {
return rows, err
}
}
}
} else if strings.HasSuffix(line, variable.SlowLogSQLSuffixStr) {
Expand Down Expand Up @@ -168,31 +173,35 @@ func getOneLine(reader *bufio.Reader) ([]byte, error) {
}

type slowQueryTuple struct {
time time.Time
txnStartTs uint64
user string
connID uint64
queryTime float64
processTime float64
waitTime float64
backOffTime float64
requestCount uint64
totalKeys uint64
processKeys uint64
db string
indexIDs string
isInternal bool
digest string
statsInfo string
avgProcessTime float64
p90ProcessTime float64
maxProcessTime float64
avgWaitTime float64
p90WaitTime float64
maxWaitTime float64
memMax int64
succ bool
sql string
time time.Time
txnStartTs uint64
user string
host string
connID uint64
queryTime float64
processTime float64
waitTime float64
backOffTime float64
requestCount uint64
totalKeys uint64
processKeys uint64
db string
indexIDs string
digest string
statsInfo string
avgProcessTime float64
p90ProcessTime float64
maxProcessTime float64
maxProcessAddress string
avgWaitTime float64
p90WaitTime float64
maxWaitTime float64
maxWaitAddress string
memMax int64
prevStmt string
sql string
isInternal bool
succ bool
}

func (st *slowQueryTuple) setFieldValue(tz *time.Location, field, value string) error {
Expand Down Expand Up @@ -295,6 +304,7 @@ func (st *slowQueryTuple) convertToDatumRow() []types.Datum {
} else {
record = append(record, types.NewIntDatum(0))
}
record = append(record, types.NewStringDatum(st.prevStmt))
record = append(record, types.NewStringDatum(st.sql))
return record
}
Expand Down
4 changes: 2 additions & 2 deletions infoschema/slow_log_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ func (s *testSuite) TestParseSlowLogFile(c *C) {
# Cop_wait_avg: 0.05 Cop_wait_p90: 0.6 Cop_wait_max: 0.8
# Mem_max: 70724
# Succ: false
# Prev_stmt: update t set i = 1;
select * from t;`)
reader := bufio.NewReader(slowLog)
loc, err := time.LoadLocation("Asia/Shanghai")
Expand All @@ -54,8 +55,7 @@ select * from t;`)
}
recordString += str
}
expectRecordString := "2019-04-28 15:24:04.309074,405888132465033227,,0,0.216905,0.021,0,0,1,637,0,,,1,42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772,t1:1,t2:2,0.1,0.2,0.03,0.05,0.6,0.8,70724,0,select * from t;"

expectRecordString := "2019-04-28 15:24:04.309074,405888132465033227,,0,0.216905,0.021,0,0,1,637,0,,,1,42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772,t1:1,t2:2,0.1,0.2,0.03,0.05,0.6,0.8,70724,0,update t set i = 1;,select * from t;"
c.Assert(expectRecordString, Equals, recordString)

// fix sql contain '# ' bug
Expand Down
5 changes: 3 additions & 2 deletions infoschema/tables_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -505,6 +505,7 @@ func (s *testSuite) TestSlowQuery(c *C) {
# Cop_wait_avg: 0.05 Cop_wait_p90: 0.6 Cop_wait_max: 0.8
# Mem_max: 70724
# Succ: true
# Prev_stmt: update t set i = 2;
select * from t_slim;`))
c.Assert(f.Sync(), IsNil)
c.Assert(err, IsNil)
Expand All @@ -513,10 +514,10 @@ select * from t_slim;`))
tk.MustExec("set time_zone = '+08:00';")
re := tk.MustQuery("select * from information_schema.slow_query")
re.Check(testutil.RowsWithSep("|",
"2019-02-12 19:33:56.571953|406315658548871171|root@127.0.0.1|6|4.895492|0.161|0.101|0.092|1|100001|100000|test||0|42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772|t1:1,t2:2|0.1|0.2|0.03|0.05|0.6|0.8|70724|1|select * from t_slim;"))
"2019-02-12 19:33:56.571953|406315658548871171|root@127.0.0.1|6|4.895492|0.161|0.101|0.092|1|100001|100000|test||0|42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772|t1:1,t2:2|0.1|0.2|0.03|0.05|0.6|0.8|70724|1|update t set i = 2;|select * from t_slim;"))
tk.MustExec("set time_zone = '+00:00';")
re = tk.MustQuery("select * from information_schema.slow_query")
re.Check(testutil.RowsWithSep("|", "2019-02-12 11:33:56.571953|406315658548871171|root@127.0.0.1|6|4.895492|0.161|0.101|0.092|1|100001|100000|test||0|42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772|t1:1,t2:2|0.1|0.2|0.03|0.05|0.6|0.8|70724|1|select * from t_slim;"))
re.Check(testutil.RowsWithSep("|", "2019-02-12 11:33:56.571953|406315658548871171|root@127.0.0.1|6|4.895492|0.161|0.101|0.092|1|100001|100000|test||0|42a1c8aae6f133e934d4bf0147491709a8812ea05ff8819ec522780fe657b772|t1:1,t2:2|0.1|0.2|0.03|0.05|0.6|0.8|70724|1|update t set i = 2;|select * from t_slim;"))

// Test for long query.
_, err = f.Write([]byte(`
Expand Down
1 change: 1 addition & 0 deletions session/tidb.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,7 @@ func runStmt(ctx context.Context, sctx sessionctx.Context, s sqlexec.Statement)
// then it could include the transaction commit time.
if rs == nil {
s.(*executor.ExecStmt).LogSlowQuery(origTxnCtx.StartTS, err == nil)
sessVars.PrevStmt = s.OriginText()
}
}()

Expand Down
15 changes: 14 additions & 1 deletion sessionctx/variable/session.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -343,11 +343,14 @@ type SessionVars struct {
// StartTime is the start time of the last query.
StartTime time.Time

// DurationParse is the duration of pasing SQL string to AST of the last query.
// DurationParse is the duration of parsing SQL string to AST of the last query.
DurationParse time.Duration

// DurationCompile is the duration of compiling AST to execution plan of the last query.
DurationCompile time.Duration

// PrevStmt is used to store the previous executed statement in the current session.
PrevStmt string
}

// ConnectionInfo present connection used by audit.
Expand Down Expand Up @@ -853,6 +856,10 @@ const (
SlowLogMemMax = "Mem_max"
// SlowLogSucc is used to indicate whether this sql execute successfully.
SlowLogSucc = "Succ"
// SlowLogPrevStmt is used to show the previous executed statement.
SlowLogPrevStmt = "Prev_stmt"
// SlowLogPrevStmtPrefix is the prefix of Prev_stmt in slow log file.
SlowLogPrevStmtPrefix = SlowLogPrevStmt + SlowLogSpaceMarkStr
)

// SlowQueryLogItems is a collection of items that should be included in the
Expand All @@ -870,6 +877,7 @@ type SlowQueryLogItems struct {
ExecDetail execdetails.ExecDetails
MemMax int64
Succ bool
PrevStmt string
}

// SlowLogFormat uses for formatting slow log.
Expand All @@ -890,6 +898,7 @@ type SlowQueryLogItems struct {
// # Cop_wait: Avg_time: 10ms P90_time: 20ms Max_time: 30ms Max_Addr: 10.6.131.79
// # Memory_max: 4096
// # Succ: true
// # Prev_stmt: begin;
// select * from t_slim;
func (s *SessionVars) SlowLogFormat(logItems *SlowQueryLogItems) string {
var buf bytes.Buffer
Expand Down Expand Up @@ -955,6 +964,10 @@ func (s *SessionVars) SlowLogFormat(logItems *SlowQueryLogItems) string {

writeSlowLogItem(&buf, SlowLogSucc, strconv.FormatBool(logItems.Succ))

if logItems.PrevStmt != "" {
writeSlowLogItem(&buf, SlowLogPrevStmt, logItems.PrevStmt)
}

buf.WriteString(logItems.SQL)
if len(logItems.SQL) == 0 || logItems.SQL[len(logItems.SQL)-1] != ';' {
buf.WriteString(";")
Expand Down

0 comments on commit c1c80d6

Please sign in to comment.