topsql: introduce pub/sub mode

config/config.go

@@ -439,12 +439,17 @@ func (s *Security) ClusterSecurity() tikvcfg.Security {
 
 // Status is the status section of the config.
 type Status struct {
-	StatusHost      string `toml:"status-host" json:"status-host"`
-	MetricsAddr     string `toml:"metrics-addr" json:"metrics-addr"`
-	StatusPort      uint   `toml:"status-port" json:"status-port"`
-	MetricsInterval uint   `toml:"metrics-interval" json:"metrics-interval"`
-	ReportStatus    bool   `toml:"report-status" json:"report-status"`
-	RecordQPSbyDB   bool   `toml:"record-db-qps" json:"record-db-qps"`
+	StatusHost            string `toml:"status-host" json:"status-host"`
+	MetricsAddr           string `toml:"metrics-addr" json:"metrics-addr"`
+	StatusPort            uint   `toml:"status-port" json:"status-port"`
+	MetricsInterval       uint   `toml:"metrics-interval" json:"metrics-interval"`
+	ReportStatus          bool   `toml:"report-status" json:"report-status"`
+	RecordQPSbyDB         bool   `toml:"record-db-qps" json:"record-db-qps"`
+	GRPCKeepAliveTime     uint   `toml:"grpc-keepalive-time" json:"grpc-keepalive-time"`
+	GRPCKeepAliveTimeout  uint   `toml:"grpc-keepalive-timeout" json:"grpc-keepalive-timeout"`
+	GRPCConcurrentStreams uint   `toml:"grpc-concurrent-streams" json:"grpc-concurrent-streams"`
+	GRPCInitialWindowSize int    `toml:"grpc-initial-window-size" json:"grpc-initial-window-size"`
+	GRPCMaxSendMsgSize    int    `toml:"grpc-max-send-msg-size" json:"grpc-max-send-msg-size"`
 }
 
 // Performance is the performance section of the config.
@@ -658,11 +663,16 @@ var defaultConf = Config{
 		EnableSlowLog:       *NewAtomicBool(logutil.DefaultTiDBEnableSlowLog),
 	},
 	Status: Status{
-		ReportStatus:    true,
-		StatusHost:      DefStatusHost,
-		StatusPort:      DefStatusPort,
-		MetricsInterval: 15,
-		RecordQPSbyDB:   false,
+		ReportStatus:          true,
+		StatusHost:            DefStatusHost,
+		StatusPort:            DefStatusPort,
+		MetricsInterval:       15,
+		RecordQPSbyDB:         false,
+		GRPCKeepAliveTime:     10,
+		GRPCKeepAliveTimeout:  3,
+		GRPCConcurrentStreams: 1024,
+		GRPCInitialWindowSize: 2 * 1024 * 1024,
+		GRPCMaxSendMsgSize:    10 * 1024 * 1024,
 	},
 	Performance: Performance{
 		MaxMemory:             0,

config/config.toml.example

@@ -238,6 +238,23 @@ metrics-interval = 15
 # Record statements qps by database name if it is enabled.
 record-db-qps = false
 
+# After a duration of this time in seconds if the server doesn't see any activity it pings
+# the client to see if the transport is still alive.
+grpc-keepalive-time = 10
+
+# After having pinged for keepalive check, the server waits for a duration of timeout in seconds
+# and if no activity is seen even after that the connection is closed.
+grpc-keepalive-timeout = 3
+
+## The number of max concurrent streams/requests on a client connection.
+# grpc-concurrent-streams = 1024
+
+## Sets window size for stream. The default value is 2MB.
+# grpc-initial-window-size = 2097152
+
+## Set maximum message length in bytes that gRPC can send. `-1` means unlimited. The default value is 10MB.
+# grpc-max-send-msg-size = 10485760
+
 [performance]
 # Max CPUs to use, 0 use number of CPUs in the machine.
 max-procs = 0

ddl/ddl_worker.go

@@ -477,7 +477,7 @@ func newMetaWithQueueTp(txn kv.Transaction, tp workerType) *meta.Meta {
 }
 
 func (w *worker) setDDLLabelForTopSQL(job *model.Job) {
-	if !variable.TopSQLEnabled() || job == nil {
+	if !topsql.InstanceEnabled() || job == nil {
 		return
 	}
 

distsql/request_builder.go

@@ -35,6 +35,7 @@ import (
 	"github.com/pingcap/tidb/util/codec"
 	"github.com/pingcap/tidb/util/memory"
 	"github.com/pingcap/tidb/util/ranger"
+	"github.com/pingcap/tidb/util/topsql"
 	"github.com/pingcap/tipb/go-tipb"
 )
 
@@ -291,7 +292,7 @@ func (builder *RequestBuilder) SetFromInfoSchema(pis interface{}) *RequestBuilde
 
 // SetResourceGroupTagger sets the request resource group tagger.
 func (builder *RequestBuilder) SetResourceGroupTagger(sc *stmtctx.StatementContext) *RequestBuilder {
-	if variable.TopSQLEnabled() {
+	if topsql.InstanceEnabled() {
 		builder.Request.ResourceGroupTagger = sc.GetResourceGroupTagger()
 	}
 	return builder

domain/globalconfigsync/globalconfig_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/tidb/domain/globalconfigsync"
 	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/store/mockstore"
+	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/util/testbridge"
 	"github.com/stretchr/testify/require"
 	"go.etcd.io/etcd/integration"
@@ -67,11 +68,8 @@ func TestStoreGlobalConfig(t *testing.T) {
 	defer cluster.Terminate(t)
 	domain.GetGlobalConfigSyncer().SetEtcdClient(cluster.RandClient())
 
-	se, err := session.CreateSession4Test(store)
-	require.NoError(t, err)
-
-	_, err = se.Execute(context.Background(), "set @@global.tidb_enable_top_sql=1;")
-	require.NoError(t, err)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("set @@global.tidb_enable_top_sql=1;")
 
 	for i := 0; i < 20; i++ {
 		resp, err := cluster.RandClient().Get(context.Background(), "/global/config/enable_resource_metering")

domain/sysvar_cache.go

@@ -26,6 +26,8 @@ import (
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/sqlexec"
 	"github.com/pingcap/tidb/util/stmtsummary"
+	"github.com/pingcap/tidb/util/topsql/reporter"
+	"github.com/pingcap/tidb/util/topsql/tracecpu"
 	storekv "github.com/tikv/client-go/v2/kv"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -201,36 +203,34 @@ func (do *Domain) checkEnableServerGlobalVar(name, sVal string) {
 		err = stmtsummary.StmtSummaryByDigestMap.SetMaxSQLLength(sVal, false)
 	case variable.TiDBCapturePlanBaseline:
 		variable.CapturePlanBaseline.Set(sVal, false)
-	case variable.TiDBEnableTopSQL:
-		variable.TopSQLVariable.Enable.Store(variable.TiDBOptOn(sVal))
 	case variable.TiDBTopSQLPrecisionSeconds:
 		var val int64
 		val, err = strconv.ParseInt(sVal, 10, 64)
 		if err != nil {
 			break
 		}
-		variable.TopSQLVariable.PrecisionSeconds.Store(val)
+		tracecpu.PrecisionSeconds.Store(val)
 	case variable.TiDBTopSQLMaxStatementCount:
 		var val int64
 		val, err = strconv.ParseInt(sVal, 10, 64)
 		if err != nil {
 			break
 		}
-		variable.TopSQLVariable.MaxStatementCount.Store(val)
+		reporter.MaxStatementCount.Store(val)
 	case variable.TiDBTopSQLMaxCollect:
 		var val int64
 		val, err = strconv.ParseInt(sVal, 10, 64)
 		if err != nil {
 			break
 		}
-		variable.TopSQLVariable.MaxCollect.Store(val)
+		reporter.MaxCollect.Store(val)
 	case variable.TiDBTopSQLReportIntervalSeconds:
 		var val int64
 		val, err = strconv.ParseInt(sVal, 10, 64)
 		if err != nil {
 			break
 		}
-		variable.TopSQLVariable.ReportIntervalSeconds.Store(val)
+		reporter.ReportIntervalSeconds.Store(val)
 	case variable.TiDBRestrictedReadOnly:
 		variable.RestrictedReadOnly.Store(variable.TiDBOptOn(sVal))
 	case variable.TiDBStoreLimit:

executor/adapter.go

@@ -314,7 +314,7 @@ func (a *ExecStmt) RebuildPlan(ctx context.Context) (int64, error) {
 }
 
 func (a *ExecStmt) setPlanLabelForTopSQL(ctx context.Context) context.Context {
-	if a.Plan == nil || !variable.TopSQLEnabled() {
+	if a.Plan == nil || !topsql.InstanceEnabled() {
 		return ctx
 	}
 	vars := a.Ctx.GetSessionVars()

executor/executor.go

@@ -1730,7 +1730,7 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
 			goCtx = pprof.WithLabels(goCtx, pprof.Labels("sql", util.QueryStrForLog(prepareStmt.NormalizedSQL)))
 			pprof.SetGoroutineLabels(goCtx)
 		}
-		if variable.TopSQLEnabled() && prepareStmt.SQLDigest != nil {
+		if topsql.InstanceEnabled() && prepareStmt.SQLDigest != nil {
 			topsql.AttachSQLInfo(goCtx, prepareStmt.NormalizedSQL, prepareStmt.SQLDigest, "", nil, vars.InRestrictedSQL)
 		}
 	}
@@ -1910,7 +1910,7 @@ func FillVirtualColumnValue(virtualRetTypes []*types.FieldType, virtualColumnInd
 }
 
 func setResourceGroupTaggerForTxn(sc *stmtctx.StatementContext, snapshot kv.Snapshot) {
-	if snapshot != nil && variable.TopSQLEnabled() {
+	if snapshot != nil && topsql.InstanceEnabled() {
 		snapshot.SetOption(kv.ResourceGroupTagger, sc.GetResourceGroupTagger())
 	}
 }

executor/executor_test.go

@@ -82,6 +82,8 @@ import (
 	"github.com/pingcap/tidb/util/testkit"
 	"github.com/pingcap/tidb/util/testutil"
 	"github.com/pingcap/tidb/util/timeutil"
+	"github.com/pingcap/tidb/util/topsql"
+	topsqlmock "github.com/pingcap/tidb/util/topsql/tracecpu/mock"
 	"github.com/pingcap/tipb/go-tipb"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"
@@ -8718,11 +8720,8 @@ func (s *testResourceTagSuite) TestResourceGroupTag(c *C) {
 	tk.MustExec("create table t(a int, b int, unique index idx(a));")
 	tbInfo := testGetTableByName(c, tk.Se, "test", "t")
 
-	// Enable Top SQL
-	variable.TopSQLVariable.Enable.Store(true)
-	config.UpdateGlobal(func(conf *config.Config) {
-		conf.TopSQL.ReceiverAddress = "mock-agent"
-	})
+	topsql.SetupTopSQL(topsqlmock.NewTopSQLCollector())
+	defer topsql.Close()
 
 	c.Assert(failpoint.Enable("github.com/pingcap/tidb/store/mockstore/unistore/unistoreRPCClientSendHook", `return(true)`), IsNil)
 	defer failpoint.Disable("github.com/pingcap/tidb/store/mockstore/unistore/unistoreRPCClientSendHook")

executor/main_test.go

@@ -57,6 +57,9 @@ func TestMain(m *testing.M) {
 		goleak.IgnoreTopFunction("go.etcd.io/etcd/pkg/logutil.(*MergeLogger).outputLoop"),
 		goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"),
 		goleak.IgnoreTopFunction("gopkg.in/natefinch/lumberjack%2ev2.(*Logger).millRun"),
+		goleak.IgnoreTopFunction("time.Sleep"),
+		goleak.IgnoreTopFunction("runtime/pprof.readProfile"),
+		goleak.IgnoreTopFunction("github.com/pingcap/tidb/util/topsql/tracecpu.(*sqlCPUProfiler).startAnalyzeProfileWorker"),
 	}
 	callback := func(i int) int {
 		testDataMap.GenerateOutputIfNeeded()

executor/prepared.go

@@ -30,7 +30,6 @@ import (
 	"github.com/pingcap/tidb/planner"
 	plannercore "github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/sessionctx"
-	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/types"
 	driver "github.com/pingcap/tidb/types/parser_driver"
 	"github.com/pingcap/tidb/util"
@@ -194,7 +193,7 @@ func (e *PrepareExec) Next(ctx context.Context, req *chunk.Chunk) error {
 		SchemaVersion: ret.InfoSchema.SchemaMetaVersion(),
 	}
 	normalizedSQL, digest := parser.NormalizeDigest(prepared.Stmt.Text())
-	if variable.TopSQLEnabled() {
+	if topsql.InstanceEnabled() {
 		ctx = topsql.AttachSQLInfo(ctx, normalizedSQL, digest, "", nil, vars.InRestrictedSQL)
 	}
 

executor/set_test.go

@@ -38,6 +38,8 @@ import (
 	"github.com/pingcap/tidb/util/mock"
 	"github.com/pingcap/tidb/util/testkit"
 	"github.com/pingcap/tidb/util/testutil"
+	"github.com/pingcap/tidb/util/topsql/reporter"
+	"github.com/pingcap/tidb/util/topsql/tracecpu"
 )
 
 func (s *testSerialSuite1) TestSetVar(c *C) {
@@ -1497,27 +1499,21 @@ func (s *testSerialSuite) TestSetTopSQLVariables(c *C) {
 	}()
 
 	tk := testkit.NewTestKit(c, s.store)
-	tk.MustExec("set @@global.tidb_enable_top_sql='On';")
-	tk.MustQuery("select @@global.tidb_enable_top_sql;").Check(testkit.Rows("1"))
-	c.Assert(variable.TopSQLVariable.Enable.Load(), IsTrue)
-	tk.MustExec("set @@global.tidb_enable_top_sql='off';")
-	tk.MustQuery("select @@global.tidb_enable_top_sql;").Check(testkit.Rows("0"))
-	c.Assert(variable.TopSQLVariable.Enable.Load(), IsFalse)
 
 	tk.MustExec("set @@global.tidb_top_sql_precision_seconds=2;")
 	tk.MustQuery("select @@global.tidb_top_sql_precision_seconds;").Check(testkit.Rows("2"))
-	c.Assert(variable.TopSQLVariable.PrecisionSeconds.Load(), Equals, int64(2))
+	c.Assert(tracecpu.PrecisionSeconds.Load(), Equals, int64(2))
 	_, err := tk.Exec("set @@global.tidb_top_sql_precision_seconds='abc';")
 	c.Assert(err.Error(), Equals, "[variable:1232]Incorrect argument type to variable 'tidb_top_sql_precision_seconds'")
 	tk.MustExec("set @@global.tidb_top_sql_precision_seconds='-1';")
 	tk.MustQuery("select @@global.tidb_top_sql_precision_seconds;").Check(testkit.Rows("1"))
 	tk.MustExec("set @@global.tidb_top_sql_precision_seconds=2;")
 	tk.MustQuery("select @@global.tidb_top_sql_precision_seconds;").Check(testkit.Rows("2"))
-	c.Assert(variable.TopSQLVariable.PrecisionSeconds.Load(), Equals, int64(2))
+	c.Assert(tracecpu.PrecisionSeconds.Load(), Equals, int64(2))
 
 	tk.MustExec("set @@global.tidb_top_sql_max_statement_count=20;")
 	tk.MustQuery("select @@global.tidb_top_sql_max_statement_count;").Check(testkit.Rows("20"))
-	c.Assert(variable.TopSQLVariable.MaxStatementCount.Load(), Equals, int64(20))
+	c.Assert(reporter.MaxStatementCount.Load(), Equals, int64(20))
 	_, err = tk.Exec("set @@global.tidb_top_sql_max_statement_count='abc';")
 	c.Assert(err.Error(), Equals, "[variable:1232]Incorrect argument type to variable 'tidb_top_sql_max_statement_count'")
 	tk.MustExec("set @@global.tidb_top_sql_max_statement_count='-1';")
@@ -1528,11 +1524,11 @@ func (s *testSerialSuite) TestSetTopSQLVariables(c *C) {
 
 	tk.MustExec("set @@global.tidb_top_sql_max_statement_count=20;")
 	tk.MustQuery("select @@global.tidb_top_sql_max_statement_count;").Check(testkit.Rows("20"))
-	c.Assert(variable.TopSQLVariable.MaxStatementCount.Load(), Equals, int64(20))
+	c.Assert(reporter.MaxStatementCount.Load(), Equals, int64(20))
 
 	tk.MustExec("set @@global.tidb_top_sql_max_collect=20000;")
 	tk.MustQuery("select @@global.tidb_top_sql_max_collect;").Check(testkit.Rows("20000"))
-	c.Assert(variable.TopSQLVariable.MaxCollect.Load(), Equals, int64(20000))
+	c.Assert(reporter.MaxCollect.Load(), Equals, int64(20000))
 	_, err = tk.Exec("set @@global.tidb_top_sql_max_collect='abc';")
 	c.Assert(err.Error(), Equals, "[variable:1232]Incorrect argument type to variable 'tidb_top_sql_max_collect'")
 	tk.MustExec("set @@global.tidb_top_sql_max_collect='-1';")
@@ -1545,11 +1541,11 @@ func (s *testSerialSuite) TestSetTopSQLVariables(c *C) {
 
 	tk.MustExec("set @@global.tidb_top_sql_max_collect=20000;")
 	tk.MustQuery("select @@global.tidb_top_sql_max_collect;").Check(testkit.Rows("20000"))
-	c.Assert(variable.TopSQLVariable.MaxCollect.Load(), Equals, int64(20000))
+	c.Assert(reporter.MaxCollect.Load(), Equals, int64(20000))
 
 	tk.MustExec("set @@global.tidb_top_sql_report_interval_seconds=120;")
 	tk.MustQuery("select @@global.tidb_top_sql_report_interval_seconds;").Check(testkit.Rows("120"))
-	c.Assert(variable.TopSQLVariable.ReportIntervalSeconds.Load(), Equals, int64(120))
+	c.Assert(reporter.ReportIntervalSeconds.Load(), Equals, int64(120))
 	_, err = tk.Exec("set @@global.tidb_top_sql_report_interval_seconds='abc';")
 	c.Assert(err.Error(), Equals, "[variable:1232]Incorrect argument type to variable 'tidb_top_sql_report_interval_seconds'")
 	tk.MustExec("set @@global.tidb_top_sql_report_interval_seconds='5000';")
@@ -1558,7 +1554,7 @@ func (s *testSerialSuite) TestSetTopSQLVariables(c *C) {
 
 	tk.MustExec("set @@global.tidb_top_sql_report_interval_seconds=120;")
 	tk.MustQuery("select @@global.tidb_top_sql_report_interval_seconds;").Check(testkit.Rows("120"))
-	c.Assert(variable.TopSQLVariable.ReportIntervalSeconds.Load(), Equals, int64(120))
+	c.Assert(reporter.ReportIntervalSeconds.Load(), Equals, int64(120))
 
 	// Test for hide top sql variable in show variable.
 	tk.MustQuery("show variables like '%top_sql%'").Check(testkit.Rows())

executor/update.go

@@ -26,12 +26,12 @@ import (
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	plannercore "github.com/pingcap/tidb/planner/core"
-	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
 	"github.com/pingcap/tidb/util/execdetails"
 	"github.com/pingcap/tidb/util/memory"
+	"github.com/pingcap/tidb/util/topsql"
 	"github.com/tikv/client-go/v2/txnkv/txnsnapshot"
 )
 
@@ -271,7 +271,7 @@ func (e *UpdateExec) updateRows(ctx context.Context) (int, error) {
 				txn.GetSnapshot().SetOption(kv.CollectRuntimeStats, e.stats.SnapshotRuntimeStats)
 			}
 		}
-		if variable.TopSQLEnabled() {
+		if topsql.InstanceEnabled() {
 			txn, err := e.ctx.Txn(true)
 			if err == nil {
 				txn.SetOption(kv.ResourceGroupTagger, e.ctx.GetSessionVars().StmtCtx.GetResourceGroupTagger())

metrics/metrics.go

@@ -154,6 +154,7 @@ func RegisterMetrics() {
 	prometheus.MustRegister(SmallTxnWriteDuration)
 	prometheus.MustRegister(TxnWriteThroughput)
 	prometheus.MustRegister(LoadSysVarCacheCounter)
+	prometheus.MustRegister(TopSQLProfileCounter)
 	prometheus.MustRegister(TopSQLIgnoredCounter)
 	prometheus.MustRegister(TopSQLReportDurationHistogram)
 	prometheus.MustRegister(TopSQLReportDataHistogram)

metrics/topsql.go

@@ -18,6 +18,14 @@ import "github.com/prometheus/client_golang/prometheus"
 
 // Top SQL metrics.
 var (
+	TopSQLProfileCounter = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "tidb",
+			Subsystem: "topsql",
+			Name:      "profile_total",
+			Help:      "Counter of profiling",
+		})
+
 	TopSQLIgnoredCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "tidb",

server/conn.go

@@ -85,6 +85,7 @@ import (
 	"github.com/pingcap/tidb/util/hack"
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/memory"
+	"github.com/pingcap/tidb/util/topsql"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tikv/client-go/v2/util"
 	"go.uber.org/zap"
@@ -1244,7 +1245,7 @@ func (cc *clientConn) dispatch(ctx context.Context, data []byte) error {
 	cc.lastPacket = data
 	cmd := data[0]
 	data = data[1:]
-	if variable.TopSQLEnabled() {
+	if topsql.InstanceEnabled() {
 		defer pprof.SetGoroutineLabels(ctx)
 	}
 	if variable.EnablePProfSQLCPU.Load() {