photoprism/internal/tidb/server.go

365 lines
9.6 KiB
Go
Raw Normal View History

// 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,
// See the License for the specific language governing permissions and
// limitations under the License.
package tidb
import (
"context"
"flag"
"fmt"
"os"
"runtime"
"strconv"
"strings"
"time"
"github.com/opentracing/opentracing-go"
"github.com/pingcap/errors"
"github.com/pingcap/parser/mysql"
"github.com/pingcap/parser/terror"
pd "github.com/pingcap/pd/client"
pumpcli "github.com/pingcap/tidb-tools/tidb-binlog/pump_client"
"github.com/pingcap/tidb/config"
"github.com/pingcap/tidb/ddl"
"github.com/pingcap/tidb/domain"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/metrics"
plannercore "github.com/pingcap/tidb/planner/core"
"github.com/pingcap/tidb/privilege/privileges"
"github.com/pingcap/tidb/server"
"github.com/pingcap/tidb/session"
"github.com/pingcap/tidb/sessionctx/binloginfo"
"github.com/pingcap/tidb/sessionctx/variable"
"github.com/pingcap/tidb/statistics"
"github.com/pingcap/tidb/store/mockstore"
"github.com/pingcap/tidb/store/tikv"
"github.com/pingcap/tidb/store/tikv/gcworker"
"github.com/pingcap/tidb/util/logutil"
"github.com/pingcap/tidb/util/printer"
xserver "github.com/pingcap/tidb/x-server"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/push"
log "github.com/sirupsen/logrus"
)
var (
cfg *config.Config
storage kv.Storage
dom *domain.Domain
svr *server.Server
xsvr *xserver.Server
graceful bool
)
// Start the TiDB server using the configuration provided
func Start(ctx context.Context, path string, port uint, host string, debug bool) {
if err := logutil.SetLevel("fatal"); err != nil {
log.Error(err)
}
registerStores()
registerMetrics()
loadConfig()
cfg.Log.Level = log.GetLevel().String()
// cfg.Security.SkipGrantTable = true
if debug {
cfg.Log.Level = "error"
host = "0.0.0.0"
}
cfg.Path = path
cfg.Store = "mocktikv"
if host == "" {
host = "localhost"
}
cfg.Host = host
cfg.Port = port
cfg.Status.ReportStatus = false
validateConfig()
setGlobalVars()
setupTracing()
if debug {
printInfo()
}
setupBinlogClient()
2019-05-02 14:10:05 +02:00
// setupMetrics()
createStoreAndDomain()
createServer()
go runServer()
<-ctx.Done()
serverShutdown(true)
cleanup()
2019-06-04 00:22:25 +02:00
log.Info("tidb server shutdown complete")
}
func registerStores() {
err := session.RegisterStore("tikv", tikv.Driver{})
terror.MustNil(err)
tikv.NewGCHandlerFunc = gcworker.NewGCWorker
err = session.RegisterStore("mocktikv", mockstore.MockDriver{})
terror.MustNil(err)
}
func registerMetrics() {
metrics.RegisterMetrics()
}
func createStoreAndDomain() {
fullPath := fmt.Sprintf("%s://%s", cfg.Store, cfg.Path)
var err error
storage, err = session.NewStore(fullPath)
terror.MustNil(err)
// Bootstrap a session to load information schema.
dom, err = session.BootstrapSession(storage)
terror.MustNil(err)
}
func setupBinlogClient() {
if !cfg.Binlog.Enable {
return
}
if cfg.Binlog.IgnoreError {
binloginfo.SetIgnoreError(true)
}
client, err := pumpcli.NewPumpsClient(cfg.Path, parseDuration(cfg.Binlog.WriteTimeout), pd.SecurityOption{
CAPath: cfg.Security.ClusterSSLCA,
CertPath: cfg.Security.ClusterSSLCert,
KeyPath: cfg.Security.ClusterSSLKey,
})
terror.MustNil(err)
binloginfo.SetPumpsClient(client)
log.Infof("create pumps client success, ignore binlog error %v", cfg.Binlog.IgnoreError)
}
// 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.Infof("start Prometheus push client with server addr %s and interval %s", addr, interval)
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"
for {
err := push.AddFromGatherer(
job,
map[string]string{"instance": instanceName()},
addr,
prometheus.DefaultGatherer,
)
if err != nil {
log.Errorf("could not push metrics to Prometheus Pushgateway: %v", err)
}
time.Sleep(interval)
}
}
func instanceName() string {
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.Fatalf("invalid lease duration %s", lease)
}
return dur
}
func hasRootPrivilege() bool {
return os.Geteuid() == 0
}
func flagBoolean(name string, defaultVal bool, usage string) *bool {
if defaultVal == false {
// Fix #4125, golang do not print default false value in usage, so we append it.
usage = fmt.Sprintf("%s (default false)", usage)
return flag.Bool(name, defaultVal, usage)
}
return flag.Bool(name, defaultVal, usage)
}
func loadConfig() {
cfg = config.GetGlobalConfig()
}
func validateConfig() {
if cfg.Security.SkipGrantTable && !hasRootPrivilege() {
log.Error("TiDB run with skip-grant-table need root privilege.")
os.Exit(-1)
}
if _, ok := config.ValidStorage[cfg.Store]; !ok {
nameList := make([]string, 0, len(config.ValidStorage))
for k, v := range config.ValidStorage {
if v {
nameList = append(nameList, k)
}
}
log.Errorf("\"store\" should be in [%s] only", strings.Join(nameList, ", "))
os.Exit(-1)
}
if cfg.Store == "mocktikv" && cfg.RunDDL == false {
log.Errorf("can't disable DDL on mocktikv")
os.Exit(-1)
}
if cfg.Log.File.MaxSize > config.MaxLogFileSize {
log.Errorf("log max-size should not be larger than %d MB", config.MaxLogFileSize)
os.Exit(-1)
}
if cfg.XProtocol.XServer {
log.Error("X Server is not available")
os.Exit(-1)
}
cfg.OOMAction = strings.ToLower(cfg.OOMAction)
// lower_case_table_names is allowed to be 0, 1, 2
if cfg.LowerCaseTableNames < 0 || cfg.LowerCaseTableNames > 2 {
log.Errorf("lower-case-table-names should be 0 or 1 or 2.")
os.Exit(-1)
}
}
func setGlobalVars() {
ddlLeaseDuration := parseDuration(cfg.Lease)
session.SetSchemaLease(ddlLeaseDuration)
runtime.GOMAXPROCS(int(cfg.Performance.MaxProcs))
statsLeaseDuration := parseDuration(cfg.Performance.StatsLease)
session.SetStatsLease(statsLeaseDuration)
domain.RunAutoAnalyze = cfg.Performance.RunAutoAnalyze
statistics.FeedbackProbability = cfg.Performance.FeedbackProbability
statistics.MaxQueryFeedbackCount = int(cfg.Performance.QueryFeedbackLimit)
statistics.RatioOfPseudoEstimate = cfg.Performance.PseudoEstimateRatio
ddl.RunWorker = cfg.RunDDL
ddl.EnableSplitTableRegion = cfg.SplitTable
plannercore.AllowCartesianProduct = cfg.Performance.CrossJoin
privileges.SkipWithGrant = cfg.Security.SkipGrantTable
variable.ForcePriority = int32(mysql.Str2Priority(cfg.Performance.ForcePriority))
variable.SysVars[variable.TIDBMemQuotaQuery].Value = strconv.FormatInt(cfg.MemQuotaQuery, 10)
variable.SysVars["lower_case_table_names"].Value = strconv.Itoa(cfg.LowerCaseTableNames)
plannercore.SetPreparedPlanCache(cfg.PreparedPlanCache.Enabled)
if plannercore.PreparedPlanCacheEnabled() {
plannercore.PreparedPlanCacheCapacity = cfg.PreparedPlanCache.Capacity
}
if cfg.TiKVClient.GrpcConnectionCount > 0 {
tikv.MaxConnectionCount = cfg.TiKVClient.GrpcConnectionCount
}
tikv.GrpcKeepAliveTime = time.Duration(cfg.TiKVClient.GrpcKeepAliveTime) * time.Second
tikv.GrpcKeepAliveTimeout = time.Duration(cfg.TiKVClient.GrpcKeepAliveTimeout) * time.Second
tikv.CommitMaxBackoff = int(parseDuration(cfg.TiKVClient.CommitTimeout).Seconds() * 1000)
}
func printInfo() {
// Make sure the TiDB info is always printed.
printer.PrintTiDBInfo()
}
func createServer() {
var driver server.IDriver
driver = server.NewTiDBDriver(storage)
var err error
svr, err = server.NewServer(cfg, driver)
// Both domain and storage have started, so we have to clean them before exiting.
terror.MustNil(err, closeDomainAndStorage)
if cfg.XProtocol.XServer {
xcfg := &xserver.Config{
Addr: fmt.Sprintf("%s:%d", cfg.XProtocol.XHost, cfg.XProtocol.XPort),
Socket: cfg.XProtocol.XSocket,
TokenLimit: cfg.TokenLimit,
}
xsvr, err = xserver.NewServer(xcfg)
terror.MustNil(err, closeDomainAndStorage)
}
}
func serverShutdown(isgraceful bool) {
if isgraceful {
log.Info("graceful database shutdown")
graceful = true
} else {
log.Info("database shutdown")
}
if xsvr != nil {
xsvr.Close() // Should close xserver before server.
}
svr.Close()
log.Info("database server closed")
}
func setupTracing() {
tracingCfg := cfg.OpenTracing.ToTracingConfig()
tracer, _, err := tracingCfg.New("TiDB")
if err != nil {
log.Fatal("cannot initialize Jaeger Tracer", err)
}
opentracing.SetGlobalTracer(tracer)
}
func runServer() {
err := svr.Run()
if err != nil {
log.Errorf("Server failed to run: %v", err)
}
}
func closeDomainAndStorage() {
dom.Close()
if err := storage.Close(); err != nil {
log.Error(errors.Trace(err))
} else {
log.Info("database storage closed")
}
}
func cleanup() {
if graceful {
svr.GracefulDown()
} else {
svr.KillAllConnections()
}
closeDomainAndStorage()
}