metrics: add db QPS metric

config/config.go

@@ -61,7 +61,8 @@ type Config struct {
 	TxnLocalLatches  TxnLocalLatches `toml:"txn-local-latches" json:"txn-local-latches"`
 	// Set sys variable lower-case-table-names, ref: https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html.
 	// TODO: We actually only support mode 2, which keeps the original case, but the comparison is case-insensitive.
-	LowerCaseTableNames int `toml:"lower-case-table-names" json:"lower-case-table-names"`
+	LowerCaseTableNames int   `toml:"lower-case-table-names" json:"lower-case-table-names"`
+	DbQPSMetricSwitch   int64 `toml:"db-qps-metric-switch" json:"db-qps-metric-switch"`
 
 	Log                 Log               `toml:"log" json:"log"`
 	Security            Security          `toml:"security" json:"security"`
@@ -282,6 +283,7 @@ var defaultConf = Config{
 		Capacity: 2048000,
 	},
 	LowerCaseTableNames: 2,
+	DbQPSMetricSwitch:   1,
 	Log: Log{
 		Level:  "info",
 		Format: "text",

config/config.toml.example

@@ -35,6 +35,9 @@ token-limit = 1000
 # Valid options: ["log", "cancel"]
 oom-action = "log"
 
+# whether tidb enable db qps metric
+db-qps-metric-switch = 1
+
 # Set the memory quota for a query in bytes. Default: 32GB
 mem-quota-query = 34359738368
 

executor/compiler.go

@@ -52,6 +52,7 @@ func (c *Compiler) Compile(ctx context.Context, stmtNode ast.StmtNode) (*ExecStm
 	}
 
 	CountStmtNode(stmtNode, c.Ctx.GetSessionVars().InRestrictedSQL)
+	DbCountStmtNode(stmtNode, c.Ctx.GetSessionVars().InRestrictedSQL)
 	isExpensive := logExpensiveQuery(stmtNode, finalPlan)
 
 	return &ExecStmt{
@@ -125,6 +126,120 @@ func CountStmtNode(stmtNode ast.StmtNode, inRestrictedSQL bool) {
 	metrics.StmtNodeCounter.WithLabelValues(GetStmtLabel(stmtNode)).Inc()
 }
 
+// DbCountStmtNode records the number of statements with the same type and db.
+func DbCountStmtNode(stmtNode ast.StmtNode, inRestrictedSQL bool) {
+	cfg := config.GetGlobalConfig()
+
+	if inRestrictedSQL || cfg.DbQPSMetricSwitch != 1 {
+		return
+	}
+
+	dbLabels := getStmtDbLabel(stmtNode)
+	typeLabel := GetStmtLabel(stmtNode)
+
+	for _, dbLabel := range dbLabels {
+		metrics.DbStmtNodeCounter.WithLabelValues(dbLabel, typeLabel).Inc()
+	}
+}
+
+func getStmtDbLabel(stmtNode ast.StmtNode) []string {
+	dbLabelMap := make(map[string]int, 0)
+
+	switch x := stmtNode.(type) {
+	case *ast.AlterTableStmt:
+		dbLabel := x.Table.Schema.O
+		dbLabelMap[dbLabel] = 0
+	case *ast.CreateIndexStmt:
+		dbLabel := x.Table.Schema.O
+		dbLabelMap[dbLabel] = 0
+	case *ast.CreateTableStmt:
+		dbLabel := x.Table.Schema.O
+		dbLabelMap[dbLabel] = 0
+	case *ast.InsertStmt:
+		dbLabels := getDbFromResultNode(x.Table.TableRefs)
+		for _, db := range dbLabels {
+			dbLabelMap[db] = 0
+		}
+	case *ast.DropIndexStmt:
+		dbLabel := x.Table.Schema.O
+		dbLabelMap[dbLabel] = 0
+	case *ast.DropTableStmt:
+		tables := x.Tables
+		for _, table := range tables {
+			dbLabel := table.Schema.O
+			if _, ok := dbLabelMap[dbLabel]; !ok {
+				dbLabelMap[dbLabel] = 0
+			}
+		}
+	case *ast.SelectStmt:
+		dbLabels := getDbFromResultNode(x)
+		for _, db := range dbLabels {
+			dbLabelMap[db] = 0
+		}
+	case *ast.UpdateStmt:
+		if x.TableRefs != nil {
+			dbLabels := getDbFromResultNode(x.TableRefs.TableRefs)
+			for _, db := range dbLabels {
+				dbLabelMap[db] = 0
+			}
+		}
+	case *ast.DeleteStmt:
+		if x.TableRefs != nil {
+			dbLabels := getDbFromResultNode(x.TableRefs.TableRefs)
+			for _, db := range dbLabels {
+				dbLabelMap[db] = 0
+			}
+		}
+	}
+
+	dbLabels := make([]string, 0)
+
+	for k := range dbLabelMap {
+		dbLabels = append(dbLabels, k)
+	}
+
+	return dbLabels
+}
+
+func getDbFromResultNode(resultNode ast.ResultSetNode) []string { //may have duplicate db name
+	var dbLabels []string
+
+	if resultNode == nil {
+		return dbLabels
+	}
+
+	switch x := resultNode.(type) {
+	case *ast.TableSource:
+		return getDbFromResultNode(x.Source)
+	case *ast.SelectStmt:
+		if x.From != nil {
+			return getDbFromResultNode(x.From.TableRefs)
+		}
+	case *ast.TableName:
+		dbLabels = append(dbLabels, x.DBInfo.Name.O)
+	case *ast.Join:
+		if x.Left != nil {
+			dbs := getDbFromResultNode(x.Left)
+			if dbs != nil {
+				for _, db := range dbs {
+					dbLabels = append(dbLabels, db)
+				}
+			}
+		}
+
+		if x.Right != nil {
+			dbs := getDbFromResultNode(x.Right)
+			if dbs != nil {
+				for _, db := range dbs {
+					dbLabels = append(dbLabels, db)
+				}
+			}
+		}
+	}
+
+	return dbLabels
+}
+
 // GetStmtLabel generates a label for a statement.
 func GetStmtLabel(stmtNode ast.StmtNode) string {
 	switch x := stmtNode.(type) {

executor/prepared.go

@@ -225,6 +225,7 @@ func (e *ExecuteExec) Build() error {
 	}
 	e.stmtExec = stmtExec
 	CountStmtNode(e.stmt, e.ctx.GetSessionVars().InRestrictedSQL)
+	DbCountStmtNode(e.stmt, e.ctx.GetSessionVars().InRestrictedSQL)
 	logExpensiveQuery(e.stmt, e.plan)
 	return nil
 }

executor/set_test.go

@@ -259,6 +259,13 @@ func (s *testSuite2) TestSetVar(c *C) {
 	_, err = tk.Exec("set global tidb_slow_log_threshold = 0")
 	c.Assert(err, NotNil)
 
+	tk.MustExec("set tidb_enable_db_qps_metric = 1")
+	tk.MustQuery(`select @@session.tidb_enable_db_qps_metric;`).Check(testkit.Rows("1"))
+	tk.MustExec("set tidb_enable_db_qps_metric = 0")
+	tk.MustQuery(`select @@session.tidb_enable_db_qps_metric;`).Check(testkit.Rows("0"))
+	_, err = tk.Exec("set global tidb_enable_db_qps_metric = 0")
+	c.Assert(err, NotNil)
+
 	tk.MustExec("set tidb_query_log_max_len = 0")
 	tk.MustQuery("select @@session.tidb_query_log_max_len;").Check(testkit.Rows("0"))
 	tk.MustExec("set tidb_query_log_max_len = 20")

go.mod

@@ -84,7 +84,6 @@ require (
 	google.golang.org/genproto v0.0.0-20190108161440-ae2f86662275 // indirect
 	google.golang.org/grpc v1.17.0
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
-	gopkg.in/stretchr/testify.v1 v1.2.2 // indirect
 	sourcegraph.com/sourcegraph/appdash v0.0.0-20180531100431-4c381bd170b4
 	sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
 )

go.sum

@@ -218,8 +218,6 @@ gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8=
 gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
-gopkg.in/stretchr/testify.v1 v1.2.2 h1:yhQC6Uy5CqibAIlk1wlusa/MJ3iAN49/BsR/dCCKz3M=
-gopkg.in/stretchr/testify.v1 v1.2.2/go.mod h1:QI5V/q6UbPmuhtm10CaFZxED9NreB8PnFYN9JcR6TxU=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=

metrics/executor.go

@@ -36,4 +36,13 @@ var (
 			Name:      "statement_total",
 			Help:      "Counter of StmtNode.",
 		}, []string{LblType})
+
+	// DbStmtNodeCounter records the number of statement with the same type and db.
+	DbStmtNodeCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "tidb",
+			Subsystem: "executor",
+			Name:      "statement_db_total",
+			Help:      "Counter of db StmtNode.",
+		}, []string{LblDb, LblType})
 )

metrics/metrics.go

@@ -98,6 +98,7 @@ func RegisterMetrics() {
 	prometheus.MustRegister(StatementPerTransaction)
 	prometheus.MustRegister(StatsInaccuracyRate)
 	prometheus.MustRegister(StmtNodeCounter)
+	prometheus.MustRegister(DbStmtNodeCounter)
 	prometheus.MustRegister(StoreQueryFeedbackCounter)
 	prometheus.MustRegister(TiKVBackoffCounter)
 	prometheus.MustRegister(TiKVBackoffHistogram)

metrics/session.go

@@ -106,6 +106,7 @@ const (
 	LblError       = "error"
 	LblRollback    = "rollback"
 	LblType        = "type"
+	LblDb          = "db"
 	LblResult      = "result"
 	LblSQLType     = "sql_type"
 	LblGeneral     = "general"

sessionctx/variable/session.go

@@ -286,6 +286,9 @@ type SessionVars struct {
 	// SkipUTF8Check check on input value.
 	SkipUTF8Check bool
 
+	// TiDBEnableDbQPSMetric indicates if we should record the dp qps metric data.
+	TiDBDbQPSMetricSwitch bool
+
 	// BatchInsert indicates if we should split insert data into multiple batches.
 	BatchInsert bool
 
@@ -667,6 +670,8 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		atomic.StoreUint32(&ProcessGeneralLog, uint32(tidbOptPositiveInt32(val, DefTiDBGeneralLog)))
 	case TiDBSlowLogThreshold:
 		atomic.StoreUint64(&config.GetGlobalConfig().Log.SlowThreshold, uint64(tidbOptInt64(val, logutil.DefaultSlowThreshold)))
+	case TiDBEnableDbQPSMetric:
+		atomic.StoreInt64(&config.GetGlobalConfig().DbQPSMetricSwitch, tidbOptInt64(val, DefEnableDbQPSMetric))
 	case TiDBQueryLogMaxLen:
 		atomic.StoreUint64(&config.GetGlobalConfig().Log.QueryLogMaxLen, uint64(tidbOptInt64(val, logutil.DefaultQueryLogMaxLen)))
 	case TiDBRetryLimit:

sessionctx/variable/sysvar.go

@@ -672,6 +672,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TiDBOptimizerSelectivityLevel, strconv.Itoa(DefTiDBOptimizerSelectivityLevel)},
 	{ScopeGlobal | ScopeSession, TiDBEnableWindowFunction, boolToIntStr(DefEnableWindowFunction)},
 	/* The following variable is defined as session scope but is actually server scope. */
+	{ScopeSession, TiDBEnableDbQPSMetric, "ON"},
 	{ScopeSession, TiDBGeneralLog, strconv.Itoa(DefTiDBGeneralLog)},
 	{ScopeSession, TiDBSlowLogThreshold, strconv.Itoa(logutil.DefaultSlowThreshold)},
 	{ScopeSession, TiDBQueryLogMaxLen, strconv.Itoa(logutil.DefaultQueryLogMaxLen)},

sessionctx/variable/tidb_vars.go

@@ -185,6 +185,9 @@ const (
 	// the input string values are valid, we can skip the check.
 	TiDBSkipUTF8Check = "tidb_skip_utf8_check"
 
+	// tidb_db_qps_metric_switch is used to control whether db qps monitoring data needs to recorded.
+	TiDBEnableDbQPSMetric = "tidb_enable_db_qps_metric"
+
 	// tidb_hash_join_concurrency is used for hash join executor.
 	// The hash join outer executor starts multiple concurrent join workers to probe the hash table.
 	TiDBHashJoinConcurrency = "tidb_hash_join_concurrency"
@@ -278,6 +281,7 @@ const (
 	DefTiDBForcePriority             = mysql.NoPriority
 	DefTiDBUseRadixJoin              = false
 	DefEnableWindowFunction          = false
+	DefEnableDbQPSMetric             = 1
 )
 
 // Process global variables.

sessionctx/variable/varsutil.go

@@ -340,7 +340,7 @@ func ValidateSetSystemVar(vars *SessionVars, name string, value string) (string,
 			return "0", nil
 		}
 		return value, ErrWrongValueForVar.GenWithStackByArgs(name, value)
-	case AutocommitVar, TiDBSkipUTF8Check, TiDBOptAggPushDown,
+	case AutocommitVar, TiDBSkipUTF8Check, TiDBEnableDbQPSMetric, TiDBOptAggPushDown,
 		TiDBOptInSubqToJoinAndAgg,
 		TiDBBatchInsert, TiDBDisableTxnAutoRetry, TiDBEnableStreaming,
 		TiDBBatchDelete, TiDBBatchCommit, TiDBEnableCascadesPlanner, TiDBEnableWindowFunction: