// Copyright 2015 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
	"context"
	"flag"
	"fmt"
	"io/fs"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/opentracing/opentracing-go"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/br/pkg/storage"
	"github.com/pingcap/tidb/pkg/bindinfo"
	"github.com/pingcap/tidb/pkg/config"
	"github.com/pingcap/tidb/pkg/ddl"
	"github.com/pingcap/tidb/pkg/domain"
	"github.com/pingcap/tidb/pkg/executor"
	"github.com/pingcap/tidb/pkg/executor/importer"
	"github.com/pingcap/tidb/pkg/executor/mppcoordmanager"
	"github.com/pingcap/tidb/pkg/extension"
	_ "github.com/pingcap/tidb/pkg/extension/_import"
	"github.com/pingcap/tidb/pkg/extension/serverless/audit"
	"github.com/pingcap/tidb/pkg/keyspace"
	"github.com/pingcap/tidb/pkg/kv"
	"github.com/pingcap/tidb/pkg/metrics"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	"github.com/pingcap/tidb/pkg/parser/terror"
	parsertypes "github.com/pingcap/tidb/pkg/parser/types"
	plannercore "github.com/pingcap/tidb/pkg/planner/core"
	"github.com/pingcap/tidb/pkg/plugin"
	"github.com/pingcap/tidb/pkg/privilege/privileges"
	"github.com/pingcap/tidb/pkg/resourcemanager"
	"github.com/pingcap/tidb/pkg/server"
	"github.com/pingcap/tidb/pkg/session"
	"github.com/pingcap/tidb/pkg/session/txninfo"
	"github.com/pingcap/tidb/pkg/sessionctx"
	"github.com/pingcap/tidb/pkg/sessionctx/variable"
	"github.com/pingcap/tidb/pkg/standby"
	"github.com/pingcap/tidb/pkg/statistics"
	kvstore "github.com/pingcap/tidb/pkg/store"
	"github.com/pingcap/tidb/pkg/store/copr"
	"github.com/pingcap/tidb/pkg/store/driver"
	"github.com/pingcap/tidb/pkg/store/mockstore"
	"github.com/pingcap/tidb/pkg/table/temptable"
	"github.com/pingcap/tidb/pkg/tidbmanager"
	"github.com/pingcap/tidb/pkg/tidbworker"
	"github.com/pingcap/tidb/pkg/util"
	"github.com/pingcap/tidb/pkg/util/backoff"
	"github.com/pingcap/tidb/pkg/util/cgmon"
	"github.com/pingcap/tidb/pkg/util/chunk"
	"github.com/pingcap/tidb/pkg/util/cpuprofile"
	"github.com/pingcap/tidb/pkg/util/deadlockhistory"
	"github.com/pingcap/tidb/pkg/util/disk"
	"github.com/pingcap/tidb/pkg/util/domainutil"
	"github.com/pingcap/tidb/pkg/util/external"
	"github.com/pingcap/tidb/pkg/util/kvcache"
	"github.com/pingcap/tidb/pkg/util/logutil"
	"github.com/pingcap/tidb/pkg/util/memory"
	"github.com/pingcap/tidb/pkg/util/metricsutil"
	"github.com/pingcap/tidb/pkg/util/printer"
	"github.com/pingcap/tidb/pkg/util/redact"
	"github.com/pingcap/tidb/pkg/util/sem"
	"github.com/pingcap/tidb/pkg/util/serverless"
	remotequery "github.com/pingcap/tidb/pkg/util/serverless/remote-query"
	"github.com/pingcap/tidb/pkg/util/servicescope"
	"github.com/pingcap/tidb/pkg/util/signal"
	stmtsummaryv2 "github.com/pingcap/tidb/pkg/util/stmtsummary/v2"
	"github.com/pingcap/tidb/pkg/util/sys/linux"
	storageSys "github.com/pingcap/tidb/pkg/util/sys/storage"
	"github.com/pingcap/tidb/pkg/util/systimemon"
	"github.com/pingcap/tidb/pkg/util/tiflashcompute"
	"github.com/pingcap/tidb/pkg/util/topsql"
	"github.com/pingcap/tidb/pkg/util/versioninfo"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/push"
	tikvconfig "github.com/tikv/client-go/v2/config"
	"github.com/tikv/client-go/v2/tikv"
	"github.com/tikv/client-go/v2/txnkv/transaction"
	"github.com/tikv/client-go/v2/util/grpcutil"
	pd "github.com/tikv/pd/client"
	"go.uber.org/automaxprocs/maxprocs"
	"go.uber.org/zap"
)

// Flag Names
const (
	nmVersion          = "V"
	nmConfig           = "config"
	nmConfigCheck      = "config-check"
	nmConfigStrict     = "config-strict"
	nmStore            = "store"
	nmStorePath        = "path"
	nmHost             = "host"
	nmAdvertiseAddress = "advertise-address"
	nmPort             = "P"
	nmCors             = "cors"
	nmSocket           = "socket"
	nmRunDDL           = "run-ddl"
	nmLogLevel         = "L"
	nmLogFile          = "log-file"
	nmLogSlowQuery     = "log-slow-query"
	nmLogGeneral       = "log-general"
	nmReportStatus     = "report-status"
	nmStatusHost       = "status-host"
	nmStatusPort       = "status"
	nmMetricsAddr      = "metrics-addr"
	nmMetricsInterval  = "metrics-interval"
	nmDdlLease         = "lease"
	nmTokenLimit       = "token-limit"
	nmPluginDir        = "plugin-dir"
	nmPluginLoad       = "plugin-load"
	nmRepairMode       = "repair-mode"
	nmRepairList       = "repair-list"
	nmTempDir          = "temp-dir"
	nmClusterCa        = "cluster-ca"
	nmClusterCert      = "cluster-cert"
	nmClusterKey       = "cluster-key"
	nmSQLCA            = "sql-ca"
	nmSQLCert          = "sql-cert"
	nmSQLKey           = "sql-key"

	nmRedact = "redact"

	nmProxyProtocolNetworks      = "proxy-protocol-networks"
	nmProxyProtocolHeaderTimeout = "proxy-protocol-header-timeout"
	nmProxyProtocolFallbackable  = "proxy-protocol-fallbackable"
	nmAffinityCPU                = "affinity-cpus"

	nmInitializeSecure            = "initialize-secure"
	nmInitializeInsecure          = "initialize-insecure"
	nmInitializeSQLFile           = "initialize-sql-file"
	nmDisconnectOnExpiredPassword = "disconnect-on-expired-password"
	nmKeyspaceName                = "keyspace-name"
	nmTiDBServiceScope            = "tidb-service-scope"

	// Serverless flag names
	nmWaitKeyspaceEnabled = "wait-keyspace-enabled"
	nmStandby             = "standby"
	nmActivationTimeout   = "activation-timeout"
	nmMaxIdleSeconds      = "max-idle-seconds"
	nmKeyspaceActivate    = "keyspace-activate"
	nmTiDBWorkerAPIAddr   = "tidb-worker-api-addr"

	// nmEnableOnlyRunUpgrade used to activate existing keyspaces when upgrading TiDB.
	nmEnableOnlyRunUpgrade = "enable-only-run-upgrade"

	nmExportID = "export-id"
)

const (
	exitCodeOK  = 0
	exitCodeErr = 1
	exitCodeInt = 130
)

var (
	version      *bool
	configPath   *string
	configCheck  *bool
	configStrict *bool

	// Base
	store            *string
	storePath        *string
	host             *string
	advertiseAddress *string
	port             *string
	cors             *string
	socket           *string
	enableBinlog     *bool
	runDDL           *bool
	ddlLease         *string
	tokenLimit       *int
	pluginDir        *string
	pluginLoad       *string
	affinityCPU      *string
	repairMode       *bool
	repairList       *string
	tempDir          *string

	// Log
	logLevel     *string
	logFile      *string
	logSlowQuery *string
	logGeneral   *string

	// Status
	reportStatus    *bool
	statusHost      *string
	statusPort      *string
	metricsAddr     *string
	metricsInterval *uint

	// subcommand collect-log
	redactFlag *bool

	// PROXY Protocol
	proxyProtocolNetworks      *string
	proxyProtocolHeaderTimeout *uint
	proxyProtocolFallbackable  *bool

	// Bootstrap and security
	initializeSecure            *bool
	initializeInsecure          *bool
	initializeSQLFile           *string
	disconnectOnExpiredPassword *bool
	keyspaceName                *string
	serviceScope                *string
	help                        *bool
	clusterCA                   *string
	clusterCert                 *string
	clusterKey                  *string
	sqlCA                       *string
	sqlCert                     *string
	sqlKey                      *string

	// Serverless flags
	waitKeyspaceEnabled *bool
	// Standby
	standbyMode       *bool
	activationTimeout *uint
	maxIdleSeconds    *uint
	// Keyspace Activate
	keyspaceActivateMode *bool
	// TiDB Worker
	tidbWorkerAPIAddr *string

	// If enableOnlyRunUpgrade is true, Serverless TiDB can only do upgrade process.
	enableOnlyRunUpgrade *bool
	exportID             *string
)

func initFlagSet() *flag.FlagSet {
	fset := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
	version = flagBoolean(fset, nmVersion, false, "print version information and exit")
	configPath = fset.String(nmConfig, "", "config file path")
	configCheck = flagBoolean(fset, nmConfigCheck, false, "check config file validity and exit")
	configStrict = flagBoolean(fset, nmConfigStrict, false, "enforce config file validity")

	// Base
	store = fset.String(nmStore, "unistore", "registered store name, [tikv, mocktikv, unistore]")
	storePath = fset.String(nmStorePath, "/tmp/tidb", "tidb storage path")
	host = fset.String(nmHost, "0.0.0.0", "tidb server host")
	advertiseAddress = fset.String(nmAdvertiseAddress, "", "tidb server advertise IP")
	port = fset.String(nmPort, "4000", "tidb server port")
	cors = fset.String(nmCors, "", "tidb server allow cors origin")
	socket = fset.String(nmSocket, "/tmp/tidb-{Port}.sock", "The socket file to use for connection.")
	runDDL = flagBoolean(fset, nmRunDDL, true, "run ddl worker on this tidb-server")
	ddlLease = fset.String(nmDdlLease, "45s", "schema lease duration, very dangerous to change only if you know what you do")
	tokenLimit = fset.Int(nmTokenLimit, 1000, "the limit of concurrent executed sessions")
	pluginDir = fset.String(nmPluginDir, "/data/deploy/plugin", "the folder that hold plugin")
	pluginLoad = fset.String(nmPluginLoad, "", "wait load plugin name(separated by comma)")
	affinityCPU = fset.String(nmAffinityCPU, "", "affinity cpu (cpu-no. separated by comma, e.g. 1,2,3)")
	repairMode = flagBoolean(fset, nmRepairMode, false, "enable admin repair mode")
	repairList = fset.String(nmRepairList, "", "admin repair table list")
	tempDir = fset.String(nmTempDir, config.DefTempDir, "tidb temporary directory")

	// Log
	logLevel = fset.String(nmLogLevel, "info", "log level: info, debug, warn, error, fatal")
	logFile = fset.String(nmLogFile, "", "log file path")
	logSlowQuery = fset.String(nmLogSlowQuery, "", "slow query file path")
	logGeneral = fset.String(nmLogGeneral, "", "general log file path")

	// Status
	reportStatus = flagBoolean(fset, nmReportStatus, true, "If enable status report HTTP service.")
	statusHost = fset.String(nmStatusHost, "0.0.0.0", "tidb server status host")
	statusPort = fset.String(nmStatusPort, "10080", "tidb server status port")
	metricsAddr = fset.String(nmMetricsAddr, "", "prometheus pushgateway address, leaves it empty will disable prometheus push.")
	metricsInterval = fset.Uint(nmMetricsInterval, 15, "prometheus client push interval in second, set \"0\" to disable prometheus push.")

	// subcommand collect-log
	redactFlag = flagBoolean(fset, nmRedact, false, "remove sensitive words from marked tidb logs, if `./tidb-server --redact=xxx collect-log <input> <output>` subcommand is used")

	// PROXY Protocol
	proxyProtocolNetworks = fset.String(nmProxyProtocolNetworks, "", "proxy protocol networks allowed IP or *, empty mean disable proxy protocol support")
	proxyProtocolHeaderTimeout = fset.Uint(nmProxyProtocolHeaderTimeout, 5, "proxy protocol header read timeout, unit is second. (Deprecated: as proxy protocol using lazy mode, header read timeout no longer used)")
	proxyProtocolFallbackable = flagBoolean(fset, nmProxyProtocolFallbackable, false, "enable proxy protocol fallback mode. If it is enabled, connection will return the client IP address when the client does not send PROXY Protocol Header and it will not return any error. (Note: This feature it does NOT follow the PROXY Protocol SPEC)")

	// Bootstrap and security
	initializeSecure = flagBoolean(fset, nmInitializeSecure, false, "bootstrap tidb-server in secure mode")
	initializeInsecure = flagBoolean(fset, nmInitializeInsecure, true, "bootstrap tidb-server in insecure mode")
	initializeSQLFile = fset.String(nmInitializeSQLFile, "", "SQL file to execute on first bootstrap")
	disconnectOnExpiredPassword = flagBoolean(fset, nmDisconnectOnExpiredPassword, true, "the server disconnects the client when the password is expired")
	keyspaceName = fset.String(nmKeyspaceName, "", "keyspace name.")
	serviceScope = fset.String(nmTiDBServiceScope, "", "tidb service scope")
	help = fset.Bool("help", false, "show the usage")

	// mTLS
	clusterCA = fset.String(nmClusterCa, "", "cluster CA file path")
	clusterCert = fset.String(nmClusterCert, "", "cluster cert file path")
	clusterKey = fset.String(nmClusterKey, "", "cluster key file path")
	sqlCA = flag.String(nmSQLCA, "", "SQL ca file path")
	sqlCert = flag.String(nmSQLCert, "", "SQL cert file path")
	sqlKey = flag.String(nmSQLKey, "", "SQL key file path")
	// export mode
	exportID = flag.String(nmExportID, "", "export id")

	// Serverless flags
	waitKeyspaceEnabled = flagBoolean(fset, nmWaitKeyspaceEnabled, true, "wait for keyspace to become enabled during bootstrap")
	// Standby
	standbyMode = flagBoolean(fset, nmStandby, false, "start tidb-server as standby")
	activationTimeout = fset.Uint(nmActivationTimeout, 0, "max time in second allowed for tidb to activate from standby, 0 means no limit")
	maxIdleSeconds = fset.Uint(nmMaxIdleSeconds, 0, "max idle seconds for a connection, 0 means no limit")
	// Keyspace Activate
	keyspaceActivateMode = flagBoolean(fset, nmKeyspaceActivate, false, "start tidb-server as keyspaceActivate")
	// TiDB Worker
	tidbWorkerAPIAddr = fset.String(nmTiDBWorkerAPIAddr, "", "tidb worker API server address")

	// If enableOnlyRunUpgrade is true, Serverless TiDB can only do upgrade process.
	enableOnlyRunUpgrade = flagBoolean(fset, nmEnableOnlyRunUpgrade, false, "only run upgrade and exit")

	session.RegisterMockUpgradeFlag(fset)
	// Ignore errors; CommandLine is set for ExitOnError.
	// nolint:errcheck
	fset.Parse(os.Args[1:])
	if *help {
		fset.Usage()
		os.Exit(0)
	}
	return fset
}

func main() {
	// Enable gRPC client metrics based on config
	cfg := config.GetGlobalConfig()
	if cfg.PDClient.EnableGRPCClientMetrics {
		grpcutil.EnableGRPCClientMetrics()
	}
	fset := initFlagSet()
	if args := fset.Args(); len(args) != 0 {
		if args[0] == "collect-log" && len(args) > 1 {
			output := "-"
			if len(args) > 2 {
				output = args[2]
			}
			terror.MustNil(redact.DeRedactFile(*redactFlag, args[1], output))
			return
		}
	}
	config.InitializeConfig(*configPath, *configCheck, *configStrict, overrideConfig, fset)
	if *version {
		setVersions()
		fmt.Println(printer.GetTiDBInfo())
		os.Exit(0)
	}

	mainErrHandler := func(err error) { terror.MustNil(err) }

	quit := make(chan struct{})
	defer close(quit)
	serverless.StartMemoryScaler(quit)

	var standbyController server.StandbyController
	if config.GetGlobalConfig().StandByMode {
		mgrCli, err := createMrgClient()
		mainErrHandler(err)
		standbyController = standby.NewLoadKeyspaceController(mgrCli)
	}

	// If running standby mode, overwrite the keyspace config and error handlers.
	if standbyController != nil {
		standbyController.WaitForActivate()
		// replace mainErrHandler to make sure standby handler can exit gracefully.
		mainErrHandler = func(err error) {
			if err != nil {
				standbyController.EndStandby(err)
				terror.MustNil(err)
			}
		}
		// need to validate config again in case of config change via standby
		mainErrHandler(config.GetGlobalConfig().Valid())
	} else {
		// If not set keyspace in config, try to get keyspace name from env and update config.
		keyspace.GetKeyspaceNameBySettings()
	}

	registerStores()

	// load keyspace and set metric labels.
	keyspaceMeta, pdCli, err := getServerlessInfo()
	mainErrHandler(err)

	updateConfigForServerless(keyspaceMeta)
	err = metricsutil.RegisterMetricsWithLabels(config.GetGlobalConfig().MetricsLabels)
	terror.MustNil(err)

	var keyspaceID uint32
	if keyspaceMeta != nil {
		keyspaceID = keyspaceMeta.GetId()
		exportID := config.GetGlobalConfig().ExportID
		if config.GetGlobalConfig().EnableRULimit {
			if exportID != "" {
				log.Info("setting up serverless resource control", zap.String("exportID", exportID))
				config.DefaultResourceGroup = exportID
			} else {
				log.Info("setting up serverless resource control", zap.Uint32("keyspaceID", keyspaceID))
				config.DefaultResourceGroup = strconv.FormatUint(uint64(keyspaceID), 10)
			}
			tikv.EnableResourceControl()
		}
		if keyspaceMeta.Config != nil {
			if clusterID, ok := keyspaceMeta.Config[serverless.LabelClusterID]; ok {
				// Rewrite the AutoScalerCluster with keyspace meta.
				config.UpdateGlobal(func(c *config.Config) {
					c.AutoScalerClusterID = clusterID
				})
			}
			if IsBootstrappedForRestore, ok := keyspaceMeta.Config[serverless.LabelIsBootstrappedForRestore]; ok {
				config.UpdateGlobal(func(c *config.Config) {
					c.IsBootstrappedForRestore = IsBootstrappedForRestore
				})
			}
			if isBranch, ok := keyspaceMeta.Config[serverless.LabelIsBranch]; ok {
				config.UpdateGlobal(func(c *config.Config) {
					c.IsBranch, _ = strconv.ParseBool(isBranch)
				})
			}
			if isBranchBootstrapped, ok := keyspaceMeta.Config[serverless.LabelIsBranchBootstrapped]; ok {
				config.UpdateGlobal(func(c *config.Config) {
					c.IsBranchBootstrapped = isBranchBootstrapped
				})
			}
		}
		keyspace.SetUsernamePolicy(keyspace.NewPrefixPolicy(keyspaceMeta.GetName()))
	}

	// Serverless ===================
	updateConfigForServerless(keyspaceMeta)
	temptable.InitGloablTemporaryTableIDAllocator()

	err = setupExternalStorage(keyspaceMeta)
	terror.MustNil(err)

	if variable.EnableTmpStorageOnOOM.Load() {
		config.GetGlobalConfig().UpdateTempStoragePath()
		err := disk.InitializeTempDir()
		mainErrHandler(err)
		checkTempStorageQuota()
	}
	setupLog(keyspaceMeta.GetId())
	memory.InitMemoryHook()
	setupAuditLog()
	setupExtensions()
	setupStmtSummary()

	err = cpuprofile.StartCPUProfiler()
	mainErrHandler(err)

	if config.GetGlobalConfig().DisaggregatedTiFlash && config.GetGlobalConfig().UseAutoScaler {
		err = tiflashcompute.InitGlobalTopoFetcher(
			config.GetGlobalConfig().TiFlashComputeAutoScalerType,
			config.GetGlobalConfig().TiFlashComputeAutoScalerAddr,
			config.GetGlobalConfig().AutoScalerClusterID,
			keyspaceID,
			config.GetGlobalConfig().IsTiFlashComputeFixedPool)
		mainErrHandler(err)
	}

	// Enable failpoints in tikv/client-go if the test API is enabled.
	// It appears in the main function to be set before any use of client-go to prevent data race.
	if _, err := failpoint.Status("github.com/pingcap/tidb/pkg/server/enableTestAPI"); err == nil {
		warnMsg := "tikv/client-go failpoint is enabled, this should NOT happen in the production environment"
		logutil.BgLogger().Warn(warnMsg)
		tikv.EnableFailpoints()
	}
	setGlobalVars()
	setCPUAffinity()
	cgmon.StartCgroupMonitor()
	setupTracing() // Should before createServer and after setup config.
	printInfo()
	setupMetrics()

	if err = initTiDBWorkerManager(keyspaceMeta); err != nil {
		if config.GetGlobalConfig().TiDBWorker.Role != config.RoleMaster {
			logutil.BgLogger().Error("failed to initialize tidb worker manager as worker, exit tidb server", zap.Error(err))
			mainErrHandler(err)
		}
		logutil.BgLogger().Error("failed to initialize tidb worker manager, fallback to local mode", zap.Error(err))
		config.UpdateGlobal(func(conf *config.Config) {
			conf.TiDBWorker.Enable = false
		})
	}

	executor.Start()
	resourcemanager.InstanceResourceManager.Start()

	driverOpenOpts := &kv.DriverOpenOption{KeyspaceMeta: keyspaceMeta, PdCli: pdCli}
	storage, dom := createStoreDDLOwnerMgrAndDomain(driverOpenOpts)

	// ----------------- only run upgrade -----------------
	logutil.BgLogger().Info("finished upgrade.")
	if config.GetGlobalConfig().EnableOnlyRunUpgrade {
		closeDDLOwnerMgrDomainAndStorage(storage, dom)
		os.Exit(0)
	}
	// ----------------- only run upgrade end -----------------

	importer.SetupGetEtcdClientWithCodec(storage.GetCodec())

	svr := createServer(storage, dom)
	if standbyController != nil {
		svr.StandbyController = standbyController
		standbyController.OnServerCreated(svr)

		logutil.BgLogger().Info("standby controller starts to wait stats load")
		<-dom.StatsHandle().InitStatsDone
		standbyController.EndStandby(nil)
	}

	if config.GetGlobalConfig().KeyspaceActivateMode {
		// TODO: graceful shutdown
		os.Exit(0)
	}

	// Setup global tidb worker service client
	se, err := dom.SysSessionPool().Get()
	mainErrHandler(err)
	defer dom.SysSessionPool().Put(se)
	if err = initTiDBWorkerService(se.(sessionctx.Context)); err != nil {
		logutil.BgLogger().Error("failed to initialize tidb worker service, fallback to local mode", zap.Error(err))
		tidbworker.GlobalTiDBWorkerManager = nil
		config.UpdateGlobal(func(conf *config.Config) {
			conf.TiDBWorker.Enable = false
		})
	}

	// remote query worker
	if config.GetGlobalConfig().TiDBWorker.Role == config.RoleRemoteQueryWorker && config.GetGlobalConfig().TiDBWorker.ExecID != "" {
		logutil.BgLogger().Info("remote query worker mode")
		remoteQueryWorker, err := remotequery.NewExecutor(config.GetGlobalConfig().TiDBWorker.ExecID)
		if err != nil {
			logutil.BgLogger().Error("create remote query executor failed", zap.Error(err))
			os.Exit(1)
		}
		s, err := session.CreateSessionWithDomain(storage, dom)
		if err != nil {
			logutil.BgLogger().Error("create session with domain failed", zap.Error(err))
			os.Exit(1)
		}
		err = remoteQueryWorker.Execute(context.TODO(), s)
		if err != nil {
			logutil.BgLogger().Error("remote query execute failed", zap.Error(err))
		}
		os.Exit(0)
	}

	exited := make(chan struct{})
	exitCode := exitCodeOK
	signal.SetupSignalHandler(func(sig os.Signal) {
		svr.Close()
		cleanup(svr, storage, dom)
		cpuprofile.StopCPUProfiler()
		resourcemanager.InstanceResourceManager.Stop()
		executor.Stop()
		if sig == syscall.SIGINT {
			exitCode = exitCodeInt
		}
		close(exited)
	})
	topsql.SetupTopProfiling(keyspace.GetKeyspaceNameBytesBySettings(), svr, dom)
	terror.MustNil(svr.RunWithStore(dom, storage))
	<-exited
	if err := syncLog(); err != nil {
		exitCode = exitCodeErr
	}
	os.Exit(exitCode)
}

const (
	waitKeyspaceEnabledTimeout          = 60 * time.Second
	waitKeyspaceEnabledRetryInterval    = 50 * time.Millisecond
	waitKeyspaceEnabledRetryMaxInterval = 3 * time.Second
)

func getServerlessInfo() (*keyspacepb.KeyspaceMeta, pd.Client, error) {
	// load keyspace and set metric labels.
	cfg := config.GetGlobalConfig()
	if keyspace.IsKeyspaceNameEmpty(cfg.KeyspaceName) || strings.ToLower(cfg.Store) != "tikv" {
		return nil, nil, nil
	}

	activateRequest := standby.GetActivateRequest()
	if activateRequest.KeyspaceID != "" {
		metrics.SetConstLabels(map[string]string{
			metricsutil.KeyspaceIDLabelKey: activateRequest.KeyspaceID,
			metricsutil.TenantIDIDLabelKey: activateRequest.TenantID,
			metricsutil.ClusterIDLabelKey:  activateRequest.ClusterID,
			metricsutil.ProjectIDLabelKey:  activateRequest.ProjectID,
		})
	}

	log.Info("serverless cluster info loading...", zap.Any("keyspace", cfg.KeyspaceName))
	pdEtcdAddrs, _, _, err := tikvconfig.ParsePath("tikv://" + cfg.Path)
	if err != nil {
		return nil, nil, err
	}
	pdCli, err := driver.GetPDClient(cfg.KeyspaceName, pdEtcdAddrs)
	if err != nil {
		return nil, nil, err
	}

	start := time.Now()
	retry := 0
	// Backoff in case there are many TiDBs activing at the same time.
	backoffer := backoff.NewExponential(waitKeyspaceEnabledRetryInterval, 2, waitKeyspaceEnabledRetryMaxInterval)
	var keyspaceMeta *keyspacepb.KeyspaceMeta
	for {
		keyspaceMeta, err = pdCli.LoadKeyspace(context.TODO(), cfg.KeyspaceName)
		if err != nil {
			return nil, nil, err
		}

		if !*waitKeyspaceEnabled || keyspaceMeta.State == keyspacepb.KeyspaceState_ENABLED {
			break
		}

		if time.Since(start) >= waitKeyspaceEnabledTimeout {
			return nil, nil, errors.New("keyspace is not enabled")
		}
		if retry%5 == 0 {
			log.Info("waiting for keyspace to be enabled, retrying")
		}
		time.Sleep(backoffer.Backoff(retry))
		retry += 1
	}

	metrics.SetServerlessLabels(keyspaceMeta.Config[serverless.LabelTenantID],
		keyspaceMeta.Config[serverless.LabelProjectID],
		keyspaceMeta.Config[serverless.LabelClusterID])
	log.Info("serverless cluster info loaded",
		zap.String("tenant-id", metrics.ServerlessTenantID),
		zap.String("project-id", metrics.ServerlessProjectID),
		zap.String("cluster-id", metrics.ServerlessClusterID),
		zap.Stringer("keyspace-meta", keyspaceMeta),
	)

	//  TODO(ystaticy): check keyspace config migration status
	//if isInETCDMigration, exists := keyspaceMeta.Config[keyspace.KeyspaceMetaConfigIsInETCDMigration]; exists {
	//	if isInETCDMigration == "true" {
	//		return nil, nil, keyspace.ErrMigrationEtcd
	//	}
	//}

	return keyspaceMeta, pdCli, nil
}

func syncLog() error {
	if err := log.Sync(); err != nil {
		// Don't complain about /dev/stdout as Fsync will return EINVAL.
		if pathErr, ok := err.(*fs.PathError); ok {
			if pathErr.Path == "/dev/stdout" {
				return nil
			}
		}
		fmt.Fprintln(os.Stderr, "sync log err:", err)
		return err
	}
	return nil
}

func checkTempStorageQuota() {
	// check capacity and the quota when EnableTmpStorageOnOOM is enabled
	c := config.GetGlobalConfig()
	if c.TempStorageQuota >= 0 {
		capacityByte, err := storageSys.GetTargetDirectoryCapacity(c.TempStoragePath)
		if err != nil {
			log.Fatal(err.Error())
		} else if capacityByte < uint64(c.TempStorageQuota) {
			log.Fatal(fmt.Sprintf("value of [tmp-storage-quota](%d byte) exceeds the capacity(%d byte) of the [%s] directory", c.TempStorageQuota, capacityByte, c.TempStoragePath))
		}
	}
}

func setCPUAffinity() {
	if affinityCPU == nil || len(*affinityCPU) == 0 {
		return
	}
	var cpu []int
	for _, af := range strings.Split(*affinityCPU, ",") {
		af = strings.TrimSpace(af)
		if len(af) > 0 {
			c, err := strconv.Atoi(af)
			if err != nil {
				fmt.Fprintf(os.Stderr, "wrong affinity cpu config: %s", *affinityCPU)
				os.Exit(1)
			}
			cpu = append(cpu, c)
		}
	}
	err := linux.SetAffinity(cpu)
	if err != nil {
		fmt.Fprintf(os.Stderr, "set cpu affinity failure: %v", err)
		os.Exit(1)
	}
	if len(cpu) < runtime.GOMAXPROCS(0) {
		log.Info("cpu number less than maxprocs", zap.Int("cpu number ", len(cpu)), zap.Int("maxprocs", runtime.GOMAXPROCS(0)))
		runtime.GOMAXPROCS(len(cpu))
	}
}

func registerStores() {
	err := kvstore.Register("tikv", driver.TiKVDriver{})
	terror.MustNil(err)
	err = kvstore.Register("mocktikv", mockstore.MockTiKVDriver{})
	terror.MustNil(err)
	err = kvstore.Register("unistore", mockstore.EmbedUnistoreDriver{})
	terror.MustNil(err)
}

func createStoreDDLOwnerMgrAndDomain(driverOpenOption *kv.DriverOpenOption) (kv.Storage, *domain.Domain) {
	cfg := config.GetGlobalConfig()
	var fullPath string
	keyspaceMeta := driverOpenOption.KeyspaceMeta
	if keyspaceMeta == nil {
		fullPath = fmt.Sprintf("%s://%s", cfg.Store, cfg.Path)
	} else {
		startUpKeyspaceName := keyspaceMeta.Name
		fullPath = fmt.Sprintf("%s://%s?keyspaceName=%s", cfg.Store, cfg.Path, startUpKeyspaceName)
	}
	var err error
	storage, err := kvstore.New(fullPath, driverOpenOption)
	terror.MustNil(err)

	session.RegisterBatchDisttask(storage) // need register before disttask manager init

	copr.GlobalMPPFailedStoreProber.Run()
	mppcoordmanager.InstanceMPPCoordinatorManager.Run()
	// Bootstrap a session to load information schema.
	err = ddl.StartOwnerManager(context.Background(), storage)
	terror.MustNil(err)
	dom, err := session.BootstrapSession(storage)
	terror.MustNil(err)
	return storage, dom
}

func initTiDBWorkerManager(keyspaceMeta *keyspacepb.KeyspaceMeta) error {
	if keyspaceMeta == nil {
		return nil
	}
	cfg := config.GetGlobalConfig()
	workerConfig := cfg.TiDBWorker
	if !workerConfig.Enable || tidbworker.GlobalTiDBWorkerManager != nil {
		return nil
	}

	// When running as master without enabling SafePointV2, need to disable GC drop table.
	if workerConfig.Role == config.RoleMaster && !keyspace.IsKeyspaceUseKeyspaceLevelGC(keyspaceMeta) {
		config.UpdateGlobal(func(conf *config.Config) {
			conf.SkipGCWorker = true
		})
	}
	// GCV2 worker must have SafePointV2 enabled.
	if workerConfig.Role == config.RoleGCV2Worker && !keyspace.IsKeyspaceUseKeyspaceLevelGC(keyspaceMeta) {
		return fmt.Errorf("must enable safe point V2 to run as GCV2 worker")
	}

	log.Info("[tidb-worker] init global TiDB worker manager")
	ctx := context.Background()
	if err := tidbworker.InitManager(ctx, keyspaceMeta, workerConfig); err != nil {
		return errors.Trace(err)
	}
	return nil
}

func initTiDBWorkerService(sctx sessionctx.Context) error {
	if !tidbworker.IsMaster() {
		return nil
	}
	if tidbworker.UseKeyspaceLevelGC() {
		return tidbworker.GlobalTiDBWorkerManager.InitializeGCV2(context.TODO())
	}
	return tidbworker.GlobalTiDBWorkerManager.InitializeGC(context.TODO(), sctx)
}

// Prometheus push.
const zeroDuration = time.Duration(0)

// pushMetric pushes metrics in background.
func pushMetric(addr string, interval time.Duration) {
	if interval == zeroDuration || len(addr) == 0 {
		log.Info("disable Prometheus push client")
		return
	}
	log.Info("start prometheus push client", zap.String("server addr", addr), zap.String("interval", interval.String()))
	go prometheusPushClient(addr, interval)
}

// prometheusPushClient pushes metrics to Prometheus Pushgateway.
func prometheusPushClient(addr string, interval time.Duration) {
	// TODO: TiDB do not have uniq name, so we use host+port to compose a name.
	job := "tidb"
	pusher := push.New(addr, job)
	pusher = pusher.Gatherer(prometheus.DefaultGatherer)
	pusher = pusher.Grouping("instance", instanceName())
	for {
		err := pusher.Push()
		if err != nil {
			log.Error("could not push metrics to prometheus pushgateway", zap.String("err", err.Error()))
		}
		time.Sleep(interval)
	}
}

func pushMeteringMetric(addr string, interval time.Duration) {
	if interval == zeroDuration || len(addr) == 0 {
		log.Info("disable Prometheus push client for metering")
		return
	}
	log.Info("start prometheus push client for metering", zap.String("server addr", addr), zap.String("interval", interval.String()))
	go prometheusPushClientMetering(addr, interval)
}

func prometheusPushClientMetering(addr string, interval time.Duration) {
	job := "tidb"
	pusher := push.New(addr, job).Grouping("instance", instanceName()).Collector(metrics.StmtNodeCounter)
	for {
		err := pusher.Push()
		if err != nil {
			log.Error("push metrics for metering failed", zap.Error(err))
		}
		time.Sleep(interval)
	}

}

func instanceName() string {
	cfg := config.GetGlobalConfig()
	hostname, err := os.Hostname()
	if err != nil {
		return "unknown"
	}
	return fmt.Sprintf("%s_%d", hostname, cfg.Port)
}

// parseDuration parses lease argument string.
func parseDuration(lease string) time.Duration {
	dur, err := time.ParseDuration(lease)
	if err != nil {
		dur, err = time.ParseDuration(lease + "s")
	}
	if err != nil || dur < 0 {
		log.Fatal("invalid lease duration", zap.String("lease", lease))
	}
	return dur
}

func flagBoolean(fset *flag.FlagSet, name string, defaultVal bool, usage string) *bool {
	if !defaultVal {
		// Fix #4125, golang do not print default false value in usage, so we append it.
		usage = fmt.Sprintf("%s (default false)", usage)
		return fset.Bool(name, defaultVal, usage)
	}
	return fset.Bool(name, defaultVal, usage)
}

// overrideConfig considers command arguments and overrides some config items in the Config.
func overrideConfig(cfg *config.Config, fset *flag.FlagSet) {
	actualFlags := make(map[string]bool)
	fset.Visit(func(f *flag.Flag) {
		actualFlags[f.Name] = true
	})

	// Base
	if actualFlags[nmHost] {
		cfg.Host = *host
	}
	if actualFlags[nmAdvertiseAddress] {
		var err error
		if len(strings.Split(*advertiseAddress, " ")) > 1 {
			err = errors.Errorf("Only support one advertise-address")
		}
		terror.MustNil(err)
		cfg.AdvertiseAddress = *advertiseAddress
	}
	if len(cfg.AdvertiseAddress) == 0 && cfg.Host == "0.0.0.0" {
		cfg.AdvertiseAddress = util.GetLocalIP()
	}
	if len(cfg.AdvertiseAddress) == 0 {
		cfg.AdvertiseAddress = cfg.Host
	}
	var err error
	if actualFlags[nmPort] {
		var p int
		p, err = strconv.Atoi(*port)
		terror.MustNil(err)
		cfg.Port = uint(p)
	}
	if actualFlags[nmCors] {
		cfg.Cors = *cors
	}
	if actualFlags[nmStore] {
		cfg.Store = *store
	}
	if actualFlags[nmStorePath] {
		cfg.Path = *storePath
	}
	if actualFlags[nmSocket] {
		cfg.Socket = *socket
	}
	if actualFlags[nmRunDDL] {
		cfg.Instance.TiDBEnableDDL.Store(*runDDL)
	}
	if actualFlags[nmDdlLease] {
		cfg.Lease = *ddlLease
	}
	if actualFlags[nmTokenLimit] {
		cfg.TokenLimit = uint(*tokenLimit)
	}
	if actualFlags[nmPluginLoad] {
		cfg.Instance.PluginLoad = *pluginLoad
	}
	if actualFlags[nmPluginDir] {
		cfg.Instance.PluginDir = *pluginDir
	}

	if actualFlags[nmRepairMode] {
		cfg.RepairMode = *repairMode
	}
	if actualFlags[nmRepairList] {
		if cfg.RepairMode {
			cfg.RepairTableList = stringToList(*repairList)
		}
	}
	if actualFlags[nmTempDir] {
		cfg.TempDir = *tempDir
	}

	// Log
	if actualFlags[nmLogLevel] {
		cfg.Log.Level = *logLevel
	}
	if actualFlags[nmLogFile] {
		cfg.Log.File.Filename = *logFile
	}
	if actualFlags[nmLogSlowQuery] {
		cfg.Log.SlowQueryFile = *logSlowQuery
	}
	if actualFlags[nmLogGeneral] {
		cfg.Log.GeneralLogFile = *logGeneral
	}

	// Status
	if actualFlags[nmReportStatus] {
		cfg.Status.ReportStatus = *reportStatus
	}
	if actualFlags[nmStatusHost] {
		cfg.Status.StatusHost = *statusHost
	}
	if actualFlags[nmStatusPort] {
		var p int
		p, err = strconv.Atoi(*statusPort)
		terror.MustNil(err)
		cfg.Status.StatusPort = uint(p)
	}
	if actualFlags[nmMetricsAddr] {
		cfg.Status.MetricsAddr = *metricsAddr
	}
	if actualFlags[nmMetricsInterval] {
		cfg.Status.MetricsInterval = *metricsInterval
	}

	// PROXY Protocol
	if actualFlags[nmProxyProtocolNetworks] {
		cfg.ProxyProtocol.Networks = *proxyProtocolNetworks
	}
	if actualFlags[nmProxyProtocolHeaderTimeout] {
		cfg.ProxyProtocol.HeaderTimeout = *proxyProtocolHeaderTimeout
	}
	if actualFlags[nmProxyProtocolFallbackable] {
		cfg.ProxyProtocol.Fallbackable = *proxyProtocolFallbackable
	}

	// Sanity check: can't specify both options
	if actualFlags[nmInitializeSecure] && actualFlags[nmInitializeInsecure] {
		err = fmt.Errorf("the options -initialize-insecure and -initialize-secure are mutually exclusive")
		terror.MustNil(err)
	}
	// The option --initialize-secure=true ensures that a secure bootstrap is used.
	if actualFlags[nmInitializeSecure] {
		cfg.Security.SecureBootstrap = *initializeSecure
	}
	// The option --initialize-insecure=true/false was used.
	// Store the inverted value of this to the secure bootstrap cfg item
	if actualFlags[nmInitializeInsecure] {
		cfg.Security.SecureBootstrap = !*initializeInsecure
	}
	if actualFlags[nmDisconnectOnExpiredPassword] {
		cfg.Security.DisconnectOnExpiredPassword = *disconnectOnExpiredPassword
	}
	// Secure bootstrap initializes with Socket authentication
	// which is not supported on windows. Only the insecure bootstrap
	// method is supported.
	if runtime.GOOS == "windows" && cfg.Security.SecureBootstrap {
		err = fmt.Errorf("the option -initialize-secure is not supported on Windows")
		terror.MustNil(err)
	}
	// Initialize SQL File is used to run a set of SQL statements after first bootstrap.
	// It is important in the use case that you want to set GLOBAL variables, which
	// are persisted to the cluster and not read from a config file.
	if actualFlags[nmInitializeSQLFile] {
		if _, err := os.Stat(*initializeSQLFile); err != nil {
			err = fmt.Errorf("can not access -initialize-sql-file %s", *initializeSQLFile)
			terror.MustNil(err)
		}
		cfg.InitializeSQLFile = *initializeSQLFile
	}

	if actualFlags[nmClusterCa] {
		if *clusterCA != "" && (*clusterCert == "" || *clusterKey == "") {
			err = fmt.Errorf("cluster-ca requires both cluster-cert and cluster-key")
			terror.MustNil(err)
		}

		cfg.Security.ClusterSSLCA = *clusterCA
		cfg.Security.ClusterSSLCert = *clusterCert
		cfg.Security.ClusterSSLKey = *clusterKey
	}

	if actualFlags[nmSQLCA] {
		if *sqlCA != "" && (*sqlCert == "" || *clusterKey == "") {
			err = fmt.Errorf("sql-ca requires both sql-cert and sql-key")
			terror.MustNil(err)
		}

		cfg.Security.SSLCA = *sqlCA
		cfg.Security.SSLCert = *sqlCert
		cfg.Security.SSLKey = *sqlKey
	}

	if actualFlags[nmKeyspaceName] {
		cfg.KeyspaceName = *keyspaceName
	}

	if actualFlags[nmTiDBServiceScope] {
		err = servicescope.CheckServiceScope(*serviceScope)
		terror.MustNil(err)
		cfg.Instance.TiDBServiceScope = *serviceScope
	}

	if actualFlags[nmStandby] {
		cfg.StandByMode = *standbyMode
	}

	if actualFlags[nmKeyspaceActivate] {
		cfg.KeyspaceActivateMode = *keyspaceActivateMode
	}

	if actualFlags[nmActivationTimeout] {
		cfg.ActivationTimeout = *activationTimeout
	}

	if actualFlags[nmMaxIdleSeconds] {
		cfg.MaxIdleSeconds = *maxIdleSeconds
	}

	if actualFlags[nmEnableOnlyRunUpgrade] {
		cfg.EnableOnlyRunUpgrade = *enableOnlyRunUpgrade
	}

	if actualFlags[nmExportID] {
		cfg.ExportID = *exportID
	}
}

func setVersions() {
	cfg := config.GetGlobalConfig()
	if len(cfg.ServerVersion) > 0 {
		mysql.ServerVersion = cfg.ServerVersion
	}
	if len(cfg.TiDBEdition) > 0 {
		versioninfo.TiDBEdition = cfg.TiDBEdition
	}
	if len(cfg.TiDBReleaseVersion) > 0 {
		mysql.TiDBReleaseVersionFixed = cfg.TiDBReleaseVersion
	}
}

func setGlobalVars() {
	cfg := config.GetGlobalConfig()

	// config.DeprecatedOptions records the config options that should be moved to [instance] section.
	for _, deprecatedOption := range config.DeprecatedOptions {
		for oldName := range deprecatedOption.NameMappings {
			switch deprecatedOption.SectionName {
			case "":
				switch oldName {
				case "check-mb4-value-in-utf8":
					cfg.Instance.CheckMb4ValueInUTF8.Store(cfg.CheckMb4ValueInUTF8.Load())
				case "enable-collect-execution-info":
					cfg.Instance.EnableCollectExecutionInfo.Store(cfg.EnableCollectExecutionInfo)
				case "max-server-connections":
					cfg.Instance.MaxConnections = cfg.MaxServerConnections
				case "run-ddl":
					cfg.Instance.TiDBEnableDDL.Store(cfg.RunDDL)
				}
			case "log":
				switch oldName {
				case "enable-slow-log":
					cfg.Instance.EnableSlowLog.Store(cfg.Log.EnableSlowLog.Load())
				case "slow-threshold":
					cfg.Instance.SlowThreshold = cfg.Log.SlowThreshold
				case "record-plan-in-slow-log":
					cfg.Instance.RecordPlanInSlowLog = cfg.Log.RecordPlanInSlowLog
				}
			case "performance":
				if oldName == "force-priority" {
					cfg.Instance.ForcePriority = cfg.Performance.ForcePriority
				}
			case "plugin":
				switch oldName {
				case "load":
					cfg.Instance.PluginLoad = cfg.Plugin.Load
				case "dir":
					cfg.Instance.PluginDir = cfg.Plugin.Dir
				}
			default:
			}
		}
	}

	// Disable automaxprocs log
	nopLog := func(string, ...any) {}
	_, err := maxprocs.Set(maxprocs.Logger(nopLog))
	terror.MustNil(err)
	// We should respect to user's settings in config file.
	// The default value of MaxProcs is 0, runtime.GOMAXPROCS(0) is no-op.
	runtime.GOMAXPROCS(int(cfg.Performance.MaxProcs))

	util.SetGOGC(cfg.Performance.GOGC)

	schemaLeaseDuration := parseDuration(cfg.Lease)
	if schemaLeaseDuration <= 0 {
		// previous version allow set schema lease to 0, and mainly used on
		// uni-store and for test, to be compatible we set it to default value here.
		log.Warn("schema lease is invalid, use default value",
			zap.String("lease", schemaLeaseDuration.String()))
		schemaLeaseDuration = config.DefSchemaLease
	}
	session.SetSchemaLease(schemaLeaseDuration)
	statsLeaseDuration := parseDuration(cfg.Performance.StatsLease)
	session.SetStatsLease(statsLeaseDuration)
	planReplayerGCLease := parseDuration(cfg.Performance.PlanReplayerGCLease)
	session.SetPlanReplayerGCLease(planReplayerGCLease)
	bindinfo.Lease = parseDuration(cfg.Performance.BindInfoLease)
	statistics.RatioOfPseudoEstimate.Store(cfg.Performance.PseudoEstimateRatio)
	if cfg.SplitTable {
		atomic.StoreUint32(&ddl.EnableSplitTableRegion, 1)
	}
	plannercore.AllowCartesianProduct.Store(cfg.Performance.CrossJoin)
	privileges.SkipWithGrant = cfg.Security.SkipGrantTable
	if cfg.Performance.TxnTotalSizeLimit == config.DefTxnTotalSizeLimit {
		// practically deprecate the config, let the new session memory tracker take charge of it.
		kv.TxnTotalSizeLimit.Store(config.SuperLargeTxnSize)
	} else {
		kv.TxnTotalSizeLimit.Store(cfg.Performance.TxnTotalSizeLimit)
	}
	if cfg.Performance.TxnEntrySizeLimit > config.MaxTxnEntrySizeLimit {
		log.Fatal("cannot set txn entry size limit larger than 120M")
	}
	kv.TxnEntrySizeLimit.Store(cfg.Performance.TxnEntrySizeLimit)

	priority := mysql.Str2Priority(cfg.Instance.ForcePriority)
	variable.ForcePriority = int32(priority)

	variable.ProcessGeneralLog.Store(cfg.Instance.TiDBGeneralLog)
	variable.EnablePProfSQLCPU.Store(cfg.Instance.EnablePProfSQLCPU)
	variable.EnableRCReadCheckTS.Store(cfg.Instance.TiDBRCReadCheckTS)
	variable.IsSandBoxModeEnabled.Store(!cfg.Security.DisconnectOnExpiredPassword)
	atomic.StoreUint32(&variable.DDLSlowOprThreshold, cfg.Instance.DDLSlowOprThreshold)
	atomic.StoreUint64(&variable.ExpensiveQueryTimeThreshold, cfg.Instance.ExpensiveQueryTimeThreshold)
	atomic.StoreUint64(&variable.ExpensiveTxnTimeThreshold, cfg.Instance.ExpensiveTxnTimeThreshold)

	if len(cfg.ServerVersion) > 0 {
		mysql.ServerVersion = cfg.ServerVersion
		variable.SetSysVar(variable.Version, cfg.ServerVersion)
	}

	if len(cfg.TiDBEdition) > 0 {
		versioninfo.TiDBEdition = cfg.TiDBEdition
		variable.SetSysVar(variable.VersionComment, "TiDB Server (Apache License 2.0) "+versioninfo.TiDBEdition+" Edition, MySQL 8.0 compatible")
	}
	if len(cfg.VersionComment) > 0 {
		variable.SetSysVar(variable.VersionComment, cfg.VersionComment)
	}
	if len(cfg.TiDBReleaseVersion) > 0 {
		mysql.TiDBReleaseVersionFixed = cfg.TiDBReleaseVersion
	}

	variable.SetSysVar(variable.TiDBForcePriority, mysql.Priority2Str[priority])
	variable.SetSysVar(variable.TiDBOptDistinctAggPushDown, variable.BoolToOnOff(cfg.Performance.DistinctAggPushDown))
	variable.SetSysVar(variable.TiDBOptProjectionPushDown, variable.BoolToOnOff(cfg.Performance.ProjectionPushDown))
	variable.SetSysVar(variable.Port, fmt.Sprintf("%d", cfg.Port))
	cfg.Socket = strings.Replace(cfg.Socket, "{Port}", fmt.Sprintf("%d", cfg.Port), 1)
	variable.SetSysVar(variable.Socket, cfg.Socket)
	variable.SetSysVar(variable.DataDir, cfg.Path)
	variable.SetSysVar(variable.TiDBSlowQueryFile, cfg.Log.SlowQueryFile)
	variable.SetSysVar(variable.TiDBIsolationReadEngines, strings.Join(cfg.IsolationRead.Engines, ","))
	variable.SetSysVar(variable.TiDBEnforceMPPExecution, variable.BoolToOnOff(config.GetGlobalConfig().Performance.EnforceMPP))
	variable.MemoryUsageAlarmRatio.Store(cfg.Instance.MemoryUsageAlarmRatio)
	variable.SetSysVar(variable.TiDBConstraintCheckInPlacePessimistic, variable.BoolToOnOff(cfg.PessimisticTxn.ConstraintCheckInPlacePessimistic))
	if hostname, err := os.Hostname(); err == nil {
		variable.SetSysVar(variable.Hostname, hostname)
	}
	variable.GlobalLogMaxDays.Store(int32(config.GetGlobalConfig().Log.File.MaxDays))

	if cfg.Security.EnableSEM {
		terror.MustNil(sem.Enable(cfg.Security.SEMLevel))
	}

	// For CI environment we default enable prepare-plan-cache.
	if config.CheckTableBeforeDrop { // only for test
		variable.SetSysVar(variable.TiDBEnablePrepPlanCache, variable.BoolToOnOff(true))
	}
	// use server-memory-quota as max-plan-cache-memory
	plannercore.PreparedPlanCacheMaxMemory.Store(cfg.Performance.ServerMemoryQuota)
	total, err := memory.MemTotal()
	terror.MustNil(err)
	// if server-memory-quota is larger than max-system-memory or not set, use max-system-memory as max-plan-cache-memory
	if plannercore.PreparedPlanCacheMaxMemory.Load() > total || plannercore.PreparedPlanCacheMaxMemory.Load() <= 0 {
		plannercore.PreparedPlanCacheMaxMemory.Store(total)
	}

	atomic.StoreUint64(&transaction.CommitMaxBackoff, uint64(parseDuration(cfg.TiKVClient.CommitTimeout).Seconds()*1000))
	tikv.SetRegionCacheTTLWithJitter(int64(cfg.TiKVClient.RegionCacheTTL), int64(cfg.TiKVClient.RegionCacheTTLJitterSec))
	domainutil.RepairInfo.SetRepairMode(cfg.RepairMode)
	domainutil.RepairInfo.SetRepairTableList(cfg.RepairTableList)
	executor.GlobalDiskUsageTracker.SetBytesLimit(cfg.TempStorageQuota)
	if cfg.Performance.ServerMemoryQuota < 1 {
		// If MaxMemory equals 0, it means unlimited
		executor.GlobalMemoryUsageTracker.SetBytesLimit(-1)
	} else {
		executor.GlobalMemoryUsageTracker.SetBytesLimit(int64(cfg.Performance.ServerMemoryQuota))
	}
	kvcache.GlobalLRUMemUsageTracker.AttachToGlobalTracker(executor.GlobalMemoryUsageTracker)

	t, err := time.ParseDuration(cfg.TiKVClient.StoreLivenessTimeout)
	if err != nil || t < 0 {
		logutil.BgLogger().Fatal("invalid duration value for store-liveness-timeout",
			zap.String("currentValue", cfg.TiKVClient.StoreLivenessTimeout))
	}
	tikv.SetStoreLivenessTimeout(t)
	parsertypes.TiDBStrictIntegerDisplayWidth = cfg.DeprecateIntegerDisplayWidth
	deadlockhistory.GlobalDeadlockHistory.Resize(cfg.PessimisticTxn.DeadlockHistoryCapacity)
	txninfo.Recorder.ResizeSummaries(cfg.TrxSummary.TransactionSummaryCapacity)
	txninfo.Recorder.SetMinDuration(time.Duration(cfg.TrxSummary.TransactionIDDigestMinDuration) * time.Millisecond)
	chunk.InitChunkAllocSize(cfg.TiDBMaxReuseChunk, cfg.TiDBMaxReuseColumn)

	if len(cfg.Instance.TiDBServiceScope) > 0 {
		variable.ServiceScope.Store(strings.ToLower(cfg.Instance.TiDBServiceScope))
	}

	variable.EnableDistTask.Store(cfg.EnableDistTask)
	variable.EnableFastReorg.Store(cfg.EnableFastReorg)
}

func setupLog(keyspaceID uint32) {
	cfg := config.GetGlobalConfig()
	err := logutil.InitLogger(cfg.Log.ToLogConfig(), keyspace.WrapZapcoreWithKeyspace(keyspaceID))
	terror.MustNil(err)

	// trigger internal http(s) client init.
	util.InternalHTTPClient()
}

func setupAuditLog() {
	auditLogCfg := config.GetGlobalConfig().AuditLog
	err := audit.RegisterForServerless(&auditLogCfg)
	terror.MustNil(err)
}

func setupExtensions() *extension.Extensions {
	err := extension.Setup()
	terror.MustNil(err)

	extensions, err := extension.GetExtensions()
	terror.MustNil(err)

	return extensions
}

func printInfo() {
	// Make sure the TiDB info is always printed.
	level := log.GetLevel()
	log.SetLevel(zap.InfoLevel)
	printer.PrintTiDBInfo()
	log.SetLevel(level)
}

func createServer(storage kv.Storage, dom *domain.Domain) *server.Server {
	cfg := config.GetGlobalConfig()
	driver := server.NewTiDBDriver(storage)
	svr, err := server.NewServer(cfg, driver)
	// Both domain and storage have started, so we have to clean them before exiting.
	if err != nil {
		closeDDLOwnerMgrDomainAndStorage(storage, dom)
		log.Fatal("failed to create the server", zap.Error(err), zap.Stack("stack"))
	}
	svr.SetDomain(dom)
	go dom.ExpensiveQueryHandle().SetSessionManager(svr).Run()
	go dom.MemoryUsageAlarmHandle().SetSessionManager(svr).Run()
	go dom.ServerMemoryLimitHandle().SetSessionManager(svr).Run()
	dom.InfoSyncer().SetSessionManager(svr)
	return svr
}

func setupMetrics() {
	cfg := config.GetGlobalConfig()
	// Enable the mutex profile, 1/10 of mutex blocking event sampling.
	runtime.SetMutexProfileFraction(10)
	systimeErrHandler := func() {
		metrics.TimeJumpBackCounter.Inc()
	}
	go systimemon.StartMonitor(time.Now, systimeErrHandler)

	pushMetric(cfg.Status.MetricsAddr, time.Duration(cfg.Status.MetricsInterval)*time.Second)
	pushMeteringMetric(cfg.Status.MeteringMetricsAddr, time.Duration(cfg.Status.MeteringMetricsInterval)*time.Second)
}

func setupTracing() {
	cfg := config.GetGlobalConfig()
	tracingCfg := cfg.OpenTracing.ToTracingConfig()
	tracingCfg.ServiceName = "TiDB"
	tracer, _, err := tracingCfg.NewTracer()
	if err != nil {
		log.Fatal("setup jaeger tracer failed", zap.String("error message", err.Error()))
	}
	opentracing.SetGlobalTracer(tracer)
}

func closeDDLOwnerMgrDomainAndStorage(storage kv.Storage, dom *domain.Domain) {
	tikv.StoreShuttingDown(1)
	dom.Close()
	ddl.CloseOwnerManager()
	copr.GlobalMPPFailedStoreProber.Stop()
	mppcoordmanager.InstanceMPPCoordinatorManager.Stop()
	err := storage.Close()
	terror.Log(errors.Trace(err))
}

// The amount of time we wait for the ongoing txt to finished.
// We should better provider a dynamic way to set this value.
var gracefulCloseConnectionsTimeout = 15 * time.Second

func cleanup(svr *server.Server, storage kv.Storage, dom *domain.Domain) {
	dom.StopAutoAnalyze()

	drainClientWait := gracefulCloseConnectionsTimeout
	if svr.GetForceShutdown() {
		drainClientWait = 0
	}
	cancelClientWait := time.Second * 1
	svr.DrainClients(drainClientWait, cancelClientWait)

	// Kill sys processes such as auto analyze. Otherwise, tidb-server cannot exit until auto analyze is finished.
	// See https://github.com/pingcap/tidb/issues/40038 for details.
	svr.KillSysProcesses()
	plugin.Shutdown(context.Background())
	topsql.Close()
	closeDDLOwnerMgrDomainAndStorage(storage, dom)
	disk.CleanUp()
	closeStmtSummary()
	cgmon.StopCgroupMonitor()
}

func stringToList(repairString string) []string {
	if len(repairString) <= 0 {
		return []string{}
	}
	if repairString[0] == '[' && repairString[len(repairString)-1] == ']' {
		repairString = repairString[1 : len(repairString)-1]
	}
	return strings.FieldsFunc(repairString, func(r rune) bool {
		return r == ',' || r == ' ' || r == '"'
	})
}

func setupStmtSummary() {
	instanceCfg := config.GetGlobalConfig().Instance
	if instanceCfg.StmtSummaryEnablePersistent {
		err := stmtsummaryv2.Setup(&stmtsummaryv2.Config{
			Filename:       instanceCfg.StmtSummaryFilename,
			FileMaxSize:    instanceCfg.StmtSummaryFileMaxSize,
			FileMaxDays:    instanceCfg.StmtSummaryFileMaxDays,
			FileMaxBackups: instanceCfg.StmtSummaryFileMaxBackups,
		})
		if err != nil {
			logutil.BgLogger().Error("failed to setup statements summary", zap.Error(err))
		}
	}
}

func closeStmtSummary() {
	instanceCfg := config.GetGlobalConfig().Instance
	if instanceCfg.StmtSummaryEnablePersistent {
		stmtsummaryv2.Close()
	}
}

const defaultMgrRequestTimeout = 10 * time.Second

func createMrgClient() (tidbmanager.Client, error) {
	cfg := config.GetGlobalConfig()

	// If the manager notifier option is not enabled, we will not create the manager client.
	if !cfg.EnableManagerNotifier {
		return nil, nil
	}

	security := cfg.Security
	clusterSecurity := security.ClusterSecurity()
	tlsConfig, err := clusterSecurity.ToTLSConfig()
	if err != nil {
		return nil, err
	}

	return tidbmanager.NewClient(
		cfg.ManagerAddr,
		tlsConfig,
		os.Getenv("POD_NAME"),
		os.Getenv("POD_IP"),
		os.Getenv("NAMESPACE"),
	), nil
}

func setupExternalStorage(keyspaceMeta *keyspacepb.KeyspaceMeta) error {
	cfg := config.GetGlobalConfig()
	path := cfg.ExternalStoragePath
	if len(path) == 0 {
		path = cfg.TempDir
	}
	namespace := "NullKeyspace"
	if keyspaceMeta != nil {
		namespace = keyspaceMeta.Name
		if keyspaceMeta.Config != nil && len(keyspaceMeta.Config[serverless.LabelTenantID]) > 0 {
			// for serverless cluster, we use `org-{tenant-id}/tidb-external-storage/{cluster-id}` as namespace to isolate data.
			namespace = fmt.Sprintf("org-%s/tidb-external-storage/%s",
				keyspaceMeta.Config[serverless.LabelTenantID],
				keyspaceMeta.Config[serverless.LabelClusterID])
		}
	}
	logutil.BgLogger().Info("initialize external storage", zap.String("path", path), zap.String("namespace", namespace))

	opts := &storage.ExternalStorageOptions{
		RoleExpiryWindow: cfg.ExternalStorageConfig.RoleExpiryWindow,
		RoleDuration:     cfg.ExternalStorageConfig.RoleDuration,
	}

	return external.CreateExternalStorage(path, namespace, opts)
}

func updateConfigForServerless(keyspaceMeta *keyspacepb.KeyspaceMeta) {
	if keyspaceMeta == nil {
		return
	}
	config.UpdateGlobal(func(conf *config.Config) {
		conf.MetricsLabels = make(map[string]string)
		if keyspaceMeta.Config != nil {
			conf.MetricsLabels[metricsutil.ClusterIDLabelKey] = keyspaceMeta.Config[serverless.LabelClusterID]
			conf.MetricsLabels[metricsutil.ProjectIDLabelKey] = keyspaceMeta.Config[serverless.LabelProjectID]
			conf.MetricsLabels[metricsutil.TenantIDIDLabelKey] = keyspaceMeta.Config[serverless.LabelTenantID]

			conf.StmtSummaryAdditionalInfo[serverless.LabelClusterID] = keyspaceMeta.Config[serverless.LabelClusterID]
			conf.StmtSummaryAdditionalInfo[serverless.LabelProjectID] = keyspaceMeta.Config[serverless.LabelProjectID]
			conf.StmtSummaryAdditionalInfo[serverless.LabelTenantID] = keyspaceMeta.Config[serverless.LabelTenantID]

			conf.Log.SlowLogAdditionalInfo[serverless.SlowLogServerlessClusterIDKey] = keyspaceMeta.Config[serverless.LabelClusterID]
			conf.Log.SlowLogAdditionalInfo[serverless.SlowLogServerlessProjectIDKey] = keyspaceMeta.Config[serverless.LabelProjectID]
			conf.Log.SlowLogAdditionalInfo[serverless.SlowLogServerlessTenantIDKey] = keyspaceMeta.Config[serverless.LabelTenantID]
		}
		conf.MetricsLabels[metricsutil.KeyspaceIDLabelKey] = fmt.Sprintf("%d", keyspaceMeta.GetId())
	})
}